Spaces:
Sleeping
Sleeping
File size: 5,185 Bytes
2e818da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | from fastapi.testclient import TestClient
from app.main import app
from app.schemas.project import Project, ProjectFile
from app.services.project_service import ProjectService
import app.routers.control_room as control_room_router
def test_control_room_endpoint_combines_activity_artifacts_and_memory(monkeypatch, tmp_path):
svc = ProjectService(root=str(tmp_path / "registry"))
project = Project(project_id="p1", name="Optimizers", created_at=1, updated_at=2)
project.files = [ProjectFile(filename="adam.pdf", file_id="a" * 64)]
svc.save(project)
monkeypatch.setattr(control_room_router, "_project_service", svc)
class FakeActivity:
def list(self, project_id: str, limit: int = 50):
return []
class FakeMemory:
async def memory_status(self, project_id: str, include_liveness: bool = False):
return {"project_id": project_id, "state": "empty", "profile_dataset": True, "project_dataset": True}
monkeypatch.setattr(control_room_router, "_activity", FakeActivity())
monkeypatch.setattr(control_room_router, "StudentMemoryService", FakeMemory)
monkeypatch.setattr(control_room_router, "_artifact_roots", lambda: {
"graphs": str(tmp_path / "graphs"),
"sessions": str(tmp_path / "sessions"),
"citation_graphs": str(tmp_path / "citation_graphs"),
"cache": str(tmp_path / "cache"),
"annotations": str(tmp_path / "annotations"),
"summaries": str(tmp_path / "summaries"),
"cognee": str(tmp_path / "cognee"),
})
(tmp_path / "graphs").mkdir()
(tmp_path / "graphs" / "p1.json").write_text("[]", encoding="utf-8")
(tmp_path / "citation_graphs").mkdir()
(tmp_path / "citation_graphs" / "p1.json").write_text("{}", encoding="utf-8")
response = TestClient(app).get("/api/projects/p1/control-room?fast=false")
assert response.status_code == 200
body = response.json()
assert body["project"]["project_id"] == "p1"
assert body["memory"]["state"] == "empty"
artifact_status = {artifact["id"]: artifact["status"] for artifact in body["artifacts"]}
assert artifact_status["project_graph"] == "saved"
assert artifact_status["citation_graph"] == "saved"
# Task 5: the observability projection rides along on the same response,
# every previously-existing key stays present and untouched.
assert body["observability"]["status"] in {"not_instrumented", "not_observed", "degraded", "observed"}
assert set(body["observability"].keys()) == {
"status", "latest_trace_id", "pipeline_version", "stages", "retrieval", "exporter", "signoz_url",
}
def test_control_room_endpoint_returns_student_facing_project_status(monkeypatch, tmp_path):
svc = ProjectService(root=str(tmp_path / "registry"))
project = Project(project_id="p2", name="Attention Work", created_at=1, updated_at=2)
project.files = [
ProjectFile(filename="attention.pdf", file_id="attention"),
ProjectFile(filename="adam.pdf", file_id="adam"),
]
svc.save(project)
monkeypatch.setattr(control_room_router, "_project_service", svc)
class FakeActivity:
def list(self, project_id: str, limit: int = 50):
from app.services.project_activity import ProjectActivityEvent
return [
ProjectActivityEvent(
project_id=project_id,
event_type="citation_graph_refresh",
message="Citation network synced",
)
]
class FakeMemory:
async def memory_status(self, project_id: str):
return {"project_id": project_id, "state": "ready"}
class FakeProjectIntelligence:
def status(self, project):
return {
"brief": {
"text": "This project is about attention models and optimizers.",
"updated_at": 3,
"source": "commit",
},
"stats": {
"papers": 2,
"notes": 1,
"pinned_figures": 1,
"highlighted_passages": 2,
"commits": 1,
"recommendation_adds": 0,
"verified_citation_links": 1,
"unresolved_verified_references": 4,
"connectors": 2,
},
"settings_summary": {"manual_profiles": 0, "custom_skills": 0},
}
monkeypatch.setattr(control_room_router, "_activity", FakeActivity())
monkeypatch.setattr(control_room_router, "StudentMemoryService", FakeMemory)
monkeypatch.setattr(control_room_router, "ProjectIntelligenceService", FakeProjectIntelligence)
response = TestClient(app).get("/api/projects/p2/control-room")
assert response.status_code == 200
body = response.json()
assert body["brief"]["text"].startswith("This project is about")
assert body["stats"]["papers"] == 2
assert body["stats"]["pinned_figures"] == 1
assert body["live_log"][0]["message"] == "Citation network synced"
assert "debug_artifacts" in body
assert "observability" in body
|