606 lines
28 KiB
Python
606 lines
28 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 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
|
|
|
|
|
|
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_dropped_from_store(self, client, app):
|
|
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 == {} # kept clean
|
|
|
|
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_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 |