Spaces:
Sleeping
Sleeping
| 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 | |