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, pending=0): self.flushes = 0; self._pending = pending
    def flush(self): self.flushes += 1; return 0
    def pending(self): return self._pending


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 server'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 server can now see the camera is down


class _NoFrameRuntime:
    # step() returns None on a tick with no frame (dead/unplugged camera), a list otherwise.
    camera_id = "cam-dead"
    def __init__(self): self.steps = 0
    def step(self, now): self.steps += 1; return None


def test_no_frame_counts_as_erroring_not_a_healthy_frame():
    # A camera that returns no frame every tick (unplugged USB) must NOT tick frames++ and must
    # surface as erroring — otherwise a monitoring outage looks perfectly green (the whole point
    # of run_once returning None instead of []).
    rt, cl = _NoFrameRuntime(), _Client()
    ticks = iter(float(t) for t in range(1000))
    counters = run_loop(rt, 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)
    assert rt.steps == 10
    assert counters["frames"] == 0            # no frame was ever counted as healthy
    m = cl.metrics[0]
    assert m["fps"] == 0.0
    assert m["cameras_erroring"] == 1         # the dead camera is visible to the server


def test_heartbeat_reports_configured_camera_count_when_one_failed_to_start():
    # build_runtimes skips a camera that fails to construct at boot, so `runtime` holds only the
    # started cameras. The heartbeat must still report the CONFIGURED total and count the missing
    # one as erroring — otherwise the server can never see that a configured camera is dark.
    rt, cl, q = _Runtime(), _Client(), _Queue()   # 1 runtime started...
    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,
             configured_cameras=2)                # ...but 2 were configured
    m = cl.metrics[0]
    assert m["cameras_total"] == 2                 # the missing camera is not hidden
    assert m["cameras_erroring"] == 1              # and shows as erroring


def test_heartbeat_reports_queue_depth_and_echoes_config_version():
    class _VersionClient(_Client):
        def fetch_config(self):
            self.configs += 1
            return {"version": 7, "zones": [], "thresholds": {}}
    rt, cl, q = _Runtime(), _VersionClient(), _Queue(pending=4)
    ticks = iter(float(t) for t in range(1000))
    # sync at frame 5 (captures version 7) BEFORE the heartbeat at frame 10.
    run_loop(rt, cl, q, clock=lambda: next(ticks), sleep=lambda s: None,
             iterations=10, sync_every=5, heartbeat_every=10, flush_every=99)
    m = cl.metrics[0]
    assert m["queue_pending"] == 4            # transmitted so the dashboard can show backlog
    assert m["config_version"] == 7           # echoed so config-drift detection actually works


def test_heartbeat_omits_config_version_until_a_config_is_fetched():
    # Before any sync, the device does not know its version — it must OMIT config_version rather
    # than send a fake 0 (which made the server's drift indicator fire on every device forever).
    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 "config_version" not in cl.metrics[0]


def test_time_based_heartbeat_fires_on_elapsed_wall_time_not_iteration_count():
    # Production uses heartbeat_interval_s so the cadence is predictable no matter how slow a tick
    # is. With the clock advancing 10s/tick and a 30s interval, a beat fires every ~3 ticks.
    rt, cl, q = _Runtime(), _Client(), _Queue()
    ticks = iter(float(t * 10) for t in range(1000))  # t = 0,10,20,30,...
    run_loop(rt, cl, q, clock=lambda: next(ticks), sleep=lambda s: None,
             iterations=10, sync_every=99, heartbeat_every=None, flush_every=99,
             heartbeat_interval_s=30)
    # ticks at 0,10,20,30,...,90. Beats when now-last_beat >= 30: at 30, 60, 90 -> 3 beats.
    assert cl.beats == 3


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


# ---------------------------------------------------------------------------
# 2026-08-09: 로컬 경광등 + 임계값 적용.
# 이 두 가지는 루프 안에서만 배선되므로 여기서 고정한다.
# ---------------------------------------------------------------------------

class _TunableRuntime(_Runtime):
    def __init__(self):
        super().__init__()
        self.thresholds = None
        self.events = []

    def step(self, now):
        self.steps += 1
        return list(self.events)

    def apply_thresholds(self, thresholds, log=None):
        self.thresholds = dict(thresholds)
        return self.thresholds


class _FakeBeacon:
    def __init__(self, pin=7):
        self.pin = pin
        self.mode = "latch"
        self.hold_s = 10.0
        self.active_high = True
        self.min_rank = 0
        self.fires = []
        self.ticks = 0
        self.released = False

    def fire(self, severity, now):
        self.fires.append((severity, now))
        return True

    def tick(self, now):
        self.ticks += 1

    def release(self):
        self.released = True


class _Ev:
    def __init__(self, severity="critical"):
        self.severity = severity


def test_thresholds_from_config_are_applied_to_the_runtime():
    # 서버는 늘 이 값을 보내고 있었고, 엣지는 파싱한 뒤 버렸다. 그 배선이 살아 있는지.
    class Cl(_Client):
        def fetch_config(self):
            self.configs += 1
            return {"zones": [], "thresholds": {"fall_prone_s": 4.5}}

    rt = _TunableRuntime()
    counters = run_loop(rt, Cl(), _Queue(), clock=lambda: 0.0, sleep=lambda s: None,
                        iterations=5, sync_every=5, heartbeat_every=99, flush_every=99,
                        build_beacons_fn=lambda cfg, log=None: [])
    assert rt.thresholds == {"fall_prone_s": 4.5}
    assert counters["tuned"] == 1


def test_beacon_fires_on_the_tick_that_produced_the_event():
    # 서버 디스패처를 기다리지 않는다 — 업링크가 죽어 있어도 현장에서는 울려야 한다.
    b = _FakeBeacon()
    rt = _TunableRuntime()
    rt.events = [_Ev("critical")]
    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,
                        beacons=[b], build_beacons_fn=lambda cfg, log=None: [b])
    assert len(b.fires) == 3
    assert b.fires[0][0] == "critical"
    assert counters["beacon_fires"] == 3
    assert b.ticks == 3, "유지 시간 해제를 위해 매 틱 tick() 이 불린다"


def test_beacon_failure_never_stops_detection():
    # 경광등 배선 하나 때문에 그 카메라가 아무것도 못 보게 되는 것이 최악의 거래다.
    class Boom(_FakeBeacon):
        def fire(self, severity, now):
            raise RuntimeError("gpio busy")

    rt = _TunableRuntime()
    rt.events = [_Ev("critical")]
    counters = run_loop(rt, _Client(), _Queue(), clock=lambda: 0.0, sleep=lambda s: None,
                        iterations=4, sync_every=99, heartbeat_every=99, flush_every=99,
                        beacons=[Boom()], build_beacons_fn=lambda cfg, log=None: [])
    assert rt.steps == 4
    assert counters["errors"] >= 1


def test_beacons_are_not_reopened_when_the_wiring_is_unchanged():
    # 동기화마다 무조건 재생성하면 ~150틱마다 전 현장의 경광등이 한 번씩 깜빡인다.
    b = _FakeBeacon()
    rt = _TunableRuntime()
    run_loop(rt, _Client(), _Queue(), clock=lambda: 0.0, sleep=lambda s: None,
             iterations=10, sync_every=2, heartbeat_every=99, flush_every=99,
             beacons=[b], build_beacons_fn=lambda cfg, log=None: [_FakeBeacon()])
    assert b.released is False, "같은 배선이면 기존 비콘을 유지한다"


def test_changed_wiring_releases_the_old_beacon():
    b = _FakeBeacon(pin=7)
    rt = _TunableRuntime()
    run_loop(rt, _Client(), _Queue(), clock=lambda: 0.0, sleep=lambda s: None,
             iterations=4, sync_every=2, heartbeat_every=99, flush_every=99,
             beacons=[b], build_beacons_fn=lambda cfg, log=None: [_FakeBeacon(pin=13)])
    assert b.released is True, "핀이 바뀌면 이전 핀을 반드시 놓아준다"