| """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 |
| import gemini_config as gc |
|
|
|
|
| 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" |
|
|
