diff --git a/recognization/code/test_engine.py b/recognization/code/test_engine.py index 9832991..ec42cd2 100644 --- a/recognization/code/test_engine.py +++ b/recognization/code/test_engine.py @@ -407,6 +407,72 @@ class LocalLaneTests(unittest.TestCase): engine.shutdown() +class DecisionPolicyTests(unittest.TestCase): + """The two jobs the text lane does, and the boundary between them.""" + + def policy(self): + from identity_lanes import DecisionPolicy + + return DecisionPolicy() + + def test_local_answer_stands_when_the_text_agrees(self): + decision = self.policy().decide([("sku_a", 5.0)], [("sku_a", 0.9)]) + self.assertEqual(decision["best_sku"], "sku_a") + self.assertEqual(decision["source"], "local") + + def test_text_vetoes_a_local_answer_naming_another_row(self): + """The near-unknown case: real geometry, different words. + + The veto blocks the local answer, and the text lane then names its own + row - the two jobs compose, so the result is the text's SKU, never the + vetoed local one. + """ + decision = self.policy().decide([("sku_a", 5.0)], [("sku_b", 0.9)]) + self.assertEqual(decision["best_sku"], "sku_b") + self.assertEqual(decision["source"], "text") + + def test_text_mentioning_the_local_candidate_is_not_a_veto(self): + """A crop reading "マヨネーズ" scores every mayonnaise row. + + The local candidate is usually one of them, so a veto needs the text to + *prefer* another row, not merely to mention it: here the text mentions + sku_b but explains sku_a better, so the local answer stands. + """ + decision = self.policy().decide([("sku_a", 5.0)], [("sku_b", 0.4), ("sku_a", 0.45)]) + self.assertEqual(decision["best_sku"], "sku_a") + self.assertEqual(decision["source"], "local") + + def test_the_lanes_own_order_wins_over_the_value_order(self): + """Order and value are different quantities on the local lane. + + The order comes from `rank_key` (inliers weighted by hue); the value is + the margin over the runner-up. Re-sorting by value names a different SKU + than the lane chose - here sku_b has the larger value but sku_a is the + lane's own first choice, and sku_a is what must be answered. + """ + decision = self.policy().decide([("sku_a", 3.0), ("sku_b", 5.0)], []) + self.assertEqual(decision["best_sku"], "sku_a") + + def test_text_fills_in_when_local_has_nothing(self): + decision = self.policy().decide([], [("sku_b", 0.9)]) + self.assertEqual(decision["best_sku"], "sku_b") + self.assertEqual(decision["source"], "text") + + def test_text_below_its_own_line_does_not_fill_in(self): + decision = self.policy().decide([], [("sku_b", 0.3)]) + self.assertIsNone(decision["best_sku"]) + + def test_a_close_text_pair_does_not_fill_in(self): + """The margin gate: two rows the text explains equally must not name one.""" + decision = self.policy().decide([], [("sku_b", 0.9), ("sku_c", 0.85)]) + self.assertIsNone(decision["best_sku"]) + + def test_no_evidence_is_rejected(self): + decision = self.policy().decide([], []) + self.assertIsNone(decision["best_sku"]) + self.assertEqual(decision["identity_level"], "rejected") + + class CatalogImageTests(unittest.TestCase): def test_catalog_image_path_is_resolved_through_inventory(self): engine, _, _, _ = make_engine() diff --git a/recognization/code/test_live.py b/recognization/code/test_live.py index b8ea38f..071b900 100644 --- a/recognization/code/test_live.py +++ b/recognization/code/test_live.py @@ -95,6 +95,29 @@ class CosineEngine(RecognitionEngine): return scores +class LocalScaleEngine(RecognitionEngine): + """The shipped default scale, with the lanes' evidence injected. + + `local` is the branch that was wrong twice - it re-sorted the lane's own + order by value, and its early return on an empty local observation made the + text lane's "answer on its own" job unreachable - so it is tested here + rather than only through the real lane, which the fake matcher cannot drive. + """ + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.lane_answers = [] + + @property + def identity_scale(self): + return "local" + + def score_crop_lanes(self, crop, method=None): + if self.lane_answers: + return self.lane_answers.pop(0) + return {}, [] + + class FakeInventory: """Stateful stand-in for InventoryManager, so stock assertions are real.""" @@ -291,6 +314,58 @@ class CosineScaleTests(unittest.TestCase): engine.shutdown() +class LocalScaleTests(unittest.TestCase): + """The shipped default scale, which the fake matcher cannot reach. + + `make_session` builds a `matcher`-backend engine, so `scale == "local"` in + `_on_classified` never runs in the rest of this file - and that branch is + the one that was wrong twice. These tests drive it with injected lane + evidence, so a regression in the local decision path fails here. + """ + + def test_a_text_only_read_still_labels_the_track(self): + """The text lane answers on its own when the local lane has nothing. + + The early return on an empty local observation made this unreachable: + a SKU with no enrolled image can only ever be named by its text, so + dropping that path silently removed the lane's fill-in job. + """ + session, engine, _, _ = make_session(engine_cls=LocalScaleEngine) + engine.lane_answers = [({}, [("sku_a", 0.9)])] + result = process(session, 1) + self.assertTrue( + wait_for(lambda: any( + track["identity"]["sku_id"] == "sku_a" for track in process(session, 2)["tracks"] + )), + "a text-only read must label the track", + ) + engine.shutdown() + + def test_a_text_veto_blocks_a_local_accept(self): + """A local match whose package text names another row must not be shown. + + The sibling shares the artwork, so the geometry is real and the local + score is high; only the words differ. The text here is below its own + accept line, so it vetoes without filling in: the local answer must not + reach the wire at all. + + `scores` is a dict keyed by SKU, which is what `score_crop` returns and + what `_on_classified` iterates; a list here raises inside the future + callback, where the failure is swallowed and the test passes for the + wrong reason. + """ + session, engine, _, _ = make_session(engine_cls=LocalScaleEngine) + engine.lane_answers = [({"sku_a": 5.0}, [("other", 0.3)])] * 4 + for frame_id in range(1, 14): + result = process(session, frame_id) + for track in result["tracks"]: + self.assertNotEqual( + track["identity"]["sku_id"], "sku_a", + "a vetoed local answer must not be labelled", + ) + engine.shutdown() + + class TagPassIntegrationTests(unittest.TestCase): """The tag pass must reach the live result without blocking frames."""