| """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("<input>Useful answer"), "Useful answer") |
|
|
| def test_self_closing_forbidden_element_does_not_swallow_answer(self): |
| self.assertEqual(sanitize_answer_html("<script/>Useful answer"), "Useful answer") |
|
|
| def test_malformed_authority_does_not_raise(self): |
| answer = sanitize_answer_html('<a href="https://[broken">Useful answer</a>') |
| self.assertIn("Useful answer", answer) |
| self.assertNotIn("href=", answer) |
|
|
| def test_valid_url_punctuation_is_preserved(self): |
| url = "https://example.org/article_(library)" |
| self.assertIn(url, sanitize_answer_html(f'<a href="{url}">article</a>', allowed_urls=[url])) |
|
|
| def test_active_content_and_unverified_links_are_removed(self): |
| result = sanitize_answer_html( |
| '<p onclick="alert(1)">Answer</p><script>alert(2)</script>' |
| '<a href="https://not-verified.example">citation</a>' |
| '<a href="javascript:alert(3)">bad</a>', 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() |
|
|
