75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
from cim_suite.modules.da12.domain.models import LimitsTable, SensorTable, StationSettings, StatsTable
|
|
from cim_suite.modules.da12.protocol import messages as m
|
|
|
|
|
|
def _rec(sid, **kw):
|
|
base = dict(
|
|
serial=f"S{sid}", name="", type_text="01", scale=1.0, offset=0.0,
|
|
time="", average=1.0, value=1.0, alarm="OK", disp=0, dp=1, calc=0,
|
|
)
|
|
base.update(kw)
|
|
return m.SensorRecord(id=sid, **base)
|
|
|
|
|
|
def test_sensor_upsert_and_get():
|
|
t = SensorTable()
|
|
t.upsert(_rec(1, name="A"))
|
|
t.upsert(_rec(1, name="B")) # replaces
|
|
assert len(t) == 1 and t.get(1).name == "B"
|
|
|
|
|
|
def test_sensor_rejects_out_of_range_ids():
|
|
t = SensorTable()
|
|
t.upsert(_rec(0))
|
|
t.upsert(_rec(17))
|
|
assert len(t) == 0
|
|
|
|
|
|
def test_sensor_apply_partial_updates():
|
|
t = SensorTable()
|
|
t.upsert(_rec(1, value=1.0, average=1.0))
|
|
t.apply_current(1, 9.9)
|
|
assert t.get(1).value == 9.9
|
|
t.apply_average(m.AverageUpdate(1, "12:00:00", 8.8, "HW"))
|
|
assert t.get(1).average == 8.8 and t.get(1).alarm == "HW"
|
|
|
|
|
|
def test_sensor_all_sorted_and_remove():
|
|
t = SensorTable()
|
|
for sid in (3, 1, 2):
|
|
t.upsert(_rec(sid))
|
|
assert [r.id for r in t.all()] == [1, 2, 3]
|
|
t.remove(2)
|
|
assert [r.id for r in t.all()] == [1, 3]
|
|
|
|
|
|
def test_limits_and_stats_tables():
|
|
lim = LimitsTable()
|
|
lim.upsert(m.AlarmLimits(1, 1, 0, 0.0, 1.0, 9.0, 10.0))
|
|
assert lim.get(1).hi_alarm == 10.0
|
|
st = StatsTable()
|
|
st.upsert(m.Statistics(1, 9.0, "01:00", 1.0, "02:00", 5.0, 5.0))
|
|
assert st.get(1).maximum == 9.0
|
|
|
|
|
|
def test_station_settings_ordered():
|
|
s = StationSettings()
|
|
s.upsert(m.SettingLine(2, "B\tvb", 0))
|
|
s.upsert(m.SettingLine(1, "A\tva", -1))
|
|
assert [line.row for line in s.all()] == [1, 2]
|
|
|
|
|
|
def test_apply_current_stamps_updated_at():
|
|
t = SensorTable()
|
|
t.upsert(_rec(1))
|
|
t.apply_current(1, 5.0, at=123.0)
|
|
assert t.get(1).value == 5.0
|
|
assert t.get(1).updated_at == 123.0
|
|
|
|
|
|
def test_apply_current_without_at_leaves_timestamp_none():
|
|
t = SensorTable()
|
|
t.upsert(_rec(1))
|
|
t.apply_current(1, 5.0)
|
|
assert t.get(1).updated_at is None
|