| """Unit tests for the agentic orchestrator native (google.genai) fallback.""" |
| 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 |
|
|
|
|
| def test_tool_registry_is_complete(): |
| names = set(orch._TOOL_MAP) |
| for expected in ( |
| "get_current_alerts", |
| "get_crowd_statistics", |
| "search_person_in_camera_feeds", |
| "create_tactical_issue", |
| "get_interaction_graph", |
| "locate_nearest_emergency_services", |
| "list_enrolled_faces", |
| "set_ai_model_state", |
| "get_active_issues", |
| "get_platform_overview", |
| "broadcast_alert", |
| "set_signage_state", |
| "update_issue", |
| "get_live_camera_faces", |
| ): |
| assert expected in names |
| assert len(orch._ALL_TOOLS) >= 18 |
| |
| assert all(callable(fn) for fn in orch._TOOL_MAP.values()) |
|
|
|
|
| def test_parse_retry_delay_extracts_seconds(): |
| msg = "429 RESOURCE_EXHAUSTED ... 'retryDelay': '58s' ..." |
| assert gc.parse_retry_delay(msg) >= 58 |
| |
| assert gc.parse_retry_delay("no retry hint here") == 5.0 |
|
|
|
|
| def test_gemini_model_tiers(): |
| assert gc.get_model("default") == "gemini-3.5-flash" |
| assert gc.get_model("pro") == "gemini-3.1-pro-preview" |
| assert gc.get_model("lite") == "gemini-3.1-flash-lite" |
| assert gc.fallback_chain("default") == ["gemini-3.5-flash", "gemini-3.1-flash-lite"] |
| |
| assert gc.fallback_chain("pro") == [ |
| "gemini-3.1-pro-preview", |
| "gemini-3.5-flash", |
| "gemini-3.1-flash-lite", |
| ] |
|
|
|
|
| def test_generate_with_fallback_degrades_on_rate_limit(): |
| """A quota-exhausted primary model should transparently fall through to a working one.""" |
| 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 f"ok:{model}" |
|
|
| class _FakeClient: |
| models = _FakeModels() |
|
|
| result = gc.generate_with_fallback(_FakeClient(), tier="pro", contents="hi", config=None) |
| assert result == "ok:gemini-3.1-flash-lite" |
| |
| assert calls[-1] == "gemini-3.1-flash-lite" |
| assert "gemini-3.1-pro-preview" in calls |
|
|
|
|
| def test_search_person_is_deterministic_and_honest(): |
| """Search must reflect real injected data, not guesses, and be repeatable.""" |
| detections = [{"name": "Vidit", "camId": "cam-02", "confidence": 0.9, "timestamp": "10:00:00"}] |
| orch.inject_context( |
| alerts=[], sos_events=[], vision_engine=None, rooms=[], |
| detections_history=detections, issues=[], |
| ) |
| |
| r1 = orch.search_person_in_camera_feeds("Vidit") |
| r2 = orch.search_person_in_camera_feeds("Vidit") |
| assert r1["found"] is False and r1 == r2 |
| assert r1["status"] == "RECENTLY_SEEN" |
| assert r1["last_seen_camera"] == "cam-02" |
| |
| miss = orch.search_person_in_camera_feeds("Nobody") |
| assert miss["found"] is False |
| assert miss["status"] == "NOT_FOUND" |
|
|
|
|
| def test_get_live_camera_faces_uses_face_results_only(): |
| class FakeVE: |
| face_results = { |
| "cam-01": [{"name": "MK", "confidence": 0.9, "found": True}], |
| } |
| browser_feeds = {"cam-01": {"status": "ACTIVE", "last_seen": "now", "source": "browser"}} |
| camera_indices = {"cam-01": 0} |
|
|
| def get_ai_status(self): |
| return {"facial": True} |
|
|
| ve = FakeVE() |
| orch.inject_context( |
| alerts=[], sos_events=[], vision_engine=ve, rooms=[], |
| detections_history=[{"name": "Vidit", "camId": "cam-01"}], issues=[], |
| ) |
| live = orch.get_live_camera_faces() |
| assert live["total_live_faces"] == 1 |
| assert live["live_persons"][0]["name"] == "MK" |
| assert live["live_persons"][0]["status"] == "LIVE_DETECTED" |
| assert live["active_feed_count"] >= 1 |
| hist = orch.get_known_persons() |
| assert hist["data_type"] == "SIGHTING_HISTORY" |
| assert len(hist["persons"]) == 1 |
| assert hist["persons"][0]["name"] == "Vidit" |
|
|
|
|
| def test_friendly_model_error_messages(): |
| quota = orch._friendly_model_error(Exception("429 RESOURCE_EXHAUSTED quota")) |
| assert "rate-limited" in quota.lower() |
|
|
| auth = orch._friendly_model_error(Exception("Invalid API key provided")) |
| assert "authentication" in auth.lower() |
|
|
| generic = orch._friendly_model_error(Exception("something else broke")) |
| assert "failed" in generic.lower() |
|
|
|
|
| def test_create_tactical_issue_deduplicates_same_subject(): |
| issues = [] |
| orch.inject_context( |
| alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=issues |
| ) |
| orch._ctx["allow_issue_creation"] = True |
| first = orch.create_tactical_issue( |
| "Missing Person Database Match: MK", |
| "Database match for MK (Confidence: 0.744)", |
| "HIGH", |
| '{"person_name":"MK","confidence":0.744}', |
| ) |
| second = orch.create_tactical_issue( |
| "Database Match: MK (Pending Live Verification)", |
| "Database match for MK (Confidence: 0.763)", |
| "HIGH", |
| '{"person_name":"MK","confidence":0.763}', |
| ) |
| assert first["success"] is True |
| assert second["success"] is True |
| assert second.get("deduplicated") is True |
| assert len(issues) == 1 |
| assert issues[0]["metadata"].get("confidence") == 0.763 |
|
|
|
|
| def test_create_tactical_issue_blocked_without_explicit_intent(): |
| issues = [] |
| orch.inject_context( |
| alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=issues |
| ) |
| orch._ctx["allow_issue_creation"] = False |
| result = orch.create_tactical_issue( |
| "Missing Person Database Match: MK", |
| "Database match for MK", |
| "HIGH", |
| '{"person_name":"MK","confidence":0.744}', |
| ) |
| assert result["success"] is False |
| assert result.get("blocked") is True |
| assert len(issues) == 0 |
|
|
|
|
| def test_prompt_requests_issue_creation(): |
| assert orch.prompt_requests_issue_creation("Search MK and raise an issue if found") |
| assert orch.prompt_requests_issue_creation("Please create an issue for this match") |
| assert not orch.prompt_requests_issue_creation("Search the live camera feed for enrolled faces") |
| assert not orch.prompt_requests_issue_creation("Run a face scan on the current camera feed") |
|
|
|
|
| def test_create_tactical_issue_uses_injected_db(): |
| issues = [] |
| orch.inject_context( |
| alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=issues |
| ) |
| orch._ctx["allow_issue_creation"] = True |
| result = orch.create_tactical_issue("Test", "A test issue", "HIGH", "{}") |
| assert result["success"] is True |
| assert issues and issues[0]["title"] == "Test" |
|
|
|
|
| def test_platform_overview_and_issue_update(): |
| issues = [{"id": "ISS-1", "title": "X", "status": "ONGOING", "progress": 0, "desc": "d"}] |
| orch.inject_context( |
| alerts=[{"id": "a1"}], sos_events=[], vision_engine=None, rooms=[], |
| detections_history=[{"name": "Vidit"}], issues=issues, |
| incident_logs=[{"msg": "boot"}], signage_state={"s6": True, "s1": False}, |
| ) |
| ov = orch.get_platform_overview() |
| assert ov["active_alerts"] == 1 |
| assert ov["active_issues"] == 1 |
| assert ov["live_face_count"] == 0 |
| assert ov["active_signage"] == ["s6"] |
|
|
| |
| res = orch.update_issue("ISS-1", status="RESOLVED", progress=100, note="done") |
| assert res["success"] is True |
| assert issues[0]["status"] == "RESOLVED" and issues[0]["progress"] == 100 |
|
|
| |
| orch._action_handlers.clear() |
| alert_res = orch.broadcast_alert("fire", "Smoke in corridor", "Corridor B", "critical") |
| assert alert_res["success"] is False |
|
|
|
|
| def test_action_handlers_invoked(): |
| captured = {} |
| orch.set_action_handlers( |
| broadcast_alert=lambda t, m, l, s: captured.update({"alert": (t, m, l, s)}) or {"id": "x"}, |
| set_signage=lambda sid, active: captured.update({"signage": (sid, active)}), |
| ) |
| orch.inject_context(alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=[]) |
| assert orch.broadcast_alert("medical", "Patient down", "Hall", "high")["success"] is True |
| assert captured["alert"] == ("medical", "Patient down", "Hall", "high") |
| assert orch.set_signage_state("s3", True)["success"] is True |
| assert captured["signage"] == ("s3", True) |
| orch._action_handlers.clear() |
|
|
|
|
| def test_broadcast_alert_message_location_order(): |
| captured = {} |
| orch.set_action_handlers( |
| broadcast_alert=lambda t, m, l, s: captured.update({"alert": (t, m, l, s)}) or {"id": "x"}, |
| ) |
| orch.inject_context(alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=[]) |
| orch.broadcast_alert("fire", "Smoke in corridor", "Building A", "critical") |
| assert captured["alert"][1] == "Smoke in corridor" |
| assert captured["alert"][2] == "Building A" |
| orch._action_handlers.clear() |
|
|
|
|
| def test_dispatch_personnel_requires_linked_issue(): |
| issues = [{"id": "ISS-OTHER", "title": "Unrelated", "desc": "elsewhere", "metadata": {}}] |
| orch.inject_context(alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=issues) |
| res = orch.dispatch_personnel_to_camera("cam-99", 2) |
| assert res["success"] is False |
| assert "create_tactical_issue" in res["message"] |
|
|
|
|
| def test_adk_tool_registry_parity(): |
| native_names = set(orch._TOOL_MAP) |
| adk_names = orch.adk_tool_names() |
| missing = native_names - adk_names |
| assert not missing, f"ADK agents missing tools: {sorted(missing)}" |
|
|
|
|
| def test_decode_data_url_image(): |
| raw = b"fake-image-bytes" |
| import base64 as b64 |
| data_url = "data:image/png;base64," + b64.b64encode(raw).decode() |
| decoded, mime = orch._decode_data_url_image(data_url) |
| assert decoded == raw |
| assert mime == "image/png" |
|
|
|
|
| def test_update_issue_accepts_float_progress(): |
| issues = [{"id": "ISS-1", "title": "X", "status": "ONGOING", "progress": 0, "desc": "d"}] |
| orch.inject_context(alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=issues) |
| res = orch.update_issue("ISS-1", progress=50.6) |
| assert res["success"] is True |
| assert issues[0]["progress"] == 51 |
|
|
|
|
| def test_evacuation_plan_marked_unverified(): |
| plan = orch.generate_evacuation_plan("PLATFORM 1", 100, "fire") |
| assert plan.get("verified") is False |
| assert "TEMPLATE" in plan.get("note", "") |
|
|
