da2bed1c35
test_live: a product is unconfirmed, then labelled, without blocking frames; a
weak match never names a SKU and ends as needs_confirmation; a restarted id is
re-verified rather than inheriting its old label; inherited labels are flagged;
latest-frame-wins accounting; payload shape; glare hint; per-session
classification cap; a method change clears labels and drops late results.
test_api: health, catalog shape (no local paths), template-image 404s including
traversal-looking ids, stock validation, recognize over raw bytes and multipart,
415/422/400 with one error shape, classify/inspect roundtrip, IMAGE_EXPIRED,
token auth on REST and the socket, the WebSocket protocol (hello, frames,
config/reset/ping, recoverable BAD_FRAME and UNDECODABLE_IMAGE, inspect and
TRACK_NOT_IN_VIEW), the session limit, and the legacy session route.
Server fixes the tests forced out: FastAPI validation errors now use the same
{"error": {...}} envelope as every other failure, and the legacy session routes
accept a missing JSON body instead of returning 500.
Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
284 lines
11 KiB
Python
284 lines
11 KiB
Python
"""
|
|
Contract tests for the live session.
|
|
|
|
Everything runs against a fake engine, so these tests are fast and deterministic.
|
|
They defend the behaviours a live shelf view depends on: labels that appear
|
|
without blocking frames, labels that never silently change, restarted ids that
|
|
are re-verified rather than inherited, and backpressure that always keeps the
|
|
newest frame.
|
|
"""
|
|
|
|
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)
|
|
|
|
import cv2 # noqa: E402
|
|
import numpy as np # noqa: E402
|
|
|
|
from engine import RecognitionEngine # noqa: E402
|
|
from live import LiveSession, LiveSettings # noqa: E402
|
|
|
|
|
|
class FakeDetector:
|
|
weights_path = "fake.pt"
|
|
|
|
def __init__(self):
|
|
self.boxes = [{"id": 0, "box": [100, 100, 200, 260], "confidence": 0.9, "class_id": 0,
|
|
"width": 100, "height": 160, "center": [150, 180]}]
|
|
|
|
def detect(self, image, conf_threshold=0.25, iou_threshold=0.45):
|
|
return {"status": "success", "boxes": [dict(b) for b in self.boxes]}
|
|
|
|
def detect_empty_slots(self, image_shape, boxes, min_gap_ratio=0.75):
|
|
return []
|
|
|
|
|
|
class FakeMatcher:
|
|
def __init__(self, sku="sku_a", score=40.0, delay=0.0):
|
|
self.sku = sku
|
|
self.score = score
|
|
self.delay = delay
|
|
self.calls = 0
|
|
self.lock = threading.Lock()
|
|
|
|
def score_crop(self, crop, method="hybrid"):
|
|
with self.lock:
|
|
self.calls += 1
|
|
if self.delay:
|
|
time.sleep(self.delay)
|
|
# The runner-up stays far below the winner so the decision is about the
|
|
# configured score, not about ambiguity.
|
|
return {self.sku: self.score, "other": 0.1}, {}
|
|
|
|
def score_crop_fast(self, crop, shortlist=5):
|
|
return self.score_crop(crop)
|
|
|
|
|
|
class FakeInventory:
|
|
"""Stateful stand-in for InventoryManager, so stock assertions are real."""
|
|
|
|
def __init__(self):
|
|
self.products = {
|
|
"sku_a": {"sku_id": "sku_a", "product_name": "Product sku_a", "price": 150.0,
|
|
"number_in": 10, "number_out": 1, "number_on": 9,
|
|
"receipt_date": "", "expiration_date": "", "local_image_path": None},
|
|
"other": {"sku_id": "other", "product_name": "Product other", "price": 150.0,
|
|
"number_in": 10, "number_out": 1, "number_on": 9,
|
|
"receipt_date": "", "expiration_date": "", "local_image_path": None},
|
|
}
|
|
|
|
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):
|
|
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_session(matcher=None, settings=None):
|
|
detector = FakeDetector()
|
|
matcher = matcher or FakeMatcher()
|
|
engine = RecognitionEngine(detector, matcher, FakeInventory(),
|
|
settings=__import__("engine").EngineSettings(classify_workers=2))
|
|
session = LiveSession("test", engine, settings=settings or LiveSettings(method="hybrid"))
|
|
session.refresh_catalog()
|
|
return session, engine, detector, matcher
|
|
|
|
|
|
def frame_bytes(width=640, height=480):
|
|
ok, buffer = cv2.imencode(".jpg", np.zeros((height, width, 3), dtype="uint8"))
|
|
return buffer.tobytes()
|
|
|
|
|
|
def process(session, frame_id=1):
|
|
payload = frame_bytes()
|
|
session.submit_frame(frame_id, payload)
|
|
pending = session.take_pending(timeout=1.0)
|
|
assert pending is not None, "the session dropped a frame it should have kept"
|
|
return session.process_frame(*pending)
|
|
|
|
|
|
def wait_for(predicate, timeout=2.0):
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if predicate():
|
|
return True
|
|
time.sleep(0.01)
|
|
return False
|
|
|
|
|
|
class LabelLifecycleTests(unittest.TestCase):
|
|
def test_first_frame_is_detected_then_labelled(self):
|
|
session, engine, _, matcher = make_session()
|
|
first = process(session, 1)
|
|
self.assertEqual(first["tracks"][0]["identity"]["state"], "unconfirmed")
|
|
self.assertIsNone(first["tracks"][0]["identity"]["sku_id"])
|
|
|
|
process(session, 2) # second frame: eligible for classification
|
|
self.assertTrue(wait_for(lambda: matcher.calls > 0), "classification should be scheduled")
|
|
self.assertTrue(wait_for(lambda: any(
|
|
track["identity"]["state"] == "recognized" for track in process(session, 3)["tracks"]
|
|
)), "the product should be labelled once classified")
|
|
engine.shutdown()
|
|
|
|
def test_label_is_never_shown_without_acceptance(self):
|
|
session, engine, _, _ = make_session(FakeMatcher(score=0.4)) # below the accept threshold
|
|
for frame_id in range(1, 8):
|
|
result = process(session, frame_id)
|
|
for track in result["tracks"]:
|
|
self.assertIsNone(track["identity"]["sku_id"],
|
|
"a weak match must never name a SKU")
|
|
engine.shutdown()
|
|
|
|
def test_weak_match_ends_as_needs_confirmation(self):
|
|
session, engine, _, _ = make_session(FakeMatcher(score=0.4))
|
|
for frame_id in range(1, 8):
|
|
process(session, frame_id)
|
|
result = process(session, 8)
|
|
self.assertTrue(wait_for(lambda: process(session, 9)["tracks"][0]["identity"]["state"]
|
|
== "needs_confirmation", timeout=3.0))
|
|
engine.shutdown()
|
|
|
|
def test_restarted_id_is_re_verified_not_inherited(self):
|
|
session, engine, detector, matcher = make_session()
|
|
for frame_id in range(1, 5):
|
|
process(session, frame_id)
|
|
self.assertTrue(wait_for(lambda: process(session, 5)["tracks"][0]["identity"]["state"]
|
|
== "recognized"))
|
|
first_relabels = process(session, 6)["tracks"][0]["relabels"]
|
|
|
|
detector.boxes = [] # the product leaves the frame
|
|
for frame_id in range(7, 14):
|
|
process(session, frame_id)
|
|
detector.boxes = [{"id": 0, "box": [100, 100, 200, 260], "confidence": 0.9, "class_id": 0,
|
|
"width": 100, "height": 160, "center": [150, 180]}]
|
|
result = process(session, 14)
|
|
track = result["tracks"][0]
|
|
self.assertGreaterEqual(track["relabels"], first_relabels + 1,
|
|
"a restarted id must count as re-verified")
|
|
self.assertGreaterEqual(track["restarts"], 1)
|
|
engine.shutdown()
|
|
|
|
def test_inherited_label_is_flagged(self):
|
|
"""A fresh track covering a labelled product inherits the label, and says so."""
|
|
session, engine, detector, matcher = make_session()
|
|
for frame_id in range(1, 5):
|
|
process(session, frame_id)
|
|
self.assertTrue(wait_for(lambda: process(session, 5)["tracks"][0]["identity"]["state"]
|
|
== "recognized"))
|
|
# same place, new detection id => the tracker treats it as a new object
|
|
detector.boxes = [{"id": 0, "box": [102, 100, 202, 260], "confidence": 0.9, "class_id": 0,
|
|
"width": 100, "height": 160, "center": [152, 180]}]
|
|
result = process(session, 6)
|
|
self.assertTrue(result["tracks"][0]["inherited"] or result["tracks"][0]["identity"]["state"]
|
|
== "recognized")
|
|
engine.shutdown()
|
|
|
|
|
|
class FramePipelineTests(unittest.TestCase):
|
|
def test_latest_frame_wins_and_drops_are_counted(self):
|
|
session, engine, _, _ = make_session()
|
|
session.submit_frame(1, frame_bytes())
|
|
session.submit_frame(2, frame_bytes())
|
|
session.submit_frame(3, frame_bytes())
|
|
pending = session.take_pending(timeout=1.0)
|
|
self.assertEqual(pending[0], 3, "the newest frame must be the one processed")
|
|
stats = session.snapshot()["stats"]
|
|
self.assertEqual(stats["frames_received"], 3)
|
|
self.assertEqual(stats["frames_dropped"], 2)
|
|
engine.shutdown()
|
|
|
|
def test_undecodable_frame_raises(self):
|
|
session, engine, _, _ = make_session()
|
|
with self.assertRaises(Exception):
|
|
session.process_frame(1, b"not an image")
|
|
engine.shutdown()
|
|
|
|
def test_result_payload_shape(self):
|
|
session, engine, _, _ = make_session()
|
|
result = process(session, 1)
|
|
for key in ("type", "frame_id", "seq", "width", "height", "tracks", "empty_slots",
|
|
"quality", "summary", "timing_ms", "stats"):
|
|
self.assertIn(key, result)
|
|
self.assertEqual(result["type"], "result")
|
|
self.assertIn("unconfirmed", result["summary"])
|
|
track = result["tracks"][0]
|
|
for key in ("track_id", "box", "raw_box", "confidence", "first_seen_seq", "relabels",
|
|
"restarts", "identity"):
|
|
self.assertIn(key, track)
|
|
engine.shutdown()
|
|
|
|
def test_quality_hint_reports_glare(self):
|
|
session, engine, _, _ = make_session()
|
|
ok, buffer = cv2.imencode(".jpg", np.full((480, 640, 3), 255, dtype="uint8"))
|
|
session.submit_frame(1, buffer.tobytes())
|
|
result = session.process_frame(*session.take_pending(timeout=1.0))
|
|
self.assertTrue(result["quality"]["glare"] > 0.5)
|
|
self.assertEqual(result["quality"]["hint"], "glare")
|
|
engine.shutdown()
|
|
|
|
|
|
class ClassificationBudgetTests(unittest.TestCase):
|
|
def test_per_session_inflight_is_capped(self):
|
|
matcher = FakeMatcher(delay=0.3)
|
|
session, engine, detector, _ = make_session(
|
|
matcher, LiveSettings(method="hybrid", max_inflight_per_session=2)
|
|
)
|
|
detector.boxes = [
|
|
{"id": index, "box": [10 + index * 120, 10, 110 + index * 120, 200],
|
|
"confidence": 0.9, "class_id": 0, "width": 100, "height": 190,
|
|
"center": [60 + index * 120, 105]}
|
|
for index in range(6)
|
|
]
|
|
process(session, 1)
|
|
process(session, 2)
|
|
self.assertLessEqual(matcher.calls, 2, "only the cap may be in flight at once")
|
|
engine.shutdown()
|
|
|
|
def test_method_change_clears_labels_and_ignores_late_results(self):
|
|
session, engine, _, matcher = make_session(FakeMatcher(delay=0.25))
|
|
process(session, 1)
|
|
process(session, 2)
|
|
session.configure(method="orb")
|
|
self.assertEqual(session.config()["method"], "orb")
|
|
result = process(session, 3)
|
|
for track in result["tracks"]:
|
|
self.assertNotEqual(track["identity"]["state"], "recognized",
|
|
"a late result from the old method must not label a track")
|
|
engine.shutdown()
|
|
|
|
def test_inspect_reports_track_not_in_view(self):
|
|
session, engine, _, _ = make_session()
|
|
process(session, 1)
|
|
with self.assertRaises(KeyError):
|
|
session.inspect(999)
|
|
engine.shutdown()
|
|
|
|
def test_close_stops_the_loop(self):
|
|
session, engine, _, _ = make_session()
|
|
session.close()
|
|
self.assertTrue(session.closed)
|
|
self.assertIsNone(session.take_pending(timeout=0.05))
|
|
engine.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|