jobe/opcda-diagnozer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🚧 WIP

Browse Source
This commit is contained in:
Jonas Bengtsson
2026-10-08 06:30:54 +02:00
parent 76b1cc5ccf
commit a1af8d68c9
12 changed files with 966 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_config.py
+96
View File
@@ -23,6 +23,13 @@ def test_example_config_loads() -> None:
assert config.tags[0].name == "Channel1.Device1.Counter"
assert config.tags[0].expect_change is True
assert config.tags[1].expect_change is False
# D16 device-link indicators (spec 6.1); normal reading configurable per device
assert [d.name for d in config.devices] == ["Conveyor 3", "Conveyor 4", "Pump A", "Boiler 2"]
assert config.devices[0].tag == "Channel1.Device3.Connected"
assert config.devices[0].ok_values == ("true",) # default: boolean connected-flag
assert config.devices[1].ok_values == ("true", "1")
assert config.devices[2].ok_values == ("false", "0") # inverted fault flag: 0 = ok
assert config.devices[3].ok_values == ("RUN",) # string state
assert config.poll.interval_s == 1.0
assert config.poll.timeout_ms == 5000
assert config.latency.warn_ms == 1000
@@ -65,6 +72,95 @@ def test_duplicate_tags_rejected(tmp_path: Path) -> None:
load_config(p)
def test_devices_missing_keys_rejected(tmp_path: Path) -> None:
p = _write(tmp_path, "opc:\n server: X\ntags:\n - name: t\ndevices:\n - name: Device\n")
with pytest.raises(ConfigError, match="'tag' is required"):
load_config(p)
def test_devices_blank_name_rejected(tmp_path: Path) -> None:
p = _write(
tmp_path,
"opc:\n server: X\ntags:\n - name: t\ndevices:\n - { name: '', tag: Ind.Tag }\n",
)
with pytest.raises(ConfigError, match="non-empty"):
load_config(p)
def test_devices_duplicate_indicator_rejected(tmp_path: Path) -> None:
p = _write(
tmp_path,
"opc:\n server: X\ntags:\n - name: t\n"
"devices:\n - { name: A, tag: Ind.Tag }\n - { name: B, tag: Ind.Tag }\n",
)
with pytest.raises(ConfigError, match="one device per indicator"):
load_config(p)
def test_devices_indicator_in_tags_rejected(tmp_path: Path) -> None:
"""An indicator is polled automatically; listing it in tags: double-reports."""
p = _write(
tmp_path,
"opc:\n server: X\n"
"tags:\n - name: Ind.Tag\ndevices:\n - { name: A, tag: Ind.Tag }\n",
)
with pytest.raises(ConfigError, match="remove them from"):
load_config(p)
def test_devices_empty_is_allowed(tmp_path: Path) -> None:
"""devices: is optional (no link monitoring) and may be an empty list."""
p = _write(tmp_path, "opc:\n server: X\ntags:\n - name: t\ndevices: []\n")
assert load_config(p).devices == ()
def test_devices_ok_values_default_is_true(tmp_path: Path) -> None:
p = _write(
tmp_path,
"opc:\n server: X\ntags:\n - name: t\ndevices:\n - { name: A, tag: I.Tag }\n",
)
device = load_config(p).devices[0]
assert device.ok_values == ("true",)
def test_devices_ok_values_normalized(tmp_path: Path) -> None:
"""Booleans/ints/strings all render to comparison strings; scalar = shorthand."""
p = _write(
tmp_path,
"opc:\n server: X\ntags:\n - name: t\n"
"devices:\n"
" - { name: A, tag: I.One, ok_values: [true, 1] }\n"
" - { name: B, tag: I.Two, ok_values: [false, 0] }\n"
" - { name: C, tag: I.Three, ok_values: [\"RUN\"] }\n"
" - { name: D, tag: I.Four, ok_values: 1 }\n",
)
devices = {d.name: d for d in load_config(p).devices}
assert devices["A"].ok_values == ("true", "1")
assert devices["B"].ok_values == ("false", "0")
assert devices["C"].ok_values == ("RUN",)
assert devices["D"].ok_values == ("1",) # single scalar shorthand
def test_devices_ok_values_empty_rejected(tmp_path: Path) -> None:
p = _write(
tmp_path,
"opc:\n server: X\ntags:\n - name: t\n"
"devices:\n - { name: A, tag: I.Tag, ok_values: [] }\n",
)
with pytest.raises(ConfigError, match="must not be empty"):
load_config(p)
def test_devices_ok_values_non_scalar_rejected(tmp_path: Path) -> None:
p = _write(
tmp_path,
"opc:\n server: X\ntags:\n - name: t\n"
"devices:\n - { name: A, tag: I.Tag, ok_values: [[1]] }\n",
)
with pytest.raises(ConfigError, match="scalars"):
load_config(p)
def test_latency_threshold_order_rejected(tmp_path: Path) -> None:
p = _write(
tmp_path,
tests/test_read_cli.py
+180
View File
@@ -121,6 +121,186 @@ def test_read_error_tag_reports_error_not_crash(
assert "0xC0040008" in captured.out
# --------------------------------------------------------------------------- #
# per-tag error rendering (_per_tag_error) — no COM needed
# --------------------------------------------------------------------------- #
def _fake_com_error(hresult: int) -> Exception:
"""A stand-in with the pywin32 com_error shape for the hresult check."""
err = Exception("server rejected the call")
err.hresult = hresult # type: ignore[attr-defined]
return err
def test_per_tag_error_keeps_hresult_for_com_errors() -> None:
from opcda_diagnozer.opc.automation import _per_tag_error
rendered = _per_tag_error("cannot add tag 'X'", _fake_com_error(0x80020009))
assert "cannot add tag 'X'" in rendered
assert "HRESULT 0x80020009" in rendered # decoded COM suffix preserved
def test_per_tag_error_names_plain_exceptions() -> None:
"""A plain ValueError must show its type+text, not a bare 'failed'."""
from opcda_diagnozer.opc.automation import _per_tag_error
rendered = _per_tag_error("read of tag 'X' failed", ValueError("tz math broke"))
assert "read of tag 'X' failed" in rendered
assert "ValueError" in rendered
assert "tz math broke" in rendered
assert "HRESULT" not in rendered # never fake a COM suffix for non-COM errors
def test_timestamp_failure_never_fails_a_read() -> None:
"""A degenerate device timestamp is metadata: value/quality stay readable.
Reproduces the alias-tag failure shape end-to-end through the adapter's
read() with a stubbed COM item: a TimeStamp whose astimezone() explodes
(zero FILETIME 1601 / DST math on Matrikon aliases) must yield a GOOD
result with device_time=None — never a per-tag error.
"""
from datetime import timezone as _tz
from opcda_diagnozer.opc import automation
from opcda_diagnozer.opc.base import OPCDataSource
class _PoisonedItem:
"""A COM item stand-in: reads succeed, TimeStamp cannot be normalized."""
ItemID = "P5002.Connected"
def __init__(self) -> None:
self.read_calls: list[int] = []
def Read(self, source: int) -> None:
self.read_calls.append(source)
@property
def Value(self):
return 1
@property
def Quality(self):
return 0xC0
@property
def TimeStamp(self):
class _Bad:
def astimezone(self, tz):
raise ValueError("astimezone() cannot be applied to a naive datetime")
return _Bad()
class _Items:
def AddItem(self, item_id: str, handle: int):
assert item_id == "P5002.Connected"
return _PoisonedItem()
class _Group:
OPCItems = _Items()
class _Parent:
OPCGroups = None # set below
Parent = _Parent()
class _Groups:
def Add(self, name: str):
return _Group()
class _Opc: # minimal stand-in for the wrapper's dispatch object
OPCGroups = _Groups()
class _ClientStub(automation.AutomationOpcClient):
def __init__(self) -> None:
super().__init__(server="stub")
self._opc = _Opc() # read() requires a connected-looking client
def _release_and_uninit(self) -> None: # no COM to release on the stub
self._opc = None
client = _ClientStub()
results = client.read(["P5002.Connected"], source=OPCDataSource.DEVICE)
assert len(results) == 1
result = results[0]
assert result.error is None # the poison never surfaced
assert result.value == 1
assert result.quality_raw == 0xC0
assert result.device_time is None # metadata dropped, not the read
def test_timestamp_property_explosion_never_fails_a_read() -> None:
"""The frozen-exe failure shape: the TimeStamp PROPERTY itself explodes.
Missing win32timezone made pywin32 raise SystemError('InvokeTypes ...
exception set') when *accessing* item.TimeStamp (not in astimezone) — the
guard must therefore wrap the property fetch too, and the read must still
return the value/quality.
"""
from opcda_diagnozer.opc import automation
from opcda_diagnozer.opc.base import OPCDataSource
class _ExplodingTsItem:
ItemID = "Bucket Brigade.Int1"
def Read(self, source: int) -> None:
return None
@property
def Value(self):
return 42
@property
def Quality(self):
return 0xC0
@property
def TimeStamp(self):
# the exact frozen-exe signature: a non-COM SystemError raised by
# pywin32's conversion inside the property access
raise SystemError(
"<built-in method InvokeTypes of PyIDispatch object> "
"returned a result with an exception set"
)
class _Items:
def AddItem(self, item_id: str, handle: int):
return _ExplodingTsItem()
class _Group:
OPCItems = _Items()
class _Parent:
OPCGroups = None
Parent = _Parent()
class _Groups:
def Add(self, name: str):
return _Group()
class _Opc:
OPCGroups = _Groups()
class _ClientStub(automation.AutomationOpcClient):
def __init__(self) -> None:
super().__init__(server="stub")
self._opc = _Opc()
def _release_and_uninit(self) -> None:
self._opc = None
client = _ClientStub()
results = client.read(["Bucket Brigade.Int1"], source=OPCDataSource.DEVICE)
assert len(results) == 1
result = results[0]
assert result.error is None # the timestamp explosion never surfaced
assert result.value == 42
assert result.quality_raw == 0xC0
assert result.device_time is None
def test_probe_reads_config_tags(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
code, fake = _run_cli(monkeypatch, tmp_path, [
"probe", "-c", str(tmp_path / "config.yaml"),