diff --git a/.gitignore b/.gitignore index b772e0908..8d76a1a8b 100644 --- a/.gitignore +++ b/.gitignore @@ -171,6 +171,7 @@ cython_debug/ Thumbs.db ehthumbs.db Desktop.ini +*.stackdump # Database files *.db diff --git a/pyproject.toml b/pyproject.toml index 62e59156b..fc45d549c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -166,6 +166,7 @@ addopts = "-v --tb=short --strict-markers" markers = [ "unit: Unit tests", "integration: Integration tests", + "regression: Endpoint and scenario regression tests", "e2e: End-to-end tests", "e2e_config: End-to-end tests requiring real configuration (not included in default test suite)", "api: Tests requiring external API credentials", diff --git a/tests/unit/server/conftest.py b/tests/unit/server/conftest.py new file mode 100644 index 000000000..668cb542a --- /dev/null +++ b/tests/unit/server/conftest.py @@ -0,0 +1,101 @@ +"""Fixtures for server endpoint tests.""" + +import asyncio +import time +from concurrent.futures import ThreadPoolExecutor + +import pytest +import pytest_asyncio + +from server.config import config +from server.utils import health_check + + +async def _async_wait_for_in_flight(timeout: float = 5.0) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + futures = list(health_check._DEPENDENCY_STATUS_IN_FLIGHT.values()) + if not futures or all(future.done() for future in futures): + return + await asyncio.sleep(0.05) + pending = [ + name + for name, future in health_check._DEPENDENCY_STATUS_IN_FLIGHT.items() + if not future.done() + ] + if pending: + raise RuntimeError( + f"dependency probes still in flight after {timeout:g}s: {pending}" + ) + + +def _reset_probe_executor() -> None: + executor = health_check._DEPENDENCY_PROBE_EXECUTOR + try: + executor.shutdown(wait=True, cancel_futures=True) + except Exception: + pass + finally: + health_check._DEPENDENCY_PROBE_EXECUTOR = ThreadPoolExecutor( + max_workers=2, + thread_name_prefix="powermem-dependency-probe", + ) + + +def _clear_probe_state() -> None: + health_check._DEPENDENCY_STATUS_CACHE.clear() + health_check._DEPENDENCY_STATUS_LOCKS.clear() + health_check._DEPENDENCY_STATUS_IN_FLIGHT.clear() + + +@pytest_asyncio.fixture +async def isolated_dependency_probe_state(): + await _async_wait_for_in_flight() + _clear_probe_state() + _reset_probe_executor() + yield + await _async_wait_for_in_flight() + _clear_probe_state() + _reset_probe_executor() + + +@pytest.fixture +def status_endpoint_settings(monkeypatch): + monkeypatch.setattr(config, "auth_enabled", False) + monkeypatch.setattr(config, "dependency_check_timeout_seconds", 0.05) + monkeypatch.setattr(config, "dependency_status_cache_ttl_seconds", 10.0) + + +@pytest.fixture +def system_app(status_endpoint_settings, monkeypatch): + pytest.importorskip("fastapi", exc_type=ImportError) + from fastapi import FastAPI + + from server.api.v1.system import router + from server.middleware.auth import verify_api_key + + monkeypatch.setattr( + "server.api.v1.system.auto_config", + lambda: { + "vector_store": {"provider": "sqlite"}, + "llm": {"provider": "noop"}, + }, + ) + + app = FastAPI() + app.state.service_ready = True + app.state.storage_type = "sqlite" + app.state.service_startup_error = None + app.dependency_overrides[verify_api_key] = lambda: "anonymous" + app.include_router(router, prefix="/api/v1") + return app + + +@pytest_asyncio.fixture +async def async_client(system_app): + httpx = pytest.importorskip("httpx", exc_type=ImportError) + from httpx import ASGITransport, AsyncClient + + transport = ASGITransport(app=system_app) + async with AsyncClient(transport=transport, base_url="http://testserver") as client: + yield client diff --git a/tests/unit/server/test_system_status_dependency_probes.py b/tests/unit/server/test_system_status_dependency_probes.py new file mode 100644 index 000000000..ba1d5c8f5 --- /dev/null +++ b/tests/unit/server/test_system_status_dependency_probes.py @@ -0,0 +1,285 @@ +"""Endpoint-level tests for /system/status dependency probes (Issue #1111).""" + +import asyncio +import functools +import json +import threading +import time + +import pytest + +from server.config import config +from server.models.response import DependencyStatus +from server.utils import health_check + +pytestmark = pytest.mark.usefixtures("isolated_dependency_probe_state") + + +def dependency_status(body: dict, name: str) -> dict: + data = body.get("data") + assert data is not None, f"response missing data: {body!r}" + dependencies = data.get("dependencies") + assert dependencies is not None, f"response missing dependencies: {body!r}" + dep = dependencies.get(name) + assert dep is not None, f"response missing dependency {name!r}: {body!r}" + return dep + + +def async_timeout(seconds: float = 10): + def decorator(test_fn): + @functools.wraps(test_fn) + async def wrapper(*args, **kwargs): + return await asyncio.wait_for(test_fn(*args, **kwargs), timeout=seconds) + + return wrapper + + return decorator + + +async def await_threading_event(event: threading.Event, timeout: float = 1.0) -> bool: + loop = asyncio.get_running_loop() + try: + await asyncio.wait_for( + loop.run_in_executor(None, event.wait, timeout), + timeout=timeout + 0.1, + ) + except asyncio.TimeoutError: + pass + return event.is_set() + + +@pytest.mark.asyncio +@async_timeout(10) +async def test_blocked_status_probe_does_not_block_health_endpoint( + async_client, + monkeypatch, +): + release = threading.Event() + probe_blocked = threading.Event() + + def blocked_database_probe(): + probe_blocked.set() + release.wait(timeout=2.0) + return DependencyStatus(name="database", status="healthy") + + def fast_llm_probe(): + return DependencyStatus(name="llm", status="disabled") + + monkeypatch.setattr(health_check, "_check_database_sync", blocked_database_probe) + monkeypatch.setattr(health_check, "_check_llm_sync", fast_llm_probe) + + status_task = asyncio.create_task(async_client.get("/api/v1/system/status")) + try: + assert await await_threading_event(probe_blocked, timeout=1.0) + + start = time.monotonic() + health_response = await async_client.get("/api/v1/system/health") + health_elapsed = time.monotonic() - start + + assert health_response.status_code == 200 + assert health_elapsed < 0.2 + assert health_response.json()["data"]["status"] == "healthy" + + status_response = await status_task + + assert status_response.status_code == 200 + body = status_response.json() + database = dependency_status(body, "database") + assert database["status"] == "degraded" + assert "timed out" in database["error_message"] + finally: + release.set() + if not status_task.done(): + await status_task + + +@pytest.mark.asyncio +@async_timeout(10) +async def test_status_dependency_probes_use_dedicated_executor_and_timeout_in_parallel( + async_client, + monkeypatch, +): + calls = {"database": 0, "llm": 0} + executor_submits: list[object] = [] + default_executor_calls: list[object] = [] + original_submit = health_check._DEPENDENCY_PROBE_EXECUTOR.submit + original_run_in_executor = asyncio.BaseEventLoop.run_in_executor + + def tracked_submit(fn, *args, **kwargs): + future = original_submit(fn, *args, **kwargs) + executor_submits.append(fn) + return future + + async def tracked_run_in_executor(self, executor, func, *args): + if executor is None: + default_executor_calls.append(func) + return await original_run_in_executor(self, executor, func, *args) + + def slow_database_probe(): + calls["database"] += 1 + time.sleep(0.2) + return DependencyStatus(name="database", status="healthy") + + def slow_llm_probe(): + calls["llm"] += 1 + time.sleep(0.2) + return DependencyStatus(name="llm", status="healthy") + + monkeypatch.setattr(health_check._DEPENDENCY_PROBE_EXECUTOR, "submit", tracked_submit) + monkeypatch.setattr(asyncio.BaseEventLoop, "run_in_executor", tracked_run_in_executor) + monkeypatch.setattr(health_check, "_check_database_sync", slow_database_probe) + monkeypatch.setattr(health_check, "_check_llm_sync", slow_llm_probe) + monkeypatch.setattr(config, "dependency_status_cache_ttl_seconds", 0.0) + + start = time.monotonic() + response = await async_client.get("/api/v1/system/status") + elapsed = time.monotonic() - start + + assert response.status_code == 200 + assert elapsed < 0.2 + assert calls == {"database": 1, "llm": 1} + assert len(executor_submits) == 2 + assert default_executor_calls == [] + + body = response.json() + assert dependency_status(body, "database")["status"] == "degraded" + assert dependency_status(body, "llm")["status"] == "degraded" + + +@pytest.mark.asyncio +@async_timeout(10) +async def test_repeated_status_polls_reuse_blocked_database_probe( + async_client, + monkeypatch, +): + calls = {"database": 0} + started = threading.Event() + release = threading.Event() + + def blocked_database_probe(): + calls["database"] += 1 + started.set() + release.wait(timeout=2.0) + return DependencyStatus(name="database", status="healthy") + + def fast_llm_probe(): + return DependencyStatus(name="llm", status="disabled") + + monkeypatch.setattr(health_check, "_check_database_sync", blocked_database_probe) + monkeypatch.setattr(health_check, "_check_llm_sync", fast_llm_probe) + monkeypatch.setattr(config, "dependency_status_cache_ttl_seconds", 0.0) + + try: + first = await async_client.get("/api/v1/system/status") + assert await await_threading_event(started, timeout=1.0) + assert first.status_code == 200 + assert dependency_status(first.json(), "database")["status"] == "degraded" + + second = await async_client.get("/api/v1/system/status") + assert second.status_code == 200 + assert dependency_status(second.json(), "database")["status"] == "degraded" + assert calls == {"database": 1} + finally: + release.set() + + +@pytest.mark.asyncio +@async_timeout(10) +async def test_status_dependency_cache_reuses_within_ttl_then_refreshes( + async_client, + monkeypatch, +): + calls = {"database": 0, "llm": 0} + + def database_probe(): + calls["database"] += 1 + return DependencyStatus(name="database", status="healthy") + + def llm_probe(): + calls["llm"] += 1 + return DependencyStatus(name="llm", status="disabled") + + monkeypatch.setattr(health_check, "_check_database_sync", database_probe) + monkeypatch.setattr(health_check, "_check_llm_sync", llm_probe) + monkeypatch.setattr(config, "dependency_status_cache_ttl_seconds", 0.1) + + first = await async_client.get("/api/v1/system/status") + assert first.status_code == 200 + assert dependency_status(first.json(), "database")["status"] == "healthy" + + second = await async_client.get("/api/v1/system/status") + assert second.status_code == 200 + assert calls == {"database": 1, "llm": 1} + + await asyncio.sleep(0.11) + + third = await async_client.get("/api/v1/system/status") + assert third.status_code == 200 + assert calls == {"database": 2, "llm": 2} + + +@pytest.mark.asyncio +@async_timeout(10) +async def test_status_endpoint_surfaces_cached_degraded_dependency( + async_client, + monkeypatch, +): + calls = {"database": 0, "llm": 0} + + def slow_database_probe(): + calls["database"] += 1 + time.sleep(0.2) + return DependencyStatus(name="database", status="healthy") + + def fast_llm_probe(): + calls["llm"] += 1 + return DependencyStatus(name="llm", status="disabled") + + monkeypatch.setattr(health_check, "_check_database_sync", slow_database_probe) + monkeypatch.setattr(health_check, "_check_llm_sync", fast_llm_probe) + monkeypatch.setattr(config, "dependency_status_cache_ttl_seconds", 60.0) + + first = await async_client.get("/api/v1/system/status") + assert first.status_code == 200 + first_body = first.json() + assert first_body["data"]["status"] == "degraded" + assert dependency_status(first_body, "database")["status"] == "degraded" + + second = await async_client.get("/api/v1/system/status") + assert second.status_code == 200 + second_body = second.json() + assert second_body["data"]["status"] == "degraded" + assert dependency_status(second_body, "database")["status"] == "degraded" + assert calls == {"database": 1, "llm": 1} + assert ( + dependency_status(second_body, "database")["error_message"] + == dependency_status(first_body, "database")["error_message"] + ) + + +@pytest.mark.asyncio +@async_timeout(10) +async def test_status_endpoint_falls_back_when_dependency_probe_raises( + async_client, + monkeypatch, +): + internal_error = "dependency probe failed" + + def exploding_database_probe(): + raise RuntimeError(internal_error) + + def fast_llm_probe(): + return DependencyStatus(name="llm", status="disabled") + + monkeypatch.setattr(health_check, "_check_database_sync", exploding_database_probe) + monkeypatch.setattr(health_check, "_check_llm_sync", fast_llm_probe) + + response = await async_client.get("/api/v1/system/status") + + assert response.status_code == 200 + body = response.json() + assert body["success"] is True + assert body["data"]["dependencies"] == {} + assert body["data"]["status"] == "degraded" + assert body["message"] != "System status retrieved successfully" + assert internal_error not in json.dumps(body)