from __future__ import annotations import unittest from unittest.mock import patch from fastapi import APIRouter, Request, Response from fastapi.responses import PlainTextResponse from fastapi.testclient import TestClient from starlette.background import BackgroundTask from starlette.responses import StreamingResponse from govoplan_core.auth import get_api_principal from govoplan_core.core.registry import PlatformRegistry from govoplan_core.server.fastapi import create_govoplan_app from govoplan_core.server.platform import create_platform_router from govoplan_core.server.conditional_requests import conditional_json_get_middleware class ConditionalBufferTests(unittest.IsolatedAsyncioTestCase): async def test_unknown_length_overflow_replays_exact_bytes_without_eager_drain(self) -> None: chunks = [b'{"value":"', b'', b'a' * 32, b'b' * 32, b'c' * 32, b'"}'] consumed = [] async def body(): for chunk in chunks: consumed.append(chunk) yield chunk background = BackgroundTask(lambda: None) response = StreamingResponse(body(), media_type="application/json", background=background) async def route(request): return response request = Request({"type": "http", "method": "GET", "headers": [(b'if-none-match', b'*')]}) with patch("govoplan_core.server.conditional_requests.MAX_CONDITIONAL_JSON_BYTES", 64, create=True): result = await conditional_json_get_middleware(request, route) self.assertIs(result, response) self.assertEqual(4, len(consumed)) self.assertEqual(200, result.status_code) self.assertNotIn("etag", result.headers) self.assertIn("private", result.headers["cache-control"]) self.assertIn("Authorization", result.headers["vary"]) self.assertIs(background, result.background) self.assertEqual(b"".join(chunks), b"".join([chunk async for chunk in result.body_iterator])) self.assertEqual(chunks, consumed) async def test_known_large_body_is_not_consumed(self) -> None: consumed = [] async def body(): consumed.append(True) yield b"x" * 65 response = StreamingResponse(body(), media_type="application/json", headers={"Content-Length": "65"}) async def route(request): return response request = Request({"type": "http", "method": "GET", "headers": []}) with patch("govoplan_core.server.conditional_requests.MAX_CONDITIONAL_JSON_BYTES", 64, create=True): result = await conditional_json_get_middleware(request, route) self.assertIs(result, response) self.assertEqual([], consumed) self.assertEqual("65", result.headers["content-length"]) self.assertEqual(b"x" * 65, b"".join([chunk async for chunk in result.body_iterator])) async def test_matching_small_response_still_runs_current_route_authorization(self) -> None: calls = [] async def route(request): calls.append(True) if len(calls) > 1: return Response(status_code=403) return StreamingResponse(iter([b'{"ok":true}']), media_type="application/json") request = Request({"type": "http", "method": "GET", "headers": []}) first = await conditional_json_get_middleware(request, route) conditional = Request({"type": "http", "method": "GET", "headers": [(b'if-none-match', first.headers['etag'].encode())]}) second = await conditional_json_get_middleware(conditional, route) self.assertEqual(403, second.status_code) self.assertEqual(2, len(calls)) class ConditionalRequestTests(unittest.TestCase): def _client(self) -> TestClient: router = APIRouter() @router.get("/json") def json_payload(value: str = "alpha") -> dict[str, str]: return {"value": value} @router.get("/cookie") def cookie_payload(response: Response) -> dict[str, bool]: response.set_cookie("govoplan-test", "changed") return {"ok": True} @router.get("/text") def text_payload() -> PlainTextResponse: return PlainTextResponse("plain") app = create_govoplan_app( title="conditional request test", version="test", registry=PlatformRegistry(), api_router=router, ) return TestClient(app) def test_json_get_receives_private_etag_and_matching_request_returns_304(self) -> None: with self._client() as client: first = client.get("/json", headers={"X-Request-ID": "request-1"}) self.assertEqual(200, first.status_code, first.text) self.assertEqual({"value": "alpha"}, first.json()) etag = first.headers.get("etag") self.assertIsNotNone(etag) self.assertTrue(etag.startswith('W/"sha256-'), etag) self.assertIn("private", first.headers.get("cache-control", "")) self.assertIn("no-cache", first.headers.get("cache-control", "")) self.assertIn("authorization", first.headers.get("vary", "").lower()) self.assertIn( "x-govoplan-validity-mode", first.headers.get("vary", "").lower(), ) self.assertEqual("request-1", first.headers["X-Correlation-ID"]) second = client.get("/json", headers={"If-None-Match": etag or "", "X-Request-ID": "request-2"}) self.assertEqual(304, second.status_code, second.text) self.assertEqual(b"", second.content) self.assertEqual(etag, second.headers.get("etag")) self.assertEqual("request-2", second.headers["X-Correlation-ID"]) historical = client.get( "/json", headers={ "X-Govoplan-Validity-Mode": "at", "X-Govoplan-Valid-At": "2025-02-03T10:30:00Z", }, ) self.assertEqual(200, historical.status_code, historical.text) self.assertEqual("at", historical.headers["X-Govoplan-Validity-Mode"]) self.assertIn( "x-govoplan-valid-at", historical.headers.get("vary", "").lower(), ) def test_changed_json_body_does_not_match_previous_etag(self) -> None: with self._client() as client: first = client.get("/json?value=alpha") etag = first.headers.get("etag") self.assertIsNotNone(etag) changed = client.get("/json?value=beta", headers={"If-None-Match": etag or ""}) self.assertEqual(200, changed.status_code, changed.text) self.assertEqual({"value": "beta"}, changed.json()) self.assertNotEqual(etag, changed.headers.get("etag")) def test_conditional_middleware_skips_cookie_and_non_json_responses(self) -> None: with self._client() as client: cookie = client.get("/cookie") self.assertEqual(200, cookie.status_code, cookie.text) self.assertNotIn("etag", cookie.headers) text = client.get("/text") self.assertEqual(200, text.status_code, text.text) self.assertNotIn("etag", text.headers) def test_platform_endpoints_return_304_for_matching_if_none_match(self) -> None: api_router = APIRouter(prefix="/api/v1") api_router.include_router(create_platform_router()) app = create_govoplan_app( title="conditional platform endpoint test", version="test", registry=PlatformRegistry(), api_router=api_router, ) app.dependency_overrides[get_api_principal] = lambda: object() with TestClient(app) as client: for path in ( "/api/v1/platform/status", "/api/v1/platform/modules", "/api/v1/platform/public-modules", ): first = client.get(path) self.assertEqual(200, first.status_code, first.text) etag = first.headers.get("etag") self.assertIsNotNone(etag, path) second = client.get(path, headers={"If-None-Match": etag or ""}) self.assertEqual(304, second.status_code, second.text) self.assertEqual(b"", second.content) self.assertEqual(etag, second.headers.get("etag")) if __name__ == "__main__": unittest.main()