43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
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
|
|
def fetch_config(self): self.configs += 1; return {"zones": [], "thresholds": {}}
|
|
def send_heartbeat(self, m): self.beats += 1; 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
|