"""Regression coverage for the unauthenticated Gateway health probes.""" from __future__ import annotations from types import SimpleNamespace import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from app.gateway import auth_middleware from app.gateway.app import app as gateway_app from app.gateway.auth.token_source import get_explicit_or_session_token from app.gateway.auth_middleware import PUBLIC_HEALTH_PATHS, AuthMiddleware def _health_only_app() -> FastAPI: app = FastAPI() async def health_check() -> dict[str, str]: return {"status": "healthy", "service": "deer-flow-gateway"} for path in PUBLIC_HEALTH_PATHS: app.add_api_route(path, health_check, methods=["GET", "HEAD"]) app.add_middleware(AuthMiddleware) return app @pytest.mark.parametrize("path", sorted(PUBLIC_HEALTH_PATHS)) @pytest.mark.parametrize("method", ["GET", "HEAD"]) def test_health_probe_is_public_when_auth_is_enabled( monkeypatch: pytest.MonkeyPatch, path: str, method: str, ) -> None: monkeypatch.setattr(auth_middleware, "is_auth_disabled", lambda: False) with TestClient(_health_only_app()) as client: client.cookies.set("access_token", "invalid-token") response = client.request( method, path, headers={"Authorization": "Bearer invalid-token"}, ) assert response.status_code == 200 if method == "GET": assert response.json() == {"status": "healthy", "service": "deer-flow-gateway"} else: assert response.content == b"" @pytest.mark.parametrize("path", sorted(PUBLIC_HEALTH_PATHS)) def test_health_probe_allows_a_trailing_slash( monkeypatch: pytest.MonkeyPatch, path: str, ) -> None: monkeypatch.setattr(auth_middleware, "is_auth_disabled", lambda: False) with TestClient(_health_only_app(), follow_redirects=False) as client: response = client.get(f"{path}/") # FastAPI redirects to the canonical route; importantly, auth must not turn # the request into a 401 before routing can perform that redirect. assert response.status_code == 307 @pytest.mark.parametrize("path", sorted(PUBLIC_HEALTH_PATHS)) def test_health_probe_is_public_behind_a_reverse_proxy_prefix( monkeypatch: pytest.MonkeyPatch, path: str, ) -> None: monkeypatch.setattr(auth_middleware, "is_auth_disabled", lambda: False) with TestClient(_health_only_app(), root_path="/deerflow") as client: response = client.get(f"/deerflow{path}") assert response.status_code == 200 def test_gateway_registers_every_public_health_path_for_get_and_head() -> None: methods_by_path = { route.path: route.methods for route in gateway_app.routes if hasattr(route, "methods") } for path in PUBLIC_HEALTH_PATHS: assert {"GET", "HEAD"} <= methods_by_path[path] def test_non_health_route_still_requires_auth(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(auth_middleware, "is_auth_disabled", lambda: False) app = _health_only_app() @app.get("/private") async def private() -> dict[str, bool]: return {"ok": True} with TestClient(app) as client: response = client.get("/private") assert response.status_code == 401 def _auth_request(*, authorization: str = "", cookie: str = "") -> SimpleNamespace: headers = {"authorization": authorization} if authorization else {} cookies = {"access_token": cookie} if cookie else {} return SimpleNamespace(headers=headers, cookies=cookies) def test_explicit_bearer_token_wins_over_shared_cookie() -> None: """Another tab's cookie must not override this tab's explicit token.""" request = _auth_request(authorization="Bearer current-tab", cookie="other-tab") assert get_explicit_or_session_token(request) == "current-tab" def test_cookie_remains_the_fallback_for_cookie_only_clients() -> None: request = _auth_request(cookie="browser-session") assert get_explicit_or_session_token(request) == "browser-session" def test_malformed_authorization_header_does_not_hide_valid_cookie() -> None: request = _auth_request(authorization="Basic abc", cookie="browser-session") assert get_explicit_or_session_token(request) == "browser-session"