From 0c0988816f5bdbf85ff878b66acd274cfc7e80b8 Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Fri, 30 Jan 2026 03:41:47 +0000 Subject: [PATCH] Initial commit: Add OpenAPI Mock Server project --- .tests/test_middleware.py | 219 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 219 insertions(+) create mode 100644 .tests/test_middleware.py diff --git a/.tests/test_middleware.py b/.tests/test_middleware.py new file mode 100644 index 0000000..8f05db1 --- /dev/null +++ b/.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