329 lines
13 KiB
Python
329 lines
13 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].
|
|
Module addressing (FR-4.3.11 v1.5) numbers modules 0-based WITHIN
|
|
THEIR TYPE: DI0=DI_8, DI1=DI_16, DO0=DO_8, DO1=DO_16, AI0=AI_2,
|
|
AO0=AO_2 — regardless of physical slot interleaving. Channels are
|
|
0-based on the module.
|
|
"""
|
|
|
|
def test_module_addressing_di(self, node, datastore):
|
|
# DI0.1: DI module 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: DI module 1 (DI_16, physical slot 1), channel 0 = 9th DI
|
|
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_interleaved(self, node, datastore):
|
|
# DO0.2: DO module 0 is the DO_8 at PHYSICAL SLOT 2 — per-type
|
|
# numbering means module 0, not slot 2
|
|
scenario = _scenario()
|
|
scenario["outputs"]["third_do"] = {"type": "do", "module": 0, "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_do_module_1_is_do16(self, node):
|
|
# DO1.0: DO module 1 is the DO_16 at physical slot 3
|
|
scenario = _scenario()
|
|
scenario["outputs"]["x"] = {"type": "do", "module": 1, "channel": 0}
|
|
engine = ScenarioEngine(node, scenario)
|
|
assert engine.alias_location("x").bit == 8 # after DO_8's 8 channels
|
|
|
|
def test_module_addressing_ai_ao(self, node, datastore):
|
|
# AI0.1: AI module 0 (AI_2 at physical slot 4), channel 1 = 2nd AI word
|
|
scenario = _scenario()
|
|
scenario["inputs"]["ai2"] = {"type": "ai", "module": 0, "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
|
|
|
|
# AO0.1: AO module 0 (AO_2 at physical slot 5), channel 1 = 2nd AO word
|
|
scenario["outputs"] = {"ao2": {"type": "ao", "module": 0, "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()
|
|
# the node has only 2 DI modules
|
|
scenario["inputs"]["bad"] = {"type": "di", "module": 2, "channel": 0}
|
|
with pytest.raises(IndexError, match="DI module 2"):
|
|
ScenarioEngine(node, scenario)
|
|
|
|
def test_channel_out_of_range_rejected(self, node):
|
|
# DI 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_module_kind_routes_to_own_type(self, node):
|
|
"""Per-type numbering (v1.5): 'do' module numbers address the DO
|
|
modules regardless of physical slot — DO0 is the DO_8 at slot 2.
|
|
A 'do' alias on module 1 (the DO_16) is valid and distinct."""
|
|
scenario = _scenario()
|
|
scenario["outputs"]["do8"] = {"type": "do", "module": 0, "channel": 0}
|
|
scenario["outputs"]["do16"] = {"type": "do", "module": 1, "channel": 0}
|
|
engine = ScenarioEngine(node, scenario)
|
|
assert engine.alias_location("do8").bit == 0
|
|
assert engine.alias_location("do16").bit == 8
|
|
|
|
def test_analog_kind_on_digital_module_rejected(self, node):
|
|
# per-type routing: 'ai' module 9 is out of range (only 1 AI module)
|
|
scenario = _scenario()
|
|
scenario["inputs"]["bad"] = {"type": "ai", "module": 9, "channel": 0}
|
|
with pytest.raises(IndexError):
|
|
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"}) |