File size: 4,672 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 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """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
|