317 lines
12 KiB
Python
317 lines
12 KiB
Python
"""Scenario engine tests: aliases, macros, pulses, ramps, assertions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from wago_sim.scenario.engine import ScenarioEngine, ScenarioError, Verdict
|
|
|
|
|
|
def _scenario(**overrides) -> dict:
|
|
base = {
|
|
"name": "test_scenario",
|
|
"inputs": {
|
|
"start_button": {"type": "di", "channel": 1},
|
|
"level_sensor": {"type": "ai", "channel": 1, "scale": 1.0},
|
|
},
|
|
"outputs": {
|
|
"pump_relay": {"type": "do", "channel": 1},
|
|
},
|
|
"sequence": [],
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
class _Pump:
|
|
"""Tiny plant model inside the test: emulates PLC behavior.
|
|
|
|
PLC logic: relay follows the start button (combinatorial copy of
|
|
the input image into the output image, as a real master would do
|
|
via Modbus writes).
|
|
"""
|
|
|
|
def __init__(self, engine: ScenarioEngine) -> None:
|
|
self.engine = engine
|
|
# loc.bit == global 0-based digital channel index == Modbus address
|
|
self._bit = engine.alias_location("pump_relay").bit
|
|
|
|
def apply(self) -> None:
|
|
ds = self.engine.datastore
|
|
on = bool(self.engine.alias_value("start_button"))
|
|
# emulate a Modbus master writing the coil at the global address
|
|
result = ds.write_bits(self._bit, [on])
|
|
assert not isinstance(result, Exception), result
|
|
|
|
|
|
def _run(engine: ScenarioEngine, plant=None, max_ticks=10000):
|
|
engine.start()
|
|
result = None
|
|
for _ in range(max_ticks):
|
|
if plant is not None:
|
|
plant.apply()
|
|
result = engine.pump()
|
|
if result is not None:
|
|
return result
|
|
raise AssertionError("scenario did not finish")
|
|
|
|
|
|
class TestAliases:
|
|
def test_di_alias_maps_channel(self, node, datastore):
|
|
scenario = _scenario()
|
|
engine = ScenarioEngine(node, scenario)
|
|
engine.set_alias("start_button", True)
|
|
assert datastore.read_bits(0, 1) == [True]
|
|
|
|
def test_ai_alias_applies_scale(self, node):
|
|
scenario = _scenario()
|
|
engine = ScenarioEngine(node, scenario)
|
|
engine.set_alias("level_sensor", 100)
|
|
loc = node.channel("ai", 1)
|
|
assert node.datastore.get_input_word(loc.word) == 100
|
|
assert engine.alias_value("level_sensor") == 100
|
|
|
|
def test_invalid_channel_rejected(self, node):
|
|
scenario = _scenario()
|
|
scenario["inputs"]["bad"] = {"type": "di", "channel": 999}
|
|
with pytest.raises(IndexError):
|
|
ScenarioEngine(node, scenario)
|
|
|
|
def test_output_alias_read(self, node, datastore):
|
|
scenario = _scenario()
|
|
engine = ScenarioEngine(node, scenario)
|
|
bit = engine.alias_location("pump_relay").bit
|
|
assert datastore.write_bits(bit, [True]) is None
|
|
assert engine.alias_value("pump_relay") is True
|
|
|
|
|
|
class TestModuleAddressing:
|
|
"""Module type + module number + channel number addressing (DI0.1).
|
|
|
|
The pump node fixture modules: [DI_8, DI_16, DO_8, DO_16, AI_2, AO_2]
|
|
(indices 0..5). Module addressing uses 0-based module positions and
|
|
0-based channel numbers on the module.
|
|
"""
|
|
|
|
def test_module_addressing_di(self, node, datastore):
|
|
# DI0.1: module slot 0 (DI_8), channel 1 (0-based) = 2nd DI
|
|
scenario = _scenario()
|
|
scenario["inputs"]["second_di"] = {"type": "di", "module": 0, "channel": 1}
|
|
engine = ScenarioEngine(node, scenario)
|
|
engine.set_alias("second_di", True)
|
|
# global bit index 1 -> Modbus bit address 1
|
|
assert datastore.read_bits(1, 1) == [True]
|
|
assert engine.alias_value("second_di") is True
|
|
|
|
def test_module_addressing_across_modules(self, node, datastore):
|
|
# DI1.0: module slot 1 (DI_16), channel 0 = 9th DI globally
|
|
# (8 channels on module 0 come first)
|
|
scenario = _scenario()
|
|
scenario["inputs"]["di16_first"] = {"type": "di", "module": 1, "channel": 0}
|
|
engine = ScenarioEngine(node, scenario)
|
|
engine.set_alias("di16_first", True)
|
|
assert datastore.read_bits(8, 1) == [True]
|
|
|
|
def test_module_addressing_do(self, node, datastore):
|
|
# DO2.2: module slot 2 (DO_8), channel 2 = 3rd DO globally
|
|
scenario = _scenario()
|
|
scenario["outputs"]["third_do"] = {"type": "do", "module": 2, "channel": 2}
|
|
engine = ScenarioEngine(node, scenario)
|
|
bit = engine.alias_location("third_do").bit
|
|
assert bit == 2
|
|
assert datastore.write_bits(bit, [True]) is None
|
|
assert engine.alias_value("third_do") is True
|
|
|
|
def test_module_addressing_ai_ao(self, node, datastore):
|
|
# AI4.1: module slot 4 (AI_2), channel 1 = 2nd AI word
|
|
scenario = _scenario()
|
|
scenario["inputs"]["ai2"] = {"type": "ai", "module": 4, "channel": 1, "scale": 1.0}
|
|
engine = ScenarioEngine(node, scenario)
|
|
engine.set_alias("ai2", 0x1234)
|
|
loc = node.channel("ai", 2) # global 2nd AI
|
|
assert datastore.get_input_word(loc.word) == 0x1234
|
|
|
|
# AO5.1: module slot 5 (AO_2), channel 1 = 2nd AO word
|
|
scenario["outputs"] = {"ao2": {"type": "ao", "module": 5, "channel": 1}}
|
|
engine2 = ScenarioEngine(node, scenario)
|
|
bit_word = node.channel("ao", 2).word
|
|
assert datastore.write_words(bit_word, [0xBEEF]) is None
|
|
assert engine2.alias_value("ao2") == 0xBEEF
|
|
|
|
def test_module_equivalent_to_pool_addressing(self, node):
|
|
"""DI0.0 (module addressing) == di channel 1 (pool addressing)."""
|
|
scenario = _scenario()
|
|
scenario["inputs"]["m"] = {"type": "di", "module": 0, "channel": 0}
|
|
scenario["inputs"]["p"] = {"type": "di", "channel": 1}
|
|
engine = ScenarioEngine(node, scenario)
|
|
engine.set_alias("m", True)
|
|
assert engine.alias_value("p") is True
|
|
|
|
def test_module_out_of_range_rejected(self, node):
|
|
scenario = _scenario()
|
|
scenario["inputs"]["bad"] = {"type": "di", "module": 99, "channel": 0}
|
|
with pytest.raises(IndexError, match="module 99"):
|
|
ScenarioEngine(node, scenario)
|
|
|
|
def test_channel_out_of_range_rejected(self, node):
|
|
# module 0 is a DI_8 -> channels 0-7 only
|
|
scenario = _scenario()
|
|
scenario["inputs"]["bad"] = {"type": "di", "module": 0, "channel": 8}
|
|
with pytest.raises(IndexError, match="channel 8"):
|
|
ScenarioEngine(node, scenario)
|
|
|
|
def test_kind_mismatch_rejected(self, node):
|
|
# module 0 is DI-only -> DO addressing must fail
|
|
scenario = _scenario()
|
|
scenario["outputs"]["bad"] = {"type": "do", "module": 0, "channel": 0}
|
|
from wago_sim.modules import ChannelMismatch
|
|
with pytest.raises(ChannelMismatch):
|
|
ScenarioEngine(node, scenario)
|
|
|
|
def test_analog_kind_on_digital_module_rejected(self, node):
|
|
# module 0 is DI-only -> AI addressing must fail
|
|
scenario = _scenario()
|
|
scenario["inputs"]["bad"] = {"type": "ai", "module": 0, "channel": 0}
|
|
from wago_sim.modules import ChannelMismatch
|
|
with pytest.raises(ChannelMismatch):
|
|
ScenarioEngine(node, scenario)
|
|
|
|
|
|
class TestSequence:
|
|
def test_set_and_assert_pass(self, node):
|
|
scenario = _scenario(sequence=[
|
|
{"set": {"start_button": True}},
|
|
{"assert": {"label": "relay on", "conditions": {"start_button": "ON"}, "within": 0}},
|
|
])
|
|
# plant copies the input directly to the relay
|
|
engine = ScenarioEngine(node, scenario)
|
|
plant = _Pump(engine)
|
|
result = _run(engine, plant)
|
|
assert result.verdict is Verdict.PASS
|
|
|
|
def test_assert_timeout_fails(self, node):
|
|
scenario = _scenario(sequence=[
|
|
{"set": {"start_button": True}},
|
|
{"assert": {"label": "relay on", "conditions": {"pump_relay": "ON"}, "within": 100}},
|
|
])
|
|
engine = ScenarioEngine(node, scenario)
|
|
# no plant -> relay never turns on
|
|
result = _run(engine)
|
|
assert result.verdict is Verdict.FAIL
|
|
assert result.steps[-1].verdict is Verdict.FAIL
|
|
|
|
def test_wait_step(self, node):
|
|
scenario = _scenario(sequence=[
|
|
{"wait": 50},
|
|
{"assert": {"label": "done waiting", "conditions": {"start_button": "OFF"}, "within": 0}},
|
|
])
|
|
engine = ScenarioEngine(node, scenario)
|
|
result = _run(engine)
|
|
assert result.verdict is Verdict.PASS
|
|
assert result.duration_ms >= 50
|
|
|
|
def test_pulse_macro(self, node):
|
|
scenario = _scenario(
|
|
macros={
|
|
"press": [
|
|
{"set": {"start_button": True}},
|
|
{"wait": 30},
|
|
{"set": {"start_button": False}},
|
|
]
|
|
},
|
|
sequence=[
|
|
{"macro": "press"},
|
|
{"wait": 100},
|
|
{"assert": {"label": "released", "conditions": {"start_button": "OFF"}, "within": 0}},
|
|
],
|
|
)
|
|
engine = ScenarioEngine(node, scenario)
|
|
seen_on = False
|
|
engine.start()
|
|
result = None
|
|
for _ in range(500):
|
|
result = engine.pump()
|
|
if engine.alias_value("start_button"):
|
|
seen_on = True
|
|
if result is not None:
|
|
break
|
|
assert result.verdict is Verdict.PASS
|
|
assert seen_on
|
|
assert engine.alias_value("start_button") is False
|
|
|
|
def test_unknown_step_rejected(self, node):
|
|
scenario = _scenario(sequence=[{"frobnicate": {}}])
|
|
engine = ScenarioEngine(node, scenario)
|
|
engine.start()
|
|
with pytest.raises(ScenarioError):
|
|
engine.pump()
|
|
|
|
def test_report_pass_finishes(self, node):
|
|
scenario = _scenario(sequence=[{"report": "PASS"}])
|
|
engine = ScenarioEngine(node, scenario)
|
|
result = _run(engine)
|
|
assert result.verdict is Verdict.PASS
|
|
|
|
def test_report_fail(self, node):
|
|
scenario = _scenario(sequence=[{"report": "FAIL"}])
|
|
engine = ScenarioEngine(node, scenario)
|
|
result = _run(engine)
|
|
assert result.verdict is Verdict.FAIL
|
|
|
|
|
|
class TestRamps:
|
|
def test_ramp_reaches_target(self, node):
|
|
scenario = _scenario(sequence=[
|
|
{"ramp": {"name": "level_sensor", "to": 1000, "over": 100}},
|
|
{"wait": 120},
|
|
{"assert": {"label": "level", "conditions": {"level_sensor": ">= 999"}, "within": 0}},
|
|
])
|
|
engine = ScenarioEngine(node, scenario)
|
|
result = _run(engine)
|
|
assert result.verdict is Verdict.PASS
|
|
assert engine.alias_value("level_sensor") == 1000
|
|
|
|
def test_ramp_progresses_linearly(self, node):
|
|
scenario = _scenario(sequence=[
|
|
{"ramp": {"name": "level_sensor", "to": 1000, "over": 1000}},
|
|
{"wait": 100000},
|
|
])
|
|
engine = ScenarioEngine(node, scenario)
|
|
engine.start()
|
|
samples = []
|
|
for _ in range(6):
|
|
engine.pump(100)
|
|
samples.append(engine.alias_value("level_sensor"))
|
|
# ramp starts on the first tick; +100 per 100 ms thereafter
|
|
assert samples == [0, 100, 200, 300, 400, 500]
|
|
|
|
|
|
class TestEdgeDetection:
|
|
def test_rising_edge_latched(self, node, datastore):
|
|
# a pending wait keeps the scenario alive while we flip the output
|
|
scenario = _scenario(sequence=[{"wait": 10000}])
|
|
engine = ScenarioEngine(node, scenario)
|
|
bit = engine.alias_location("pump_relay").bit
|
|
engine.pump()
|
|
assert datastore.write_bits(bit, [True]) is None
|
|
engine.pump()
|
|
assert engine.consume_rising_edge("pump_relay") is True
|
|
assert engine.consume_rising_edge("pump_relay") is False
|
|
|
|
|
|
class TestConditionMatcher:
|
|
def test_operators(self, node):
|
|
scenario = _scenario()
|
|
engine = ScenarioEngine(node, scenario)
|
|
engine.set_alias("level_sensor", 500)
|
|
m = engine._matcher
|
|
resolve = engine._resolve_alias
|
|
assert m.evaluate({"level_sensor": 500})
|
|
assert m.evaluate({"level_sensor": ">= 499"})
|
|
assert not m.evaluate({"level_sensor": ">= 501"})
|
|
assert m.evaluate({"level_sensor": "< 501"})
|
|
assert m.evaluate({"level_sensor": "!= 499"})
|
|
assert m.evaluate({"level_sensor": "== 500"})
|
|
assert not m.evaluate({"level_sensor": "> 500"})
|
|
# boolean forms
|
|
assert m.evaluate({"start_button": "OFF"}) |