File size: 2,747 Bytes
c794b6b | 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 | """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 # noqa: E402
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
|