""" Contract tests for reconciliation and the compliance rule family. Pure functions over evidence, so these run without OCR, weights, or a camera. The rules that matter most here are the fail-closed ones: unreadable evidence must never become a violation. """ from __future__ import annotations import os import sys import unittest current_dir = os.path.dirname(os.path.abspath(__file__)) if current_dir not in sys.path: sys.path.insert(0, current_dir) from reconcile import ( # noqa: E402 ComplianceSettings, ComplianceTracker, RECONCILED_CONFIRMED, RECONCILED_MISMATCH, RECONCILED_NO_TAG, RECONCILED_TAG_ONLY, RECONCILED_UNVERIFIED, RULE_MISSING_TAG, RULE_MISSING_TAG_GROUP, RULE_ORPHAN_TAG, RULE_PRICE_MISMATCH, RULE_UNREADABLE_TAGS, evaluate_compliance, longest_common_substring_ratio, normalize_name, reconcile, summarize_findings, ) from tags import TAG_STATE_ABSENT, TAG_STATE_READABLE, TAG_STATE_UNREADABLE # noqa: E402 def accepted(sku="sku_003"): return {"state": "recognized", "sku_id": sku, "identity_level": "accepted", "confidence": 0.9} def tag(state=TAG_STATE_READABLE, **overrides): payload = { "id": "tag_1", "box": [100, 300, 200, 340], "state": state, "price": None, "regular_price": None, "name": None, "confidence": 0.8, "observations": 3, "settled": True, } payload.update(overrides) return payload def product(price=168.0, name="Tropicana エッセンシャルズ マルチビタミン", jan=None): payload = {"sku_id": "sku_003", "product_name": name, "price": price} if jan: payload["jan"] = jan return payload class NameMatchingTests(unittest.TestCase): def test_normalization_strips_noise(self): self.assertEqual(normalize_name("Ｔｒｏｐｉｃａｎａ エッセンシャルズ"), "tropicanaエッセンシャルズ") self.assertEqual(normalize_name("キュービー<310g>"), "キュービー") def test_lcs_ratio(self): self.assertEqual(longest_common_substring_ratio("tropicana", "tropicana"), 1.0) self.assertGreater(longest_common_substring_ratio("tropicanaエッセンシャルズ", "tropicana"), 0.5) self.assertEqual(longest_common_substring_ratio("", "x"), 0.0) class ReconcileTests(unittest.TestCase): def test_price_agreement_confirms(self): result = reconcile(accepted(), tag(price=168), product(price=168.0)) self.assertEqual(result["state"], RECONCILED_CONFIRMED) self.assertTrue(any(signal["kind"] == "price" and signal["agrees"] for signal in result["signals"])) def test_regular_price_agreement_confirms(self): result = reconcile(accepted(), tag(price=140, regular_price=168), product(price=168.0)) self.assertEqual(result["state"], RECONCILED_CONFIRMED) def test_price_disagreement_is_a_mismatch_with_both_values(self): result = reconcile(accepted(), tag(price=269), product(price=168.0)) self.assertEqual(result["state"], RECONCILED_MISMATCH) self.assertIn("269", result["message"]) self.assertIn("168", result["message"]) def test_name_agreement_confirms_even_without_a_price(self): result = reconcile(accepted(), tag(name="Tropicana エッセンシャルズ"), product(price=None, name="Tropicana エッセンシャルズ マルチビタミン")) self.assertEqual(result["state"], RECONCILED_CONFIRMED) def test_jan_disagreement_is_a_mismatch(self): result = reconcile(accepted(), tag(price=168, jan="4902105000001"), product(price=168.0, jan="4902105000002")) self.assertEqual(result["state"], RECONCILED_MISMATCH) def test_jan_agreement_confirms(self): result = reconcile(accepted(), tag(price=None, jan="4902105000001"), product(price=None, jan="4902105000001")) self.assertEqual(result["state"], RECONCILED_CONFIRMED) def test_tag_read_but_nothing_comparable(self): result = reconcile(accepted(), tag(name="コーヒー"), product(price=None, name=None)) self.assertEqual(result["state"], RECONCILED_UNVERIFIED) def test_unreadable_tag_is_unverified_not_missing(self): result = reconcile(accepted(), tag(state=TAG_STATE_UNREADABLE), product()) self.assertEqual(result["state"], RECONCILED_UNVERIFIED) def test_absent_tag(self): result = reconcile(accepted(), tag(state=TAG_STATE_ABSENT), product()) self.assertEqual(result["state"], RECONCILED_NO_TAG) def test_readable_tag_without_a_product(self): result = reconcile(None, tag(price=269), None) self.assertEqual(result["state"], RECONCILED_TAG_ONLY) def test_nothing_to_reconcile_returns_none(self): self.assertIsNone(reconcile(None, None, None)) self.assertIsNone(reconcile({"state": "unconfirmed"}, tag(state=TAG_STATE_ABSENT), product())) # --------------------------------------------------------------------------- compliance def band(row_index, box): class Band: pass instance = Band() instance.row_index = row_index instance.box = box return instance def track(track_id, box, sku="sku_003", state="recognized"): return { "track_id": track_id, "box": box, "raw_box": box, "identity": {"state": state, "sku_id": sku if state == "recognized" else None}, } BANDS = [band(1, [0, 300, 700, 340]), band(2, [0, 600, 700, 640])] class ComplianceTests(unittest.TestCase): def test_missing_tag_on_a_covered_row_is_a_violation(self): tracks = [ track(1, [10, 100, 90, 300]), track(2, [110, 100, 190, 300]), ] tags = { 1: tag(state=TAG_STATE_READABLE, price=168), 2: tag(state=TAG_STATE_ABSENT), } findings = evaluate_compliance(tracks, tags, [tags[1]], BANDS) rules = [finding.rule for finding in findings] self.assertIn(RULE_MISSING_TAG, rules) missing = [finding for finding in findings if finding.rule == RULE_MISSING_TAG][0] self.assertEqual(missing.count, 1, "one finding per row, counting the facings") def test_a_row_with_poor_tag_coverage_is_not_evaluated(self): """The real failure mode: our detector found tags for only a few of many facings. Claiming the rest are missing would be a false violation.""" tracks = [track(index, [10 + index * 100, 100, 90 + index * 100, 300]) for index in range(6)] tags = {index: tag(state=TAG_STATE_ABSENT) for index in range(6)} tags[0] = tag(state=TAG_STATE_READABLE, price=168, id="tag_a") findings = evaluate_compliance(tracks, tags, [tags[0]], BANDS) self.assertEqual([finding.rule for finding in findings], [RULE_UNREADABLE_TAGS]) self.assertEqual(findings[0].severity, "hint") self.assertEqual(summarize_findings(findings)["violations"], 0) def test_unreadable_row_yields_a_hint_never_a_violation(self): """The demo image case: tags exist but cannot be read. Nothing may be claimed.""" tracks = [ track(1, [10, 100, 90, 300]), track(2, [110, 100, 190, 300]), track(3, [210, 100, 290, 300]), ] tags = { 1: tag(state=TAG_STATE_UNREADABLE, price=None, id="tag_1"), 2: tag(state=TAG_STATE_UNREADABLE, price=None, id="tag_2"), 3: tag(state=TAG_STATE_UNREADABLE, price=None, id="tag_3"), } findings = evaluate_compliance(tracks, tags, [tags[1], tags[2], tags[3]], BANDS) self.assertEqual([finding.rule for finding in findings], [RULE_UNREADABLE_TAGS]) self.assertEqual(findings[0].severity, "hint") self.assertEqual(summarize_findings(findings)["violations"], 0) def test_a_row_with_no_tag_evidence_at_all_is_silent(self): tracks = [track(1, [10, 100, 90, 300])] tags = {1: tag(state=TAG_STATE_ABSENT)} findings = evaluate_compliance(tracks, tags, [], BANDS) self.assertEqual(findings, []) def test_group_of_consecutive_facings_without_tags(self): # 5 facings: 2 tagged, 3 not. Coverage counts facings with *any* tag # evidence, so give the row enough evidence to be judged at all. tracks = [track(index, [10 + index * 100, 100, 90 + index * 100, 300]) for index in range(6)] tags = { 0: tag(state=TAG_STATE_READABLE, price=168, id="tag_a"), 1: tag(state=TAG_STATE_READABLE, price=178, id="tag_b"), 2: tag(state=TAG_STATE_ABSENT), 3: tag(state=TAG_STATE_ABSENT), 4: tag(state=TAG_STATE_ABSENT), 5: tag(state=TAG_STATE_READABLE, price=188, id="tag_c"), } findings = evaluate_compliance(tracks, tags, [tags[0], tags[1], tags[5]], BANDS) groups = [finding for finding in findings if finding.rule == RULE_MISSING_TAG_GROUP] self.assertEqual(len(groups), 1) self.assertEqual(groups[0].count, 3) self.assertEqual(groups[0].row_index, 1) self.assertIn("3 facings", groups[0].detail) def test_two_missing_facings_are_not_a_group(self): tracks = [track(index, [10 + index * 100, 100, 90 + index * 100, 300]) for index in range(3)] tags = { 0: tag(state=TAG_STATE_READABLE, price=168, id="tag_a"), 1: tag(state=TAG_STATE_ABSENT), 2: tag(state=TAG_STATE_READABLE, price=188, id="tag_b"), } findings = evaluate_compliance(tracks, tags, [tags[0], tags[2]], BANDS) self.assertNotIn(RULE_MISSING_TAG_GROUP, [finding.rule for finding in findings]) missing = [finding for finding in findings if finding.rule == RULE_MISSING_TAG] self.assertEqual(len(missing), 1) self.assertEqual(missing[0].count, 1) def test_orphan_tag(self): tracks = [track(1, [10, 100, 90, 300])] tags = {1: tag(state=TAG_STATE_READABLE, price=168, id="tag_a")} all_tags = [tags[1], tag(state=TAG_STATE_READABLE, price=999, id="tag_lonely", box=[300, 600, 400, 640])] findings = evaluate_compliance(tracks, tags, all_tags, BANDS) orphans = [finding for finding in findings if finding.rule == RULE_ORPHAN_TAG] self.assertEqual(len(orphans), 1) self.assertEqual(orphans[0].box, [300, 600, 400, 640]) def test_price_mismatch_becomes_a_finding(self): tracks = [track(1, [10, 100, 90, 300])] tags = {1: tag(state=TAG_STATE_READABLE, price=269)} reconciliations = {1: reconcile(accepted(), tags[1], product(price=168.0))} findings = evaluate_compliance(tracks, tags, [tags[1]], BANDS, reconciliations) mismatches = [finding for finding in findings if finding.rule == RULE_PRICE_MISMATCH] self.assertEqual(len(mismatches), 1) self.assertIn("269", mismatches[0].detail) def test_summary_counts_only_violations(self): findings = [ evaluate_compliance( [track(1, [10, 100, 90, 300])], {1: tag(state=TAG_STATE_READABLE, price=269)}, [tag(state=TAG_STATE_READABLE, price=269)], BANDS, {1: reconcile(accepted(), tag(price=269), product(price=168.0))}, ) ][0] summary = summarize_findings(findings) self.assertGreaterEqual(summary["violations"], 1) self.assertIn(RULE_PRICE_MISMATCH, summary["by_rule"]) class StabilityTests(unittest.TestCase): def test_findings_need_to_persist(self): tracker = ComplianceTracker(ComplianceSettings(stability_frames=3, stability_window=5)) from reconcile import Finding finding = Finding(rule=RULE_MISSING_TAG, severity="violation", row_index=1, box=[0, 0, 10, 10]) first = tracker.update([finding]) self.assertEqual(len(first), 0, "a single frame must not publish a violation") tracker.update([finding]) stable = tracker.update([finding]) self.assertEqual(len(stable), 1) def test_a_transient_finding_never_publishes(self): """The rule is windowed: a finding must appear in 3 of the last 5 frames. Two sightings spread over six frames is a flicker, not a violation.""" tracker = ComplianceTracker(ComplianceSettings(stability_frames=3, stability_window=5)) from reconcile import Finding finding = Finding(rule=RULE_MISSING_TAG, severity="violation", row_index=1, box=[0, 0, 10, 10]) published = 0 for index in range(6): published += len(tracker.update([finding] if index in (0, 2) else [])) self.assertEqual(published, 0) def test_a_finding_present_in_three_of_five_frames_publishes(self): tracker = ComplianceTracker(ComplianceSettings(stability_frames=3, stability_window=5)) from reconcile import Finding finding = Finding(rule=RULE_MISSING_TAG, severity="violation", row_index=1, box=[0, 0, 10, 10]) published = 0 for index in range(5): published += len(tracker.update([finding] if index in (0, 2, 4) else [])) self.assertGreater(published, 0) if __name__ == "__main__": unittest.main()