"""Focused API regressions. Pure tests also run with stdlib unittest alone. Run the complete suite after installing requirements-dev.txt. Integration cases are skipped, explicitly, when FastAPI/Pydantic settings are not installed. """ import asyncio import importlib.util import json import os from pathlib import Path import sys import tempfile import unittest from concurrent.futures import ThreadPoolExecutor from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from src.services.html_sanitizer import sanitize_answer_html from src.services.runtime_store import JsonRuntimeStore HAS_API_DEPS = all(importlib.util.find_spec(name) is not None for name in ( "fastapi", "pydantic_settings", "httpx", "cachetools", )) class SanitizerRegressions(unittest.TestCase): def test_forbidden_void_does_not_swallow_remaining_answer(self): self.assertEqual(sanitize_answer_html("Useful answer"), "Useful answer") def test_self_closing_forbidden_element_does_not_swallow_answer(self): self.assertEqual(sanitize_answer_html("' 'citation' 'bad', allowed_urls=[], ) self.assertNotIn("onclick", result) self.assertNotIn("alert", result) self.assertNotIn("href=", result) self.assertIn("Answer", result) class RuntimeStoreRegressions(unittest.TestCase): def test_partial_and_invalid_files_keep_defaults(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "config.json" store = JsonRuntimeStore(path, default={"maintenance_mode": True, "max_results": 9}) path.write_text('{"announcement": "Welcome"}') self.assertEqual(store.load()["max_results"], 9) path.write_text("[]") self.assertTrue(store.load()["maintenance_mode"]) path.write_text("{broken") self.assertEqual(store.load()["max_results"], 9) def test_parallel_replacements_leave_one_complete_object(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "config.json" store = JsonRuntimeStore(path) payloads = [{"value": value, "body": "x" * 8000} for value in range(20)] with ThreadPoolExecutor(max_workers=8) as executor: list(executor.map(store.save, payloads)) self.assertIn(json.loads(path.read_text()), payloads) self.assertFalse(list(Path(directory).glob("*.tmp"))) if os.name == "posix": self.assertEqual(path.stat().st_mode & 0o777, 0o600) def test_failed_serialization_preserves_previous_value(self): with tempfile.TemporaryDirectory() as directory: store = JsonRuntimeStore(Path(directory) / "config.json") store.save({"value": "original"}) with self.assertRaises(TypeError): store.save({"value": object()}) self.assertEqual(store.load(), {"value": "original"}) @unittest.skipUnless(HAS_API_DEPS, "FastAPI/Pydantic/httpx dependencies are not installed") class APIIntegrationRegressions(unittest.TestCase): def setUp(self): from fastapi.testclient import TestClient import app from src.api import admin from src.config import Settings self.app_module = app self.admin = admin self.temp = tempfile.TemporaryDirectory() self.settings = Settings( _env_file=None, environment="test", admin_password="regression-password", origin_token="regression-origin", runtime_dir=self.temp.name, allowed_cors_origins="https://frontend.example", maintenance_mode=False, ) self.patchers = [ patch.object(app, "settings", self.settings), patch.object(admin, "get_settings", return_value=self.settings), ] for patcher in self.patchers: patcher.start() self.addCleanup(self.temp.cleanup) for patcher in self.patchers: self.addCleanup(patcher.stop) admin.admin_sessions.clear() admin._login_attempts.clear() admin._login_lockouts.clear() self.client = TestClient(app.app, base_url="https://testserver") def login(self): response = self.client.post("/admin/auth", json={"password": "regression-password"}) self.assertEqual(response.status_code, 200, response.text) def test_logout_revokes_persisted_cookie(self): self.login() token = self.client.cookies.get("admin_session") self.client.get("/admin/logout", follow_redirects=False) self.client.cookies.set("admin_session", token) self.assertEqual(self.client.get("/admin/config").status_code, 401) def test_lockout_survives_five_minute_attempt_window(self): with patch.object(self.admin.time, "time", return_value=1000): for _ in range(5): self.client.post("/admin/auth", json={"password": "wrong"}) with patch.object(self.admin.time, "time", return_value=1301): response = self.client.post("/admin/auth", json={"password": "regression-password"}) self.assertEqual(response.status_code, 429) self.assertGreater(int(response.headers["retry-after"]), 0) with patch.object(self.admin.time, "time", return_value=1901): self.login() def test_invalid_config_does_not_partially_save(self): self.login() before = self.client.get("/admin/config").json() response = self.client.post("/admin/config-bulk", json={"settings": { "announcement": "must not be saved", "max_results": "abc", }}) self.assertEqual(response.status_code, 422) self.assertEqual(self.client.get("/admin/config").json(), before) def test_config_alias_has_identical_environment_defaults(self): self.settings.maintenance_mode = True self.settings.max_results = 9 self.assertEqual(self.client.get("/config").json(), self.client.get("/admin/public-config").json()) self.assertTrue(self.client.get("/config").json()["maintenance_mode"]) def test_all_origin_subpaths_require_token(self): self.assertEqual(self.client.get("/agent/test").status_code, 401) self.assertEqual(self.client.get("/agent/rag-status").status_code, 401) def test_readiness_does_not_claim_ready_during_initialization(self): with patch.object(self.app_module.rag_service, "is_ready", return_value=False): response = self.client.get("/health/ready") self.assertEqual(response.status_code, 503) self.assertFalse(response.json()["rag_ready"]) self.assertEqual(self.client.get("/health/live").status_code, 200) def test_feedback_accepts_existing_frontend_session_field(self): from src.api.feedback import FeedbackRequest payload = FeedbackRequest(request_id="request_123", feedback="up", session_id="session_123") self.assertEqual(payload.session_id, "session_123") def test_streamed_body_limit_rejects_without_content_length(self): from app import RequestBodyLimitMiddleware received = [] async def downstream(scope, receive, send): received.append(await receive()) async def scenario(): chunks = iter([ {"type": "http.request", "body": b"x" * 6, "more_body": True}, {"type": "http.request", "body": b"y" * 6, "more_body": False}, ]) sent = [] async def receive(): return next(chunks) async def send(message): sent.append(message) middleware = RequestBodyLimitMiddleware(downstream, max_bytes=10) await middleware({"type": "http", "method": "POST", "headers": []}, receive, send) return sent sent = asyncio.run(scenario()) self.assertEqual(sent[0]["status"], 413) self.assertFalse(received) if __name__ == "__main__": unittest.main()