File size: 4,269 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 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 121 | """Phase 2 — Priority 1 core bug fixes (Bugs A–D)."""
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
import gemini_config as gc # noqa: E402
class _EmptyVE:
camera_indices = {}
browser_feeds = {}
face_results = {}
def get_ai_status(self):
return {"facial": True}
def test_bug_a_empty_feeds_report_zero_cameras():
"""Bug A: tool returns empty → agent report says zero cameras."""
orch.inject_context(
alerts=[], sos_events=[], vision_engine=_EmptyVE(), rooms=[],
detections_history=[], issues=[],
)
feeds = orch.get_active_camera_feeds()
assert feeds["count"] == 0
assert feeds["cameras"] == []
report = orch.report_active_cameras_from_tool()
assert report["count"] == 0
assert "0 active camera" in report["report"].lower()
assert report["source_tool"] == "get_active_camera_feeds"
def test_bug_a_report_matches_feed_count_with_one_camera():
class _OneCamVE(_EmptyVE):
browser_feeds = {"cam-01": {"status": "ACTIVE", "last_seen": "now", "source": "browser"}}
orch.inject_context(
alerts=[], sos_events=[], vision_engine=_OneCamVE(), rooms=[],
detections_history=[], issues=[],
)
report = orch.report_active_cameras_from_tool()
assert report["count"] == 1
assert "cam-01" in report["report"]
def test_bug_b_location_agent_excludes_known_persons():
"""Bug B: LocationAgent must not use get_known_persons for live presence."""
names = orch.location_agent_tool_names()
assert "get_known_persons" not in names
assert "who_is_on_camera_now" in names
assert "get_live_camera_faces" in names
assert not orch.agents_must_not_use_known_persons_for_live()
def test_bug_b_who_is_on_camera_ignores_history():
class _VE:
face_results = {"cam-01": [{"name": "MK", "confidence": 0.9}]}
browser_feeds = {"cam-01": {}}
camera_indices = {}
def get_ai_status(self):
return {"facial": True}
orch.inject_context(
alerts=[], sos_events=[], vision_engine=_VE(), rooms=[],
detections_history=[{"name": "Urvi", "camId": "cam-02"}], issues=[],
)
live = orch.who_is_on_camera_now()
assert live["count"] == 1
assert live["persons"][0]["name"] == "MK"
assert all(p["name"] != "Urvi" for p in live["persons"])
def test_bug_c_normalize_stream_step_drops_empty():
assert orch._normalize_stream_step({"agent": "OrchestratorAgent", "content": " ", "step_type": "thinking"}) is None
step = orch._normalize_stream_step({"agent": "OrchestratorAgent", "content": "OK", "step_type": "response"})
assert step is not None
assert step["content"] == "OK"
def test_bug_d_rate_limit_only_when_chain_exhausted():
exc = Exception("429 RESOURCE_EXHAUSTED quota")
exhausted = orch._friendly_model_error(exc, chain_exhausted=True)
partial = orch._friendly_model_error(exc, chain_exhausted=False)
assert "rate-limited" in exhausted.lower()
assert "rate-limited" not in partial.lower() or "temporary overload" in partial.lower()
def test_bug_d_valid_key_no_false_quota_on_non_rate_error():
os.environ["GEMINI_API_KEY"] = "test-key-present"
msg = orch._friendly_model_error(Exception("network timeout"), chain_exhausted=True)
assert "rate-limited" not in msg.lower()
assert gc.api_key_configured() is True
def test_bug_d_fallback_success_implies_no_error_message():
"""Valid key + successful fallback → no rate-limit user message path."""
calls = []
class _RateLimitError(Exception):
pass
class _FakeModels:
def generate_content(self, model, contents, config):
calls.append(model)
if model != "gemini-3.1-flash-lite":
raise _RateLimitError("429 RESOURCE_EXHAUSTED")
return "ok"
class _FakeClient:
models = _FakeModels()
result = gc.generate_with_fallback(_FakeClient(), tier="pro", contents="hi", config=None)
assert result == "ok"
assert calls[-1] == "gemini-3.1-flash-lite"
|