openapi-mock-server/.tests/test_middleware.py

"""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