| """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) |
|
|
