210 lines
7.8 KiB
Python
210 lines
7.8 KiB
Python
"""Tests for the read/probe CLI verbs and the --server/--host overrides.
|
|
|
|
All fake-client driven — no COM needed (IMPLEMENTATION_PLAN §9). The read path
|
|
itself (OPCItem.Read) is verified live on a dev box with the Matrikon
|
|
simulation server.
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from opcda_diagnozer.main import main as cli_main
|
|
from opcda_diagnozer.opc.base import OPCDataSource, OpcError, ReadResult
|
|
from opcda_diagnozer.opc.fake import FakeOpcClient
|
|
|
|
CONFIG = "opc:\n server: Fake.Server.1\ntags:\n - name: Bucket Brigade.Int1\n - name: Bucket Brigade.Real8\n"
|
|
|
|
|
|
def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, args: list[str], client: FakeOpcClient | None = None) -> tuple[int, FakeOpcClient]:
|
|
"""Run the CLI with the fake client injected in place of the COM adapter."""
|
|
import opcda_diagnozer.main as main_mod
|
|
|
|
config = tmp_path / "config.yaml"
|
|
config.write_text(CONFIG, encoding="utf-8")
|
|
fake = client if client is not None else FakeOpcClient()
|
|
|
|
def _fake_connect(config_path: str, server=None, host=None):
|
|
assert isinstance(server, str | None)
|
|
fake.connect()
|
|
return fake
|
|
|
|
monkeypatch.setattr(main_mod, "_connect_from_config", _fake_connect)
|
|
return cli_main(args), fake
|
|
|
|
|
|
def test_read_explicit_tags(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
|
|
code, fake = _run_cli(monkeypatch, tmp_path, [
|
|
"read", "-c", str(tmp_path / "config.yaml"), "Bucket Brigade.Int1", "Bucket Brigade.Real8",
|
|
])
|
|
out = capsys.readouterr().out
|
|
assert code == 0
|
|
assert "Bucket Brigade.Int1" in out
|
|
assert "0xC0" in out # GOOD quality shown decoded
|
|
# request order preserved, source = DEVICE by default
|
|
assert fake.reads == [
|
|
(("Bucket Brigade.Int1", "Bucket Brigade.Real8"), OPCDataSource.DEVICE),
|
|
]
|
|
|
|
|
|
def test_read_from_config_tags(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
|
|
code, fake = _run_cli(monkeypatch, tmp_path, [
|
|
"read", "-c", str(tmp_path / "config.yaml"),
|
|
])
|
|
out = capsys.readouterr().out
|
|
assert code == 0
|
|
assert "Bucket Brigade.Int1" in out and "Bucket Brigade.Real8" in out
|
|
assert fake.reads[0][0] == ("Bucket Brigade.Int1", "Bucket Brigade.Real8")
|
|
|
|
|
|
def test_read_cache_source_flag(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
_, fake = _run_cli(monkeypatch, tmp_path, [
|
|
"read", "-c", str(tmp_path / "config.yaml"), "--source", "cache", "Bucket Brigade.Int1",
|
|
])
|
|
assert fake.reads[0][1] == OPCDataSource.CACHE
|
|
|
|
|
|
def test_read_json_output(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
|
|
code, _ = _run_cli(monkeypatch, tmp_path, [
|
|
"read", "-c", str(tmp_path / "config.yaml"), "--json", "Bucket Brigade.Int1",
|
|
])
|
|
out = capsys.readouterr().out
|
|
assert code == 0
|
|
payload = json.loads(out.strip().splitlines()[0])
|
|
assert payload["tag"] == "Bucket Brigade.Int1"
|
|
assert payload["quality"] == "GOOD"
|
|
assert payload["quality_class"] == "GOOD"
|
|
assert payload["device_time_utc"]
|
|
|
|
|
|
def test_read_bad_quality_tag_fails_with_nonzero_exit(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
|
|
) -> None:
|
|
bad = ReadResult(
|
|
tag="Read Error.Int1",
|
|
value=None,
|
|
quality_raw=0x1C, # BAD_OUT_OF_SERVICE (0x18 would be COMM_FAILURE)
|
|
device_time=datetime(2026, 10, 7, 9, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
code, _ = _run_cli(
|
|
monkeypatch, tmp_path,
|
|
["read", "-c", str(tmp_path / "config.yaml"), "Read Error.Int1"],
|
|
client=FakeOpcClient(read_results=[bad]),
|
|
)
|
|
captured = capsys.readouterr()
|
|
assert code == 1 # non-GOOD quality -> exit 1
|
|
assert "BAD_OUT_OF_SERVICE" in captured.out
|
|
assert "did not read GOOD" in captured.err
|
|
|
|
|
|
def test_read_error_tag_reports_error_not_crash(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
|
|
) -> None:
|
|
missing = ReadResult(
|
|
tag="No.Such.Tag",
|
|
value=None,
|
|
quality_raw=0,
|
|
device_time=None,
|
|
error="cannot add tag 'No.Such.Tag' (HRESULT 0xC0040008: ...)",
|
|
)
|
|
code, _ = _run_cli(
|
|
monkeypatch, tmp_path,
|
|
["read", "-c", str(tmp_path / "config.yaml"), "No.Such.Tag"],
|
|
client=FakeOpcClient(read_results=[missing]),
|
|
)
|
|
captured = capsys.readouterr()
|
|
assert code == 1
|
|
assert "ERROR" in captured.out
|
|
assert "0xC0040008" in captured.out
|
|
|
|
|
|
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"),
|
|
])
|
|
out = capsys.readouterr().out
|
|
assert code == 0
|
|
assert fake.reads[0][0] == ("Bucket Brigade.Int1", "Bucket Brigade.Real8")
|
|
assert "Bucket Brigade.Int1" in out
|
|
|
|
|
|
def test_probe_missing_config_is_actionable(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
|
|
) -> None:
|
|
# no monkeypatching: the real _connect_from_config must produce the hint
|
|
code = cli_main(["probe", "-c", str(tmp_path / "no-such-config.yaml")])
|
|
captured = capsys.readouterr()
|
|
assert code == 2
|
|
assert "FAILED" in captured.err
|
|
assert "--server" in captured.err # the hint points at the override
|
|
|
|
|
|
def test_server_override_flows_to_client(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
|
|
) -> None:
|
|
"""--server/--host must reach the client even when a config exists."""
|
|
import opcda_diagnozer.main as main_mod
|
|
|
|
config = tmp_path / "config.yaml"
|
|
config.write_text(CONFIG, encoding="utf-8")
|
|
seen: dict[str, object] = {}
|
|
|
|
def _fake_connect(config_path: str, server=None, host=None):
|
|
seen["server"], seen["host"] = server, host
|
|
client = FakeOpcClient()
|
|
client.connect()
|
|
return client
|
|
|
|
monkeypatch.setattr(main_mod, "_connect_from_config", _fake_connect)
|
|
code = cli_main([
|
|
"read", "-c", str(config),
|
|
"--server", "Other.Server.1", "--host", "OTHER-HOST",
|
|
"Some.Tag",
|
|
])
|
|
assert code == 0
|
|
assert seen == {"server": "Other.Server.1", "host": "OTHER-HOST"}
|
|
|
|
|
|
def test_read_without_config_and_without_server_is_actionable(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
|
|
) -> None:
|
|
# no monkeypatching: the real connect path must fail with the hint
|
|
code = cli_main([
|
|
"read", "-c", str(tmp_path / "no-such.yaml"), "Some.Tag",
|
|
])
|
|
captured = capsys.readouterr()
|
|
assert code == 2
|
|
assert "--server" in captured.err
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform != "win32", reason="live read needs Windows + pywin32")
|
|
def test_live_automation_read_smoke() -> None:
|
|
"""Live smoke test — requires the Matrikon simulation server on this dev box."""
|
|
pytest.importorskip("pythoncom")
|
|
from opcda_diagnozer.opc.automation import AutomationOpcClient
|
|
|
|
client = AutomationOpcClient(server="Matrikon.OPC.Simulation.1")
|
|
try:
|
|
client.connect()
|
|
results = client.read(
|
|
["Bucket Brigade.Int1", "Read Error.Int1", "No.Such.Tag"],
|
|
source=OPCDataSource.DEVICE,
|
|
)
|
|
by_tag = {r.tag: r for r in results}
|
|
good = by_tag["Bucket Brigade.Int1"]
|
|
assert good.quality_raw == 0xC0
|
|
assert good.device_time is not None
|
|
# unknown tag -> per-tag error, never an exception
|
|
assert by_tag["No.Such.Tag"].error is not None
|
|
# "Read Error.Int1" from DEVICE returns the error code as the value
|
|
# with GOOD quality (verified live 2026-10-07); the BAD quality only
|
|
# appears on the CACHE path — the diagnozer's quality monitoring
|
|
# therefore reads CACHE for that tag in the poll loop (Phase 1).
|
|
assert by_tag["Read Error.Int1"].value is not None
|
|
except OpcError as exc:
|
|
pytest.skip(f"simulation server not usable on this machine: {exc}")
|
|
finally:
|
|
client.close() |