| 1 | """Darwin Cortex - smoke tests. |
| 2 | |
| 3 | Tests that the FastAPI app boots and /ask returns a valid-shaped response |
| 4 | even with all engines down (graceful degradation). |
| 5 | """ |
| 6 | import sys |
| 7 | import os |
| 8 | |
| 9 | sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) |
| 10 | |
| 11 | import pytest |
| 12 | import asyncio |
| 13 | from unittest.mock import AsyncMock, patch |
| 14 | from fastapi.testclient import TestClient |
| 15 | |
| 16 | from models import EngineStatus, Hit |
| 17 | |
| 18 | |
| 19 | def _make_down_result(engine: str): |
| 20 | return ([], EngineStatus(engine=engine, status="down", ms=0)) |
| 21 | |
| 22 | |
| 23 | @pytest.fixture |
| 24 | def client(): |
| 25 | """FastAPI test client with all adapters mocked as down.""" |
| 26 | with patch("adapters.rag.RagAdapter.query", new_callable=AsyncMock, return_value=_make_down_result("rag")), \ |
| 27 | patch("adapters.cbm.CbmAdapter.query", new_callable=AsyncMock, return_value=_make_down_result("cbm")), \ |
| 28 | patch("adapters.serena.SerenaAdapter.query", new_callable=AsyncMock, return_value=_make_down_result("serena")), \ |
| 29 | patch("adapters.memgraph.MemgraphAdapter.query", new_callable=AsyncMock, return_value=_make_down_result("memgraph")), \ |
| 30 | patch("adapters.inbox.InboxAdapter.query", new_callable=AsyncMock, return_value=_make_down_result("inbox")): |
| 31 | from main import app |
| 32 | with TestClient(app) as c: |
| 33 | yield c |
| 34 | |
| 35 | |
| 36 | class TestSmoke: |
| 37 | def test_root_endpoint(self, client): |
| 38 | resp = client.get("/") |
| 39 | assert resp.status_code == 200 |
| 40 | data = resp.json() |
| 41 | assert data["service"] == "darwin-cortex" |
| 42 | |
| 43 | def test_ask_all_engines_down_returns_valid_shape(self, client): |
| 44 | """Even with all engines down, /ask must return a valid AskResponse.""" |
| 45 | resp = client.post("/ask", json={"query": "is Hermes running?", "synthesize": False}) |
| 46 | assert resp.status_code == 200 |
| 47 | data = resp.json() |
| 48 | |
| 49 | assert "answer" in data |
| 50 | assert "receipts" in data |
| 51 | assert "hits" in data |
| 52 | assert "conflicts" in data |
| 53 | assert "engines" in data |
| 54 | assert "took_ms" in data |
| 55 | |
| 56 | engines = {e["engine"]: e["status"] for e in data["engines"]} |
| 57 | for engine in ("rag", "cbm", "serena", "memgraph", "inbox"): |
| 58 | assert engine in engines |
| 59 | assert engines[engine] == "down" |
| 60 | |
| 61 | assert data["hits"] == [] |
| 62 | |
| 63 | def test_ask_mode_search_no_answer(self, client): |
| 64 | """mode=search should return answer=null.""" |
| 65 | resp = client.post("/ask", json={"query": "test", "mode": "search", "synthesize": False}) |
| 66 | assert resp.status_code == 200 |
| 67 | data = resp.json() |
| 68 | assert data["answer"] is None |
| 69 | |
| 70 | def test_health_endpoint(self, client): |
| 71 | resp = client.get("/health") |
| 72 | assert resp.status_code == 200 |
| 73 | data = resp.json() |
| 74 | assert "engines" in data |
| 75 | assert "ok" in data |
| 76 | assert data["ok"] is False |
| 77 | |
| 78 | def test_ask_with_scope_filter(self, client): |
| 79 | """scope filter limits which engines are queried.""" |
| 80 | resp = client.post("/ask", json={ |
| 81 | "query": "test", |
| 82 | "scope": ["rag"], |
| 83 | "synthesize": False, |
| 84 | }) |
| 85 | assert resp.status_code == 200 |
| 86 | data = resp.json() |
| 87 | engine_names = [e["engine"] for e in data["engines"]] |
| 88 | assert engine_names == ["rag"] |
| 89 | |
| 90 | |
| 91 | class TestAskWithHits: |
| 92 | """Tests with mocked engine hits to verify ranking and response shape.""" |
| 93 | |
| 94 | def test_ask_returns_ranked_hits(self): |
| 95 | """Hits from adapters are ranked and returned.""" |
| 96 | from datetime import datetime, timezone, timedelta |
| 97 | |
| 98 | fresh_hit = Hit( |
| 99 | engine="rag", |
| 100 | title="Hermes stopped", |
| 101 | path="/var/log/hermes", |
| 102 | snippet="Hermes stopped - service inactive", |
| 103 | score=0.8, |
| 104 | last_seen=datetime.now(timezone.utc) - timedelta(hours=2), |
| 105 | confidence=0.9, |
| 106 | authority_tier="live-infra", |
| 107 | ) |
| 108 | stale_hit = Hit( |
| 109 | engine="memgraph", |
| 110 | title="Hermes active", |
| 111 | path="memory://hermes", |
| 112 | snippet="Hermes is active and running", |
| 113 | score=0.9, |
| 114 | last_seen=datetime.now(timezone.utc) - timedelta(days=40), |
| 115 | confidence=0.85, |
| 116 | authority_tier="agent-memory", |
| 117 | ) |
| 118 | |
| 119 | rag_result = ([fresh_hit], EngineStatus(engine="rag", status="up", ms=100)) |
| 120 | mem_result = ([stale_hit], EngineStatus(engine="memgraph", status="up", ms=50)) |
| 121 | |
| 122 | with patch("adapters.rag.RagAdapter.query", new_callable=AsyncMock, return_value=rag_result), \ |
| 123 | patch("adapters.cbm.CbmAdapter.query", new_callable=AsyncMock, return_value=_make_down_result("cbm")), \ |
| 124 | patch("adapters.serena.SerenaAdapter.query", new_callable=AsyncMock, return_value=_make_down_result("serena")), \ |
| 125 | patch("adapters.memgraph.MemgraphAdapter.query", new_callable=AsyncMock, return_value=mem_result), \ |
| 126 | patch("adapters.inbox.InboxAdapter.query", new_callable=AsyncMock, return_value=_make_down_result("inbox")): |
| 127 | from main import app |
| 128 | with TestClient(app) as client: |
| 129 | resp = client.post("/ask", json={ |
| 130 | "query": "is Hermes running?", |
| 131 | "synthesize": False, |
| 132 | }) |
| 133 | assert resp.status_code == 200 |
| 134 | data = resp.json() |
| 135 | |
| 136 | hits = data["hits"] |
| 137 | assert len(hits) >= 2 |
| 138 | assert hits[0]["engine"] == "rag", ( |
| 139 | f"Expected live-infra (rag) first, got {hits[0]['engine']}. " |
| 140 | "Darwin's decay valve is not working!" |
| 141 | ) |