Initial commit: Add OpenAPI Mock Server project

.tests/test_middleware.py

@@ -0,0 +1,219 @@
+"""Tests for middleware components."""
+
+import asyncio
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from starlette.requests import Request
+from starlette.responses import Response
+
+from openapi_mock.server.server import (
+    ResponseDelayMiddleware,
+    AuthMiddleware,
+)
+
+
+class TestResponseDelayMiddleware:
+    """Tests for ResponseDelayMiddleware."""
+
+    def test_no_delay(self):
+        """Test middleware with no delay configured."""
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(ResponseDelayMiddleware, delay_range=None)
+        client = TestClient(app)
+
+        import time
+        start = time.time()
+        response = client.get("/test")
+        elapsed = time.time() - start
+
+        assert response.status_code == 200
+        assert elapsed < 0.1
+
+    def test_with_delay(self):
+        """Test middleware with configured delay."""
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(ResponseDelayMiddleware, delay_range=(0.1, 0.2))
+        client = TestClient(app)
+
+        import time
+        start = time.time()
+        response = client.get("/test")
+        elapsed = time.time() - start
+
+        assert response.status_code == 200
+        assert elapsed >= 0.1
+
+    def test_delay_range(self):
+        """Test middleware with delay range."""
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(ResponseDelayMiddleware, delay_range=(0.05, 0.15))
+        client = TestClient(app)
+
+        import time
+        times = []
+        for _ in range(5):
+            start = time.time()
+            response = client.get("/test")
+            elapsed = time.time() - start
+            times.append(elapsed)
+            assert response.status_code == 200
+
+        assert all(t >= 0.05 for t in times)
+        assert all(t <= 0.2 for t in times)
+
+
+class TestAuthMiddleware:
+    """Tests for AuthMiddleware."""
+
+    def test_auth_disabled(self):
+        """Test middleware with auth disabled."""
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(AuthMiddleware, auth_type="none")
+        client = TestClient(app)
+
+        response = client.get("/test")
+        assert response.status_code == 200
+
+    def test_bearer_missing(self):
+        """Test Bearer auth with missing token."""
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(AuthMiddleware, auth_type="bearer")
+        client = TestClient(app)
+
+        response = client.get("/test")
+        assert response.status_code == 401
+        assert "Authorization" in response.json()["error"]
+
+    def test_bearer_invalid(self):
+        """Test Bearer auth with invalid token format."""
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(AuthMiddleware, auth_type="bearer")
+        client = TestClient(app)
+
+        response = client.get("/test", headers={"Authorization": "InvalidFormat"})
+        assert response.status_code == 401
+
+    def test_api_key_missing(self):
+        """Test API key auth with missing key."""
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(AuthMiddleware, auth_type="api_key")
+        client = TestClient(app)
+
+        response = client.get("/test")
+        assert response.status_code == 401
+
+    def test_api_key_valid(self):
+        """Test API key auth with valid key."""
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(
+            AuthMiddleware,
+            auth_type="api_key",
+            api_keys=["valid-key-123"]
+        )
+        client = TestClient(app)
+
+        response = client.get("/test", headers={"X-API-Key": "valid-key-123"})
+        assert response.status_code == 200
+
+    def test_api_key_in_header(self):
+        """Test API key in Authorization header."""
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(
+            AuthMiddleware,
+            auth_type="api_key",
+            api_keys=["header-key-456"]
+        )
+        client = TestClient(app)
+
+        response = client.get("/test", headers={"Authorization": "ApiKey header-key-456"})
+        assert response.status_code == 200
+
+    def test_basic_missing(self):
+        """Test Basic auth with missing credentials."""
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(AuthMiddleware, auth_type="basic")
+        client = TestClient(app)
+
+        response = client.get("/test")
+        assert response.status_code == 401
+
+    def test_basic_invalid_format(self):
+        """Test Basic auth with invalid format."""
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(AuthMiddleware, auth_type="basic")
+        client = TestClient(app)
+
+        response = client.get("/test", headers={"Authorization": "Basic not-base64"})
+        assert response.status_code == 401
+
+    def test_basic_valid(self):
+        """Test Basic auth with valid credentials."""
+        import base64
+        app = FastAPI()
+
+        @app.get("/test")
+        async def test_endpoint():
+            return {"message": "test"}
+
+        app.add_middleware(AuthMiddleware, auth_type="basic")
+        client = TestClient(app)
+
+        credentials = base64.b64encode(b"user:password").decode()
+        response = client.get("/test", headers={"Authorization": f"Basic {credentials}"})
+        assert response.status_code == 200