product-recognition-demo/recognization/code/test_observe.py

"""Contract tests for identity ranking and session-local observe().

No OpenCV, YOLO, or network. These tests defend replay, isolation,
stale frames, backpressure, unknown/reject, and live-vs-still mechanics.
They do not measure shelf accuracy or latency.
"""

from __future__ import annotations

import os
import sys
import threading
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 identity import (
    IDENTITY_ACCEPTED,
    IDENTITY_AMBIGUOUS,
    IDENTITY_REJECTED,
    IDENTITY_UNKNOWN,
    decide_identity,
)
from observe import (
    CAPTURE_SAMPLED,
    CAPTURE_STILL,
    FrameRequest,
    ObserveError,
    PerceptionAdapter,
    PerceptionService,
    ScanService,
)


def _req(session, frame_id, payload, sequence, captured_at=1.0, kind=CAPTURE_SAMPLED):
    return FrameRequest(
        frame_id=frame_id,
        frame_bytes=payload,
        sequence=sequence,
        captured_at=captured_at,
        session_id=session.session_id,
        zone_id=session.zone_id,
        catalog_version=session.catalog_version,
        capture_kind=kind,
    )


class IdentityRankingTests(unittest.TestCase):
    def test_zero_scores_are_rejected_not_a_sku(self):
        ranked = decide_identity({"sku_a": 0.0, "sku_b": 0.0})
        self.assertEqual(ranked["identity_level"], IDENTITY_REJECTED)
        self.assertIsNone(ranked["best_sku"])
        self.assertEqual(ranked["status"], IDENTITY_REJECTED)

    def test_empty_scores_are_rejected(self):
        ranked = decide_identity({})
        self.assertEqual(ranked["identity_level"], IDENTITY_REJECTED)
        self.assertIsNone(ranked["best_sku"])

    def test_weak_unique_score_is_unknown(self):
        ranked = decide_identity({"sku_a": 0.4, "sku_b": 0.0})
        self.assertEqual(ranked["identity_level"], IDENTITY_UNKNOWN)
        self.assertIsNone(ranked["best_sku"])

    def test_close_scores_are_ambiguous(self):
        ranked = decide_identity({"sku_a": 20.0, "sku_b": 19.5})
        self.assertEqual(ranked["identity_level"], IDENTITY_AMBIGUOUS)
        self.assertIsNone(ranked["best_sku"])

    def test_unique_strong_score_is_accepted(self):
        ranked = decide_identity({"sku_a": 40.0, "sku_b": 5.0})
        self.assertEqual(ranked["identity_level"], IDENTITY_ACCEPTED)
        self.assertEqual(ranked["best_sku"], "sku_a")
        self.assertEqual(ranked["status"], "success")


class SessionObserveTests(unittest.TestCase):
    def setUp(self):
        def detect(frame_bytes: bytes):
            sku = frame_bytes.decode("ascii", errors="ignore")
            box = {"id": 0, "box": [0, 0, 10, 10], "sku_hint": sku}
            return {
                "width": 10,
                "height": 10,
                "boxes": [box],
                "empty_slots": [],
                "quality": {"coverage": 1.0},
            }

        def classify(box):
            hint = box.get("sku_hint", "")
            if hint.startswith("unknown"):
                return {"sku_a": 0.0, "sku_b": 0.0}
            if hint.startswith("ambig"):
                return {"sku_a": 20.0, "sku_b": 19.8}
            if hint.startswith("weak"):
                return {"sku_a": 0.4, "sku_b": 0.0}
            if hint.startswith("sku_b"):
                return {"sku_a": 2.0, "sku_b": 40.0}
            return {"sku_a": 40.0, "sku_b": 2.0}

        adapter = PerceptionAdapter(
            detect_fn=detect,
            classify_fn=classify,
            model_version="fake",
            matcher_method="hybrid",
            feature_extractor="fake",
        )
        self.scans = ScanService(PerceptionService(adapter), max_in_flight=1, max_queued=1)

    def test_replay_same_frame_id_and_content_is_idempotent(self):
        session = self.scans.start_session("zone-1", "stock-v1")
        first = self.scans.observe_frame(_req(session, "f1", b"sku_a", 1))
        second = self.scans.observe_frame(_req(session, "f1", b"sku_a", 1))
        self.assertIs(first, second)
        self.assertEqual(len(session.observations), 1)
        self.assertEqual(first.identities[0]["best_sku"], "sku_a")

    def test_same_frame_id_different_content_is_rejected(self):
        session = self.scans.start_session("zone-1", "stock-v1")
        self.scans.observe_frame(_req(session, "f1", b"sku_a", 1))
        with self.assertRaises(ObserveError):
            self.scans.observe_frame(_req(session, "f1", b"sku_b", 1))

    def test_sessions_do_not_share_observations(self):
        a = self.scans.start_session("zone-1", "stock-v1")
        b = self.scans.start_session("zone-1", "stock-v1")
        self.scans.observe_frame(_req(a, "f1", b"sku_a", 1))
        self.scans.observe_frame(_req(b, "f1", b"sku_b", 1))
        self.assertEqual(a.observations["f1"].identities[0]["best_sku"], "sku_a")
        self.assertEqual(b.observations["f1"].identities[0]["best_sku"], "sku_b")
        self.assertEqual(a.selected_frame_id, "f1")
        self.assertEqual(b.selected_frame_id, "f1")
        self.assertIsNot(a.observations["f1"], b.observations["f1"])

    def test_stale_sequence_cannot_overwrite_selected_state(self):
        session = self.scans.start_session("zone-1", "stock-v1")
        newer = self.scans.observe_frame(_req(session, "f2", b"sku_a", 2))
        older = self.scans.observe_frame(_req(session, "f1", b"sku_b", 1))
        self.assertEqual(session.selected_frame_id, "f2")
        self.assertTrue(newer.selected)
        self.assertFalse(older.selected)
        self.assertGreaterEqual(session.dropped_stale, 1)
        self.assertIn("f1", session.observations)

    def test_better_later_frame_replaces_selected_best(self):
        session = self.scans.start_session("zone-1", "stock-v1")
        weak = self.scans.observe_frame(_req(session, "f1", b"weak", 1))
        strong = self.scans.observe_frame(_req(session, "f2", b"sku_a", 2))
        self.assertFalse(weak.selected)
        self.assertTrue(strong.selected)
        self.assertEqual(session.selected_frame_id, "f2")

    def test_unknown_zero_score_is_not_selected_as_sku(self):
        session = self.scans.start_session("zone-1", "stock-v1")
        obs = self.scans.observe_frame(_req(session, "f1", b"unknown", 1))
        self.assertEqual(obs.identities[0]["identity_level"], IDENTITY_REJECTED)
        self.assertIsNone(obs.identities[0]["best_sku"])

    def test_backpressure_rejects_when_queue_full(self):
        started = threading.Event()
        release = threading.Event()

        def slow_detect(frame_bytes: bytes):
            started.set()
            release.wait(timeout=2)
            return {
                "width": 1,
                "height": 1,
                "boxes": [],
                "empty_slots": [],
                "quality": {},
            }

        scans = ScanService(
            PerceptionService(PerceptionAdapter(detect_fn=slow_detect)),
            max_in_flight=1,
            max_queued=0,
        )
        session = scans.start_session("zone-1", "stock-v1")
        errors = []

        def run():
            try:
                scans.observe_frame(_req(session, "slow", b"sku_a", 1))
            except ObserveError as exc:
                errors.append(str(exc))

        worker = threading.Thread(target=run)
        worker.start()
        self.assertTrue(started.wait(timeout=2))
        with self.assertRaises(ObserveError) as ctx:
            scans.observe_frame(_req(session, "other", b"sku_b", 2))
        self.assertIn("backpressure", str(ctx.exception))
        release.set()
        worker.join(timeout=2)
        self.assertTrue(session.dropped_stale >= 1)

    def test_finalize_is_immutable_and_rescan_creates_revision(self):
        session = self.scans.start_session("zone-1", "stock-v1")
        self.scans.observe_frame(_req(session, "f1", b"sku_a", 1))
        self.scans.review(session.session_id, actor="op", reason="looks-right")
        snap = self.scans.finalize(session.session_id)
        self.assertEqual(snap["status"], "finalized")
        with self.assertRaises(ObserveError):
            self.scans.observe_frame(_req(session, "f2", b"sku_b", 2))
        with self.assertRaises(ObserveError):
            self.scans.cancel(session.session_id)
        child = self.scans.rescan(session.session_id)
        self.assertEqual(child.status, "draft")
        self.assertEqual(child.revision, session.revision + 1)
        self.assertEqual(child.review["previous_session_id"], session.session_id)
        self.assertEqual(len(child.observations), 0)

    def test_live_vs_still_comparator_does_not_sum_frames(self):
        sampled = self.scans.start_session("zone-1", "stock-v1", capture_kind=CAPTURE_SAMPLED)
        still = self.scans.start_session("zone-1", "stock-v1", capture_kind=CAPTURE_STILL)
        self.scans.observe_frame(_req(sampled, "s1", b"sku_a", 1, kind=CAPTURE_SAMPLED))
        self.scans.observe_frame(_req(sampled, "s2", b"sku_a", 2, kind=CAPTURE_SAMPLED))
        self.scans.observe_frame(_req(still, "still1", b"sku_a", 1, kind=CAPTURE_STILL))
        comparison = self.scans.compare_live_vs_still(sampled.session_id, still.session_id)
        self.assertEqual(comparison["sampled"]["observation_count"], 2)
        self.assertEqual(comparison["still"]["observation_count"], 1)
        self.assertEqual(comparison["sampled"]["selected"]["accepted"], 1)
        self.assertEqual(comparison["still"]["selected"]["accepted"], 1)
        self.assertEqual(sampled.selected_frame_id, "s2")
        self.assertIn("not measured shelf accuracy", comparison["note"])

    def test_expired_session_rejects_new_frames(self):
        clock = {"t": 100.0}

        def now():
            return clock["t"]

        scans = ScanService(self.scans.perception, ttl_s=10, clock=now)
        session = scans.start_session("zone-1", "stock-v1")
        clock["t"] = 120.0
        with self.assertRaises(ObserveError):
            scans.observe_frame(_req(session, "f1", b"sku_a", 1))


class SessionHttpTests(unittest.TestCase):
    def setUp(self):
        from fastapi import FastAPI
        from fastapi.testclient import TestClient
        from observe import ScanService, PerceptionService, PerceptionAdapter, CAPTURE_SAMPLED

        def detect(frame_bytes: bytes):
            return {
                "width": 8,
                "height": 8,
                "boxes": [{"id": 0, "box": [0, 0, 8, 8], "sku_hint": frame_bytes.decode()}],
                "empty_slots": [],
                "quality": {},
            }

        def classify(box):
            hint = box.get("sku_hint", "")
            if hint == "sku_b":
                return {"sku_a": 2.0, "sku_b": 40.0}
            return {"sku_a": 40.0, "sku_b": 2.0}

        self.scans = ScanService(
            PerceptionService(PerceptionAdapter(detect_fn=detect, classify_fn=classify))
        )
        app = FastAPI()

        @app.post("/api/sessions")
        def start(payload: dict):
            session = self.scans.start_session(
                payload.get("zone_id", "zone-1"),
                payload.get("catalog_version", "stock-v1"),
                capture_kind=payload.get("capture_kind", CAPTURE_SAMPLED),
            )
            return {"status": "success", "session": session.snapshot()}

        @app.post("/api/sessions/{session_id}/frames")
        def frames(session_id: str, payload: dict):
            import base64
            session = self.scans.get_session(session_id)
            request = FrameRequest(
                frame_id=payload["frame_id"],
                frame_bytes=base64.b64decode(payload["frame_bytes_b64"]),
                sequence=int(payload["sequence"]),
                captured_at=1.0,
                session_id=session.session_id,
                zone_id=session.zone_id,
                catalog_version=session.catalog_version,
                capture_kind=session.capture_kind,
            )
            obs = self.scans.observe_frame(request)
            return {"status": "success", "observation": obs.as_dict(), "session": session.snapshot()}

        self.client = TestClient(app)

    def test_http_sessions_are_isolated(self):
        import base64
        a = self.client.post("/api/sessions", json={"zone_id": "zone-1", "catalog_version": "stock-v1"}).json()
        b = self.client.post("/api/sessions", json={"zone_id": "zone-1", "catalog_version": "stock-v1"}).json()
        aid = a["session"]["session_id"]
        bid = b["session"]["session_id"]
        oa = self.client.post(
            f"/api/sessions/{aid}/frames",
            json={"frame_id": "f1", "sequence": 1, "frame_bytes_b64": base64.b64encode(b"sku_a").decode()},
        ).json()
        ob = self.client.post(
            f"/api/sessions/{bid}/frames",
            json={"frame_id": "f1", "sequence": 1, "frame_bytes_b64": base64.b64encode(b"sku_b").decode()},
        ).json()
        self.assertEqual(oa["observation"]["identities"][0]["best_sku"], "sku_a")
        self.assertEqual(ob["observation"]["identities"][0]["best_sku"], "sku_b")
        self.assertNotEqual(aid, bid)



if __name__ == "__main__":
    unittest.main()