#!/usr/bin/env python3 """ Tests for Story 7.01: FastAPI App Bootstrap AIVA RLM Nexus PRD v2 — Track A, Module 7 Black box tests (BB1-BB3): verify public API behaviour from the outside. White box tests (WB1-WB3): verify internal app configuration and module structure. All tests use httpx.ASGITransport — zero real server or network I/O. """ from __future__ import annotations import inspect import sys from datetime import datetime, timezone import httpx import pytest sys.path.insert(0, "/mnt/e/genesis-system") from api.aiva_memory_api import app # --------------------------------------------------------------------------- # Shared async client fixture # --------------------------------------------------------------------------- @pytest.fixture def client() -> httpx.AsyncClient: """Return an AsyncClient wired to the ASGI app — no real server needed.""" transport = httpx.ASGITransport(app=app) return httpx.AsyncClient(transport=transport, base_url="http://test") # --------------------------------------------------------------------------- # BB1: GET /health → 200, body has status="ok" # --------------------------------------------------------------------------- class TestBB1_HealthReturns200: """BB1: GET /health returns HTTP 200 and status field equals 'ok'.""" @pytest.mark.anyio async def test_status_code_200(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") assert response.status_code == 200 @pytest.mark.anyio async def test_status_field_is_ok(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") body = response.json() assert body["status"] == "ok" @pytest.mark.anyio async def test_response_content_type_is_json(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") assert "application/json" in response.headers["content-type"] # --------------------------------------------------------------------------- # BB2: GET /health → services dict has postgres, redis, qdrant keys # --------------------------------------------------------------------------- class TestBB2_ServicesDict: """BB2: GET /health response has 'services' dict with exactly 3 Elestio keys.""" @pytest.mark.anyio async def test_services_key_present(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") body = response.json() assert "services" in body @pytest.mark.anyio async def test_services_has_postgres_key(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") services = response.json()["services"] assert "postgres" in services @pytest.mark.anyio async def test_services_has_redis_key(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") services = response.json()["services"] assert "redis" in services @pytest.mark.anyio async def test_services_has_qdrant_key(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") services = response.json()["services"] assert "qdrant" in services @pytest.mark.anyio async def test_services_has_exactly_three_keys(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") services = response.json()["services"] assert set(services.keys()) == {"postgres", "redis", "qdrant"} @pytest.mark.anyio async def test_all_service_values_are_connected(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") services = response.json()["services"] for key, value in services.items(): assert value == "connected", ( f"Expected services['{key}'] == 'connected', got '{value}'" ) # --------------------------------------------------------------------------- # BB3: GET /unknown_path → 404 (FastAPI default behaviour) # --------------------------------------------------------------------------- class TestBB3_UnknownPathReturns404: """BB3: Unregistered routes return HTTP 404.""" @pytest.mark.anyio async def test_unknown_path_returns_404(self, client: httpx.AsyncClient): async with client: response = await client.get("/unknown_path") assert response.status_code == 404 @pytest.mark.anyio async def test_another_unknown_path_returns_404(self, client: httpx.AsyncClient): async with client: response = await client.get("/api/v1/nonexistent") assert response.status_code == 404 # --------------------------------------------------------------------------- # WB1: FastAPI app title and version match spec # --------------------------------------------------------------------------- class TestWB1_AppMetadata: """WB1: The FastAPI instance carries the correct title and version.""" def test_app_title(self): assert app.title == "AIVA Memory API" def test_app_version(self): assert app.version == "1.0.0" def test_app_is_fastapi_instance(self): from fastapi import FastAPI as _FastAPI assert isinstance(app, _FastAPI) # --------------------------------------------------------------------------- # WB2: timestamp is valid ISO 8601 string # --------------------------------------------------------------------------- class TestWB2_TimestampFormat: """WB2: The timestamp field is a valid ISO 8601 string with timezone info.""" @pytest.mark.anyio async def test_timestamp_present(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") body = response.json() assert "timestamp" in body @pytest.mark.anyio async def test_timestamp_parseable_by_fromisoformat(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") ts_str = response.json()["timestamp"] # Must not raise parsed = datetime.fromisoformat(ts_str) assert parsed is not None @pytest.mark.anyio async def test_timestamp_is_timezone_aware(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") ts_str = response.json()["timestamp"] parsed = datetime.fromisoformat(ts_str) assert parsed.tzinfo is not None, "timestamp must be timezone-aware" @pytest.mark.anyio async def test_timestamp_is_utc(self, client: httpx.AsyncClient): async with client: response = await client.get("/health") ts_str = response.json()["timestamp"] parsed = datetime.fromisoformat(ts_str) utc_offset = parsed.utcoffset() assert utc_offset is not None # UTC offset must be zero seconds assert utc_offset.total_seconds() == 0.0, ( f"Expected UTC (+00:00) but got offset={utc_offset}" ) # --------------------------------------------------------------------------- # WB3: Module has uvicorn config with port 8765 # --------------------------------------------------------------------------- class TestWB3_UvicornConfig: """WB3: The module's __main__ block configures uvicorn on port 8765.""" def test_module_source_contains_port_8765(self): import api.aiva_memory_api as module source = inspect.getsource(module) assert "8765" in source, ( "Module must contain a uvicorn.run(..., port=8765) call" ) def test_module_source_contains_uvicorn_run(self): import api.aiva_memory_api as module source = inspect.getsource(module) assert "uvicorn.run" in source, ( "Module must contain uvicorn.run in the __main__ block" ) def test_module_source_contains_main_guard(self): import api.aiva_memory_api as module source = inspect.getsource(module) assert '__name__ == "__main__"' in source or "__name__ == '__main__'" in source, ( "uvicorn.run must be inside if __name__ == '__main__': guard" ) def test_no_sqlite3_in_module(self): import api.aiva_memory_api as module source = inspect.getsource(module) assert "import sqlite3" not in source, ( "sqlite3 is FORBIDDEN in aiva_memory_api.py (Rule 7)" ) # --------------------------------------------------------------------------- # Run summary # --------------------------------------------------------------------------- if __name__ == "__main__": result = pytest.main([__file__, "-v", "--tb=short"]) sys.exit(result)