yovision/Brain/tests/test_runtime.py

from __future__ import annotations

import unittest

from datetime import datetime, timezone

from Brain.yovision_brain.domain import Box, Detection
from Brain.yovision_brain.runtime import DemoEngine, EventIngressUnavailable
from Brain.yovision_brain.source import FramePacket, SyntheticSource, cv2, np


class _TwoFrameSource:
    fixture = True
    mode = "test"
    label = "test"
    source_ref = "safe-source"

    def __init__(self) -> None:
        self.index = 0

    def read(self) -> FramePacket:
        boxes = (Box(0.10, 0.4, 0.20, 0.8), Box(0.60, 0.4, 0.70, 0.8))
        detection = Detection("P-1", "person", boxes[min(self.index, 1)])
        self.index += 1
        return FramePacket(np.zeros((180, 320, 3), dtype=np.uint8), datetime.now(timezone.utc), (detection,))

    def close(self) -> None:
        return


class _Ingress:
    def __init__(self, fail: bool = False) -> None:
        self.items = []
        self.fail = fail

    def submit(self, value: object) -> None:
        if self.fail:
            raise RuntimeError("disk unavailable")
        self.items.append(value)

    def start(self) -> None:
        return

    def stop(self) -> None:
        return

    def status(self) -> dict[str, object]:
        return {"enabled": True, "queued": len(self.items)}


@unittest.skipIf(cv2 is None, "pinned OpenCV package is not installed")
class RuntimeTests(unittest.TestCase):
    def test_rejects_non_finite_fps(self) -> None:
        with self.assertRaisesRegex(ValueError, "fps"):
            DemoEngine(SyntheticSource(width=320, height=180), fps=float("nan"))

    def test_synthetic_step_produces_safe_state_and_jpeg(self) -> None:
        engine = DemoEngine(SyntheticSource(width=320, height=180), fps=2.0)
        engine.step()
        state = engine.state()
        self.assertTrue(state["source"]["fixture"])
        self.assertEqual(state["detector"]["name"], "scripted_fixture")
        self.assertEqual(state["frame"]["width"], 320)
        self.assertGreater(len(engine.frame_jpeg() or b""), 100)
        self.assertNotIn("url", state["source"])

    def test_zone_update_increments_version(self) -> None:
        engine = DemoEngine(SyntheticSource(width=320, height=180))
        updated = engine.update_zone({"name": "新区域", "points": [{"x": 0.1, "y": 0.1}, {"x": 0.9, "y": 0.1}, {"x": 0.5, "y": 0.9}]})
        self.assertEqual(updated.version, 2)
        self.assertEqual(engine.state()["zone"]["name"], "新区域")

    def test_event_is_persisted_before_it_is_exposed(self) -> None:
        ingress = _Ingress()
        engine = DemoEngine(_TwoFrameSource(), event_ingress=ingress)
        engine.step()
        engine.step()
        state = engine.state()
        self.assertEqual(1, len(ingress.items))
        self.assertEqual("outbox_persisted", state["events"][0]["delivery_status"])
        self.assertEqual(1, state["event_ingress"]["queued"])

    def test_outbox_failure_does_not_claim_delivery(self) -> None:
        engine = DemoEngine(_TwoFrameSource(), event_ingress=_Ingress(fail=True))
        engine.step()
        with self.assertRaises(EventIngressUnavailable):
            engine.step()
        self.assertEqual([], engine.state()["events"])


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