""" 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, "barcode": None, "flags": [], "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_barcode_disagreement_is_a_mismatch(self): result = reconcile(accepted(), tag(price=168, barcode="4902105000001"), product(price=168.0, jan="4902105000002")) self.assertEqual(result["state"], RECONCILED_MISMATCH) def test_barcode_agreement_confirms(self): result = reconcile(accepted(), tag(price=None, barcode="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}, } # Wide enough that separate products can sit apart in a row. BANDS = [band(1, [0, 300, 3000, 340]), band(2, [0, 600, 3000, 640])] class ComplianceTests(unittest.TestCase): def test_missing_tag_on_a_covered_row_is_a_violation(self): # Four separate products (wide gaps), three with tag evidence - 75%, above # the 70% bar - and one with none. tracks = [ track(index, [10 + index * 300, 100, 140 + index * 300, 300], sku=f"sku_{index}") for index in range(4) ] 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_READABLE, price=188, id="tag_c"), } findings = evaluate_compliance(tracks, tags, [tags[0], tags[1], tags[3]], 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 products") 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 * 300, 100, 90 + index * 300, 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): # Coverage counts *products* with any tag evidence, so the row needs enough # of it to be judged at all (bar is 70%: 7 of 10 products). tracks = [ track(index, [10 + index * 300, 100, 70 + index * 300, 300], sku=f"sku_{index}") for index in range(10) ] tags = {index: tag(state=TAG_STATE_READABLE, price=168 + index, id=f"tag_{index}") for index in range(10)} for absent in (2, 3, 4): tags[absent] = tag(state=TAG_STATE_ABSENT, id=None) findings = evaluate_compliance(tracks, tags, [t for t in tags.values() if t["id"]], 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 products", groups[0].detail) def test_a_product_whose_tag_sits_elsewhere_in_the_row_is_not_missing(self): """Measured false positive: a run of ピュアセレクト facings was split by geometry, and the piece was reported as missing the tag that was sitting further along its own run. Identity and the row's tags both guard it.""" tracks = [ track(0, [10, 100, 160, 300], sku="sku_a"), track(1, [170, 100, 320, 300], sku="sku_a"), track(2, [330, 100, 460, 300], sku="sku_a"), ] tags = { 0: tag(state=TAG_STATE_READABLE, price=209, id="tag_a"), 1: tag(state=TAG_STATE_ABSENT), 2: tag(state=TAG_STATE_ABSENT), } findings = evaluate_compliance(tracks, tags, [tags[0]], BANDS) self.assertEqual(findings, [], "the product's tag is on its own run") def test_two_missing_products_are_not_a_group(self): tracks = [ track(index, [10 + index * 300, 100, 140 + index * 300, 300], sku=f"sku_{index}") for index in range(4) ] 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"), 3: tag(state=TAG_STATE_READABLE, price=198, id="tag_c"), } findings = evaluate_compliance(tracks, tags, [tags[0], tags[2], tags[3]], 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_facings_of_one_product_share_a_tag(self): """The measured domain rule: this store prints one tag per product, not per facing - a row carries ~16 facings but only ~4 tags. Counting facings would accuse a perfectly labelled shelf.""" # Four facings of the same product (touching, same size) under one tag. tracks = [track(index, [10 + index * 80, 100, 90 + index * 80, 300]) for index in range(4)] tags = { 0: tag(state=TAG_STATE_READABLE, price=269, id="tag_a"), 1: tag(state=TAG_STATE_ABSENT), 2: tag(state=TAG_STATE_ABSENT), 3: tag(state=TAG_STATE_ABSENT), } findings = evaluate_compliance(tracks, tags, [tags[0]], BANDS) self.assertEqual(findings, [], "one tag covers the whole product: nothing is wrong") 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): # A reportable mismatch carries an anchor: the tag's JAN was read, so we # know which tag we are comparing. tracks = [track(1, [10, 100, 90, 300])] # The anchor is the tag's own arithmetic agreeing, not a price field. tags = {1: tag(state=TAG_STATE_READABLE, price=269, flags=["tax_consistent"])} 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_a_mismatch_without_an_anchor_is_only_a_hint(self): """Measured noise: a wrong identity plus a garbled tag read (name match 0.11, no JAN) produced a confident "tag says 90, catalog says 188" that was really two bad reads disagreeing. Without an anchor it is a hint.""" tracks = [track(1, [10, 100, 90, 300])] tags = {1: tag(state=TAG_STATE_READABLE, price=90, name="ラスお要をおいしくだべおまにすくg)")} reconciliations = {1: reconcile(accepted(), tags[1], product(price=188.0))} findings = evaluate_compliance(tracks, tags, [tags[1]], BANDS, reconciliations) self.assertNotIn(RULE_PRICE_MISMATCH, [finding.rule for finding in findings]) hints = [finding for finding in findings if finding.severity == "hint"] self.assertTrue(hints) self.assertEqual(summarize_findings(findings)["violations"], 0) def test_summary_counts_only_violations(self): findings = [ evaluate_compliance( [track(1, [10, 100, 90, 300])], {1: tag(state=TAG_STATE_READABLE, price=269, flags=["tax_consistent"])}, [tag(state=TAG_STATE_READABLE, price=269, flags=["tax_consistent"])], 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()