#!/usr/bin/env python3 """ Unit tests for mr-number-lookup. Exercise the whole device path (adb control, navigation, full-history capture, vision extraction, consolidation, multi-axis rating, result mapping, and the screening record) **without** a real device, adb, app, vision, or network. Run from this directory: python3 -m unittest mr_lookup_test -v """ import unittest from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import sys sys.path.insert(0, str(Path(__file__).parent)) import mr_lookup class TestDecideResultFallback(unittest.TestCase): """The deterministic fallback heuristic (used only if the SDK profile fails).""" def test_denied_on_negative_flags(self): extracted = {"reports": ["no show last week", "was rude over text"], "red_flags": ["cop vibes"], "suggested_result": "approved"} self.assertEqual(mr_lookup.decide_result(extracted), "denied") def test_denied_on_hyphenated_variant(self): # The historical bug: 'no-show' (hyphen) must still match. extracted = {"reports": ["total no-show, ghosted me"], "red_flags": [], "report_count": 1} self.assertEqual(mr_lookup.decide_result(extracted), "denied") def test_never_approves_over_model_denied(self): # Even with clean-looking text, a model 'denied' is honored (the real bug). extracted = {"report_count": 3, "reports": ["seemed ok"], "suggested_result": "denied"} self.assertEqual(mr_lookup.decide_result(extracted), "denied") def test_deposit_is_not_negative(self): # 'deposit' must NOT trip the negative keywords. extracted = {"report_count": 1, "reports": ["always sends a deposit, great client"], "suggested_result": "approved"} self.assertEqual(mr_lookup.decide_result(extracted), "approved") def test_falls_back_to_suggested(self): self.assertEqual(mr_lookup.decide_result({"report_count": 0, "suggested_result": "not_found"}), "not_found") def test_pending_default(self): self.assertEqual(mr_lookup.decide_result({}), "pending") class TestRatingMapping(unittest.TestCase): """Pure score/grade/result mapping + the safety override.""" def test_grade_bands(self): self.assertEqual(mr_lookup.grade_from_score(90), "A") self.assertEqual(mr_lookup.grade_from_score(75), "B") self.assertEqual(mr_lookup.grade_from_score(60), "C") self.assertEqual(mr_lookup.grade_from_score(45), "D") self.assertEqual(mr_lookup.grade_from_score(20), "F") def test_result_from_score(self): self.assertEqual(mr_lookup.result_from_score(80), "approved") self.assertEqual(mr_lookup.result_from_score(55), "pending") self.assertEqual(mr_lookup.result_from_score(30), "denied") self.assertEqual(mr_lookup.result_from_score(None), "pending") def test_profile_prefers_recommendation(self): prof = {"score": 90, "recommended_result": "pending", "axes": {"safety": {"score": 90}}} self.assertEqual(mr_lookup.result_from_profile(prof), "pending") def test_profile_safety_override_forces_denied(self): # High overall score but a law-enforcement/safety signal → denied regardless. prof = {"score": 88, "recommended_result": "approved", "axes": {"safety": {"score": 10}}} self.assertEqual(mr_lookup.result_from_profile(prof), "denied") def test_profile_none_is_pending(self): self.assertEqual(mr_lookup.result_from_profile(None), "pending") class TestMergeReports(unittest.TestCase): """Consolidation across multiple screenshots: dedupe + counts.""" def test_dedupes_and_unions(self): extractions = [ {"reports": ["paid deposit", "On time"], "red_flags": ["none"], "classification": "Personal Line", "report_count": 4}, {"reports": ["paid deposit", " on time ", "ghosted once"], "red_flags": ["ghosting"], "report_count": 4}, ] merged = mr_lookup.merge_reports(extractions, "+15551112222") # 'paid deposit' and 'On time'/'on time' dedupe case/space-insensitively → 3 unique self.assertEqual(merged["captured_count"], 3) self.assertEqual(merged["declared_count"], 4) self.assertEqual(merged["classification"], "Personal Line") self.assertIn("ghosting", merged["red_flags"]) # Clean caller → no critical safety flags promoted. self.assertEqual(merged["safety_flags"], []) class TestSafetyFlags(unittest.TestCase): """Deterministic promotion of critical safety signals out of the report text.""" def test_spanish_police_is_promoted_and_folded(self): # The real-world miss: 'Es policía' (accented Spanish) buried in red_flags. flags = mr_lookup.detect_safety_flags(["Es policía"], ["law enforcement impersonation (Es policía)"]) self.assertEqual(len(flags), 1) f = flags[0] self.assertEqual(f["category"], "law_enforcement") self.assertEqual(f["severity"], "critical") self.assertEqual(f["icon"], "🚔") self.assertIn("Es policía", f["evidence"]) def test_english_cop_synonyms(self): for term in ("He's a cop", "suspected LEO", "undercover sting", "this is the police"): flags = mr_lookup.detect_safety_flags([term], []) self.assertTrue(flags and flags[0]["category"] == "law_enforcement", term) def test_no_false_positive_on_substrings(self): # 'problem' contains 'rob', 'copy' contains 'cop' — must NOT trip (word boundaries). flags = mr_lookup.detect_safety_flags(["no problem, sent a copy of ID", "policy questions"], []) self.assertEqual(flags, []) def test_distinct_categories_separated(self): flags = mr_lookup.detect_safety_flags(["he robbed me", "threatened me", "es policia"], []) cats = {f["category"] for f in flags} self.assertEqual(cats, {"robbery", "coercion", "law_enforcement"}) def test_override_forces_denied_over_clean_result(self): sf = [{"category": "law_enforcement", "severity": "critical", "icon": "🚔", "label": "Law enforcement / sting", "evidence": ["Es policía"]}] self.assertEqual(mr_lookup.apply_safety_override("approved", sf), "denied") self.assertEqual(mr_lookup.apply_safety_override("approved", []), "approved") self.assertTrue(mr_lookup.has_critical_safety_flag(sf)) class TestScreeningSignalValue(unittest.TestCase): """Verdict → the bare valueText that the people-signal carries (consumer contract).""" def _cop(self): return [{"category": "law_enforcement", "severity": "critical", "evidence": ["Es policía"]}] def _violence(self): return [{"category": "violence", "severity": "critical", "evidence": ["weapon"]}] def test_law_enforcement_is_cop_flag(self): # cop_flag is the distinct LE value (cop overrides approval) even on a 'denied' result. self.assertEqual(mr_lookup.screening_signal_value("denied", self._cop()), "cop_flag") self.assertEqual(mr_lookup.screening_signal_value("approved", self._cop()), "cop_flag") def test_other_critical_flags_ride_denied(self): # Non-LE critical flags are not cop_flag; they map to the (overridden) 'denied' result. self.assertEqual(mr_lookup.screening_signal_value("denied", self._violence()), "denied") def test_plain_verdicts_passthrough(self): self.assertEqual(mr_lookup.screening_signal_value("denied", []), "denied") self.assertEqual(mr_lookup.screening_signal_value("approved", []), "approved") self.assertEqual(mr_lookup.screening_signal_value("error", []), "error") def test_pending_and_not_found_are_none(self): # Consumers read a missing valueText as 'not_screened'. self.assertIsNone(mr_lookup.screening_signal_value("pending", [])) self.assertIsNone(mr_lookup.screening_signal_value("not_found", [])) class TestOcrGpuBackends(unittest.IsolatedAsyncioTestCase): """The on-box (Claude-API-less) path: OCR extraction + GPU OpenAI-compatible rating.""" def test_extract_json_variants(self): self.assertEqual(mr_lookup._extract_json('```json\n{"a": 1}\n```'), {"a": 1}) self.assertEqual(mr_lookup._extract_json('blah {"score": 8} trailing'), {"score": 8}) self.assertIsNone(mr_lookup._extract_json("no json here")) def test_merge_collects_ocr_and_flags_from_raw_text(self): # OCR extractions carry raw_ocr (no structured reports). A LE term in the raw text # must still trip the safety flag before the rating LLM structures anything. ex = [{"reports": [], "red_flags": [], "raw_ocr": "User reports\nEs policía\nasks for address"}] h = mr_lookup.merge_reports(ex, "+16315304426") self.assertEqual(h["reports"], []) self.assertIn("Es policía", h["ocr_text"]) self.assertTrue(any(f["category"] == "law_enforcement" for f in h["safety_flags"])) async def test_extract_from_screenshot_uses_ocr_backend(self): with patch("mr_lookup.extract_backend", return_value="ocr"), \ patch("mr_lookup.ocr_extract", return_value={"ok": True, "text": "raw screen text", "lines": ["raw screen text"]}) as m: out = await mr_lookup._extract_from_screenshot("/tmp/s.png", "+15551234567") m.assert_called_once() self.assertEqual(out["raw_ocr"], "raw screen text") self.assertEqual(out["reports"], []) async def test_rating_uses_gpu_when_url_set(self): gpu_json = '{"score": 8, "recommended_result": "denied", "reports": ["no-show, ghosted"], "axes": {"safety": {"score": 0}}}' with patch("mr_lookup.rating_llm_url", return_value="http://10.20.0.9:8000"), \ patch("mr_lookup.rating_llm_model", return_value="qwen"), \ patch("mr_lookup.openai_chat", return_value=gpu_json) as chat: prof = await mr_lookup.build_rating_profile({"ocr_text": "Es policia\nno show", "reports": []}) chat.assert_called_once() self.assertEqual(prof["score"], 8) self.assertEqual(prof["grade"], "F") # normalized from score self.assertEqual(prof["reports"], ["no-show, ghosted"]) class TestFullFlow(unittest.IsolatedAsyncioTestCase): """End-to-end device path with the expensive parts mocked.""" async def test_records_correct_wire_body_with_rating(self): phone = "+15551234567" ref = "prospect-42" shots = [Path("/tmp/s0.png"), Path("/tmp/s1.png")] fake_extracted = { "phone": phone, "report_count": 4, "reports": ["no-show, ghosted", "time waster"], "red_flags": ["no-show", "ghosting"], "classification": "Personal Line", "suggested_result": "denied", } fake_profile = { "score": 18, "grade": "F", "is_mixed": False, "axes": {"reliability": {"score": 10}, "payment": {"score": 40}, "respect": {"score": 30}, "safety": {"score": 70}}, "recommended_result": "denied", "summary": "Repeated no-shows and time-wasting.", } mock_requests = MagicMock() mock_post = mock_requests.post mock_post.return_value.json.return_value = {"id": 999, "status": "created"} mock_post.return_value.raise_for_status = MagicMock() with patch("mr_lookup.launch_app"), \ patch("mr_lookup.find_and_tap_text", return_value=True), \ patch("mr_lookup.find_edit_text_and_input", return_value=True), \ patch("mr_lookup.go_to_search", return_value=True), \ patch("mr_lookup.open_report_detail", return_value=True), \ patch("mr_lookup.expand_all_reports", return_value=True), \ patch("mr_lookup.capture_full_history", return_value=shots), \ patch("mr_lookup._extract_from_screenshot", new_callable=AsyncMock, return_value=fake_extracted), \ patch("mr_lookup.build_rating_profile", new_callable=AsyncMock, return_value=fake_profile), \ patch("mr_lookup.save_history", return_value=Path("/tmp/hist.json")), \ patch.dict("sys.modules", {"requests": mock_requests}), \ patch("mr_lookup.PEOPLE_SERVICE_TOKEN", "fake-token"), \ patch("mr_lookup.time.sleep"): out = await mr_lookup.main_async(phone=phone, ref=ref, dry_run=False) # Result comes from the rating profile (denied), score/grade surfaced. self.assertEqual(out["result"], "denied") self.assertEqual(out["score"], 18) self.assertEqual(out["grade"], "F") # The actual wire body = a people-service signal (POST /internal/people/signals). mock_post.assert_called_once() url = mock_post.call_args[0][0] self.assertTrue(url.endswith("/internal/people/signals"), url) body = mock_post.call_args[1].get("json", {}) self.assertEqual(body.get("handle"), phone) self.assertEqual(body.get("channel"), "sms") self.assertEqual(body.get("signalType"), "screening_mrnumber") self.assertEqual(body.get("sourceFeature"), "mr-number") self.assertEqual(body.get("sourceHandle"), ref) # correlation id self.assertEqual(body.get("valueText"), "denied") # bare verdict consumers switch on # Rich detail rides valueJsonb (the full history + profile). self.assertEqual(body["valueJsonb"]["rating_profile"]["score"], 18) self.assertIn("time waster", body["valueJsonb"]["reports"]) async def test_dry_run_does_not_record(self): with patch("mr_lookup.launch_app"), \ patch("mr_lookup.find_and_tap_text", return_value=True), \ patch("mr_lookup.find_edit_text_and_input", return_value=True), \ patch("mr_lookup.go_to_search", return_value=True), \ patch("mr_lookup.open_report_detail", return_value=True), \ patch("mr_lookup.expand_all_reports", return_value=False), \ patch("mr_lookup.capture_full_history", return_value=[Path("/tmp/s0.png")]), \ patch("mr_lookup._extract_from_screenshot", new_callable=AsyncMock, return_value={"report_count": 0, "reports": [], "suggested_result": "not_found"}), \ patch("mr_lookup.build_rating_profile", new_callable=AsyncMock, return_value=None), \ patch("mr_lookup.save_history", return_value=Path("/tmp/hist.json")), \ patch("mr_lookup.record_screening") as mock_record, \ patch("mr_lookup.time.sleep"): out = await mr_lookup.main_async(phone="+10000000000", ref="x", dry_run=True) mock_record.assert_not_called() # No reports + no profile → fallback heuristic → pending. self.assertEqual(out["result"], "pending") async def test_aborts_when_detail_does_not_match(self): # The stale-page bug: if we can't confirm the right caller's detail, never rate. with patch("mr_lookup.launch_app"), \ patch("mr_lookup.go_to_search", return_value=True), \ patch("mr_lookup.find_and_tap_text", return_value=True), \ patch("mr_lookup.find_edit_text_and_input", return_value=True), \ patch("mr_lookup.open_report_detail", return_value=False), \ patch("mr_lookup.take_screenshot", return_value=Path("/tmp/nomatch.png")), \ patch("mr_lookup.capture_full_history") as mock_cap, \ patch("mr_lookup.build_rating_profile", new_callable=AsyncMock) as mock_rate, \ patch("mr_lookup.record_screening") as mock_record, \ patch("mr_lookup.time.sleep"): out = await mr_lookup.main_async(phone="+16315304426", ref="7", dry_run=False) self.assertEqual(out["result"], "error") self.assertEqual(out["error"], "detail_not_loaded") mock_cap.assert_not_called() mock_rate.assert_not_called() mock_record.assert_not_called() async def test_box_path_ocr_extract_plus_gpu_rating(self): # The real on-box flow: OCR backend (no structured reports) → GPU rating extracts # the reports; a LE term in the OCR text forces cop_flag, folded into the signal. phone = "+16315304426" gpu_json = ('{"score": 3, "recommended_result": "denied", ' '"reports": ["Es policia", "no show after booking"], ' '"axes": {"safety": {"score": 0}}, "summary": "LE-adjacent; deny."}') mock_requests = MagicMock() mock_post = mock_requests.post mock_post.return_value.json.return_value = {"id": 7, "status": "created"} mock_post.return_value.raise_for_status = MagicMock() with patch("mr_lookup.launch_app"), \ patch("mr_lookup.find_and_tap_text", return_value=True), \ patch("mr_lookup.find_edit_text_and_input", return_value=True), \ patch("mr_lookup.go_to_search", return_value=True), \ patch("mr_lookup.open_report_detail", return_value=True), \ patch("mr_lookup.expand_all_reports", return_value=True), \ patch("mr_lookup.capture_full_history", return_value=[Path("/tmp/s0.png")]), \ patch("mr_lookup.extract_backend", return_value="ocr"), \ patch("mr_lookup.ocr_extract", return_value={"ok": True, "text": "User reports\nEs policia\nno show after booking", "lines": []}), \ patch("mr_lookup.rating_llm_url", return_value="http://10.20.0.9:8000"), \ patch("mr_lookup.rating_llm_model", return_value="qwen"), \ patch("mr_lookup.openai_chat", return_value=gpu_json), \ patch("mr_lookup.save_history", return_value=Path("/tmp/hist.json")), \ patch.dict("sys.modules", {"requests": mock_requests}), \ patch("mr_lookup.PEOPLE_SERVICE_TOKEN", "fake-token"), \ patch("mr_lookup.time.sleep"): out = await mr_lookup.main_async(phone=phone, ref="prospect-9", dry_run=False) self.assertEqual(out["result"], "denied") # LE flag forces denied body = mock_post.call_args[1].get("json", {}) self.assertEqual(body.get("valueText"), "cop_flag") # LE → cop_flag verdict # GPU-extracted reports folded into the recorded signal. self.assertIn("no show after booking", body["valueJsonb"]["reports"]) self.assertTrue(any(f["category"] == "law_enforcement" for f in body["valueJsonb"]["safety_flags"])) class TestEmulatorControl(unittest.TestCase): """adb controller in isolation.""" def setUp(self): self.emu = mr_lookup.MrNumberEmulator(device="emulator-test", package="com.test.mrnumber") @patch("redroid_client.device.subprocess.check_output") def test_adb_success(self, mock_check): mock_check.return_value = "ok\n" self.assertIn("ok", self.emu.adb(["shell", "echo", "ok"])) @patch("redroid_client.device.subprocess.check_output") def test_screen_size_parsed(self, mock_check): mock_check.return_value = "Physical size: 1080x1920\n" self.assertEqual(self.emu.screen_size(), (1080, 1920)) @patch("redroid_client.device.subprocess.check_output") def test_screen_size_fallback(self, mock_check): mock_check.return_value = "weird output" self.assertEqual(self.emu.screen_size(), (720, 1280)) @patch.object(mr_lookup.MrNumberEmulator, "adb") @patch.object(mr_lookup.MrNumberEmulator, "get_ui_dump") def test_find_and_tap_text(self, mock_dump, mock_adb): mock_dump.return_value = '' self.assertTrue(self.emu.find_and_tap_text(["view all"])) mock_adb.assert_called() if __name__ == "__main__": unittest.main()