"""Unit and integration tests for FastAPI application, HTTP contracts, and security.""" import asyncio from pathlib import Path import pytest import httpx from httpx import ASGITransport from backend.app import create_app, SESSION_COOKIE_NAME, CSP_POLICY from backend.artifacts import ArtifactStore from backend.confluence import ConfluenceClient from backend.containers import FakeContainerManager from backend.dev.fake_peer import ScriptedContainerPeer from backend.model import FakeModelAdapter from backend.settings import Settings from tests.backend.conftest import make_test_confluence_client_factory @pytest.fixture def test_app_env(tmp_path: Path): settings = Settings( approved_confluence_origins=["https://approved.example.com"], query_timeout_seconds=30.0, cleanup_timeout_seconds=5.0, ) store = ArtifactStore(tmp_path / "artifacts") container_mgr = FakeContainerManager(lambda: ScriptedContainerPeer(scenario="standard")) model_adapter = FakeModelAdapter() conf_factory = make_test_confluence_client_factory() app = create_app( settings=settings, container_manager=container_mgr, artifact_store=store, model_adapter=model_adapter, confluence_client_factory=conf_factory, ) return app, store, container_mgr @pytest.mark.asyncio async def test_root_endpoint(test_app_env): app, _, _ = test_app_env transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: resp = await client.get("/") assert resp.status_code == 200 assert SESSION_COOKIE_NAME in resp.cookies cookie = resp.cookies[SESSION_COOKIE_NAME] assert len(cookie) >= 16 # Security headers assert resp.headers.get("Content-Security-Policy") == CSP_POLICY assert resp.headers.get("Referrer-Policy") == "no-referrer" @pytest.mark.asyncio async def test_auth_verify_origin_enforcement(test_app_env): app, _, _ = test_app_env transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: # 1. Missing Origin header -> 403 origin_denied resp = await client.post( "/api/v1/auth/verify", json={"url": "https://approved.example.com", "pat": "valid-pat"}, ) assert resp.status_code == 403 data = resp.json() assert data["error"]["code"] == "origin_denied" # 2. Mismatched Origin header -> 403 origin_denied resp = await client.post( "/api/v1/auth/verify", json={"url": "https://approved.example.com", "pat": "valid-pat"}, headers={"Origin": "https://evil.com"}, ) assert resp.status_code == 403 data = resp.json() assert data["error"]["code"] == "origin_denied" @pytest.mark.asyncio async def test_auth_verify_success_and_failures(test_app_env): app, _, _ = test_app_env transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: headers = {"Origin": "http://testserver"} # 1. Valid credentials resp = await client.post( "/api/v1/auth/verify", json={"url": "https://approved.example.com", "pat": "valid-pat"}, headers=headers, ) assert resp.status_code == 200 assert resp.json() == {"valid": True} assert resp.headers.get("Cache-Control") == "no-store" assert SESSION_COOKIE_NAME in resp.cookies # 2. Invalid PAT -> 403 confluence_auth_failed resp = await client.post( "/api/v1/auth/verify", json={"url": "https://approved.example.com", "pat": "wrong-pat"}, headers=headers, ) assert resp.status_code == 403 assert resp.json()["error"]["code"] == "confluence_auth_failed" # 3. Disapproved Confluence URL -> 403 destination_denied resp = await client.post( "/api/v1/auth/verify", json={"url": "https://unapproved.example.com", "pat": "valid-pat"}, headers=headers, ) assert resp.status_code == 403 assert resp.json()["error"]["code"] == "destination_denied" @pytest.mark.asyncio async def test_query_and_artifact_download(test_app_env): app, store, container_mgr = test_app_env transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: headers = {"Origin": "http://testserver"} # 1. Execute query payload = { "prompt": "How do I deploy service X?", "credentials": { "url": "https://approved.example.com", "pat": "valid-pat", }, } resp = await client.post("/api/v1/query", json=payload, headers=headers) assert resp.status_code == 200 assert resp.headers.get("Cache-Control") == "no-store" assert SESSION_COOKIE_NAME in client.cookies res_data = resp.json() assert res_data["session_id"] assert "Deployment Guide" in res_data["markdown"] assert len(res_data["pages_accessed"]) == 1 assert len(res_data["artifacts"]) == 1 artifact_meta = res_data["artifacts"][0] aid = artifact_meta["id"] assert artifact_meta["name"] == "checklist.md" assert artifact_meta["size_bytes"] == 32 # 2. Download artifact with same session cookie dl_resp = await client.get(f"/api/v1/artifacts/{aid}") assert dl_resp.status_code == 200 assert dl_resp.headers.get("Cache-Control") == "no-store" assert dl_resp.headers.get("X-Content-Type-Options") == "nosniff" assert "attachment" in dl_resp.headers.get("Content-Disposition", "") assert dl_resp.content == b"# Checklist\n\n- Deploy service X\n" # 3. Download with wrong session cookie -> 404 async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as other_client: # New client has no or different session cookie other_resp = await other_client.get(f"/api/v1/artifacts/{aid}") assert other_resp.status_code == 404 assert other_resp.json()["error"]["code"] == "artifact_not_found" @pytest.mark.asyncio async def test_framework_validation_error_sanitization(test_app_env): app, _, _ = test_app_env transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: headers = {"Origin": "http://testserver"} # Send payload missing required 'prompt' field and containing a fake PAT bad_payload = {"credentials": {"url": "https://approved.example.com", "pat": "secret-secret-token"}} resp = await client.post("/api/v1/query", json=bad_payload, headers=headers) assert resp.status_code == 400 data = resp.json() assert data["error"]["code"] == "invalid_input" # Secret token must NEVER be echoed in error response! assert "secret-secret-token" not in str(data) @pytest.mark.asyncio async def test_query_origin_enforcement(test_app_env): app, _, _ = test_app_env transport = ASGITransport(app=app) payload = { "prompt": "Test query", "credentials": {"url": "https://approved.example.com", "pat": "valid-pat"}, } async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: # 1. Missing Origin header resp = await client.post("/api/v1/query", json=payload) assert resp.status_code == 403 assert resp.json()["error"]["code"] == "origin_denied" assert resp.headers.get("Cache-Control") == "no-store" # 2. Mismatched Origin header resp = await client.post( "/api/v1/query", json=payload, headers={"Origin": "https://malicious.org"}, ) assert resp.status_code == 403 assert resp.json()["error"]["code"] == "origin_denied" assert resp.headers.get("Cache-Control") == "no-store" # 3. Scheme mismatch (https Origin for http request) resp = await client.post( "/api/v1/query", json=payload, headers={"Origin": "https://testserver"}, ) assert resp.status_code == 403 assert resp.json()["error"]["code"] == "origin_denied" @pytest.mark.asyncio async def test_body_size_limits_and_chunked_streaming(test_app_env): app, _, _ = test_app_env transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: # 1. Malformed Content-Length -> 400 invalid_input resp = await client.post( "/api/v1/auth/verify", content=b"{}", headers={"Origin": "http://testserver", "Content-Length": "not-an-int"}, ) assert resp.status_code == 400 assert resp.json()["error"]["code"] == "invalid_input" assert resp.headers.get("Cache-Control") == "no-store" # 2. Declared Content-Length exceeding 128 KiB -> 413 request_too_large resp = await client.post( "/api/v1/auth/verify", content=b"{}", headers={"Origin": "http://testserver", "Content-Length": str(129 * 1024)}, ) assert resp.status_code == 413 assert resp.json()["error"]["code"] == "request_too_large" assert resp.headers.get("Cache-Control") == "no-store" # 3. Chunked/streamed body exceeding limit -> 413 request_too_large async def oversized_stream(): chunk = b"x" * 1024 for _ in range(129): # 129 KiB > 128 KiB yield chunk resp = await client.post( "/api/v1/auth/verify", content=oversized_stream(), headers={"Origin": "http://testserver", "Content-Type": "application/json"}, ) assert resp.status_code == 413 assert resp.json()["error"]["code"] == "request_too_large" assert resp.headers.get("Cache-Control") == "no-store" @pytest.mark.asyncio async def test_unexpected_fields_rejection(test_app_env): app, _, _ = test_app_env transport = ASGITransport(app=app) headers = {"Origin": "http://testserver"} async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: # 1. Extra field in auth verify payload resp = await client.post( "/api/v1/auth/verify", json={"url": "https://approved.example.com", "pat": "valid-pat", "unexpected": "bad"}, headers=headers, ) assert resp.status_code == 400 assert resp.json()["error"]["code"] == "invalid_input" # 2. Extra field in query payload resp = await client.post( "/api/v1/query", json={ "prompt": "test", "credentials": {"url": "https://approved.example.com", "pat": "valid-pat"}, "extra_key": 123, }, headers=headers, ) assert resp.status_code == 400 assert resp.json()["error"]["code"] == "invalid_input" @pytest.mark.asyncio async def test_route_404_and_envelope_codes(test_app_env): app, _, _ = test_app_env transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: # Unknown API route maps to invalid_input (not artifact_not_found) resp = await client.get("/api/v1/nonexistent") assert resp.status_code == 404 assert resp.json()["error"]["code"] == "invalid_input" assert resp.headers.get("Cache-Control") == "no-store" # Unknown artifact maps to artifact_not_found resp = await client.get("/api/v1/artifacts/missing-id") assert resp.status_code == 404 assert resp.json()["error"]["code"] == "artifact_not_found" assert resp.headers.get("Cache-Control") == "no-store" @pytest.mark.asyncio async def test_rfc6266_non_ascii_filename_download(test_app_env): app, store, _ = test_app_env transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: # 1. Obtain a server-issued session root_resp = await client.get("/") assert root_resp.status_code == 200 session_id = root_resp.cookies[SESSION_COOKIE_NAME] # 2. Stage and commit an artifact with non-ASCII Unicode characters (e.g. 'clé') import base64 staging = store.create_staging_session("q-unicode") decision, _ = staging.handle_begin("t-1", "rapport_clé_2026.pdf", 14) assert decision == "accept" b64_data = base64.b64encode(b"PDF Content OK").decode("utf-8") staging.handle_chunk("t-1", 0, b64_data) staging.handle_end("t-1", 14, 1) committed = staging.commit(session_id=session_id) assert len(committed) == 1 art_id = committed[0]["id"] # 3. Download the artifact and verify RFC 6266 Content-Disposition header resp = await client.get(f"/api/v1/artifacts/{art_id}") assert resp.status_code == 200 assert resp.headers.get("Cache-Control") == "no-store" assert resp.headers.get("X-Content-Type-Options") == "nosniff" cd = resp.headers.get("Content-Disposition", "") # Must contain ASCII fallback assert 'filename="rapport_cl__2026.pdf"' in cd # Must contain UTF-8 percent-encoded filename assert "filename*=UTF-8''rapport_cl%C3%A9_2026.pdf" in cd assert resp.content == b"PDF Content OK" def test_dev_mode_wires_fake_container_manager(tmp_path: Path): settings = Settings(dev_mode=True, artifact_storage_dir=tmp_path) app = create_app(settings=settings) assert app is not None def test_production_model_validation_failures(tmp_path: Path): # 1. Unknown provider settings_bad_provider = Settings( dev_mode=False, model_provider="unsupported_provider", artifact_storage_dir=tmp_path, ) with pytest.raises(RuntimeError, match="Unknown model_provider"): create_app(settings=settings_bad_provider) # 2. Missing OpenAI key in production mode settings_no_key = Settings( dev_mode=False, model_provider="openai", model_api_key="", artifact_storage_dir=tmp_path, ) with pytest.raises(RuntimeError, match="Missing model_api_key"): create_app(settings=settings_no_key)