71b8bdd876
reconcile() compares one accepted identity with its tag and the catalog entry and publishes a state plus the individual signals (JAN exact, name by longest common substring, price against either the sale or the regular price). It never changes the product identity: a disagreement is a second opinion shown next to the first, and only a checksummed barcode outranks the matcher. evaluate_compliance() is the first rule family: missing_price_tag, the owner's missing_tag_group (>= 3 consecutive facings without a tag), orphan_tag, price_mismatch, plus an unreadable_tags hint. Every rule fails closed on evidence quality - a row whose tags could not be read produces "move closer", never a violation. Findings must appear in 3 of the last 5 frames before they publish, so the UI never flickers. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
288 lines
12 KiB
Python
288 lines
12 KiB
Python
"""
|
||
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, 500, 340]), band(2, [0, 600, 500, 640])]
|
||
|
||
|
||
class ComplianceTests(unittest.TestCase):
|
||
def test_missing_tag_on_a_readable_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)
|
||
|
||
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):
|
||
tracks = [track(index, [10 + index * 100, 100, 90 + index * 100, 300]) for index in range(5)]
|
||
tags = {
|
||
0: tag(state=TAG_STATE_READABLE, price=168, id="tag_a"),
|
||
1: tag(state=TAG_STATE_ABSENT),
|
||
2: tag(state=TAG_STATE_ABSENT),
|
||
3: tag(state=TAG_STATE_ABSENT),
|
||
4: tag(state=TAG_STATE_READABLE, price=178, id="tag_b"),
|
||
}
|
||
findings = evaluate_compliance(tracks, tags, [tags[0], tags[4]], 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_ABSENT),
|
||
}
|
||
findings = evaluate_compliance(tracks, tags, [tags[0]], BANDS)
|
||
self.assertNotIn(RULE_MISSING_TAG_GROUP, [finding.rule for finding in findings])
|
||
self.assertEqual(
|
||
len([finding for finding in findings if finding.rule == RULE_MISSING_TAG]), 2
|
||
)
|
||
|
||
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()
|