826 lines
38 KiB
Python
826 lines
38 KiB
Python
"""Web UI API tests (spec §10): auth, node layout, values, manipulation,
|
|
diagnostics — exercised in-process via httpx ASGITransport (no socket)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from wago_sim.webui.app import build_webui
|
|
|
|
|
|
@pytest.fixture
|
|
def webui(app):
|
|
"""FastAPI app bound to the simulator app (default credentials)."""
|
|
return build_webui(app, user="admin", password="simu")
|
|
|
|
|
|
@pytest.fixture
|
|
async def client(webui):
|
|
"""Async httpx client with a valid session (default credentials)."""
|
|
transport = httpx.ASGITransport(app=webui)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
|
r = await c.post("/login", json={"user": "admin", "password": "simu"})
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["ok"] is True
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
async def anon_client(webui):
|
|
"""Async httpx client WITHOUT login."""
|
|
transport = httpx.ASGITransport(app=webui)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
class TestAuth:
|
|
async def test_login_page_served(self, anon_client):
|
|
r = await anon_client.get("/login")
|
|
assert r.status_code == 200
|
|
assert "WAGO" in r.text
|
|
|
|
async def test_root_redirects_when_anonymous(self, anon_client):
|
|
r = await anon_client.get("/", follow_redirects=False)
|
|
assert r.status_code == 303
|
|
assert r.headers["location"] == "/login"
|
|
|
|
async def test_api_requires_session(self, anon_client):
|
|
r = await anon_client.get("/api/node")
|
|
assert r.status_code == 401
|
|
|
|
async def test_login_wrong_password(self, anon_client):
|
|
r = await anon_client.post("/login", json={"user": "admin", "password": "wrong"})
|
|
assert r.status_code == 401
|
|
assert r.json()["ok"] is False
|
|
|
|
async def test_logout_invalidates_session(self, client):
|
|
r = await client.post("/api/auth/logout")
|
|
assert r.status_code == 200
|
|
r = await client.get("/api/node")
|
|
assert r.status_code == 401
|
|
|
|
async def test_login_lockout_after_failures(self, anon_client, monkeypatch):
|
|
import wago_sim.webui.auth as auth_mod
|
|
|
|
# reset the module-level throttle so suite order does not matter
|
|
monkeypatch.setattr(auth_mod, "_LOGIN_ATTEMPTS", {})
|
|
for _ in range(5):
|
|
r = await anon_client.post("/login", json={"user": "admin", "password": "nope"})
|
|
assert r.status_code == 401
|
|
# 6th attempt with CORRECT password is locked out too
|
|
r = await anon_client.post("/login", json={"user": "admin", "password": "simu"})
|
|
assert r.status_code == 401
|
|
|
|
|
|
class TestNodeAPI:
|
|
async def test_node_layout(self, client):
|
|
r = await client.get("/api/node")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["coupler"] == "750-362"
|
|
articles = [m["article"] for m in data["modules"]]
|
|
assert "750-430" in articles # DI_8 from pump_modules
|
|
# module-based addressing: channel 0 -> address like DI0.0
|
|
di_mod = next(m for m in data["modules"] if m["article"] == "750-430")
|
|
assert di_mod["channels"][0]["address"] == "DI0.0"
|
|
assert "modbus" in di_mod["channels"][0]
|
|
# image sizes from the layout are reported
|
|
assert data["image_bits"]["digital_in"] == 24 # DI_8 + DI_16
|
|
|
|
async def test_values_snapshot(self, client):
|
|
r = await client.get("/api/values")
|
|
assert r.status_code == 200
|
|
values = r.json()["values"]
|
|
kinds = {v["kind"] for v in values}
|
|
assert kinds == {"di", "do", "ai", "ao"}
|
|
|
|
|
|
class TestManipulation:
|
|
async def test_toggle_di_via_api(self, client, datastore):
|
|
r = await client.post("/api/channels/di/0/0", json={"action": "toggle"})
|
|
assert r.status_code == 200
|
|
assert r.json()["value"] is True
|
|
# the internal memory handler shows the same state (channel 1)
|
|
assert datastore.digital_input_channel(1) is True
|
|
|
|
async def test_toggle_do_forces_output_image(self, client, datastore):
|
|
# DO_8 sits at module slot 2 in the pump fixture
|
|
r = await client.post("/api/channels/do/0/0", json={"action": "toggle"})
|
|
assert r.status_code == 200
|
|
# DO channel 1 is forced ON in the OUTPUT image (master-side)
|
|
assert datastore.digital_output_channel(1) is True
|
|
# ... and the +0x0200 read-back mirror reflects it
|
|
bits = datastore.read_bits(0x0200, 1)
|
|
assert bits == [True]
|
|
|
|
async def test_ai_value_set(self, client, datastore):
|
|
# AI_2 sits at module slot 4 in the pump fixture (DI_8, DI_16,
|
|
# DO_8, DO_16, AI_2, AO_2); its first word is input word 0
|
|
r = await client.post("/api/channels/ai/0/0", json={"value": 200})
|
|
assert r.status_code == 200
|
|
assert r.json()["value"] == 200
|
|
assert datastore.get_input_word(0) == 200
|
|
|
|
async def test_ao_force_writes_output_image(self, client, datastore):
|
|
r = await client.post("/api/channels/ao/0/0", json={"value": 1234})
|
|
assert r.status_code == 200
|
|
# written at the output image word 0 -> read back at +0x0200
|
|
assert datastore.read_words(0x0200, 1) == [1234]
|
|
|
|
async def test_unknown_channel_404(self, client):
|
|
r = await client.post("/api/channels/di/99/0", json={"action": "toggle"})
|
|
assert r.status_code == 404
|
|
|
|
async def test_kind_mismatch_400(self, client):
|
|
# per-type numbering (v1.5): DO module 0 is the DO_8 — asking
|
|
# for a DO channel beyond its 8 channels is out of range (404);
|
|
# DO module 2 does not exist (only DO0, DO1)
|
|
r = await client.post("/api/channels/do/2/0", json={"action": "toggle"})
|
|
assert r.status_code == 404
|
|
r = await client.post("/api/channels/do/0/9", json={"action": "toggle"})
|
|
assert r.status_code in (400, 404) # channel out of module range
|
|
|
|
|
|
class TestDiagnostics:
|
|
async def test_diagnostics_payload(self, client):
|
|
r = await client.get("/api/diagnostics")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["watchdog"]["status"] in (0, 1, 2)
|
|
assert "mod_tcp_message_counter" in data["statistics"]
|
|
assert "led_error" in data
|
|
|
|
async def test_inject_and_clear_led_fault(self, client, app):
|
|
r = await client.post("/api/diagnostics/inject-fault",
|
|
json={"fault": "led", "code": 5, "argument": 2})
|
|
assert r.status_code == 200
|
|
assert app.registers.read_config(0x1020, 1) == [5]
|
|
assert app.registers.read_config(0x1021, 1) == [2]
|
|
r = await client.post("/api/diagnostics/clear-fault", json={"fault": "led"})
|
|
assert r.status_code == 200
|
|
assert app.registers.read_config(0x1020, 1) == [0]
|
|
|
|
async def test_inject_module_diagnosis(self, client, app):
|
|
r = await client.post("/api/diagnostics/inject-fault",
|
|
json={"fault": "module_diagnosis", "module": 2,
|
|
"channel": 1, "diagnosis": 0x0001})
|
|
assert r.status_code == 200
|
|
assert app.registers.read_config(0x1050, 3) == [2, 1, 0x0001]
|
|
|
|
async def test_reset_stats(self, client, app):
|
|
app.stats.bad_address += 3
|
|
r = await client.post("/api/diagnostics/reset-stats")
|
|
assert r.status_code == 200
|
|
assert app.stats.bad_address == 0
|
|
|
|
|
|
class TestReadOnlyMode:
|
|
async def test_readonly_blocks_writes(self, app):
|
|
ro = build_webui(app, user="admin", password="simu", readonly=True)
|
|
transport = httpx.ASGITransport(app=ro)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
|
r = await c.post("/login", json={"user": "admin", "password": "simu"})
|
|
assert r.status_code == 200
|
|
assert r.json()["readonly"] is True
|
|
r = await c.get("/api/node")
|
|
assert r.status_code == 200
|
|
r = await c.post("/api/channels/di/0/0", json={"action": "toggle"})
|
|
assert r.status_code == 403
|
|
assert "read-only" in r.json()["detail"]
|
|
r = await c.post("/api/diagnostics/reset-stats")
|
|
assert r.status_code == 403
|
|
|
|
async def test_readonly_blocks_env_control(self, app):
|
|
ro = build_webui(app, user="admin", password="simu", readonly=True)
|
|
transport = httpx.ASGITransport(app=ro)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
|
await c.post("/login", json={"user": "admin", "password": "simu"})
|
|
r = await c.get("/api/env")
|
|
assert r.status_code == 200 # status is a read
|
|
assert r.json()["state"] == "none"
|
|
r = await c.post("/api/env/start",
|
|
json={"ref": "envs.pump_station_env:PumpStationEnv"})
|
|
assert r.status_code == 403
|
|
r = await c.post("/api/env/stop")
|
|
assert r.status_code == 403
|
|
|
|
|
|
class TestEnvControl:
|
|
"""FR-10.2.15 runtime env lifecycle + FR-10.2.17 hot reload."""
|
|
|
|
REF = "envs.pump_station_env:PumpStationEnv"
|
|
|
|
async def test_env_status_none(self, client):
|
|
r = await client.get("/api/env")
|
|
assert r.status_code == 200
|
|
assert r.json() == {"ref": None, "state": "none", "reason": None,
|
|
"sim_time_ms": None, "tick_ms": None}
|
|
|
|
async def test_start_stop_cycle(self, client, app):
|
|
r = await client.post("/api/env/start", json={"ref": self.REF})
|
|
assert r.status_code == 200, r.text
|
|
st = r.json()
|
|
assert st["state"] == "running"
|
|
assert st["ref"] == self.REF
|
|
assert app.runner.env is not None
|
|
# let it tick a little, then check status shows sim time
|
|
await asyncio.sleep(0.05)
|
|
r = await client.get("/api/env")
|
|
assert r.json()["sim_time_ms"] is not None
|
|
# stop -> none
|
|
r = await client.post("/api/env/stop")
|
|
assert r.status_code == 200
|
|
assert r.json()["state"] == "none"
|
|
assert app.runner.env is None
|
|
|
|
async def test_start_refused_while_loaded(self, client):
|
|
await client.post("/api/env/start", json={"ref": self.REF})
|
|
r = await client.post("/api/env/start", json={"ref": self.REF})
|
|
assert r.status_code == 409
|
|
assert "stop" in r.json()["detail"]
|
|
await client.post("/api/env/stop")
|
|
|
|
async def test_start_requires_ref(self, client):
|
|
r = await client.post("/api/env/start", json={})
|
|
assert r.status_code == 400
|
|
|
|
async def test_pause_resume(self, client, app):
|
|
await client.post("/api/env/start", json={"ref": self.REF})
|
|
# let it tick so sim time advances
|
|
await asyncio.sleep(0.05)
|
|
t0 = (await client.get("/api/env")).json()["sim_time_ms"]
|
|
assert t0 > 0
|
|
r = await client.post("/api/env/pause")
|
|
assert r.json()["state"] == "paused"
|
|
t1 = (await client.get("/api/env")).json()["sim_time_ms"]
|
|
await asyncio.sleep(0.05)
|
|
t2 = (await client.get("/api/env")).json()["sim_time_ms"]
|
|
assert t2 == t1 # frozen while paused
|
|
r = await client.post("/api/env/resume")
|
|
assert r.json()["state"] == "running"
|
|
await asyncio.sleep(0.05)
|
|
t3 = (await client.get("/api/env")).json()["sim_time_ms"]
|
|
assert t3 > t2
|
|
# double-pause / resume-without-pause are refused
|
|
await client.post("/api/env/pause")
|
|
r = await client.post("/api/env/pause")
|
|
assert r.status_code == 409
|
|
r = await client.post("/api/env/stop")
|
|
assert r.status_code == 200
|
|
r = await client.post("/api/env/resume")
|
|
assert r.status_code == 409
|
|
r = await client.post("/api/env/pause")
|
|
assert r.status_code == 409 # nothing loaded
|
|
|
|
async def test_reset_returns_to_initial_state(self, client):
|
|
await client.post("/api/env/start", json={"ref": self.REF})
|
|
await asyncio.sleep(0.05)
|
|
st = (await client.get("/api/env")).json()
|
|
assert st["sim_time_ms"] > 0
|
|
r = await client.post("/api/env/reset")
|
|
assert r.status_code == 200
|
|
# the reset RESPONSE carries the post-reset status: asserting
|
|
# on a follow-up GET would race the running 10 ms ticker
|
|
# (which legitimately ticks between the two requests)
|
|
assert r.json()["sim_time_ms"] == 0.0 # deterministic fresh episode
|
|
await client.post("/api/env/stop")
|
|
|
|
async def test_restart_fresh_instance(self, client, app):
|
|
await client.post("/api/env/start", json={"ref": self.REF})
|
|
first = app.runner.env
|
|
await asyncio.sleep(0.02)
|
|
r = await client.post("/api/env/restart")
|
|
assert r.status_code == 200
|
|
assert r.json()["state"] == "running"
|
|
assert app.runner.env is not first # a NEW instance
|
|
assert app.runner.env.sim_time_ms < first.sim_time_ms
|
|
await client.post("/api/env/stop")
|
|
|
|
async def test_is_done_ends_ticker(self, client, app):
|
|
# an env that is done after 30 ms of sim time (TICK_MS 10 -> 3 ticks)
|
|
ref = "envs.test_done_env:PumpStationEnv"
|
|
app.runner.env = None
|
|
r = await client.post("/api/env/start", json={"ref": ref})
|
|
assert r.status_code == 200
|
|
await asyncio.sleep(0.15) # let the paced ticker finish
|
|
st = (await client.get("/api/env")).json()
|
|
assert st["state"] == "done"
|
|
assert st["reason"] == "is_done()"
|
|
assert app.runner.env is not None # instance kept for inspection
|
|
# pause on done -> refused (restart instead)
|
|
r = await client.post("/api/env/pause")
|
|
assert r.status_code == 409
|
|
# reset brings it back to running
|
|
r = await client.post("/api/env/reset")
|
|
assert r.json()["state"] == "running"
|
|
await client.post("/api/env/stop")
|
|
|
|
async def test_module_save_discards_env(self, client, app):
|
|
await client.post("/api/env/start", json={"ref": self.REF})
|
|
assert app.runner.env is not None
|
|
r = await client.put("/api/config/modules", json={
|
|
"modules": ["750-430", "750-530"], "persist": False,
|
|
})
|
|
assert r.status_code == 200
|
|
# FR-10.2.15: the rebuild discards the env (old-layout bindings)
|
|
assert app.runner.env is None
|
|
assert (await client.get("/api/env")).json()["state"] == "none"
|
|
# and the new runner points at the NEW node
|
|
assert app.runner.node is app.node
|
|
|
|
async def test_start_unknown_env_is_409(self, client):
|
|
r = await client.post("/api/env/start",
|
|
json={"ref": "envs.no_such_env:Env"})
|
|
assert r.status_code == 409
|
|
assert "cannot import" in r.json()["detail"]
|
|
|
|
async def test_hot_reload_after_source_edit(self, client, app, tmp_path,
|
|
monkeypatch):
|
|
"""FR-10.2.17: start -> edit source on disk -> restart runs the
|
|
NEW code (sys.modules cache invalidated)."""
|
|
pkg = tmp_path / "reloadpkg"
|
|
pkg.mkdir()
|
|
(pkg / "__init__.py").write_text("", encoding="utf-8")
|
|
# v1: level rises
|
|
(pkg / "reload_env.py").write_text(
|
|
"from wago_sim.scenario.plantenv import PlantEnv\n"
|
|
"class ReloadEnv(PlantEnv):\n"
|
|
" def model(self, state, dt_s):\n"
|
|
" level = float(state.by_code('AI0.0'))\n"
|
|
" return {'AI0.0': level + 1.0}\n"
|
|
"Env = ReloadEnv\n", encoding="utf-8")
|
|
sys.path.insert(0, str(tmp_path))
|
|
try:
|
|
r = await client.post("/api/env/start",
|
|
json={"ref": "reloadpkg.reload_env:ReloadEnv"})
|
|
assert r.status_code == 200, r.text
|
|
await asyncio.sleep(0.03)
|
|
v1 = app.datastore.get_input_word(0)
|
|
assert v1 > 0
|
|
# v2: level DRAINS instead — same class name
|
|
(pkg / "reload_env.py").write_text(
|
|
"from wago_sim.scenario.plantenv import PlantEnv\n"
|
|
"class ReloadEnv(PlantEnv):\n"
|
|
" def model(self, state, dt_s):\n"
|
|
" level = float(state.by_code('AI0.0'))\n"
|
|
" return {'AI0.0': max(0.0, level - 1.0)}\n"
|
|
"Env = ReloadEnv\n", encoding="utf-8")
|
|
r = await client.post("/api/env/restart")
|
|
assert r.status_code == 200
|
|
app.datastore.set_input_word(0, 100)
|
|
await asyncio.sleep(0.03)
|
|
assert app.datastore.get_input_word(0) < 100 # NEW code ran
|
|
await client.post("/api/env/stop")
|
|
finally:
|
|
sys.path.remove(str(tmp_path))
|
|
|
|
|
|
class TestCouplerConfig:
|
|
async def test_get_coupler_config(self, client, app):
|
|
r = await client.get("/api/config/coupler")
|
|
assert r.status_code == 200
|
|
cfg = r.json()
|
|
# defaults: 100 units = 10 s watchdog, full masks, no delay
|
|
assert cfg["watchdog"]["time_units"] == 100
|
|
assert cfg["watchdog"]["mask_fc_1_16"] == 0xFFFF
|
|
assert cfg["watchdog"]["type"] == 0
|
|
assert cfg["response_delay_ms"] == 0
|
|
assert cfg["idle_timeout_units"] == 600
|
|
assert len(cfg["mac_id"]) == 3
|
|
# image sizes for the pump fixture (AO_2 out / AI_2 in analog,
|
|
# DO 24 bits out / DI 24 bits in — 0x1022..0x1025 per manual §11.2.5.5)
|
|
assert cfg["image_bits"]["analog_in"] == 32
|
|
assert cfg["image_bits"]["analog_out"] == 32
|
|
assert cfg["image_bits"]["digital_in"] == 24
|
|
assert cfg["image_bits"]["digital_out"] == 24
|
|
|
|
async def test_put_coupler_config_roundtrip(self, client, app):
|
|
r = await client.put("/api/config/coupler", json={
|
|
"watchdog": {"time_units": 200, "close_on_timeout": True},
|
|
"response_delay_ms": 25,
|
|
"idle_timeout_units": 10,
|
|
})
|
|
assert r.status_code == 200
|
|
assert r.json()["applied"] == 4
|
|
# written through the internal register bank
|
|
assert app.registers.read_config(0x1000, 1) == [200]
|
|
assert app.registers.read_config(0x1037, 1) == [25]
|
|
assert app.registers.read_config(0x1030, 1) == [10]
|
|
assert app.registers.read_config(0x1009, 1) == [1]
|
|
|
|
async def test_watchdog_refusal_while_running(self, client, app):
|
|
# start the watchdog (trigger write, like the manual's flow)
|
|
app.watchdog.write(0x1003, 0x0001)
|
|
assert app.watchdog.status == 1
|
|
r = await client.put("/api/config/coupler", json={
|
|
"watchdog": {"time_units": 50},
|
|
})
|
|
assert r.status_code == 400
|
|
assert "cannot change time while running" in r.json()["detail"]
|
|
|
|
async def test_put_coupler_config_readonly_403(self, app):
|
|
ro = build_webui(app, user="admin", password="simu", readonly=True)
|
|
transport = httpx.ASGITransport(app=ro)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
|
await c.post("/login", json={"user": "admin", "password": "simu"})
|
|
r = await c.put("/api/config/coupler", json={"response_delay_ms": 1})
|
|
assert r.status_code == 403
|
|
|
|
|
|
class TestModuleConfig:
|
|
async def test_get_module_config(self, client, app):
|
|
r = await client.get("/api/config/modules")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert [m["article"] for m in data["modules"]] == [
|
|
"750-430", "750-1405", "750-530", "750-1516", "750-467", "750-554",
|
|
]
|
|
articles = {c["article"] for c in data["catalog"]}
|
|
assert "750-461" in articles # catalog has all 13 articles
|
|
assert data["max_modules"] == 64
|
|
assert data["node_config_path"] is None # fixture built without a path
|
|
|
|
async def test_rebuild_updates_layout_and_registers(self, client, app):
|
|
# swap in a bigger DI module list
|
|
r = await client.put("/api/config/modules", json={
|
|
"modules": ["750-1405", "750-1405", "750-530"], "persist": False,
|
|
})
|
|
assert r.status_code == 200
|
|
assert r.json()["image_bits"]["digital_in"] == 32
|
|
# /api/node reflects the new layout
|
|
r = await client.get("/api/node")
|
|
assert [m["article"] for m in r.json()["modules"]] == [
|
|
"750-1405", "750-1405", "750-530",
|
|
]
|
|
# register bank updated: 0x1025 digital-IN bits = 32 (two DI_16)
|
|
assert app.registers.read_config(0x1025, 1) == [32]
|
|
# 0x1024 digital-OUT bits = 8 (one DO_8)
|
|
assert app.registers.read_config(0x1024, 1) == [8]
|
|
# module description 0x2030 contains the new articles
|
|
# (2 chars/word, high byte first — reverse of _string_to_words)
|
|
words = app.registers.read_config(0x2030, 65)
|
|
text = "".join(
|
|
chr(b) for w in words for b in ((w >> 8) & 0xFF, w & 0xFF)
|
|
)
|
|
assert "750-1405" in text
|
|
# values were cleared by the rebuild (KBUS-reset semantics)
|
|
assert all(w == 0 for w in app.datastore.output_words)
|
|
|
|
async def test_rebuild_rejected_invalid(self, client):
|
|
r = await client.put("/api/config/modules", json={"modules": ["750-9999"]})
|
|
assert r.status_code == 400
|
|
assert "750-9999" in r.json()["detail"]
|
|
r = await client.put("/api/config/modules", json={"modules": []})
|
|
assert r.status_code == 400
|
|
r = await client.put("/api/config/modules", json={"modules": ["750-430"] * 65})
|
|
assert r.status_code == 400
|
|
assert "max 64" in r.json()["detail"]
|
|
|
|
async def test_rebuild_persists_yaml(self, client, app, tmp_path):
|
|
cfg_file = tmp_path / "node.yaml"
|
|
app.node_config_path = str(cfg_file)
|
|
r = await client.put("/api/config/modules", json={
|
|
"modules": ["750-430", "750-530"], "persist": True,
|
|
})
|
|
assert r.status_code == 200
|
|
assert r.json()["saved_to"] == str(cfg_file)
|
|
text = cfg_file.read_text(encoding="utf-8")
|
|
assert 'coupler: 750-362' in text
|
|
assert '"750-430"' in text and '"750-530"' in text
|
|
# and it loads back through the normal loader
|
|
from wago_sim.scenario.dsl import load_yaml, load_node_modules
|
|
modules = load_node_modules(load_yaml(cfg_file))
|
|
assert [m.article for m in modules] == ["750-430", "750-530"]
|
|
|
|
async def test_modbus_still_serves_after_rebuild(self, client, app):
|
|
"""Modbus connections survive a rebuild; new image answers."""
|
|
r = await client.put("/api/config/modules", json={
|
|
"modules": ["750-530", "750-554"], "persist": False,
|
|
})
|
|
assert r.status_code == 200
|
|
# write an output via the internal dispatch and read the mirror
|
|
app.datastore.write_words(0x0000, [0x1234])
|
|
assert app.datastore.read_words(0x0200, 1) == [0x1234]
|
|
|
|
|
|
class TestChannelNames:
|
|
async def test_names_in_layout_and_values(self, client, app):
|
|
r = await client.put("/api/config/channel-names", json={
|
|
"channel_names": {"DI0.0": "start_button", "AI0.0": "tank_level"},
|
|
"persist": False,
|
|
})
|
|
assert r.status_code == 200
|
|
# layout carries the name on the right channel
|
|
r = await client.get("/api/node")
|
|
di0 = next(c for m in r.json()["modules"] if m["index"] == 0
|
|
for c in m["channels"] if c["address"] == "DI0.0")
|
|
assert di0["channel_name"] == "start_button"
|
|
# values snapshot carries it too
|
|
r = await client.get("/api/values")
|
|
entry = next(v for v in r.json()["values"]
|
|
if (v["kind"], v["module"], v["channel"]) == ("ai", 0, 0))
|
|
assert entry["name"] == "tank_level"
|
|
|
|
async def test_names_persist_to_yaml(self, client, app, tmp_path):
|
|
cfg_file = tmp_path / "node.yaml"
|
|
app.node_config_path = str(cfg_file)
|
|
r = await client.put("/api/config/channel-names", json={
|
|
"channel_names": {"DO0.0": "pump_relay"}, "persist": True,
|
|
})
|
|
assert r.status_code == 200
|
|
text = cfg_file.read_text(encoding="utf-8")
|
|
assert "channel_names:" in text
|
|
assert "DO0.0: pump_relay" in text
|
|
# and it loads back
|
|
from wago_sim.scenario.dsl import load_yaml, load_channel_names
|
|
names = load_channel_names(load_yaml(cfg_file))
|
|
assert names == {"DO0.0": "pump_relay"}
|
|
|
|
async def test_invalid_code_rejected(self, client):
|
|
r = await client.put("/api/config/channel-names", json={
|
|
"channel_names": {"XX0.0": "nope"}, "persist": False,
|
|
})
|
|
assert r.status_code == 400
|
|
assert "invalid I/O code" in r.json()["detail"]
|
|
|
|
async def test_names_survive_module_save(self, client, app, tmp_path):
|
|
"""Module SAVE CONFIG must not wipe channel names (they persist)."""
|
|
app.node_config_path = str(tmp_path / "node.yaml")
|
|
await client.put("/api/config/channel-names", json={
|
|
"channel_names": {"DI0.0": "start_button"}, "persist": True,
|
|
})
|
|
r = await client.put("/api/config/modules", json={
|
|
"modules": ["750-430", "750-530"], "persist": True,
|
|
})
|
|
assert r.status_code == 200
|
|
assert app.node.channel_names.get("DI0.0") == "start_button"
|
|
# YAML contains both sections
|
|
text = (tmp_path / "node.yaml").read_text(encoding="utf-8")
|
|
assert "DI0.0: start_button" in text
|
|
|
|
async def test_readonly_blocks_rename(self, app):
|
|
ro = build_webui(app, user="admin", password="simu", readonly=True)
|
|
transport = httpx.ASGITransport(app=ro)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
|
await c.post("/login", json={"user": "admin", "password": "simu"})
|
|
r = await c.put("/api/config/channel-names", json={"channel_names": {}})
|
|
assert r.status_code == 403
|
|
|
|
|
|
class TestConfigExportImport:
|
|
async def test_export_shape(self, client):
|
|
r = await client.get("/api/config/export")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["format"] == "wago-sim-config"
|
|
assert data["version"] == 1
|
|
assert len(data["modules"]) == 6
|
|
assert "coupler_settings" in data
|
|
assert "channel_names" in data
|
|
|
|
async def test_export_import_roundtrip(self, client, app, tmp_path):
|
|
# configure something recognizable
|
|
await client.put("/api/config/channel-names", json={
|
|
"channel_names": {"DI0.0": "start_button"}, "persist": False,
|
|
})
|
|
await client.put("/api/config/coupler", json={"response_delay_ms": 33})
|
|
exported = (await client.get("/api/config/export")).json()
|
|
|
|
app.node_config_path = str(tmp_path / "node.yaml")
|
|
r = await client.post("/api/config/import", json=exported)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert "750-430" in body["modules"]
|
|
# names and coupler settings applied
|
|
assert app.node.channel_names.get("DI0.0") == "start_button"
|
|
assert app.registers.read_config(0x1037, 1) == [33]
|
|
# persisted to YAML
|
|
text = (tmp_path / "node.yaml").read_text(encoding="utf-8")
|
|
assert "DI0.0: start_button" in text
|
|
|
|
async def test_import_rejects_bad_payload(self, client, app):
|
|
# wrong format marker
|
|
r = await client.post("/api/config/import", json={"format": "nope"})
|
|
assert r.status_code == 400
|
|
# unknown article: nothing applied
|
|
before = [m.article for m in app.node.modules]
|
|
r = await client.post("/api/config/import", json={
|
|
"format": "wago-sim-config", "modules": ["750-9999"],
|
|
"channel_names": {}, "coupler_settings": {},
|
|
})
|
|
assert r.status_code == 400
|
|
assert [m.article for m in app.node.modules] == before # untouched
|
|
|
|
async def test_import_drops_stale_name_codes(self, client, app):
|
|
# names for channels that don't exist on the imported layout
|
|
r = await client.post("/api/config/import", json={
|
|
"format": "wago-sim-config",
|
|
"modules": ["750-530"],
|
|
"channel_names": {"DO0.0": "pump", "DI0.0": "stale_name"},
|
|
"coupler_settings": {},
|
|
"persist": False,
|
|
})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["dropped_codes"] == ["DI0.0"]
|
|
assert app.node.channel_names == {"DO0.0": "pump"}
|
|
|
|
async def test_readonly_blocks_import_export(self, app):
|
|
ro = build_webui(app, user="admin", password="simu", readonly=True)
|
|
transport = httpx.ASGITransport(app=ro)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
|
await c.post("/login", json={"user": "admin", "password": "simu"})
|
|
r = await c.get("/api/config/export")
|
|
assert r.status_code == 200 # export is a read
|
|
r = await c.post("/api/config/import", json={"format": "wago-sim-config",
|
|
"modules": ["750-430"]})
|
|
assert r.status_code == 403
|
|
|
|
|
|
class TestChannelRanges:
|
|
async def test_default_range_in_layout(self, client):
|
|
r = await client.get("/api/node")
|
|
ai = next(c for m in r.json()["modules"] if m["index"] == 4
|
|
for c in m["channels"] if c["kind"] == "ai")
|
|
assert ai["range"] == {"min": 0, "max": 4095}
|
|
# digital channels carry no range
|
|
di = next(c for m in r.json()["modules"] if m["index"] == 0
|
|
for c in m["channels"] if c["kind"] == "di")
|
|
assert di["range"] is None
|
|
|
|
async def test_put_ranges_and_layout(self, client, app):
|
|
r = await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": 0, "max": 32000},
|
|
"AO0.0": {"min": 400, "max": 2000}},
|
|
"persist": False,
|
|
})
|
|
assert r.status_code == 200
|
|
# layout reflects the configured range
|
|
r = await client.get("/api/node")
|
|
ai = next(c for m in r.json()["modules"] if m["index"] == 4
|
|
for c in m["channels"] if c["kind"] == "ai" and c["address"] == "AI0.0")
|
|
assert ai["range"] == {"min": 0, "max": 32000}
|
|
assert app.node.channel_ranges["AO0.0"] == (400, 2000)
|
|
|
|
async def test_default_range_kept_in_dense_store(self, client, app):
|
|
# the store is DENSE (FR-10.2.14): a default-equal entry is
|
|
# materialized like any other — no sparse-override dropping
|
|
r = await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": 0, "max": 4095}}, # == default
|
|
"persist": False,
|
|
})
|
|
assert r.status_code == 200
|
|
assert app.node.channel_ranges["AI0.0"] == (0, 4095)
|
|
# every analog channel of the layout is present
|
|
analog = {c for c in app.node.channel_ranges}
|
|
assert {"AI0.0", "AI0.1", "AO0.0", "AO0.1"} <= analog
|
|
# and digital codes never are
|
|
assert not any(c.startswith(("DI", "DO")) for c in analog)
|
|
|
|
async def test_dense_store_pruned_on_module_save(self, client, app, tmp_path):
|
|
app.node_config_path = str(tmp_path / "node.yaml")
|
|
await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": 0, "max": 32000}}, "persist": False,
|
|
})
|
|
r = await client.put("/api/config/modules", json={
|
|
"modules": ["750-430", "750-530"], "persist": False,
|
|
})
|
|
assert r.status_code == 200
|
|
# no analog modules left -> the range store is empty (pruned)
|
|
assert app.node.channel_ranges == {}
|
|
|
|
async def test_range_validation(self, client):
|
|
base = {"persist": False}
|
|
r = await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"XX4.0": {"min": 0, "max": 100}}, **base})
|
|
assert r.status_code == 400 # bad code
|
|
r = await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": 100, "max": 50}}, **base})
|
|
assert r.status_code == 400 # min >= max
|
|
r = await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": 0, "max": 70000}}, **base})
|
|
assert r.status_code == 400 # > 65535
|
|
r = await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": 0}}, **base})
|
|
assert r.status_code == 400 # missing max
|
|
|
|
async def test_negative_range_accepted(self, client, app):
|
|
r = await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": -32000, "max": 32000}},
|
|
"persist": False,
|
|
})
|
|
assert r.status_code == 200
|
|
assert app.node.channel_ranges["AI0.0"] == (-32000, 32000)
|
|
# layout exposes the signed range
|
|
r = await client.get("/api/node")
|
|
ai = next(c for m in r.json()["modules"] if m["index"] == 4
|
|
for c in m["channels"] if c["address"] == "AI0.0")
|
|
assert ai["range"] == {"min": -32000, "max": 32000}
|
|
|
|
async def test_range_floor_is_minus_65534(self, client):
|
|
r = await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": -65535, "max": 0}}, "persist": False})
|
|
assert r.status_code == 400 # below -65534
|
|
|
|
async def test_signed_value_display_and_write(self, client, app, datastore):
|
|
# channel with a negative-min range -> signed two's-complement view
|
|
await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": -32000, "max": 32000}},
|
|
"persist": False})
|
|
r = await client.post("/api/channels/ai/0/0", json={"value": -2000})
|
|
assert r.status_code == 200
|
|
# raw image holds the two's-complement word
|
|
assert datastore.get_input_word(0) == (-2000) & 0xFFFF
|
|
# values API shows the signed number again
|
|
r = await client.get("/api/values")
|
|
entry = next(v for v in r.json()["values"]
|
|
if (v["kind"], v["module"], v["channel"]) == ("ai", 0, 0))
|
|
assert entry["value"] == -2000
|
|
|
|
async def test_negative_ao_force(self, client, datastore):
|
|
r = await client.post("/api/channels/ao/0/0", json={"value": -3000})
|
|
assert r.status_code == 200
|
|
assert datastore.get_output_word(0) == (-3000) & 0xFFFF
|
|
# unsigned AO (default range) still displays raw
|
|
r = await client.get("/api/values")
|
|
entry = next(v for v in r.json()["values"]
|
|
if (v["kind"], v["module"], v["channel"]) == ("ao", 0, 0))
|
|
assert entry["value"] == (-3000) & 0xFFFF
|
|
|
|
async def test_value_bounds_rejected(self, client):
|
|
r = await client.post("/api/channels/ai/0/0", json={"value": -65535})
|
|
assert r.status_code == 400
|
|
r = await client.post("/api/channels/ai/0/0", json={"value": 65536})
|
|
assert r.status_code == 400
|
|
r = await client.post("/api/channels/ai/0/0", json={"value": "banana"})
|
|
assert r.status_code == 400
|
|
|
|
async def test_ranges_persist_to_yaml(self, client, app, tmp_path):
|
|
cfg_file = tmp_path / "node.yaml"
|
|
app.node_config_path = str(cfg_file)
|
|
r = await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": 0, "max": 32000}}, "persist": True,
|
|
})
|
|
assert r.status_code == 200
|
|
text = cfg_file.read_text(encoding="utf-8")
|
|
assert "channel_ranges:" in text
|
|
assert "min: 0, max: 32000" in text
|
|
# loads back
|
|
from wago_sim.scenario.dsl import load_yaml, load_channel_ranges
|
|
ranges = load_channel_ranges(load_yaml(cfg_file))
|
|
assert ranges == {"AI0.0": (0, 32000)}
|
|
|
|
async def test_ranges_yaml_lean_defaults(self, client, app, tmp_path):
|
|
# dense store, lean YAML: default-equal entries are NOT written
|
|
cfg_file = tmp_path / "node.yaml"
|
|
app.node_config_path = str(cfg_file)
|
|
r = await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": 0, "max": 4095},
|
|
"AO0.0": {"min": 0, "max": 32000}},
|
|
"persist": True,
|
|
})
|
|
assert r.status_code == 200
|
|
text = cfg_file.read_text(encoding="utf-8")
|
|
assert "AO0.0" in text and "AI0.0" not in text
|
|
assert app.node.channel_ranges["AI0.0"] == (0, 4095) # store dense
|
|
|
|
async def test_ranges_survive_module_save(self, client, app, tmp_path):
|
|
app.node_config_path = str(tmp_path / "node.yaml")
|
|
await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": 0, "max": 32000}}, "persist": True,
|
|
})
|
|
r = await client.put("/api/config/modules", json={
|
|
"modules": ["750-430", "750-467"], "persist": True,
|
|
})
|
|
assert r.status_code == 200
|
|
# AI_2 is at slot 1 now; the AI0.0 range became stale -> carried
|
|
# over but no longer matches a channel; the store keeps it (names
|
|
# semantics differ: stale name codes are dropped on import only)
|
|
text = (tmp_path / "node.yaml").read_text(encoding="utf-8")
|
|
assert "channel_ranges:" in text
|
|
|
|
async def test_export_import_carries_ranges(self, client, app, tmp_path):
|
|
await client.put("/api/config/channel-ranges", json={
|
|
"channel_ranges": {"AI0.0": {"min": 600, "max": 2000}}, "persist": False,
|
|
})
|
|
exported = (await client.get("/api/config/export")).json()
|
|
assert exported["channel_ranges"]["AI0.0"] == {"min": 600, "max": 2000}
|
|
app.node_config_path = str(tmp_path / "node.yaml")
|
|
r = await client.post("/api/config/import", json=exported)
|
|
assert r.status_code == 200
|
|
assert app.node.channel_ranges["AI0.0"] == (600, 2000)
|
|
text = (tmp_path / "node.yaml").read_text(encoding="utf-8")
|
|
assert "min: 600, max: 2000" in text
|
|
|
|
async def test_readonly_blocks_ranges(self, app):
|
|
ro = build_webui(app, user="admin", password="simu", readonly=True)
|
|
transport = httpx.ASGITransport(app=ro)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
|
await c.post("/login", json={"user": "admin", "password": "simu"})
|
|
r = await c.put("/api/config/channel-ranges", json={"channel_ranges": {}})
|
|
assert r.status_code == 403 |