| """Phase 3 — Bug A tool-layer + Bug C generation-point fixes.""" |
| import os |
| import sys |
|
|
| os.environ.setdefault("CEPHEUS_CLOUD", "1") |
| BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| if BACKEND_DIR not in sys.path: |
| sys.path.insert(0, BACKEND_DIR) |
|
|
| import agentic_orchestrator as orch |
|
|
|
|
| class _FaceOnlyVE: |
| """Simulates active camera with face_results but empty browser_feeds (original Bug A).""" |
| camera_indices = {} |
| browser_feeds = {} |
| active_cameras = {} |
| face_results = {"cam-01": [{"name": "MK", "confidence": 0.92}]} |
|
|
| def get_ai_status(self): |
| return {"facial": True} |
|
|
|
|
| class _LocalOpenCVVE: |
| camera_indices = {"cam-02": 0} |
| browser_feeds = {} |
| active_cameras = {"cam-02": object()} |
| face_results = {} |
|
|
| def get_ai_status(self): |
| return {"facial": True} |
|
|
|
|
| def test_bug_a_tool_returns_camera_from_face_results_only(): |
| """Tool layer: face_results without browser_feeds must list the camera.""" |
| orch.inject_context( |
| alerts=[], sos_events=[], vision_engine=_FaceOnlyVE(), rooms=[], |
| detections_history=[], issues=[], |
| ) |
| feeds = orch.get_active_camera_feeds() |
| assert feeds["count"] == 1 |
| assert feeds["cameras"][0]["cam_id"] == "cam-01" |
| report = orch.report_active_cameras_from_tool() |
| assert report["count"] == 1 |
| assert "cam-01" in report["report"] |
|
|
|
|
| def test_bug_a_tool_returns_local_opencv_camera(): |
| orch.inject_context( |
| alerts=[], sos_events=[], vision_engine=_LocalOpenCVVE(), rooms=[], |
| detections_history=[], issues=[], |
| ) |
| feeds = orch.get_active_camera_feeds() |
| assert feeds["count"] == 1 |
| assert feeds["cameras"][0]["cam_id"] == "cam-02" |
|
|
|
|
| def test_bug_a_e2e_active_camera_tool_and_agent_report(): |
| """End-to-end: active camera → tool lists it → agent report matches.""" |
| orch.inject_context( |
| alerts=[], sos_events=[], vision_engine=_FaceOnlyVE(), rooms=[], |
| detections_history=[], issues=[], |
| ) |
| feeds = orch.get_active_camera_feeds() |
| agent_report = orch.report_active_cameras_from_tool() |
| assert feeds["count"] == agent_report["count"] == 1 |
| assert orch.format_active_camera_report(feeds) == agent_report["report"] |
|
|
|
|
| def test_bug_c_make_text_step_blocks_empty_at_source(): |
| assert orch._make_text_step("OrchestratorAgent", " ", "thinking") is None |
| assert orch._make_text_step("OrchestratorAgent", "", "response") is None |
| step = orch._make_text_step("OrchestratorAgent", " hello ", "response") |
| assert step["content"] == "hello" |
|
|
|
|
| def test_bug_c_empty_steps_not_in_normalized_stream(): |
| assert orch._normalize_stream_step({"agent": "OrchestratorAgent", "content": "\n", "step_type": "thinking"}) is None |
|
|
