"""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 # noqa: E402 import gemini_config as gc # noqa: E402 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 # Every tool maps back to a callable 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 # Falls back to a sane default when no delay is present 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"] # Pro tier degrades Pro -> Flash -> Flash-Lite (deduped, ordered). 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" # It tried Pro and Flash before succeeding on 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=[], ) # Known recent sighting -> RECENTLY_SEEN but not live (found=false). 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" # Absent person -> not found, no fabrication. 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"] # update_issue mutates the injected list deterministically. 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 # Action tools degrade gracefully when no handler is wired. 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", "")