From 90cce64cc8c41bb7424e63e39297f634b67776c8 Mon Sep 17 00:00:00 2001 From: kiyo9w Date: Thu, 24 Sep 2026 12:43:32 +0900 Subject: [PATCH] test: defend engine and tracking contracts test_engine: detection normalization, one-caller-at-a-time detection, method validation, undecodable payloads, crop isolation, classify_boxes accepting the detection dicts the API actually hands it, summary counting, empty-slot normalization across frame sizes, the still-image store (roundtrip, TTL, bound), catalog/stock behaviour, and template-image path resolution that refuses anything outside the catalog. test_tracking: per-session id numbering, the case that matters (constructing a second tracker must not reset a running session's ids), independent sessions, pan continuity, buffer expiry producing a new id, detection index mapping, and reset scoping. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- recognization/code/test_engine.py | 284 ++++++++++++++++++++++++++++ recognization/code/test_tracking.py | 110 +++++++++++ 2 files changed, 394 insertions(+) create mode 100644 recognization/code/test_engine.py create mode 100644 recognization/code/test_tracking.py diff --git a/recognization/code/test_engine.py b/recognization/code/test_engine.py new file mode 100644 index 0000000..d32ac67 --- /dev/null +++ b/recognization/code/test_engine.py @@ -0,0 +1,284 @@ +""" +Contract tests for the recognition engine. + +No YOLO weights and no network: a fake detector and a fake matcher stand in for +the model, so these tests defend the engine's own contracts - normalization, +locking, isolation between clients, empty-slot geometry, and the still-image +store. +""" + +from __future__ import annotations + +import os +import sys +import threading +import time +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 engine import ( # noqa: E402 + EngineSettings, + ImageExpired, + ImageStore, + RecognitionEngine, + UndecodableImage, + UnknownMethod, + identity_payload, +) + + +class FakeDetector: + """Records concurrency and returns fixed boxes.""" + + weights_path = "fake.pt" + + def __init__(self, boxes=None): + self.boxes = boxes if boxes is not None else [ + {"id": 0, "box": [10, 10, 60, 90], "confidence": 0.9, "class_id": 0, + "width": 50, "height": 80, "center": [35, 50]}, + {"id": 1, "box": [70, 10, 120, 90], "confidence": 0.8, "class_id": 0, + "width": 50, "height": 80, "center": [95, 50]}, + ] + self.concurrent = 0 + self.max_concurrent = 0 + self.calls = 0 + self.lock = threading.Lock() + + def detect(self, image, conf_threshold=0.25, iou_threshold=0.45): + with self.lock: + self.calls += 1 + self.concurrent += 1 + self.max_concurrent = max(self.max_concurrent, self.concurrent) + time.sleep(0.01) + with self.lock: + self.concurrent -= 1 + return {"status": "success", "boxes": [dict(b) for b in self.boxes]} + + def detect_empty_slots(self, image_shape, boxes, min_gap_ratio=0.75): + # One gap between the two boxes, expressed in the frame the engine hands + # over. Scaling with that frame is what makes the normalization test + # meaningful: the engine must translate the answer back to the client's + # pixel space. + height, width = int(image_shape[0]), int(image_shape[1]) + x1, x2 = int(width * 0.30), int(width * 0.35) + return [{ + "id": "empty_1", + "box": [x1, int(height * 0.10), x2, int(height * 0.90)], + "width": x2 - x1, + "height": int(height * 0.80), + "missing_items_est": 1, + "center": [(x1 + x2) // 2, height // 2], + "row_index": 1, + }] + + +class FakeMatcher: + def __init__(self): + self.calls = 0 + + def score_crop(self, crop, method="hybrid"): + self.calls += 1 + time.sleep(0.01) + return {"sku_a": 40.0, "sku_b": 2.0}, {"sku_a": {"score": 40.0}} + + def score_crop_fast(self, crop, shortlist=5): + return self.score_crop(crop) + + def classify_crop(self, crop, method="hybrid"): + from identity import decide_identity + + scores, details = self.score_crop(crop, method=method) + return decide_identity(scores, details=details, method=method) + + def generate_match_visualization(self, crop, sku, method="surf", max_matches=30): + return crop + + +class FakeInventory: + def __init__(self): + self.products = { + "sku_a": {"sku_id": "sku_a", "product_name": "Alpha", "price": 100.0, + "number_in": 10, "number_out": 2, "number_on": 8, + "receipt_date": "2026-01-01", "expiration_date": "2026-12-31", + "local_image_path": None}, + "sku_b": {"sku_id": "sku_b", "product_name": "Beta", "price": 200.0, + "number_in": 5, "number_out": 1, "number_on": 4, + "receipt_date": "2026-01-02", "expiration_date": "2026-12-30", + "local_image_path": None}, + } + self.lock = threading.Lock() + + def list_all_skus(self): + return list(self.products) + + def get_product(self, sku_id): + return self.products.get(sku_id) + + def update_stock(self, sku_id, delta): + with self.lock: + product = self.products.get(sku_id) + if product is None: + return False + product["number_on"] = max(0, product["number_on"] + delta) + if delta < 0: + product["number_out"] += abs(delta) + else: + product["number_in"] += delta + return True + + +def make_engine(**overrides): + settings = EngineSettings(device="cpu", classify_workers=4, **overrides) + detector = FakeDetector() + matcher = FakeMatcher() + inventory = FakeInventory() + engine = RecognitionEngine(detector, matcher, inventory, settings=settings) + return engine, detector, matcher, inventory + + +def make_image(width=200, height=120): + import numpy as np + + return np.zeros((height, width, 3), dtype="uint8") + + +class EngineBasicsTests(unittest.TestCase): + def test_detect_normalizes_and_reindexes(self): + engine, _, _, _ = make_engine() + boxes = engine.detect(make_image()) + self.assertEqual(len(boxes), 2) + self.assertEqual([b["id"] for b in boxes], [0, 1]) + self.assertEqual(boxes[0]["width"], 50) + self.assertEqual(boxes[0]["center"], [35, 50]) + + def test_detector_calls_never_overlap(self): + engine, detector, _, _ = make_engine() + threads = [threading.Thread(target=engine.detect, args=(make_image(),)) for _ in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + self.assertEqual(detector.calls, 8) + self.assertEqual(detector.max_concurrent, 1, "the model must be used by one caller at a time") + + def test_unknown_method_is_rejected(self): + engine, _, _, _ = make_engine() + with self.assertRaises(UnknownMethod): + engine.score_crop(make_image(), method="not-a-method") + + def test_undecodable_payload_raises(self): + engine, _, _, _ = make_engine() + with self.assertRaises(UndecodableImage): + engine.decode_image(b"this is not an image") + + def test_crop_is_a_copy(self): + engine, _, _, _ = make_engine() + image = make_image() + crop = engine.crop(image, [0, 0, 10, 10]) + crop[0, 0, 0] = 255 + self.assertEqual(int(image[0, 0, 0]), 0) + + +class EngineClassificationTests(unittest.TestCase): + def test_classify_boxes_accepts_detection_dicts_without_size_fields(self): + """The API hands classify_boxes stored detections, which carry no width/height.""" + engine, _, _, _ = make_engine() + detections = [ + {"id": 0, "box": [10, 10, 60, 90], "confidence": 0.9, "identity": identity_payload(None)}, + {"id": 1, "box": [70, 10, 120, 90], "confidence": 0.8, "identity": identity_payload(None)}, + ] + rows = engine.classify_boxes(make_image(), detections, method="hybrid") + self.assertEqual(len(rows), 2) + self.assertEqual(rows[0]["identity"]["state"], "recognized") + self.assertEqual(rows[0]["identity"]["sku_id"], "sku_a") + + def test_parallel_classification_matches_sequential(self): + engine, _, matcher, _ = make_engine() + boxes = [ + {"id": index, "box": [10 + index, 10, 60 + index, 90], "confidence": 0.9, + "width": 50, "height": 80, "center": [35, 50]} + for index in range(12) + ] + parallel = engine.classify_boxes(make_image(), boxes, method="hybrid") + sequential = [engine.score_crop(engine.crop(make_image(), box["box"]), "hybrid") for box in boxes] + self.assertEqual(len(parallel), 12) + self.assertEqual(len(sequential), 12) + for row in parallel: + self.assertEqual(row["identity"]["sku_id"], "sku_a") + + def test_summary_counts_states(self): + engine, _, _, _ = make_engine() + detections = [ + {"identity": {"state": "recognized", "sku_id": "sku_a"}}, + {"identity": {"state": "recognized", "sku_id": "sku_a"}}, + {"identity": {"state": "needs_confirmation", "sku_id": None}}, + {"identity": {"state": "unrecognized", "sku_id": None}}, + ] + summary = engine.summarize(detections, []) + self.assertEqual(summary["recognized"], 2) + self.assertEqual(summary["needs_confirmation"], 1) + self.assertEqual(summary["unrecognized"], 1) + self.assertEqual(summary["by_sku"], [{"sku_id": "sku_a", "facings": 2}]) + + +class EmptySlotNormalizationTests(unittest.TestCase): + def test_slots_scale_with_the_frame(self): + """The detector's thresholds are absolute pixels, so a half-size frame must + produce the same slots (scaled) as the reference-size frame.""" + engine, _, _, _ = make_engine() + boxes_small = [ + {"box": [5, 5, 30, 45]}, + {"box": [35, 5, 60, 45]}, + ] + slots_small = engine.empty_slots((60, 100), boxes_small) + boxes_big = [ + {"box": [10, 10, 60, 90]}, + {"box": [70, 10, 120, 90]}, + ] + slots_big = engine.empty_slots((120, 200), boxes_big) + self.assertEqual(len(slots_small), 1) + self.assertEqual(len(slots_big), 1) + # the small frame is scaled up to the reference width, so the returned + # slot comes back in the small frame's own pixels + self.assertLess(slots_small[0]["box"][2], slots_big[0]["box"][2]) + + +class ImageStoreTests(unittest.TestCase): + def test_roundtrip_and_expiry(self): + store = ImageStore(max_items=2, ttl_s=0.05, max_bytes=10_000_000) + stored = store.put(make_image()) + self.assertEqual(store.get(stored.image_id).width, 200) + time.sleep(0.08) + with self.assertRaises(ImageExpired): + store.get(stored.image_id) + + def test_bounded_by_count(self): + store = ImageStore(max_items=2, ttl_s=60, max_bytes=10_000_000) + ids = [store.put(make_image()).image_id for _ in range(4)] + with self.assertRaises(ImageExpired): + store.get(ids[0]) + self.assertEqual(store.get(ids[-1]).width, 200) + + +class CatalogTests(unittest.TestCase): + def test_catalog_and_stock_update(self): + engine, _, _, _ = make_engine() + catalog = engine.catalog() + self.assertEqual(len(catalog), 2) + self.assertEqual(catalog[0]["image_url"], "/v1/catalog/sku_a/image") + product = engine.update_stock("sku_a", -1) + self.assertEqual(product["number_on"], 7) + self.assertEqual(product["number_out"], 3) + self.assertIsNone(engine.update_stock("nope", -1)) + + def test_catalog_image_path_is_resolved_through_inventory(self): + engine, _, _, _ = make_engine() + self.assertIsNone(engine.catalog_image_path("../../etc/passwd")) + self.assertIsNone(engine.catalog_image_path("sku_a")) + + +if __name__ == "__main__": + unittest.main() diff --git a/recognization/code/test_tracking.py b/recognization/code/test_tracking.py new file mode 100644 index 0000000..d94ea83 --- /dev/null +++ b/recognization/code/test_tracking.py @@ -0,0 +1,110 @@ +""" +Contract tests for session tracking. + +The important one is ID isolation: ultralytics mints track ids from a +process-global counter and resets it when a tracker is constructed, so two live +clients would otherwise be able to hand each other's cached labels to the wrong +product. +""" + +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 tracking import SessionTracker # noqa: E402 + + +def pan(tracker, start_x, frames=8, step=10, width=60, height=80, y=100): + ids = [] + for index in range(frames): + x = start_x + index * step + tracks = tracker.update([[x, y, x + width, y + height]], [0.9]) + ids.append(tracks[0].track_id if tracks else None) + return ids + + +class IdIsolationTests(unittest.TestCase): + def test_sessions_do_not_share_ids(self): + first = SessionTracker() + second = SessionTracker() + first_ids = pan(first, 100) + second_ids = pan(second, 100) + self.assertEqual(len(set(first_ids)), 1, "a panning box keeps one id") + self.assertEqual(len(set(second_ids)), 1, "a panning box keeps one id") + self.assertEqual(first_ids[0], 1) + self.assertEqual(second_ids[0], 1, "each session starts its own numbering") + + def test_constructing_a_tracker_does_not_reset_a_running_session(self): + """ultralytics resets the global id counter in BYTETracker.__init__; a + second client connecting mid-scan must not renumber the first client's + tracks (its cached labels are keyed by id).""" + first = SessionTracker() + ids = pan(first, 100, frames=6, step=5) + self.assertEqual(len(set(ids)), 1) + second = SessionTracker() # a second client connects + continued = pan(first, 100 + 6 * 5, frames=6, step=5) + self.assertEqual(set(continued), {ids[0]}, "ids must keep counting, not restart") + self.assertEqual(pan(second, 400)[0], 1, "the new session still starts at 1") + + def test_two_sessions_track_the_same_product_independently(self): + first = SessionTracker() + second = SessionTracker() + first_ids = pan(first, 100) + second_ids = pan(second, 100) + self.assertEqual(first_ids, second_ids, "same motion, same ids per session") + # ... but they are separate objects: dropping one does not touch the other + for _ in range(10): + first.update([], []) + self.assertEqual(len(set(pan(second, 100 + 8 * 10, frames=3, step=5))), 1) + + def test_two_boxes_get_distinct_ids(self): + tracker = SessionTracker() + tracks = tracker.update([[10, 10, 70, 90], [200, 10, 260, 90]], [0.9, 0.8]) + self.assertEqual(len(tracks), 2) + self.assertEqual(len({track.track_id for track in tracks}), 2) + + +class TrackLifecycleTests(unittest.TestCase): + def test_pan_continuity(self): + tracker = SessionTracker() + ids = pan(tracker, 50, frames=10, step=12) + self.assertEqual(len(set(ids)), 1) + + def test_absent_track_is_dropped_and_gets_a_new_id(self): + tracker = SessionTracker(buffer=5) + first_ids = pan(tracker, 50, frames=2) + for _ in range(8): + tracker.update([], []) + for _ in range(3): # a re-detected object is unconfirmed for a frame or two + tracks = tracker.update([[50, 100, 110, 180]], [0.9]) + self.assertTrue(tracks, "the object is tracked again") + self.assertNotEqual(tracks[0].track_id, first_ids[0], "a dropped track must not resume its old id") + + def test_detection_index_maps_back_to_the_input_box(self): + tracker = SessionTracker() + tracks = tracker.update( + [[10, 10, 70, 90], [200, 10, 260, 90], [400, 10, 460, 90]], [0.9, 0.85, 0.95] + ) + indices = sorted(track.det_index for track in tracks) + self.assertEqual(indices, [0, 1, 2]) + + def test_reset_starts_numbering_over_without_touching_others(self): + other = SessionTracker() + other_ids = pan(other, 100, frames=6, step=5) + tracker = SessionTracker() + pan(tracker, 100, frames=6, step=5) + tracker.reset() + after = pan(tracker, 300) + self.assertEqual(set(after), {1}, "reset restarts this session's numbering") + continued = pan(other, 100 + 6 * 5, frames=4, step=5) + self.assertEqual(set(continued), {other_ids[0]}, "the other session keeps counting") + + +if __name__ == "__main__": + unittest.main()