applications/redroid-mrnumber
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
redroid-mrnumber/client/mr_lookup_test.py
Natalie 15e7348413 feat(box): OCR extraction + GPU (OpenAI-compatible) rating backend, env-selectable 
Wire the on-box (Claude-API-less) path decided with the operator: EXTRACT_BACKEND=ocr
sends each screenshot to the on-box mrnumber-ocr service (raw text, no per-shot
structuring); build_rating_profile uses an OpenAI-compatible LLM on a DO GPU droplet
(RATING_LLM_URL) which extracts the reports from the raw OCR text AND produces the
multi-axis verdict. Reports are folded back into the history so the people-signal +
counts + safety flags reflect them; safety detection also scans the raw OCR lines so a
LE term forces cop_flag even before structuring. vision/Claude stays the plum-dev
default. +5 tests incl. full OCR→GPU→cop_flag flow. 33/33.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-30 00:39:06 -04:00

369 lines
20 KiB
Python
Raw Permalink Blame History

#!/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 = '<node text="View all 4 reports" bounds="[100,200][300,400]" />'
self.assertTrue(self.emu.find_and_tap_text(["view all"]))
mock_adb.assert_called()
if __name__ == "__main__":
unittest.main()
Reference in a new issue View git blame Copy permalink