77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
from cim_suite.modules.da12.domain.server_link import (
|
|
ServerLinkMonitor,
|
|
ServerLinkStatus,
|
|
)
|
|
from cim_suite.modules.da12.protocol import messages as m
|
|
|
|
|
|
def _status(incoming=0, errors=0, tsl_min=0):
|
|
return m.StationStatus(
|
|
incoming=incoming, errors=errors, time_since_last_min=tsl_min
|
|
)
|
|
|
|
|
|
def test_unknown_before_any_frame():
|
|
mon = ServerLinkMonitor()
|
|
assert mon.snapshot().status is ServerLinkStatus.UNKNOWN
|
|
|
|
|
|
def test_server_frame_connects():
|
|
mon = ServerLinkMonitor(monotonic=lambda: 0.0)
|
|
state = mon.update(_status(incoming=1))
|
|
assert state.status is ServerLinkStatus.CONNECTED
|
|
|
|
|
|
def test_errors_count_as_server_frame():
|
|
mon = ServerLinkMonitor(monotonic=lambda: 0.0)
|
|
state = mon.update(_status(incoming=0, errors=1))
|
|
assert state.status is ServerLinkStatus.CONNECTED
|
|
|
|
|
|
def test_quiet_past_timeout_disconnects():
|
|
clock = {"t": 0.0}
|
|
mon = ServerLinkMonitor(comms_timeout_s=30, monotonic=lambda: clock["t"])
|
|
mon.update(_status(incoming=1)) # connected at t=0
|
|
for step in range(1, 40):
|
|
clock["t"] = step * 1.2
|
|
mon.update(_status(incoming=0, tsl_min=int(clock["t"] // 60)))
|
|
assert mon.snapshot().status is ServerLinkStatus.DISCONNECTED
|
|
|
|
|
|
def test_reconnect_after_disconnect():
|
|
clock = {"t": 0.0}
|
|
mon = ServerLinkMonitor(comms_timeout_s=30, monotonic=lambda: clock["t"])
|
|
mon.update(_status(incoming=1))
|
|
for step in range(1, 40):
|
|
clock["t"] = step * 1.2
|
|
mon.update(_status(incoming=0, tsl_min=int(clock["t"] // 60)))
|
|
assert mon.snapshot().status is ServerLinkStatus.DISCONNECTED
|
|
clock["t"] += 1.2
|
|
state = mon.update(_status(incoming=1))
|
|
assert state.status is ServerLinkStatus.CONNECTED
|
|
|
|
|
|
def test_set_timeout_changes_threshold():
|
|
clock = {"t": 0.0}
|
|
mon = ServerLinkMonitor(comms_timeout_s=300, monotonic=lambda: clock["t"])
|
|
mon.set_timeout(10)
|
|
mon.update(_status(incoming=1)) # connected at t=0
|
|
clock["t"] = 12.0
|
|
mon.update(_status(incoming=0))
|
|
assert mon.snapshot().status is ServerLinkStatus.DISCONNECTED
|
|
|
|
|
|
def test_minute_backstop_trips_disconnect_immediately():
|
|
# A single quiet frame reporting 9 minutes since last server contact should
|
|
# disconnect at once (540s >> timeout) even with no elapsed monotonic time.
|
|
mon = ServerLinkMonitor(comms_timeout_s=120, monotonic=lambda: 0.0)
|
|
mon.update(_status(incoming=1))
|
|
state = mon.update(_status(incoming=0, tsl_min=9))
|
|
assert state.status is ServerLinkStatus.DISCONNECTED
|
|
|
|
|
|
def test_short_frame_does_not_crash_and_holds_status():
|
|
mon = ServerLinkMonitor(monotonic=lambda: 0.0)
|
|
state = mon.update(m.StationStatus()) # all-default frame: no server activity
|
|
assert state.status is ServerLinkStatus.UNKNOWN
|