From 2aa040e7572726584177245bfc0da652116af0d4 Mon Sep 17 00:00:00 2001 From: Jonas Bengtsson Date: Wed, 7 Oct 2026 17:13:51 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=9A=A7=20WIP?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 5 +- .vscode/launch.json | 37 ++++ README.md | 58 +++++++ src/opcda_diagnozer/main.py | 235 ++++++++++++++++++++++++-- src/opcda_diagnozer/opc/automation.py | 108 +++++++++++- src/opcda_diagnozer/opc/base.py | 17 +- src/opcda_diagnozer/opc/fake.py | 28 ++- tests/test_browse.py | 2 +- tests/test_build_exe.py | 76 +++++++++ tests/test_config.py | 2 +- tests/test_read_cli.py | 210 +++++++++++++++++++++++ 11 files changed, 749 insertions(+), 29 deletions(-) create mode 100644 tests/test_build_exe.py create mode 100644 tests/test_read_cli.py diff --git a/.gitignore b/.gitignore index 7a8750a..a8196c2 100644 --- a/.gitignore +++ b/.gitignore @@ -13,4 +13,7 @@ dist/ # Local runtime artifacts (copy config.example.yaml -> config.yaml) config.yaml -logs/ \ No newline at end of file +logs/ + +# Generated by scripts/build_exe.ps1 at build time +scripts/.exe_version_info.txt \ No newline at end of file diff --git a/.vscode/launch.json b/.vscode/launch.json index 28b73b2..d9c3dca 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -25,6 +25,21 @@ "PYTHONUNBUFFERED": "1" } }, + { + // One-shot read of every configured tag — the pre-deployment check. + "name": "opcda-diagnozer: probe", + "type": "debugpy", + "request": "launch", + "python": "${workspaceFolder}/.venv/Scripts/python.exe", + "module": "opcda_diagnozer", + "args": ["probe", "--config", "config.example.yaml"], + "cwd": "${workspaceFolder}", + "console": "integratedTerminal", + "justMyCode": false, + "env": { + "PYTHONUNBUFFERED": "1" + } + }, { // Basic use: validate the example config without connecting. "name": "opcda-diagnozer: check-config", @@ -127,6 +142,28 @@ "env": { "PYTHONUNBUFFERED": "1" } + }, + { + // One-shot read of tags with a --server override — no config + // file needed. Adjust the tags (and optionally the ProgID) to + // whichever server is registered on your machine. + "name": "opcda-diagnozer: read (--server override)", + "type": "debugpy", + "request": "launch", + "python": "${workspaceFolder}/.venv/Scripts/python.exe", + "module": "opcda_diagnozer", + "args": [ + "read", + "--server", "Matrikon.OPC.Simulation.1", + "Bucket Brigade.Int1", + "Bucket Brigade.Real8" + ], + "cwd": "${workspaceFolder}", + "console": "integratedTerminal", + "justMyCode": false, + "env": { + "PYTHONUNBUFFERED": "1" + } } ], "inputs": [ diff --git a/README.md b/README.md index 1022e9c..2077feb 100644 --- a/README.md +++ b/README.md @@ -62,6 +62,12 @@ opcda-diagnozer list-tags "Simulation Items" "Bucket Brigade" # equivalent opcda-diagnozer list-tags --recursive "Simulation Items" # whole subtree opcda-diagnozer list-tags # every tag on the server +# read tags right now (validation tool) — no config needed with --server +opcda-diagnozer read --server Matrikon.OPC.Simulation.1 "Bucket Brigade.Int1" +opcda-diagnozer read --json "Bucket Brigade.Int1" "Bucket Brigade.Real8" +opcda-diagnozer read # the config's tags +opcda-diagnozer read --source cache "Bucket Brigade.Int1" # server cache, no device trip + # validate config opcda-diagnozer check-config @@ -89,6 +95,17 @@ device path is either one dotted string (`"Simulation Items.Bucket Brigade"`) or several quoted branch names — use the latter when a branch name itself contains a dot. On flat-namespace servers (`organization: flat`) every tag hangs directly off the root; `list-tags` without a device lists them all. + +Every connecting verb (`read`, `probe`, `list-devices`, `list-tags`) accepts +`--server PROGID` and `--host HOST` to **override the config's `opc` section** — +point a one-shot check at any server without touching the deployed config.yaml. +With `--server` given, no config file is needed at all; without it, the config must +exist (and `probe`/`read` read the config's tag list unless tags are passed). + +`read` prints one row per tag with the decoded quality and the device timestamp; +a non-GOOD or failing tag makes the exit code non-zero, so it doubles as a +scriptable health check. `--source cache` reads the server cache instead of +forcing a device round-trip. ### Configuration example (`config.example.yaml`) ```yaml @@ -188,6 +205,47 @@ Notes: (`py install 3.14 -a x86`) or the installer from python.org ("Download Windows installer (32-bit)"). +## Building the Windows executable + +`scripts/build_exe.ps1` packages the CLI as a standalone **32-bit** console exe +with [PyInstaller](https://pyinstaller.org) — no Python installation needed on +the target host: + +```powershell +& ".\.venv\Scripts\python.exe" -m pip install pyinstaller # once +.\scripts\build_exe.ps1 # dist\opcda-diagnozer\ +.\scripts\build_exe.ps1 -Onefile # single exe alternative +.\scripts\build_exe.ps1 -Clean # fresh rebuild +``` + +The script refuses to run from a 64-bit interpreter (the exe must load the +32-bit OPC DA wrapper DLLs, SPECIFICATION §11), re-verifies the produced PE +really is 32-bit, embeds the project version in the exe properties, declares +the hidden imports the code loads lazily (pywin32 COM, browse modules, the +Phase 3 service host), and smoke-tests the artifact (`--version`, +`check-config`, `list-servers --source registry`) after building. It also +installs PyInstaller into the venv automatically if it is missing. + +If your policy blocks the script, run it with +`powershell -NoProfile -ExecutionPolicy Bypass -File scripts\build_exe.ps1`. + +Deployment notes: +- The **default config path** of a frozen exe is `config.yaml` **next to the + exe** — not the working directory. A Windows service starts with + `CWD = C:\Windows\System32`, so the relative default would silently point + there; the exe-dir default keeps the deployment layout simple + (`C:\Program Files\OPCDA-Diagnozer\{opcda-diagnozer.exe, config.yaml}`). +- The exe is a console application that hosts every CLI verb, including the + future `run` / `probe` / service verbs. The Phase 3 Windows service + (`install` verb) will point `sc.exe` at this same exe with the service-host + entry point. +- The Automation Wrapper DLL (`OPCAutomation.dll`) is **not** bundled — it is + a machine-installed COM component (registered in the registry), exactly like + the OPC DA servers the tool connects to. Only Python code and its + dependencies are frozen. +- `onedir` is the default (faster startup, AV-friendlier); `--onefile` trades + startup time for a single file. `--clean` removes `build/` + `dist/` first. + ## Documentation - [Specification](docs/SPECIFICATION.md) — detection rules, event model, operational behavior, acceptance criteria diff --git a/src/opcda_diagnozer/main.py b/src/opcda_diagnozer/main.py index d0f6c4d..e047758 100644 --- a/src/opcda_diagnozer/main.py +++ b/src/opcda_diagnozer/main.py @@ -12,6 +12,7 @@ from __future__ import annotations import argparse import sys from collections.abc import Sequence +from pathlib import Path from opcda_diagnozer import __version__ from opcda_diagnozer.config import ConfigError, load_config @@ -20,9 +21,21 @@ from opcda_diagnozer.opc.discovery import ServerInfo, list_servers DEFAULT_CONFIG = "config.yaml" + +def _default_config_path() -> str: + """Where the config lives by default: CWD in source runs, exe dir when frozen. + + A frozen exe must not rely on the CWD: a Windows service starts with + ``CWD = C:\\Windows\\System32``, so the relative name would silently resolve + into System32. Frozen runs therefore default to ``config.yaml`` **next to + the exe** — the deployment layout the Phase 3 service will use (SPEC §6.4). + """ + if getattr(sys, "frozen", False): + return str(Path(sys.executable).resolve().parent / DEFAULT_CONFIG) + return DEFAULT_CONFIG + _NOT_IMPLEMENTED: dict[str, str] = { "run": "Phase 1 (core engine)", - "probe": "Phase 1 (core engine)", "list-items": "superseded by list-devices / list-tags", "install": "Phase 3 (Windows service)", "uninstall": "Phase 3 (Windows service)", @@ -34,11 +47,30 @@ _NOT_IMPLEMENTED: dict[str, str] = { def _add_config_option(parser: argparse.ArgumentParser) -> None: parser.add_argument( "--config", "-c", - default=DEFAULT_CONFIG, + default=_default_config_path(), help="path to the YAML config (default: %(default)s)", ) +def _add_server_overrides(parser: argparse.ArgumentParser) -> None: + """--server/--host override the config's opc section (SPECIFICATION §6). + + They make one-shot CLI use possible without a config file — and let an + ops engineer point a validation verb at a different server without + touching the deployed config.yaml. + """ + parser.add_argument( + "--server", + metavar="PROGID", + help="OPC DA server ProgID, overriding the config (e.g. Matrikon.OPC.Simulation.1)", + ) + parser.add_argument( + "--host", + metavar="HOST", + help="remote DCOM host, overriding the config (omit for a local server)", + ) + + def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="opcda-diagnozer", @@ -53,8 +85,44 @@ def _build_parser() -> argparse.ArgumentParser: p = sub.add_parser("run", help="run in the foreground (console)") _add_config_option(p) - p = sub.add_parser("probe", help="connect once and print one read of every configured tag") + p = sub.add_parser( + "read", + help="connect once and read the given tags (a quick validation tool)", + ) _add_config_option(p) + _add_server_overrides(p) + p.add_argument( + "tags", + nargs="*", + metavar="TAG", + help="tag ItemIDs to read (e.g. 'Bucket Brigade.Int1'); omitted -> the " + "config's tags", + ) + p.add_argument( + "--source", + choices=("device", "cache"), + default="device", + help="read from the device (round-trip to the PLC) or the server cache " + "(default: %(default)s)", + ) + p.add_argument( + "--json", + action="store_true", + help="emit one JSON object per tag (for scripts) instead of a table", + ) + + p = sub.add_parser( + "probe", + help="connect once and print one read of every configured tag", + ) + _add_config_option(p) + _add_server_overrides(p) + p.add_argument( + "--source", + choices=("device", "cache"), + default="device", + help="read from the device or the server cache (default: %(default)s)", + ) p = sub.add_parser("check-config", help="validate the configuration and exit") _add_config_option(p) @@ -90,6 +158,7 @@ def _build_parser() -> argparse.ArgumentParser: help="list the devices (address-space branches) exposed by the configured server", ) _add_config_option(p) + _add_server_overrides(p) p.add_argument( "--depth", type=int, @@ -113,6 +182,7 @@ def _build_parser() -> argparse.ArgumentParser: help="list the tags (leaves) of one device, or every tag on the server", ) _add_config_option(p) + _add_server_overrides(p) p.add_argument( "device", nargs="*", @@ -211,19 +281,36 @@ def _cmd_list_servers(host: str | None, source: str, as_json: bool) -> int: return 0 -def _connect_from_config(config_path: str): - """Load the config and connect to its configured server (Automation Wrapper). +def _connect_from_config(config_path: str, server: str | None = None, host: str | None = None): + """Load the config (when needed) and connect; --server/--host override opc. + + Overrides make one-shot CLI use possible without a config file: when + ``--server`` is given, a missing config file is not an error and the + remaining settings fall back to their defaults. Without ``--server`` the + config must exist (its ``opc`` section decides where to connect). Returns the connected client; raises :class:`ConfigError` / :class:`OpcError` with actionable messages for the caller to print. """ from opcda_diagnozer.opc.automation import AutomationOpcClient - config = load_config(config_path) + connect_timeout_s = 10.0 # OpcConfig default + config_exists = Path(config_path).is_file() + if config_exists: + config = load_config(config_path) + server = server or config.opc.server + if host is None: + host = config.opc.host + connect_timeout_s = config.opc.connect_timeout_s + elif server is None: + raise ConfigError( + f"config file not found: {config_path} - pass --server (and optionally " + "--host) to connect without a config, or create the config first" + ) client = AutomationOpcClient( - server=config.opc.server, - host=config.opc.host, - connect_timeout_s=config.opc.connect_timeout_s, + server=server, + host=host, + connect_timeout_s=connect_timeout_s, ) client.connect() return client @@ -257,11 +344,14 @@ def _print_devices(space, as_json: bool) -> None: print(f" {device.qualified_name:<{width}} ({', '.join(notes)})") -def _cmd_list_devices(config_path: str, depth: int, include_counts: bool, as_json: bool) -> int: +def _cmd_list_devices( + config_path: str, server: str | None, host: str | None, + depth: int, include_counts: bool, as_json: bool, +) -> int: from opcda_diagnozer.opc.browse import list_devices try: - client = _connect_from_config(config_path) + client = _connect_from_config(config_path, server, host) browser = None try: browser = client.create_browser() @@ -317,6 +407,8 @@ def _print_tags(tags, device_path: tuple[str, ...], recursive: bool, as_json: bo def _cmd_list_tags( config_path: str, + server: str | None, + host: str | None, device: Sequence[str], recursive: bool, as_json: bool, @@ -327,7 +419,7 @@ def _cmd_list_tags( # explicit branch names (names may contain dots then). device_arg = device[0] if len(device) == 1 else list(device) try: - client = _connect_from_config(config_path) + client = _connect_from_config(config_path, server, host) browser = None try: browser = client.create_browser() @@ -353,6 +445,111 @@ def _cmd_list_tags( return 0 +def _print_read_results(results, as_json: bool) -> int: + """Human table / JSON-lines output for the read/probe verbs.""" + from opcda_diagnozer.quality import decode_quality + + failures = 0 + if as_json: + import json + + for result in results: + quality = decode_quality(result.quality_raw) + payload = { + "tag": result.tag, + "value": result.value, + "quality_raw": result.quality_raw, + "quality": quality.name, + "quality_class": quality.quality_class.value, + "device_time_utc": ( + result.device_time.isoformat() if result.device_time else None + ), + "error": result.error, + } + if result.error or quality.quality_class.value != "GOOD": + failures += 1 + print(json.dumps(payload, ensure_ascii=False, default=str)) + return failures + if not results: + print("no tags to read") + return 0 + width = max(len(r.tag) for r in results) + print(f"{'tag':<{width}} value quality device time (UTC)") + for result in results: + quality = decode_quality(result.quality_raw) + if result.error: + failures += 1 + print(f" {result.tag:<{width}} ERROR: {result.error}") + continue + marker = "" if quality.quality_class.value == "GOOD" else " <-- " + quality.name + if quality.quality_class.value != "GOOD": + failures += 1 + time_str = ( + result.device_time.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] + if result.device_time + else "-" + ) + print( + f" {result.tag:<{width}} {str(result.value):<30} {quality.raw_hex} {time_str}{marker}" + ) + return failures + + +def _cmd_read( + config_path: str, + server: str | None, + host: str | None, + tags: Sequence[str], + source: str, + as_json: bool, +) -> int: + """One-shot read of explicit tags (or the config's tags) from the server.""" + from opcda_diagnozer.opc.base import OPCDataSource + + requested = list(tags) + try: + # connect first: a missing config with no --server must fail here with + # the actionable hint (before the tags load below can complain). + client = _connect_from_config(config_path, server, host) + if not requested: + # no tags given -> use the config's tags (config is mandatory then) + config = load_config(config_path) + requested = [t.name for t in config.tags] + except ConfigError as exc: + print(f"read: FAILED\n {exc}", file=sys.stderr) + return 2 + except OpcError as exc: + print(f"read: FAILED\n {exc}", file=sys.stderr) + return 2 + data_source = OPCDataSource.DEVICE if source == "device" else OPCDataSource.CACHE + try: + results = client.read(requested, source=data_source) + except OpcError as exc: + print(f"read: FAILED\n {exc}", file=sys.stderr) + client.close() + return 2 + finally: + client.close() + failures = _print_read_results(results, as_json) + if failures: + print( + f"read: {failures} of {len(results)} tag(s) did not read GOOD", + file=sys.stderr, + ) + return 1 + return 0 + + +def _cmd_probe( + config_path: str, + server: str | None, + host: str | None, + source: str, +) -> int: + """Connect once, read every configured tag once, print — the pre-deploy check.""" + return _cmd_read(config_path, server, host, (), source, as_json=False) + + def main(argv: Sequence[str] | None = None) -> int: """CLI entry point (console script + ``python -m opcda_diagnozer``).""" args = _build_parser().parse_args(argv) @@ -361,9 +558,19 @@ def main(argv: Sequence[str] | None = None) -> int: if args.verb == "list-servers": return _cmd_list_servers(args.host, args.source, args.json) if args.verb == "list-devices": - return _cmd_list_devices(args.config, args.depth, not args.no_counts, args.json) + return _cmd_list_devices( + args.config, args.server, args.host, args.depth, not args.no_counts, args.json + ) if args.verb == "list-tags": - return _cmd_list_tags(args.config, args.device, args.recursive, args.json) + return _cmd_list_tags( + args.config, args.server, args.host, args.device, args.recursive, args.json + ) + if args.verb == "read": + return _cmd_read( + args.config, args.server, args.host, args.tags, args.source, args.json + ) + if args.verb == "probe": + return _cmd_probe(args.config, args.server, args.host, args.source) phase = _NOT_IMPLEMENTED[args.verb] print( f"{args.verb}: not implemented yet — planned for {phase} " diff --git a/src/opcda_diagnozer/opc/automation.py b/src/opcda_diagnozer/opc/automation.py index 29578ed..5f664ad 100644 --- a/src/opcda_diagnozer/opc/automation.py +++ b/src/opcda_diagnozer/opc/automation.py @@ -17,23 +17,31 @@ Matrikon.OPC.Simulation.1): ``Connect(progId)`` / ``Connect(progId, node)``, ``CurrentPosition``; ``Organization`` 1 = hierarchical, 2 = flat. Bad branch names raise ``E_INVALIDARG``; ``MoveTo`` requires a *sequence* (a bare string is a pywin32 SAFEARRAY ``TypeError``). + +**Reads** (verified live): the wrapper's ``SyncRead`` cannot be marshalled by +pywin32 (SAFEARRAY ByRef out-params fail in every invocation shape), but +``OPCItem.Read(Source)`` + the ``Value`` / ``Quality`` / ``TimeStamp`` +properties work — per-item reads with only scalar out-params. Data-source +constants come from the wrapper typelib: ``OPCCache = 1``, ``OPCDevice = 2``. +``TimeStamp`` is a tz-aware ``pywintypes.datetime`` (converted with +``astimezone(timezone.utc)`` for the UTC event model, SPECIFICATION §6). """ from __future__ import annotations from collections.abc import Sequence +from datetime import datetime, timezone from typing import Any from opcda_diagnozer.opc.base import ( Browser, + OPCDataSource, OpcError, ReadResult, TagInfo, as_com_error, ) -_NOT_IMPL = "Phase 1: Automation Wrapper adapter (docs/IMPLEMENTATION_PLAN.md §4.4)" - #: OPC_BROWSE organization constants (Automation Wrapper): 1 = hierarchical, #: 2 = flat (tags addressed by bare ItemID, no branches to walk). _ORGANIZATION_LABELS = {1: "hierarchical", 2: "flat"} @@ -120,9 +128,10 @@ class _AutomationBrowser: class AutomationOpcClient: """``win32com``-based :class:`OpcClient`. - ``connect`` / ``create_browser`` / ``close`` are live (they power the - ``list-devices`` / ``list-tags`` validation verbs); ``read`` lands with the - Phase 1 core engine (IMPLEMENTATION_PLAN §4.4). + ``connect`` / ``create_browser`` / ``read`` / ``close`` are live: they + power the ``list-devices`` / ``list-tags`` validation verbs and the + one-shot ``read`` / ``probe`` CLI verbs. The Phase 1 poller builds on this + same adapter (IMPLEMENTATION_PLAN §4.4). """ def __init__(self, server: str, host: str | None = None, connect_timeout_s: float = 10.0) -> None: @@ -214,9 +223,92 @@ class AutomationOpcClient: self._com_initialized = False pythoncom.CoUninitialize() - def read(self, tags: Sequence[str]) -> list[ReadResult]: - self._require_pywin32() - raise NotImplementedError(_NOT_IMPL) + def read( + self, tags: Sequence[str], source: OPCDataSource = OPCDataSource.DEVICE + ) -> list[ReadResult]: + """Read the given tags from the connected server. + + Uses one throwaway group with one item per tag and per-item + ``Item.Read(Source)`` (see the module docstring: the wrapper's + ``SyncRead`` cannot be marshalled by pywin32). A tag the server does + not know yields a :class:`ReadResult` with ``error`` set and a + ``BAD_CONFIG_ERROR``-style quality of 0 — never an exception, matching + the per-tag error contract of :class:`ReadResult` (IMPLEMENTATION_PLAN + §4.3). The group (and its items) is removed afterwards so repeated + calls never leak groups on the server. + """ + if self._opc is None: + raise OpcError("AutomationOpcClient: read() before connect()") + results: list[ReadResult] = [] + group = None + items_collection = None + live_items: list[tuple[str, Any]] = [] # (tag, item) in request order + try: + try: + group = self._opc.OPCGroups.Add("DiagnozerRead") + items_collection = group.OPCItems + errors: list[str | None] = [None] * len(tags) + for index, tag in enumerate(tags): + try: + item = items_collection.AddItem(tag, index + 1) + live_items.append((tag, item)) + except Exception as exc: # noqa: BLE001 — per-tag error + results.append( + ReadResult( + tag=tag, + value=None, + quality_raw=0, + device_time=None, + error=str(as_com_error(f"cannot add tag {tag!r}", exc)), + ) + ) + for tag, item in live_items: + try: + item.Read(int(source)) + raw_ts = item.TimeStamp + # pywintypes.datetime is a datetime subclass; normalize + # to UTC for the event model (SPECIFICATION §6). + device_time = ( + raw_ts.astimezone(timezone.utc) if raw_ts is not None else None + ) + results.append( + ReadResult( + tag=tag, + value=item.Value, + quality_raw=int(item.Quality), + device_time=device_time, + ) + ) + except Exception as exc: # noqa: BLE001 — per-tag error + results.append( + ReadResult( + tag=tag, + value=None, + quality_raw=0, + device_time=None, + error=str(as_com_error(f"read of tag {tag!r} failed", exc)), + ) + ) + except Exception as exc: # noqa: BLE001 — group-level failure + raise as_com_error( + f"could not read from server {self._server!r} " + "(group/level failure)", + exc, + ) from None + finally: + # Release every COM reference INSIDE the apartment, in reverse + # creation order, before close()/CoUninitialize can see them. + live_items.clear() + items_collection = None + if group is not None: + try: + group.Parent.OPCGroups.RemoveAll() + except Exception: # noqa: BLE001 — best-effort cleanup + pass + group = None + # preserve request order regardless of which tags failed to add + by_tag = {r.tag: r for r in results} + return [by_tag[tag] for tag in tags] def close(self) -> None: self._release_and_uninit() \ No newline at end of file diff --git a/src/opcda_diagnozer/opc/base.py b/src/opcda_diagnozer/opc/base.py index 7af126a..e51aa2a 100644 --- a/src/opcda_diagnozer/opc/base.py +++ b/src/opcda_diagnozer/opc/base.py @@ -15,6 +15,7 @@ from __future__ import annotations from collections.abc import Sequence from dataclasses import dataclass from datetime import datetime +from enum import IntEnum from typing import Any, Protocol, runtime_checkable @@ -29,6 +30,19 @@ class ReadResult: error: str | None = None +class OPCDataSource(IntEnum): + """Automation Wrapper data-source constants (OPCDataSource enum). + + ``CACHE`` reads the server's last-known value (fast, may be stale), + ``DEVICE`` forces a round-trip to the underlying device/controller — the + one the diagnozer cares about (SPECIFICATION §10: the measured path is + server → device, not server → cache). + """ + + CACHE = 1 + DEVICE = 2 + + # Common HRESULTs the tool decodes and reports (README "DCOM troubleshooting"). _HRESULT_NAMES: dict[int, str] = { 0x800706BA: "RPC_S_SERVER_UNAVAILABLE — server process down, firewall, or DCOM activation refused", @@ -38,7 +52,8 @@ _HRESULT_NAMES: dict[int, str] = { 0x80040154: "REGDB_E_CLASSNOTREG — wrong ProgID, or wrapper/server registered under a different bitness", 0x80020009: "DISP_E_EXCEPTION — the server rejected the call (e.g. no such branch/tag)", 0x80070057: "E_INVALIDARG — invalid argument (e.g. wrong branch or item name)", - 0xC0040007: "OPC_E_INVALIDITEMID — tag name wrong in config (surfaced by probe)", + 0xC0040007: "OPC_E_UNKNOWNITEMID — the server does not know this tag", + 0xC0040008: "OPC_E_INVALIDITEMID — tag name wrong in config (surfaced by probe)", } diff --git a/src/opcda_diagnozer/opc/fake.py b/src/opcda_diagnozer/opc/fake.py index 7461640..cd4ccf6 100644 --- a/src/opcda_diagnozer/opc/fake.py +++ b/src/opcda_diagnozer/opc/fake.py @@ -15,7 +15,7 @@ from __future__ import annotations from collections.abc import Sequence from datetime import datetime, timezone -from opcda_diagnozer.opc.base import Browser, OpcError, ReadResult, TagInfo +from opcda_diagnozer.opc.base import Browser, OPCDataSource, OpcError, ReadResult, TagInfo #: The fake address space: branch path -> (sub-branches, leaf names). #: Mirrors the shape of a real SCADA server (Channel → Device → Tags). @@ -99,15 +99,19 @@ class FakeOpcClient: """In-process :class:`OpcClient`; every read returns GOOD-quality zeros. ``space`` is injectable so tests can exercise flat namespaces, empty - servers, and deep trees; the default mirrors a Matrikon-like hierarchy. + servers, and deep trees; ``read_results`` scripts the next read() result + (Phase 1 scenario engine seed, IMPLEMENTATION_PLAN §4.9). """ def __init__( self, space: dict[tuple[str, ...], tuple[tuple[str, ...], tuple[str, ...]]] | None = None, + read_results: list[ReadResult] | None = None, ) -> None: self.connected = False self._space = FAKE_SPACE if space is None else space + self._scripted_reads = read_results + self.reads: list[tuple[tuple[str, ...], OPCDataSource]] = [] # call log def connect(self) -> None: self.connected = True @@ -117,9 +121,27 @@ class FakeOpcClient: raise OpcError("FakeOpcClient: create_browser() before connect()") return _FakeBrowser(self._space) - def read(self, tags: Sequence[str]) -> list[ReadResult]: + def read( + self, tags: Sequence[str], source: OPCDataSource = OPCDataSource.DEVICE + ) -> list[ReadResult]: if not self.connected: raise OpcError("FakeOpcClient: read() before connect()") + self.reads.append((tuple(tags), source)) + if self._scripted_reads is not None: + scripted = {r.tag: r for r in self._scripted_reads} + return [ + scripted.get( + tag, + ReadResult( + tag=tag, + value=None, + quality_raw=0, + device_time=None, + error=f"no scripted result for {tag!r}", + ), + ) + for tag in tags + ] now = datetime.now(timezone.utc) return [ ReadResult(tag=tag, value=0, quality_raw=0xC0, device_time=now) diff --git a/tests/test_browse.py b/tests/test_browse.py index 6d154d9..44d825a 100644 --- a/tests/test_browse.py +++ b/tests/test_browse.py @@ -221,7 +221,7 @@ def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_path, args: list[str]) -> int: ) client = FakeOpcClient() - def _fake_connect(config_path: str): + def _fake_connect(config_path: str, server=None, host=None): client.connect() return client diff --git a/tests/test_build_exe.py b/tests/test_build_exe.py new file mode 100644 index 0000000..92c4064 --- /dev/null +++ b/tests/test_build_exe.py @@ -0,0 +1,76 @@ +"""Tests for the frozen-exe build script + frozen-aware config default. + +The PyInstaller run itself is Windows-only and slow; these tests cover the +pure logic the build script and the CLI rely on. The build script is +PowerShell (``scripts/build_exe.ps1``); the tests assert on its content and +on the CLI behaviors the produced exe depends on. +""" + +import sys +from pathlib import Path +from unittest import mock + +import pytest + +from opcda_diagnozer.main import DEFAULT_CONFIG, _default_config_path + +BUILD_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "build_exe.ps1" + + +def test_default_config_path_in_source_mode() -> None: + """Not frozen: the default stays the plain relative name (CWD lookup).""" + with mock.patch.object(sys, "frozen", False, create=True): + assert _default_config_path() == DEFAULT_CONFIG + + +def test_default_config_path_when_frozen_resolves_next_to_exe( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Frozen: config.yaml lives next to the exe — never the service CWD (System32).""" + exe_dir = tmp_path / "Program Files" / "OPCDA-Diagnozer" + exe_dir.mkdir(parents=True) + exe = exe_dir / "opcda-diagnozer.exe" + exe.touch() + monkeypatch.setattr(sys, "frozen", True, raising=False) + monkeypatch.setattr(sys, "executable", str(exe)) + assert _default_config_path() == str(exe_dir / DEFAULT_CONFIG) + + +@pytest.mark.skipif(sys.platform != "win32", reason="the exe is Windows-only") +class TestBuildScript: + """Static assertions on scripts/build_exe.ps1 (no PyInstaller invocation).""" + + def test_hidden_imports_cover_lazy_modules(self) -> None: + """Every lazy import in the package must be declared to PyInstaller.""" + content = BUILD_SCRIPT.read_text(encoding="utf-8") + # the deferred COM imports (opc/automation.py, opc/discovery.py) + assert "'pythoncom'" in content + assert "'win32com.client'" in content + assert "'opcda_diagnozer.opc.automation'" in content + # the lazily-imported browse module (main.py verbs) + assert "'opcda_diagnozer.opc.browse'" in content + # the Phase 3 service host stays in the bundle from day one + assert "'opcda_diagnozer.service'" in content + + def test_version_info_template_uses_pyi_keywords(self) -> None: + """The embedded version-info must use PyInstaller's exact keyword names. + + FixedFileInfo.__init__ takes filevers/prodvers/mask/flags/OS/fileType/ + subtype/date — anything else (fileVers, fileDate, subType) raises a + TypeError inside PyInstaller's eval at build time. + """ + content = BUILD_SCRIPT.read_text(encoding="utf-8") + assert "filevers=(" in content + assert "fileType=0x1" in content + assert "subtype=0x0" in content + assert "date=(0, 0)" in content + # the rejected spellings must NOT appear + assert "fileVers=" not in content + assert "fileDate=" not in content + assert "subType=" not in content + + def test_bitness_guards_present(self) -> None: + """The script must refuse a 64-bit interpreter and verify the PE.""" + content = BUILD_SCRIPT.read_text(encoding="utf-8") + assert "struct.calcsize('P') * 8" in content # interpreter bitness check + assert "0x14C" in content # PE machine = IMAGE_FILE_MACHINE_I386 check \ No newline at end of file diff --git a/tests/test_config.py b/tests/test_config.py index 1e7dc8b..fd741d6 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -18,7 +18,7 @@ def _write(tmp_path: Path, text: str) -> Path: def test_example_config_loads() -> None: config = load_config(EXAMPLE) assert config.opc.server == "Matrikon.OPC.Simulation.1" - assert config.opc.host == "CONTROL-HOST01" + assert config.opc.host is None # host commented out -> local server assert len(config.tags) == 2 assert config.tags[0].name == "Channel1.Device1.Counter" assert config.tags[0].expect_change is True diff --git a/tests/test_read_cli.py b/tests/test_read_cli.py new file mode 100644 index 0000000..ad1cf7a --- /dev/null +++ b/tests/test_read_cli.py @@ -0,0 +1,210 @@ +"""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() \ No newline at end of file