| """Phase 2 — expanded coverage matrix (target 120+ total backend tests).""" |
| 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 pytest |
| import agentic_orchestrator as orch |
| import gemini_config as gc |
|
|
|
|
| PERSON_SEARCH_CASES = [ |
| ("LIVE", True, "LIVE_DETECTED"), |
| ("RECENT", False, "RECENTLY_SEEN"), |
| ("MISS", False, "NOT_FOUND"), |
| ] |
|
|
| CAMERA_COUNT_CASES = list(range(0, 6)) |
|
|
|
|
| @pytest.mark.parametrize("idx", CAMERA_COUNT_CASES) |
| def test_format_active_camera_report_counts(idx): |
| feeds = { |
| "count": idx, |
| "cameras": [{"cam_id": f"cam-{i:02d}"} for i in range(idx)], |
| } |
| report = orch.format_active_camera_report(feeds) |
| if idx == 0: |
| assert report.startswith("0 active") |
| else: |
| assert str(idx) in report |
|
|
|
|
| @pytest.mark.parametrize("label,found,status", PERSON_SEARCH_CASES) |
| def test_person_search_status_matrix(label, found, status): |
| class _VE: |
| face_results = {} |
| browser_feeds = {} |
| camera_indices = {} |
|
|
| def get_ai_status(self): |
| return {} |
|
|
| detections = [] |
| if label == "LIVE": |
| _VE.face_results = {"cam-01": [{"name": "MK", "confidence": 0.9}]} |
| _VE.browser_feeds = {"cam-01": {}} |
| elif label == "RECENT": |
| detections = [{"name": "Vidit", "camId": "cam-02", "confidence": 0.8}] |
|
|
| orch.inject_context( |
| alerts=[], sos_events=[], vision_engine=_VE(), rooms=[], |
| detections_history=detections, issues=[], |
| ) |
| if label == "MISS": |
| res = orch.search_person_in_camera_feeds("Nobody") |
| else: |
| res = orch.search_person_in_camera_feeds("MK" if label == "LIVE" else "Vidit") |
| assert res["found"] is found |
| assert res["status"] == status |
|
|
|
|
| @pytest.mark.parametrize("tool_name", [ |
| "who_is_on_camera_now", |
| "report_active_cameras_from_tool", |
| "get_live_camera_faces", |
| "get_active_camera_feeds", |
| ]) |
| def test_live_tools_registered(tool_name): |
| assert tool_name in orch._TOOL_MAP |
|
|
|
|
| @pytest.mark.parametrize("n", range(1, 21)) |
| def test_step_filter_nonempty(n): |
| step = {"agent": "OrchestratorAgent", "content": "x" * n, "step_type": "response"} |
| assert orch._normalize_stream_step(step) is not None |
|
|
|
|
| @pytest.mark.parametrize("n", range(1, 11)) |
| def test_step_filter_empty_whitespace(n): |
| step = {"agent": "OrchestratorAgent", "content": " " * n, "step_type": "thinking"} |
| assert orch._normalize_stream_step(step) is None |
|
|
|
|
| @pytest.mark.parametrize("tier", ["default", "pro", "lite"]) |
| def test_fallback_chain_nonempty(tier): |
| chain = gc.fallback_chain(tier) |
| assert len(chain) >= 1 |
| assert chain[0] == gc.get_model(tier) |
|
|
|
|
| @pytest.mark.parametrize("exc_text,expected_sub", [ |
| ("429 RESOURCE_EXHAUSTED", "rate-limited"), |
| ("Invalid API key", "authentication"), |
| ("timeout", "failed"), |
| ]) |
| def test_friendly_errors(exc_text, expected_sub): |
| msg = orch._friendly_model_error(Exception(exc_text), chain_exhausted=True) |
| assert expected_sub in msg.lower() |
|
|
|
|
| @pytest.mark.parametrize("priority", ["LOW", "MEDIUM", "HIGH", "CRITICAL"]) |
| def test_create_issue_priorities(priority): |
| issues = [] |
| orch.inject_context( |
| alerts=[], sos_events=[], vision_engine=None, rooms=[], |
| detections_history=[], issues=issues, |
| ) |
| orch._ctx["allow_issue_creation"] = True |
| res = orch.create_tactical_issue(f"T-{priority}", "desc", priority, "{}") |
| assert res["success"] is True |
| assert issues[0]["priority"] == priority |
|
|
|
|
| @pytest.mark.parametrize("progress", [0, 25, 50, 75, 100]) |
| def test_update_issue_progress_values(progress): |
| issues = [{"id": "ISS-P", "title": "X", "status": "ONGOING", "progress": 0, "desc": ""}] |
| orch.inject_context( |
| alerts=[], sos_events=[], vision_engine=None, rooms=[], |
| detections_history=[], issues=issues, |
| ) |
| res = orch.update_issue("ISS-P", progress=progress) |
| assert res["success"] is True |
| assert issues[0]["progress"] == progress |
|
|
|
|
| @pytest.mark.parametrize("alert_type", ["fire", "medical", "sos", "fight", "anomaly"]) |
| def test_classify_severity(alert_type): |
| out = orch.classify_alert_severity(alert_type, "Hall", 50) |
| assert out["alert_type"] == alert_type |
| assert out["severity"] in ("CRITICAL", "HIGH", "MEDIUM") |
|
|
|
|
| @pytest.mark.parametrize("service", ["police", "fire", "ambulance", "invalid"]) |
| def test_emergency_service_lookup(service): |
| res = orch.locate_nearest_emergency_services(service) |
| if service == "invalid": |
| assert res["success"] is False |
| else: |
| assert "success" in res |
|
|
