move_detect/tests/test_daemon.py

import pytest

from edge.daemon import run_loop


class _Runtime:
    def __init__(self): self.steps = 0; self.applied = None
    def step(self, now): self.steps += 1; return []
    def apply_zones(self, zones): self.applied = zones


class _Client:
    def __init__(self): self.configs = 0; self.beats = 0; self.metrics = []
    def fetch_config(self): self.configs += 1; return {"zones": [], "thresholds": {}}
    def send_heartbeat(self, m): self.beats += 1; self.metrics.append(m); return True


class _Queue:
    def __init__(self): self.flushes = 0
    def flush(self): self.flushes += 1; return 0


def test_loop_steps_and_periodics():
    rt, cl, q = _Runtime(), _Client(), _Queue()
    ticks = iter(range(1000))
    counters = run_loop(rt, cl, q, clock=lambda: float(next(ticks)),
                        sleep=lambda s: None, iterations=10,
                        sync_every=5, heartbeat_every=5, flush_every=5)
    assert rt.steps == 10
    assert cl.configs == 2 and cl.beats == 2 and q.flushes == 2  # at frames 5 and 10
    assert rt.applied == []  # zones applied from fetched config


def test_loop_survives_step_exception():
    class Boom(_Runtime):
        def step(self, now):
            self.steps += 1
            if self.steps == 2: raise RuntimeError("frame glitch")
            return []
    rt = Boom()
    counters = run_loop(rt, _Client(), _Queue(), clock=lambda: 0.0,
                        sleep=lambda s: None, iterations=3,
                        sync_every=99, heartbeat_every=99, flush_every=99)
    assert rt.steps == 3  # kept going past the exception
    assert counters["errors"] == 1


def test_loop_logs_swallowed_step_exception_rate_limited():
    # A permanently-broken pipeline must not present as a silent green unit: the swallowed
    # exception has to be logged. But it must be rate-limited — a camera failing every frame
    # cannot spam the journal thousands of times a second.
    logs = []

    class AlwaysBoom(_Runtime):
        camera_id = "cam-x"
        def step(self, now):
            self.steps += 1
            raise RuntimeError("perma-broken")

    rt = AlwaysBoom()
    counters = run_loop(rt, _Client(), _Queue(), clock=lambda: 0.0,
                        sleep=lambda s: None, iterations=5,
                        sync_every=99, heartbeat_every=99, flush_every=99,
                        log=lambda msg, exc=None: logs.append(msg))
    assert counters["errors"] == 5           # every step raised
    assert len(logs) == 1                    # logged once, not five times (rate-limited)
    assert "cam-x" in logs[0]                # names the failing camera


def test_heartbeat_reports_real_fps_and_camera_health():
    # A hardcoded fps=0.0 makes the cloud's device-health columns fiction; it can't tell a
    # live device from a dead one. The heartbeat must carry the measured fps and per-camera
    # health so a degraded device is visible remotely.
    rt, cl, q = _Runtime(), _Client(), _Queue()
    ticks = iter(float(t) for t in range(1000))
    run_loop(rt, cl, q, clock=lambda: next(ticks), sleep=lambda s: None,
             iterations=10, sync_every=99, heartbeat_every=10, flush_every=99)
    assert len(cl.metrics) == 1
    m = cl.metrics[0]
    # 10 frames over the elapsed wall time between the first tick (t=0) and the beat (t=9).
    assert m["fps"] == pytest.approx(10 / 9, rel=0.05)
    assert m["cameras_total"] == 1
    assert m["cameras_erroring"] == 0


def test_heartbeat_flags_a_failing_camera():
    class AlwaysBoom(_Runtime):
        camera_id = "cam-x"
        def step(self, now):
            self.steps += 1
            raise RuntimeError("dead")
    cl = _Client()
    ticks = iter(float(t) for t in range(1000))
    run_loop(AlwaysBoom(), cl, _Queue(), clock=lambda: next(ticks), sleep=lambda s: None,
             iterations=10, sync_every=99, heartbeat_every=10, flush_every=99,
             log=lambda *a, **k: None)
    m = cl.metrics[0]
    assert m["cameras_erroring"] == 1   # the cloud can now see the camera is down


def test_loop_survives_flush_exception():
    class BoomQueue(_Queue):
        def flush(self):
            self.flushes += 1
            raise RuntimeError("db locked")
    rt = _Runtime()
    counters = run_loop(rt, _Client(), BoomQueue(), clock=lambda: 0.0,
                        sleep=lambda s: None, iterations=5,
                        sync_every=99, heartbeat_every=99, flush_every=2)
    assert rt.steps == 5          # loop kept running past the flush exceptions
    assert counters["errors"] >= 1