"""Offline checks of actual API function bodies extracted with AST. This does not import or simulate FastAPI/Pydantic routing. It verifies bounded body handling, admin state transitions, and lifespan cancellation with minimal response/request doubles. Run full pytest separately in the installed runtime. """ import ast import asyncio from contextlib import asynccontextmanager, suppress import __future__ import hmac import json import logging from pathlib import Path import secrets import sys import tempfile from types import SimpleNamespace import unittest ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from src.services.runtime_store import JsonRuntimeStore class ResponseDouble: def __init__(self, content=None, status_code=200, headers=None, **kwargs): self.content = content self.status_code = status_code self.headers = headers or {} self.cookies = {} def set_cookie(self, key, value, **kwargs): self.cookies[key] = value def delete_cookie(self, key): self.cookies.pop(key, None) async def __call__(self, scope, receive, send): await send({"type": "http.response.start", "status": self.status_code}) await send({"type": "http.response.body", "body": json.dumps(self.content).encode()}) class HTTPExceptionDouble(Exception): def __init__(self, status_code, detail, headers=None): super().__init__(detail) self.status_code = status_code self.headers = headers or {} def extract(relative_path, names, namespace): parsed = ast.parse((ROOT / relative_path).read_text()) nodes = [node for node in parsed.body if getattr(node, "name", None) in names] if len(nodes) != len(names): raise AssertionError("One of the expected functions has moved or disappeared") for node in nodes: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name != "lifespan": node.decorator_list = [] module = ast.Module(body=nodes, type_ignores=[]) code = compile(module, str(ROOT / relative_path), "exec", flags=__future__.annotations.compiler_flag) exec(code, namespace) return namespace class OfflineAPIChecks(unittest.TestCase): def test_body_limiter_measures_chunks_and_replays_valid_body(self): namespace = extract("app.py", {"RequestBodyLimitMiddleware"}, {"JSONResponse": ResponseDouble}) middleware_class = namespace["RequestBodyLimitMiddleware"] async def check(chunks, limit, lengths=()): forwarded = [] sent = [] messages = iter({"type": "http.request", "body": chunk, "more_body": index < len(chunks)-1} for index, chunk in enumerate(chunks)) async def receive(): return next(messages) async def send(message): sent.append(message) async def downstream(scope, receive, send): forwarded.append((await receive())["body"]) middleware = middleware_class(downstream, max_bytes=limit) await middleware({"type": "http", "method": "POST", "headers": list(lengths)}, receive, send) return forwarded, sent forwarded, sent = asyncio.run(check([b"a"*6, b"b"*6], 10)) self.assertEqual(sent[0]["status"], 413) self.assertFalse(forwarded) forwarded, sent = asyncio.run(check([b"a"*20000, b"b"*20000], 524288)) self.assertEqual(forwarded, [b"a"*20000+b"b"*20000]) self.assertFalse(sent) forwarded, sent = asyncio.run(check([], 10, [(b"content-length", b"99")])) self.assertEqual(sent[0]["status"], 413) def test_admin_lockout_logout_and_defaults(self): with tempfile.TemporaryDirectory() as directory: clock = SimpleNamespace(time=lambda: 1000) settings = SimpleNamespace(admin_password="correct", environment="test", runtime_path=Path(directory), config_path=Path(directory)/"config.json", runtime_defaults={"maintenance_mode": True, "max_results": 9}) namespace = { "JsonRuntimeStore": JsonRuntimeStore, "get_settings": lambda: settings, "json": json, "time": clock, "secrets": secrets, "HTTPException": HTTPExceptionDouble, "Response": ResponseDouble, "RedirectResponse": ResponseDouble, "logger": logging.getLogger("offline"), "admin_sessions": {}, "_login_attempts": {}, "_login_lockouts": {}, "_MAX_LOGIN_ATTEMPTS": 5, "_LOCKOUT_WINDOW": 300, "_LOCKOUT_DURATION": 900, } extract("src/api/admin.py", {"admin_auth", "admin_logout_get", "_sessions_file", "_load_sessions", "_save_sessions", "_cleanup_sessions", "_require_session", "_config_store"}, namespace) request = SimpleNamespace(client=SimpleNamespace(host="one-client"), cookies={}) for attempt in range(5): with self.assertRaises(HTTPExceptionDouble) as caught: asyncio.run(namespace["admin_auth"](SimpleNamespace(password="wrong"), request)) self.assertEqual(caught.exception.status_code, 429 if attempt == 4 else 401) clock.time = lambda: 1301 with self.assertRaises(HTTPExceptionDouble) as caught: asyncio.run(namespace["admin_auth"](SimpleNamespace(password="correct"), request)) self.assertEqual(caught.exception.status_code, 429) clock.time = lambda: 1901 response = asyncio.run(namespace["admin_auth"](SimpleNamespace(password="correct"), request)) request.cookies["admin_session"] = response.cookies["admin_session"] namespace["_require_session"](request) asyncio.run(namespace["admin_logout_get"](request)) with self.assertRaises(HTTPExceptionDouble) as caught: namespace["_require_session"](request) self.assertEqual(caught.exception.status_code, 401) self.assertEqual(namespace["_config_store"]().load(), settings.runtime_defaults) def test_config_validation_rejects_silent_skips(self): ns = extract("src/api/admin.py", {"_validated_config_values"}, {"HTTPException": HTTPExceptionDouble}) validate = ns["_validated_config_values"] self.assertEqual(validate({"max_results": "9", "maintenance_mode": "true"}), {"max_results": 9, "maintenance_mode": True}) for value in ({"max_results": "abc"}, {"max_results": 2.5}, {"maintenance_mode": "perhaps"}, {"announcement": "x"*501}): with self.assertRaises(HTTPExceptionDouble) as caught: validate(value) self.assertEqual(caught.exception.status_code, 422) def test_lifespan_serves_before_initialization_completes_and_cancels_tasks(self): async def scenario(): began = asyncio.Event() cancelled = [] async def initialize(): began.set() try: await asyncio.Event().wait() finally: cancelled.append("startup") async def metrics_loop(): try: await asyncio.Event().wait() finally: cancelled.append("metrics") flushed = [] async def flush(): flushed.append(True) ns = extract("app.py", {"lifespan"}, { "asynccontextmanager": asynccontextmanager, "suppress": suppress, "asyncio": asyncio, "_initialize_services": initialize, "_metrics_flush_loop": metrics_loop, "metrics_service": SimpleNamespace(flush=flush), "logger": logging.getLogger("offline"), }) async with ns["lifespan"](None): await asyncio.wait_for(began.wait(), timeout=1) self.assertEqual(cancelled, []) self.assertCountEqual(cancelled, ["startup", "metrics"]) self.assertEqual(flushed, [True]) asyncio.run(scenario()) def test_readiness_distinguishes_initializing_maintenance_and_missing_secret(self): rag = SimpleNamespace(is_ready=lambda: False, get_stats=lambda: {"retrieval_mode": "unavailable"}) settings = SimpleNamespace(origin_token="origin", environment="production", openai_api_key="key") config = {"maintenance_mode": False} ns = extract("app.py", {"health_ready"}, { "_runtime_config": lambda: config, "rag_service": rag, "settings": settings, "_startup_state": "initializing", "LIBBEE_VERSION": "test", "JSONResponse": ResponseDouble, }) self.assertEqual(ns["health_ready"]().status_code, 503) rag.is_ready = lambda: True self.assertEqual(ns["health_ready"]().status_code, 200) settings.origin_token = "" self.assertEqual(ns["health_ready"]().content["status"], "misconfigured") config["maintenance_mode"] = True self.assertEqual(ns["health_ready"]().status_code, 503) def test_origin_subpaths_fail_closed_and_admin_cross_site_posts_fail(self): settings = SimpleNamespace(origin_token="origin-secret", environment="production") ns = extract("app.py", {"secure_origin"}, { "settings": settings, "hmac": hmac, "asyncio": asyncio, "JSONResponse": ResponseDouble, "_runtime_config": lambda: {"maintenance_mode": False}, "_HF_PREVIEW_PATHS": set(), }) async def next_handler(request): return ResponseDouble() request = SimpleNamespace(url=SimpleNamespace(path="/agent/test"), method="GET", headers={}) self.assertEqual(asyncio.run(ns["secure_origin"](request, next_handler)).status_code, 401) request.headers = {"x-libbee-origin-token": "origin-secret"} self.assertEqual(asyncio.run(ns["secure_origin"](request, next_handler)).status_code, 200) settings.origin_token = "" self.assertEqual(asyncio.run(ns["secure_origin"](request, next_handler)).status_code, 401) request.url.path = "/admin/config" request.method = "POST" request.headers = {"sec-fetch-site": "cross-site"} self.assertEqual(asyncio.run(ns["secure_origin"](request, next_handler)).status_code, 403) if __name__ == "__main__": unittest.main(verbosity=2)