From 76b1cc5ccffad99e12fadfba28f4f7eea7dd0085 Mon Sep 17 00:00:00 2001 From: Jonas Bengtsson Date: Wed, 7 Oct 2026 17:57:30 +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 --- .vscode/launch.json | 26 +- README.md | 47 ++-- config.example.yaml | 3 + src/opcda_diagnozer/config.py | 13 +- src/opcda_diagnozer/main.py | 286 +++++++++++----------- src/opcda_diagnozer/opc/automation.py | 7 +- src/opcda_diagnozer/opc/base.py | 39 +-- src/opcda_diagnozer/opc/browse.py | 260 +++++++++----------- tests/test_browse.py | 335 ++++++++++++++++---------- 9 files changed, 538 insertions(+), 478 deletions(-) diff --git a/.vscode/launch.json b/.vscode/launch.json index d9c3dca..eae2857 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -105,14 +105,14 @@ } }, { - // List the devices (address-space branches) of the configured - // server — one level, with per-device tag counts. - "name": "opcda-diagnozer: list-devices", + // Full structure tree of the configured server: every branch + // AND every tag in place, whole subtree. + "name": "opcda-diagnozer: list-items (whole tree)", "type": "debugpy", "request": "launch", "python": "${workspaceFolder}/.venv/Scripts/python.exe", "module": "opcda_diagnozer", - "args": ["list-devices", "--config", "config.example.yaml"], + "args": ["list-items", "all", "--depth", "0", "--config", "config.example.yaml"], "cwd": "${workspaceFolder}", "console": "integratedTerminal", "justMyCode": false, @@ -121,20 +121,20 @@ } }, { - // List the tags of one device (whole subtree) as JSON — the - // item_id values are what go into config `tags:`. Adjust the - // device path to the server your config points at. - "name": "opcda-diagnozer: list-tags (recursive, JSON)", + // Flat ItemID list of one subtree as JSON — the item_id values + // are what go into config `tags:`. Adjust the branch path to the + // server your config points at. + "name": "opcda-diagnozer: list-items tags (JSON)", "type": "debugpy", "request": "launch", "python": "${workspaceFolder}/.venv/Scripts/python.exe", "module": "opcda_diagnozer", "args": [ - "list-tags", - "--config", "config.example.yaml", - "Simulation Items.Bucket Brigade", - "--recursive", - "--json" + "list-items", + "tags", + "--branch", "Simulation Items.Bucket Brigade", + "--json", + "--config", "config.example.yaml" ], "cwd": "${workspaceFolder}", "console": "integratedTerminal", diff --git a/README.md b/README.md index 2077feb..53d78f7 100644 --- a/README.md +++ b/README.md @@ -33,8 +33,8 @@ See the full interpretation matrix in the [specification](docs/SPECIFICATION.md) - Windows service (auto-start, SCM crash recovery) with a foreground `run` mode for testing - Local or remote (DCOM) connections to any OPC DA server -- Server browsing: `list-servers` (discovery), `list-devices` and `list-tags` - (address-space walk) — live now +- Server browsing: `list-servers` (discovery) and `list-items` + (address-space browse: full tree / branches / tags) — live now - Latency monitoring: spikes, streaks, and p95 trends against configurable thresholds - Hang watchdog: flags RPC calls that never return — the silent killer - Quality monitoring: BAD/UNCERTAIN transitions with decoded subcodes, persistent-bad @@ -51,16 +51,15 @@ opcda-diagnozer list-servers opcda-diagnozer list-servers --host CONTROL-HOST01 # remote, via DCOM opcda-diagnozer list-servers --source registry --json # no-COM local scan -# list the devices (address-space branches) of the configured server -opcda-diagnozer list-devices -opcda-diagnozer list-devices --depth 2 # include sub-devices -opcda-diagnozer list-devices --json # script-friendly output - -# list the tags of one device (the item_id values go into config `tags:`) -opcda-diagnozer list-tags "Simulation Items.Bucket Brigade" -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 +# browse the server's item space (list-items all|branches|tags) +# all = the full structure tree: every branch AND every tag in place +# branches = the structure nodes (channels/devices/folders — all "branches" to OPC DA) +# tags = the flat ItemID list that goes into config `tags:` +opcda-diagnozer list-items # top level of the tree +opcda-diagnozer list-items --depth 0 # the whole tree (browse.max_depth guards) +opcda-diagnozer list-items branches --depth 2 # structure, two levels +opcda-diagnozer list-items tags --branch "Simulation Items" # one subtree, flat ItemIDs +opcda-diagnozer list-items --json # script-friendly output # read tags right now (validation tool) — no config needed with --server opcda-diagnozer read --server Matrikon.OPC.Simulation.1 "Bucket Brigade.Int1" @@ -88,15 +87,20 @@ hosts or when the wrapper DLL is not registered. `auto` tries the first, falls b the second. Note the bitness rule still applies: a 32-bit process enumerates the 32-bit registry view — exactly the servers this tool can load. -`list-devices` / `list-tags` browse the **connected server's** address space through -the Automation Wrapper's browser (`CreateBrowser`): devices are *branches*, tags are -*leaves*. They work locally and over DCOM (same Launch+Activation requirements). A -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. +`list-items` browses the **connected server's** address space through the +Automation Wrapper's browser (`CreateBrowser`). OPC DA defines exactly two +element kinds — **branches** (the structure: what SCADA vendors call channels, +devices, groups or folders; the protocol cannot tell them apart) and **leaves** +(the tags). The `all` view renders the full tree — including leaves that hang +directly at the server root and belong to no branch (server-internal items +like `@ClientCount`). It works locally and over DCOM (same Launch+Activation +requirements). A `--branch` path is either one dotted string +(`"Simulation Items.Bucket Brigade"`) or repeated flag segments — use the +latter when a branch name itself contains a dot. `--depth 1` (default) lists +one level; `--depth 0` walks the whole subtree, guarded by `browse.max_depth` +from the config (default 32) against server-side cycles. -Every connecting verb (`read`, `probe`, `list-devices`, `list-tags`) accepts +Every connecting verb (`read`, `probe`, `list-items`) 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 @@ -136,6 +140,9 @@ stale: enabled: true max_age_s: 120 +browse: + max_depth: 32 # address-space walk guard (list-items --depth 0) + report: heartbeat_s: 300 jsonl: { dir: "C:/ProgramData/OPCDA-Diagnozer/logs", max_mb: 50, backups: 10 } diff --git a/config.example.yaml b/config.example.yaml index 92eb161..0f18b1c 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -31,6 +31,9 @@ stale: enabled: true max_age_s: 120 # D12 device timestamp older than this while GOOD +browse: + max_depth: 32 # address-space walk guard (list-items --depth 0) + report: heartbeat_s: 300 # D15 periodic stats snapshot jsonl: diff --git a/src/opcda_diagnozer/config.py b/src/opcda_diagnozer/config.py index a06888f..24db745 100644 --- a/src/opcda_diagnozer/config.py +++ b/src/opcda_diagnozer/config.py @@ -66,6 +66,13 @@ class StaleConfig: max_age_s: float = 120.0 # D12 / D13 +@dataclass(frozen=True) +class BrowseConfig: + # Address-space walk guard: deeper SCADA trees are fine, but a walk this + # deep is almost certainly a server-side cycle (list-items --depth 0). + max_depth: int = 32 + + @dataclass(frozen=True) class JsonlConfig: dir: str = "C:/ProgramData/OPCDA-Diagnozer/logs" @@ -96,10 +103,11 @@ class Config: latency: LatencyConfig = field(default_factory=LatencyConfig) quality: QualityConfig = field(default_factory=QualityConfig) stale: StaleConfig = field(default_factory=StaleConfig) + browse: BrowseConfig = field(default_factory=BrowseConfig) report: ReportConfig = field(default_factory=ReportConfig) -_ALLOWED_TOP_LEVEL = {"opc", "poll", "tags", "latency", "quality", "stale", "report"} +_ALLOWED_TOP_LEVEL = {"opc", "poll", "tags", "latency", "quality", "stale", "browse", "report"} _TAG_KEYS = {"name", "expect_change"} @@ -207,6 +215,8 @@ def _validate(config: Config) -> None: raise ConfigError(f"quality.flapping_window_s: must be > 0, got {config.quality.flapping_window_s}") if config.stale.enabled and config.stale.max_age_s <= 0: raise ConfigError(f"stale.max_age_s: must be > 0, got {config.stale.max_age_s}") + if config.browse.max_depth < 1: + raise ConfigError(f"browse.max_depth: must be >= 1, got {config.browse.max_depth}") if config.report.heartbeat_s <= 0: raise ConfigError(f"report.heartbeat_s: must be > 0, got {config.report.heartbeat_s}") if config.report.eventlog.min_severity not in ("INFO", "WARN", "CRITICAL"): @@ -246,6 +256,7 @@ def load_config(path: str | Path) -> Config: latency=_build_section(LatencyConfig, data.get("latency"), "latency"), quality=_build_section(QualityConfig, data.get("quality"), "quality"), stale=_build_section(StaleConfig, data.get("stale"), "stale"), + browse=_build_section(BrowseConfig, data.get("browse"), "browse"), report=_build_report(data.get("report")), ) try: diff --git a/src/opcda_diagnozer/main.py b/src/opcda_diagnozer/main.py index e047758..2d42a93 100644 --- a/src/opcda_diagnozer/main.py +++ b/src/opcda_diagnozer/main.py @@ -2,9 +2,9 @@ Verbs: ``run``, ``probe``, ``check-config``, ``list-items`` plus the service verbs ``install`` / ``uninstall`` / ``start`` / ``stop``. Currently functional: -``check-config``, ``list-servers`` (OPC DA server discovery) and the browse -verbs ``list-devices`` / ``list-tags`` (devices = branches, tags = leaves of -the connected server's address space); the remaining verbs land with Phases 1–3. +``check-config``, ``list-servers`` (discovery), ``list-items`` (address-space +browsing: ``all`` / ``branches`` / ``tags``) and ``read`` / ``probe`` (one-shot +tag reads); the remaining verbs land with Phases 1–3. """ from __future__ import annotations @@ -36,7 +36,6 @@ def _default_config_path() -> str: _NOT_IMPLEMENTED: dict[str, str] = { "run": "Phase 1 (core engine)", - "list-items": "superseded by list-devices / list-tags", "install": "Phase 3 (Windows service)", "uninstall": "Phase 3 (Windows service)", "start": "Phase 3 (Windows service)", @@ -127,9 +126,6 @@ def _build_parser() -> argparse.ArgumentParser: p = sub.add_parser("check-config", help="validate the configuration and exit") _add_config_option(p) - p = sub.add_parser("list-items", help="browse the server item space (validation tool)") - _add_config_option(p) - p = sub.add_parser( "list-servers", help="enumerate the OPC DA servers available locally or on a DCOM host", @@ -154,52 +150,42 @@ def _build_parser() -> argparse.ArgumentParser: ) p = sub.add_parser( - "list-devices", - help="list the devices (address-space branches) exposed by the configured server", + "list-items", + help="browse the server's item space: the full structure tree, or a " + "flat view of its branches / tags", ) _add_config_option(p) _add_server_overrides(p) + p.add_argument( + "kind", + nargs="?", + choices=("all", "branches", "tags"), + default="all", + help="all = the full structure tree (branches AND tags in place); " + "branches = the structure nodes only (SCADA channels/devices/folders " + "are all branches to OPC DA); tags = the flat ItemID list that goes " + "into config `tags:` (default: %(default)s)", + ) + p.add_argument( + "--branch", + action="append", + metavar="PATH", + help="limit the listing to this branch subtree: one dotted string " + "('--branch \"Simulation Items.Bucket Brigade\"') or repeat the flag " + "for explicit segment names containing dots (default: the server root)", + ) p.add_argument( "--depth", type=int, default=1, metavar="N", - help="how many branch levels to descend (default: %(default)s)", - ) - p.add_argument( - "--no-counts", - action="store_true", - help="skip fetching per-device tag counts (faster on huge servers)", + help="how many branch levels to descend; 0 = the whole subtree " + "(default: %(default)s)", ) p.add_argument( "--json", action="store_true", - help="emit one JSON object per device (for scripts) instead of a table", - ) - - p = sub.add_parser( - "list-tags", - 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="*", - metavar="BRANCH", - help="device path: one or more branch names — either a single dotted string " - "in quotes ('\"Simulation Items.Bucket Brigade\"') or separate quoted " - "segments ('\"Simulation Items\" \"Bucket Brigade\"'). Default: the server root", - ) - p.add_argument( - "--recursive", "-r", - action="store_true", - help="walk the whole subtree, not just the device's direct tags", - ) - p.add_argument( - "--json", - action="store_true", - help="emit one JSON object per tag (for scripts) instead of a table", + help="emit JSON objects (for scripts) instead of a tree/list", ) p = sub.add_parser("install", help="install as a Windows service (requires admin)") @@ -316,46 +302,65 @@ def _connect_from_config(config_path: str, server: str | None = None, host: str return client -def _print_devices(space, as_json: bool) -> None: - """Human table / JSON-lines output for the list-devices verb.""" - if as_json: - import json - - for device in space.devices: - payload = { - "name": device.name, - "path": list(device.path), - "qualified_name": device.qualified_name, - "sub_devices": list(device.sub_devices), - "tags": list(device.tag_names), - } - print(json.dumps(payload, ensure_ascii=False)) - return - if not space.devices: - print(f"no devices (branches) found — organization: {space.organization}") - return - print(f"address space organization: {space.organization}") - print("devices:") - width = max(len(d.qualified_name) for d in space.devices) - for device in space.devices: - notes = [f"{len(device.tag_names)} tags"] - if device.sub_devices: - notes.append(f"{len(device.sub_devices)} sub-devices") - print(f" {device.qualified_name:<{width}} ({', '.join(notes)})") +def _print_item_tree(nodes, indent: int = 0) -> None: + """Indented structure view: branches with a trailing /, tags with their ItemID.""" + for node in nodes: + prefix = " " * indent + if node.kind == "branch": + print(f"{prefix}{node.name}/") + _print_item_tree(node.children, indent + 1) + else: + print(f"{prefix}{node.name} [ItemID: {node.item_id}]") -def _cmd_list_devices( +def _node_payload(node) -> dict: + """JSON payload for one SpaceNode (branches carry their children).""" + return { + "kind": node.kind, + "name": node.name, + "path": list(node.path), + "item_id": node.item_id, + "children": [_node_payload(c) for c in node.children], + } + + +def _cmd_list_items( config_path: str, server: str | None, host: str | None, - depth: int, include_counts: bool, as_json: bool, + kind: str, branch: Sequence[str] | None, depth: int, as_json: bool, ) -> int: - from opcda_diagnozer.opc.browse import list_devices + from opcda_diagnozer.opc.browse import ( + DEFAULT_MAX_DEPTH, flatten_branches, flatten_leaves, + split_branch_path, walk_tree, + ) + import json + + branch_list = list(branch) if branch else None + # One --branch occurrence is a *dotted* branch path ("Simulation Items.Bucket + # Brigade"); several occurrences are explicit segment names (those may + # contain dots). Same convention the fake and browse helpers share. + branch_arg = ( + branch_list[0] + if branch_list and len(branch_list) == 1 + else branch_list + ) + # --depth 0 means the whole subtree; the hard guard is browse.max_depth + # from the config (default 32) — a missing config is fine when --server + # is given, so fall back to the default guard. + max_depth = DEFAULT_MAX_DEPTH + if Path(config_path).is_file(): + try: + max_depth = load_config(config_path).browse.max_depth + except ConfigError: + pass # an invalid config surfaces in _connect_from_config below try: client = _connect_from_config(config_path, server, host) browser = None try: browser = client.create_browser() - space = list_devices(browser, depth=depth, include_tag_counts=include_counts) + nodes = walk_tree( + browser, branch_arg, depth=depth, max_depth=max_depth + ) finally: # Release the browser's COM reference *before* close() tears down the # apartment — releasing an RCW after CoUninitialize crashes the process. @@ -363,85 +368,76 @@ def _cmd_list_devices( browser.close() client.close() except ConfigError as exc: - print(f"list-devices: FAILED\n {exc}", file=sys.stderr) + print(f"list-items: FAILED\n {exc}", file=sys.stderr) return 2 except (OpcError, ValueError) as exc: - print(f"list-devices: FAILED\n {exc}", file=sys.stderr) + print(f"list-items: FAILED\n {exc}", file=sys.stderr) return 2 - _print_devices(space, as_json) - if not as_json and not space.devices: - hint = ( - "hint: the server address space is flat — every tag hangs directly off " - "the root; use: opcda-diagnozer list-tags" - if space.organization == "flat" - else "hint: no branches at the root — the server may expose only tags; " - "try: opcda-diagnozer list-tags" - ) - print(hint, file=sys.stderr) - return 0 + path = split_branch_path(branch_arg) + where = ".".join(path) or "(server root)" -def _print_tags(tags, device_path: tuple[str, ...], recursive: bool, as_json: bool) -> None: - """Human table / JSON-lines output for the list-tags verb.""" - where = ".".join(device_path) or "(server root)" - scope = " (recursive)" if recursive else "" - if as_json: - import json + if kind == "all": + if not nodes: + print(f"nothing defined under {where}") + return 0 + if as_json: + for node in nodes: + print(json.dumps(_node_payload(node), ensure_ascii=False)) + else: + scope = "whole subtree" if depth == 0 else f"depth {depth}" + print(f"items under {where} ({scope}):") + _print_item_tree(nodes) + return 0 + if kind == "tags": + tags = flatten_leaves(nodes) + if as_json: + for tag in tags: + print(json.dumps( + {"name": tag.name, "item_id": tag.item_id, "path": list(tag.path)}, + ensure_ascii=False, + )) + return 0 + if not tags: + print( + f"no tags found under {where} — increase --depth (0 = whole subtree) " + "or pick another --branch" + ) + return 0 + print(f"tags under {where}: {len(tags)}") + width = max(len(t.item_id) for t in tags) for tag in tags: - payload = { - "name": tag.name, - "item_id": tag.item_id, - "path": list(tag.path), - } - print(json.dumps(payload, ensure_ascii=False)) - return - if not tags: - print(f"no tags found under {where}{scope}") - return - print(f"tags under {where}{scope}: {len(tags)}") - width = max(len(t.item_id) for t in tags) - for tag in tags: - print(f" {tag.item_id:<{width}} (leaf: {tag.name})") + print(f" {tag.item_id:<{width}} (leaf: {tag.name})") + return 0 - -def _cmd_list_tags( - config_path: str, - server: str | None, - host: str | None, - device: Sequence[str], - recursive: bool, - as_json: bool, -) -> int: - from opcda_diagnozer.opc.browse import list_tags, split_device_path - - # A single CLI argument is a *dotted* branch path; several arguments are - # 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, server, host) - browser = None - try: - browser = client.create_browser() - tags = list_tags(browser, device=device_arg or None, recursive=recursive) - finally: - if browser is not None: - browser.close() - client.close() - except ConfigError as exc: - print(f"list-tags: FAILED\n {exc}", file=sys.stderr) - return 2 - except (OpcError, ValueError) as exc: - print(f"list-tags: FAILED\n {exc}", file=sys.stderr) - return 2 - device_path = split_device_path(device_arg or None) - _print_tags(tags, device_path, recursive, as_json) - if not as_json and not tags and not recursive: + # kind == "branches" + branches = flatten_branches(nodes) + if as_json: + for node in branches: + print(json.dumps({ + "name": node.name, + "path": list(node.path), + "tags": sum(1 for c in node.children if c.kind == "leaf"), + "sub_branches": sum(1 for c in node.children if c.kind == "branch"), + }, ensure_ascii=False)) + return 0 + if not branches: print( - "hint: the branch exists but exposes no direct leaves — try --recursive " - "to include the whole subtree", - file=sys.stderr, + f"no branches found under {where} — the space may be flat; " + "try kind 'all' or 'tags'" ) + return 0 + print(f"branches under {where}: {len(branches)}") + width = max(len(".".join(b.path)) for b in branches) + for node in branches: + dotted = ".".join(node.path) + tags = sum(1 for c in node.children if c.kind == "leaf") + subs = sum(1 for c in node.children if c.kind == "branch") + notes = [f"{tags} tags"] + if subs: + notes.append(f"{subs} sub-branches") + print(f" {dotted:<{width}} ({', '.join(notes)})") return 0 @@ -557,13 +553,9 @@ def main(argv: Sequence[str] | None = None) -> int: return _cmd_check_config(args.config) 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.server, args.host, args.depth, not args.no_counts, args.json - ) - if args.verb == "list-tags": - return _cmd_list_tags( - args.config, args.server, args.host, args.device, args.recursive, args.json + if args.verb == "list-items": + return _cmd_list_items( + args.config, args.server, args.host, args.kind, args.branch, args.depth, args.json ) if args.verb == "read": return _cmd_read( diff --git a/src/opcda_diagnozer/opc/automation.py b/src/opcda_diagnozer/opc/automation.py index 5f664ad..3659369 100644 --- a/src/opcda_diagnozer/opc/automation.py +++ b/src/opcda_diagnozer/opc/automation.py @@ -106,9 +106,10 @@ class _AutomationBrowser: if not name: continue item_id = str(self._browser.GetItemID(name)).strip() - if not item_id: - # Some servers return an empty ItemID for filtered leaves; - # fall back to the browse name so config tags stay usable. + if not item_id or item_id == f".{name}": + # Some servers return an empty ItemID for filtered/special + # leaves (observed: Matrikon's alias-root items); fall back + # to the browse name so config tags stay usable. item_id = name leaves.append(TagInfo(name=name, item_id=item_id, path=tuple(path))) return leaves diff --git a/src/opcda_diagnozer/opc/base.py b/src/opcda_diagnozer/opc/base.py index e51aa2a..aa15e31 100644 --- a/src/opcda_diagnozer/opc/base.py +++ b/src/opcda_diagnozer/opc/base.py @@ -5,9 +5,12 @@ are normalized to :class:`OpcError` carrying the raw HRESULT and a decoded name. There is deliberately **no write path** — the tool is read-only by construction (SPECIFICATION §10.5). -Browsing (devices = branches, tags = leaves) is modelled by :class:`Browser`; -the ``list-devices`` / ``list-tags`` CLI verbs walk the address space through -it without any recursion into the monitored poll path. +Browsing is modelled by :class:`Browser`. The OPC DA browse world defines +exactly two element kinds: **branches** (what SCADA vendors call channels, +devices, groups or folders — indistinguishable at the protocol level) and +**leaves** (the tags). :class:`SpaceNode` is the tree node the ``list-items`` +CLI verb renders; leaves may hang directly at the server root, belonging to +no branch at all (server-internal items like ``@ClientCount``). """ from __future__ import annotations @@ -16,7 +19,7 @@ from collections.abc import Sequence from dataclasses import dataclass from datetime import datetime from enum import IntEnum -from typing import Any, Protocol, runtime_checkable +from typing import Any, Literal, Protocol, runtime_checkable @dataclass(frozen=True) @@ -105,25 +108,25 @@ ACCESS_WRITE = 0x02 @dataclass(frozen=True) -class DeviceInfo: - """One browsable device: a branch in the server's hierarchical address space. +class SpaceNode: + """One node of the browsed address space: a branch or a leaf (tag). - The Automation Wrapper exposes no per-branch metadata beyond the name, so a - "device" here is simply the branch path plus the counts the CLI prints. + OPC DA's browse model has exactly these two element kinds — there is no + separate "device" or "folder" type: SCADA channels, devices, groups and + folders are all plain branches, indistinguishable at the protocol level. + A branch may hold leaves, sub-branches, or both; leaves may also hang + directly at the root, belonging to no branch at all. """ + kind: Literal["branch", "leaf"] name: str - #: Branch path from the server root, e.g. ``("Simulation Items", "Bucket Brigade")``. + #: For a branch: its own path from the root, e.g. ``("Simulation Items",)``. + #: For a leaf: the branch it hangs under — ``()`` means directly at the root. path: tuple[str, ...] - #: Sub-branches directly under this device (usually channels/groups). - sub_devices: tuple[str, ...] = () - #: Leaf (tag) names directly under this device. - tag_names: tuple[str, ...] = () - - @property - def qualified_name(self) -> str: - """Human-friendly dotted rendering: ``Simulation Items.Bucket Brigade``.""" - return ".".join(self.path) if self.path else self.name + #: Fully qualified ItemID — leaves only; what goes into config ``tags:``. + item_id: str | None = None + #: Child nodes — branches only; empty when the depth limit stopped here. + children: tuple["SpaceNode", ...] = () @dataclass(frozen=True) diff --git a/src/opcda_diagnozer/opc/browse.py b/src/opcda_diagnozer/opc/browse.py index 7e2e00b..0cd966e 100644 --- a/src/opcda_diagnozer/opc/browse.py +++ b/src/opcda_diagnozer/opc/browse.py @@ -1,51 +1,45 @@ -"""Browse the connected server's address space: devices (branches) and tags (leaves). +"""Browse the connected server's address space (IMPLEMENTATION_PLAN §6.2). -OPC DA has no formal "device" concept — servers publish a *hierarchical -address space* whose top-level branches almost always represent devices (or -device families), with tags as the leaves. This module maps that tree onto the -two questions ops asks (SPECIFICATION §8, validation tooling): +OPC DA's browse model defines exactly two element kinds: -- *Which devices does this server expose?* → :func:`list_devices` -- *Which tags does device X expose?* → :func:`list_tags` +- **branches** — the structure nodes. SCADA vendors map channels, devices, + groups and folders onto branches; the protocol cannot tell them apart, so + this tool reports them uniformly as branches. +- **leaves** — the tags (items), each with a fully qualified ItemID that goes + straight into config ``tags:``. -The Automation Wrapper browser (``opc.CreateBrowser()``) is the only portable -traversal API — it works over DCOM for remote servers too, with the same -Launch+Activation requirements as ``list-servers``. Traversal is one level at -a time; the ``max_depth`` guard keeps a pathological (cyclic/AB-tree) server -from stalling the CLI. +Leaves may also hang directly at the server root, belonging to no branch — +server-internal items (``@ClientCount`` on the Matrikon sim) look "wrong" in +a device-oriented view but are ordinary root-level leaves. -Flat-namespace servers report ``organization == "flat"``; there every leaf is -addressed by its bare ItemID, the root is treated as one implicit device. +The ``list-items`` CLI verb renders whatever the caller asks for +(``all`` / ``branches`` / ``tags``) through :func:`walk_tree`, bounded by +``--depth`` (1 = one level, :data:`WHOLE_SUBTREE` = everything) and by the +configured ``browse.max_depth`` (default 32) against server-side cycles. """ from __future__ import annotations from collections.abc import Sequence -from dataclasses import dataclass -from opcda_diagnozer.opc.base import Browser, DeviceInfo, OpcError, TagInfo +from opcda_diagnozer.opc.base import Browser, OpcError, SpaceNode, TagInfo -#: Walk guard: deep SCADA hierarchies exist (Channel → Device → Tag), but OPC -#: trees deeper than this are almost certainly a server-side loop. -MAX_DEPTH = 32 +#: Default walk guard: deep SCADA hierarchies exist (Channel → Device → Tag), +#: but OPC trees deeper than this are almost certainly a server-side loop. +#: Overridable via the ``browse.max_depth`` config key. +DEFAULT_MAX_DEPTH = 32 + +#: Depth value meaning "no limit" (the whole subtree, max_depth-guarded). +WHOLE_SUBTREE = 0 -@dataclass(frozen=True) -class Space: - """Result of a device-space walk: one entry per discovered branch node.""" - - organization: str - root: DeviceInfo - devices: tuple[DeviceInfo, ...] - - -def split_device_path(path: Sequence[str] | str | None) -> tuple[str, ...]: - """Normalize a user-supplied device path into branch segments. +def split_branch_path(path: Sequence[str] | str | None) -> tuple[str, ...]: + """Normalize a user-supplied branch path into segments. A **single** string is treated as dotted: ``"Simulation Items.Bucket Brigade"`` → ``("Simulation Items", "Bucket Brigade")``. **Multiple** segments are taken as explicit branch names — no dot splitting, so names that contain - a dot stay intact (they just have to be passed as separate arguments). + a dot stay intact (pass them as separate ``--branch`` arguments). Empty segments and whitespace are dropped. """ if path is None: @@ -61,134 +55,96 @@ def split_device_path(path: Sequence[str] | str | None) -> tuple[str, ...]: return tuple(parts) -def list_devices( - browser: Browser, - *, - depth: int = 1, - include_tag_counts: bool = True, -) -> Space: - """Enumerate devices (branches) starting at the server root. - - ``depth=1`` lists top-level branches (the typical "device list"); greater - depths reveal sub-devices (channels/groups). Each :class:`DeviceInfo` - carries the direct sub-branches and — when ``include_tag_counts`` is on — - the leaf names under it, so the CLI can print ``Device (3 tags, 2 sub)``. - - Raises :class:`OpcError` when the server rejects the walk (COM failure or - unsupported browse interface). The walk is bounded by ``depth`` (≤ - :data:`MAX_DEPTH`), so a pathological server can never stall the CLI. - """ - if depth < 1: - raise ValueError("depth must be >= 1") - if depth > MAX_DEPTH: - raise ValueError(f"depth must be <= {MAX_DEPTH}") - organization = browser.organization - root_branches = browser.list_branches(()) - root_leaves = browser.list_leaves(()) if include_tag_counts else [] - root = DeviceInfo( - name="(root)", - path=(), - sub_devices=tuple(root_branches), - tag_names=tuple(t.name for t in root_leaves), - ) - devices: list[DeviceInfo] = [] - _walk_level(browser, (), root_branches, depth, devices, include_tag_counts) - return Space(organization=organization, root=root, devices=tuple(devices)) - - -def _walk_level( - browser: Browser, - path: tuple[str, ...], - branches: Sequence[str], - depth: int, - devices: list[DeviceInfo], - include_tag_counts: bool, -) -> None: - if depth <= 0: - return - for branch in branches: - sub_path = path + (branch,) - sub_branches = tuple(browser.list_branches(sub_path)) - tag_names = ( - tuple(t.name for t in browser.list_leaves(sub_path)) - if include_tag_counts - else () - ) - devices.append( - DeviceInfo( - name=branch, - path=sub_path, - sub_devices=sub_branches, - tag_names=tag_names, - ) - ) - _walk_level(browser, sub_path, sub_branches, depth - 1, devices, include_tag_counts) - - -def list_tags( - browser: Browser, - device: str | Sequence[str] | None = None, - *, - recursive: bool = False, -) -> list[TagInfo]: - """List the tags of one device (branch), or of the whole server. - - ``device`` accepts either a dotted string (``"Simulation Items.Bucket - Brigade"``) or a pre-split sequence of branch names. ``None`` means the - server root; with ``recursive=True`` every leaf under it is returned - (bounded by :data:`MAX_DEPTH`). The returned :class:`TagInfo.item_id` - values are exactly what goes into the config ``tags:`` list. An unknown - device path raises :class:`OpcError` pointing at ``list-devices``. - """ - path = _require_device(browser, device, what="list-tags") - leaves = browser.list_leaves(path) - tags: list[TagInfo] = list(leaves) - if not recursive: - return sorted(tags, key=lambda t: t.item_id.lower()) - sub_branches = browser.list_branches(path) - _collect_recursive(browser, path, sub_branches, tags) - return sorted(tags, key=lambda t: t.item_id.lower()) - - -def _collect_recursive( - browser: Browser, - path: tuple[str, ...], - branches: Sequence[str], - tags: list[TagInfo], -) -> None: - for branch in branches: - sub_path = path + (branch,) - if len(sub_path) > MAX_DEPTH: - raise OpcError( - f"address space deeper than {MAX_DEPTH} levels at " - f"{'.'.join(sub_path)} — refusing to walk further (possible " - "server-side cycle)" - ) - tags.extend(browser.list_leaves(sub_path)) - _collect_recursive(browser, sub_path, browser.list_branches(sub_path), tags) - - -def _require_device( - browser: Browser, - device: str | Sequence[str] | None, - *, - what: str, +def require_branch( + browser: Browser, branch: Sequence[str] | str | None, *, what: str ) -> tuple[str, ...]: - """Resolve and validate a user-supplied device path (exists, is a branch).""" - path = split_device_path(device) + """Resolve and validate a user-supplied branch path (must exist).""" + path = split_branch_path(branch) if not path: return () siblings = browser.list_branches(path[:-1]) if path[-1] not in siblings: raise OpcError( - f"{what}: device {path[-1]!r} not found under " - f"{'.'.join(path[:-1]) or '(root)'} — run list-devices to see the " - "available branches" + f"{what}: branch {path[-1]!r} not found under " + f"{'.'.join(path[:-1]) or '(root)'} — list the levels above it " + "with list-items to see the available branch names" ) return path -def tags_for_device(browser: Browser, device: str | Sequence[str]) -> list[TagInfo]: - """List tags of a **known** device path — validation helper (probe).""" - path = _require_device(browser, device, what="list-tags") - return sorted(browser.list_leaves(path), key=lambda t: t.item_id.lower()) \ No newline at end of file +def walk_tree( + browser: Browser, + branch: Sequence[str] | str | None = None, + *, + depth: int = 1, + max_depth: int = DEFAULT_MAX_DEPTH, +) -> list[SpaceNode]: + """Return the nodes the server defines under ``branch``, keeping structure. + + ``depth`` is the number of branch levels to descend: 1 (the default) + returns the direct leaves and branches under ``branch`` (branches come + back with empty ``children``); :data:`WHOLE_SUBTREE` walks the entire + subtree. ``max_depth`` is the hard guard against server-side cycles — + it defaults to :data:`DEFAULT_MAX_DEPTH` and comes from the + ``browse.max_depth`` config key. + """ + if depth < 0: + raise ValueError(f"depth must be >= 0 (0 = whole subtree), got {depth}") + if max_depth < 1: + raise ValueError(f"max_depth must be >= 1, got {max_depth}") + if depth > max_depth: + raise ValueError(f"depth must be <= max_depth ({max_depth}), got {depth}") + root = require_branch(browser, branch, what="list-items") + return _walk_level(browser, root, depth, max_depth) + + +def _walk_level( + browser: Browser, path: tuple[str, ...], depth: int, max_depth: int +) -> list[SpaceNode]: + nodes: list[SpaceNode] = [ + SpaceNode(kind="leaf", name=leaf.name, path=path, item_id=leaf.item_id) + for leaf in browser.list_leaves(path) + ] + for branch in browser.list_branches(path): + sub = path + (branch,) + if len(sub) > max_depth: + raise OpcError( + f"address space deeper than {max_depth} levels at {'.'.join(sub)} " + "— refusing to walk further (possible server-side cycle)" + ) + if depth == 1: + children: tuple[SpaceNode, ...] = () + else: + next_depth = WHOLE_SUBTREE if depth == WHOLE_SUBTREE else depth - 1 + children = tuple(_walk_level(browser, sub, next_depth, max_depth)) + nodes.append(SpaceNode(kind="branch", name=branch, path=sub, children=children)) + return nodes + + +def flatten_leaves(nodes: Sequence[SpaceNode]) -> list[TagInfo]: + """Every leaf in the walked tree as a flat, case-insensitively sorted list.""" + tags: list[TagInfo] = [] + _collect_leaves(nodes, tags) + return sorted(tags, key=lambda t: t.item_id.lower()) + + +def _collect_leaves(nodes: Sequence[SpaceNode], tags: list[TagInfo]) -> None: + for node in nodes: + if node.kind == "leaf" and node.item_id is not None: + tags.append(TagInfo(name=node.name, item_id=node.item_id, path=node.path)) + else: + _collect_leaves(node.children, tags) + + +def flatten_branches(nodes: Sequence[SpaceNode]) -> list[SpaceNode]: + """Every branch in the walked tree, depth-first, as a flat list.""" + branches: list[SpaceNode] = [] + _collect_branches(nodes, branches) + return branches + + +def _collect_branches(nodes: Sequence[SpaceNode], out: list[SpaceNode]) -> None: + for node in nodes: + if node.kind == "branch": + out.append(node) + _collect_branches(node.children, out) \ No newline at end of file diff --git a/tests/test_browse.py b/tests/test_browse.py index 44d825a..5b0b84e 100644 --- a/tests/test_browse.py +++ b/tests/test_browse.py @@ -1,24 +1,28 @@ -"""Browse tests (opc/browse.py + fake browser + CLI verbs). +"""Browse tests (opc/browse.py walk_tree + the list-items CLI verb). Pure tests run on any OS against the fake client's address space; the COM -path is Windows-only and covered by the live smoke tests (IMPLEMENTATION_PLAN +path is Windows-only and covered by the live smoke test (IMPLEMENTATION_PLAN §9: COM adapters are import-guarded). """ +import json import sys +from pathlib import Path import pytest from opcda_diagnozer.main import main as cli_main -from opcda_diagnozer.opc.base import Browser, DeviceInfo, OpcError, TagInfo +from opcda_diagnozer.opc.base import Browser, OpcError, SpaceNode from opcda_diagnozer.opc.browse import ( - MAX_DEPTH, - list_devices, - list_tags, - split_device_path, - tags_for_device, + DEFAULT_MAX_DEPTH, + WHOLE_SUBTREE, + flatten_branches, + flatten_leaves, + require_branch, + split_branch_path, + walk_tree, ) -from opcda_diagnozer.opc.fake import FAKE_SPACE, FakeOpcClient +from opcda_diagnozer.opc.fake import FakeOpcClient @pytest.fixture @@ -29,7 +33,7 @@ def browser() -> Browser: # --------------------------------------------------------------------------- # -# split_device_path +# split_branch_path # --------------------------------------------------------------------------- # @@ -50,57 +54,99 @@ def browser() -> Browser: (("Bucket. Inc", "Device 1"), ("Bucket. Inc", "Device 1")), ], ) -def test_split_device_path(raw, expected: tuple[str, ...]) -> None: - assert split_device_path(raw) == expected +def test_split_branch_path(raw, expected: tuple[str, ...]) -> None: + assert split_branch_path(raw) == expected # --------------------------------------------------------------------------- # -# list_devices +# walk_tree # --------------------------------------------------------------------------- # -def test_list_devices_top_level(browser: Browser) -> None: - space = list_devices(browser) - assert space.organization == "hierarchical" - assert space.root.sub_devices == ("Simulation Items", "Configured Aliases") - names = [d.name for d in space.devices] - assert names == ["Simulation Items", "Configured Aliases"] - assert all(d.path == (d.name,) for d in space.devices) +def test_walk_tree_depth_one(browser: Browser) -> None: + """Default depth: direct leaves + branches, branches have no children.""" + nodes = walk_tree(browser) + assert [(n.kind, n.name) for n in nodes] == [ + ("leaf", "#MonitorACLFile"), + ("leaf", "@ClientCount"), + ("leaf", "@Clients"), + ("branch", "Simulation Items"), + ("branch", "Configured Aliases"), + ] + # depth 1: the walk stops at the branches themselves + assert all(n.children == () for n in nodes if n.kind == "branch") -def test_list_devices_depth_two_includes_sub_devices(browser: Browser) -> None: - space = list_devices(browser, depth=2) - paths = [d.path for d in space.devices] - assert ("Simulation Items", "Bucket Brigade") in paths - assert ("Simulation Items", "Random") in paths - # depth 2: the root's direct children plus their direct children - assert len(paths) == 2 + 8 # 2 top-level + 8 under Simulation Items +def test_walk_tree_depth_one_includes_root_leaves(browser: Browser) -> None: + """Root-level leaves (belonging to no branch) are ordinary tree items.""" + nodes = walk_tree(browser) + root_leaves = [n for n in nodes if n.kind == "leaf"] + assert {n.name for n in root_leaves} == {"@Clients", "@ClientCount", "#MonitorACLFile"} + assert all(n.path == () for n in root_leaves) # they hang directly at the root -def test_list_devices_reports_tag_counts(browser: Browser) -> None: - space = list_devices(browser) - by_name = {d.name: d for d in space.devices} - assert by_name["Simulation Items"].tag_names == () # no direct leaves - assert by_name["Configured Aliases"].tag_names == ("Alias1", "Alias2") - assert by_name["Simulation Items"].sub_devices[:4] == ( +def test_walk_tree_depth_two(browser: Browser) -> None: + nodes = walk_tree(browser, depth=2) + sim = next(n for n in nodes if n.name == "Simulation Items") + child_names = [c.name for c in sim.children] + assert child_names == [ "Bucket Brigade", "Random", "Read Error", "Saw-toothed Waves", - ) + "Square Waves", "Triangle Waves", "Write Error", "Write Only", + ] + # depth 2 stops inside those sub-branches too + bucket = next(c for c in sim.children if c.name == "Bucket Brigade") + assert bucket.kind == "branch" and bucket.children == () + assert len(flatten_leaves(nodes)) == 5 # root leaves only; nothing deeper read -def test_list_devices_no_counts_skips_leaves(browser: Browser) -> None: - space = list_devices(browser, include_tag_counts=False) - assert space.root.tag_names == () - assert all(d.tag_names == () for d in space.devices) +def test_walk_tree_whole_subtree(browser: Browser) -> None: + """depth=0: everything defined in the tree, in structure.""" + nodes = walk_tree(browser, depth=WHOLE_SUBTREE) + tags = flatten_leaves(nodes) + item_ids = {t.item_id for t in tags} + assert "Bucket Brigade.Int1" in item_ids # device tags + assert "Configured Aliases.Alias1" in item_ids # alias tags + assert "@ClientCount" in item_ids # root-level, no branch + # the structure is preserved: Bucket Brigade is nested under Simulation Items + sim = next(n for n in nodes if n.name == "Simulation Items") + bucket = next(c for c in sim.children if c.name == "Bucket Brigade") + assert bucket.path == ("Simulation Items", "Bucket Brigade") + assert any(c.name == "Int1" and c.item_id == "Bucket Brigade.Int1" for c in bucket.children) -@pytest.mark.parametrize("depth", [0, -1, MAX_DEPTH + 1]) -def test_list_devices_rejects_bad_depth(browser: Browser, depth: int) -> None: - with pytest.raises(ValueError): - list_devices(browser, depth=depth) +def test_walk_tree_from_subtree(browser: Browser) -> None: + """--branch limits the walk to that subtree.""" + nodes = walk_tree(browser, branch="Simulation Items.Bucket Brigade", depth=WHOLE_SUBTREE) + # only the subtree: no root leaves, no other branches + kinds = [(n.kind, n.name) for n in nodes] + assert all(n.kind == "leaf" for n in nodes) + assert {n.name for n in nodes} == { + "Int1", "Int2", "Int4", "Real4", "Real8", "String", "Boolean", + } -def test_list_devices_cycle_guard(monkeypatch: pytest.MonkeyPatch) -> None: - """A pathological self-nesting branch must not hang the recursive walk.""" +def test_walk_tree_missing_branch_is_actionable(browser: Browser) -> None: + with pytest.raises(OpcError, match="list-items"): + walk_tree(browser, branch="Simulation Items.No Such Device") + with pytest.raises(OpcError, match="not found under"): + walk_tree(browser, branch="Nope") + + +@pytest.mark.parametrize("depth", [-1, -5]) +def test_walk_tree_rejects_negative_depth(browser: Browser, depth: int) -> None: + with pytest.raises(ValueError, match=">= 0"): + walk_tree(browser, depth=depth) + + +def test_walk_tree_rejects_bad_max_depth(browser: Browser) -> None: + with pytest.raises(ValueError, match="max_depth"): + walk_tree(browser, max_depth=0) + with pytest.raises(ValueError, match="max_depth"): + walk_tree(browser, depth=33, max_depth=32) + + +def test_walk_tree_configurable_max_depth_guards_cycles() -> None: + """A pathological self-nesting branch must hit the max_depth guard, not hang.""" class _LoopBrowser: @property @@ -119,53 +165,38 @@ def test_list_devices_cycle_guard(monkeypatch: pytest.MonkeyPatch) -> None: def close(self) -> None: return None - with pytest.raises(OpcError, match="deeper than"): - list_tags(_LoopBrowser(), recursive=True) # type: ignore[arg-type] + with pytest.raises(OpcError, match="deeper than 2"): + walk_tree(_LoopBrowser(), depth=WHOLE_SUBTREE, max_depth=2) # type: ignore[arg-type] + + +def test_walk_tree_default_max_depth_constant() -> None: + """The default guard matches the documented config default (32).""" + assert DEFAULT_MAX_DEPTH == 32 # --------------------------------------------------------------------------- # -# list_tags +# flatten helpers # --------------------------------------------------------------------------- # -def test_list_tags_root_direct(browser: Browser) -> None: - tags = list_tags(browser) - assert [t.name for t in tags] == ["#MonitorACLFile", "@ClientCount", "@Clients"] - assert all(t.item_id == t.name for t in tags) # root leaves: bare ItemIDs - assert all(t.path == () for t in tags) +def test_flatten_leaves_sorted_case_insensitively(browser: Browser) -> None: + tags = flatten_leaves(walk_tree(browser, depth=WHOLE_SUBTREE)) + ids = [t.item_id for t in tags] + assert ids == sorted(ids, key=str.lower) -def test_list_tags_of_device_dotted_string(browser: Browser) -> None: - tags = list_tags(browser, device="Simulation Items.Bucket Brigade") - item_ids = [t.item_id for t in tags] - assert "Bucket Brigade.Int1" in item_ids - assert "Bucket Brigade.Real8" in item_ids - assert len(tags) == 7 +def test_flatten_branches_depth_first(browser: Browser) -> None: + branches = flatten_branches(walk_tree(browser, depth=WHOLE_SUBTREE)) + paths = [".".join(b.path) for b in branches] + # depth-first: parents before their children + assert paths.index("Simulation Items") < paths.index("Simulation Items.Bucket Brigade") + assert paths.index("Simulation Items") < paths.index("Configured Aliases") -def test_list_tags_of_device_segments(browser: Browser) -> None: - tags = list_tags(browser, device=["Simulation Items", "Random"]) - assert {t.name for t in tags} == {"Int1", "Int4", "Money", "Time"} - assert all(t.path == ("Simulation Items", "Random") for t in tags) - - -def test_list_tags_recursive(browser: Browser) -> None: - tags = list_tags(browser, device="Simulation Items", recursive=True) - item_ids = {t.item_id for t in tags} - # direct children of "Simulation Items" have no leaves; recursion gathers all - assert "Bucket Brigade.Int1" in item_ids - assert "Random.Money" in item_ids - assert "Read Error.Int4" in item_ids - assert "Saw-toothed Waves.Real8" in item_ids - assert all(t.path for t in tags) - - - -def test_tags_are_sorted_case_insensitively(browser: Browser) -> None: - tags = list_tags(browser, device="Simulation Items.Bucket Brigade") - assert [t.item_id for t in tags] == sorted( - (t.item_id for t in tags), key=lambda s: s.lower() - ) +def test_spacenode_is_frozen() -> None: + node = SpaceNode(kind="leaf", name="X", path=(), item_id="X") + with pytest.raises(AttributeError): + node.name = "Y" # type: ignore[misc] # --------------------------------------------------------------------------- # @@ -174,7 +205,6 @@ def test_tags_are_sorted_case_insensitively(browser: Browser) -> None: def test_fake_browser_matches_protocol(browser: Browser) -> None: - # runtime_checkable protocol: structural check assert isinstance(browser, Browser) @@ -199,17 +229,24 @@ def test_fake_flat_space_reports_flat_organization() -> None: client.connect() browser = client.create_browser() assert browser.organization == "flat" - space = list_devices(browser) - assert space.devices == () - assert space.root.tag_names == ("Tag1", "Tag2") + nodes = walk_tree(browser) + assert all(n.kind == "leaf" for n in nodes) + assert {n.name for n in nodes} == {"Tag1", "Tag2"} + + +def test_require_branch_root_ok_and_missing(browser: Browser) -> None: + assert require_branch(browser, None, what="list-items") == () + assert require_branch(browser, "Simulation Items", what="list-items") == ("Simulation Items",) + with pytest.raises(OpcError, match="list-items"): + require_branch(browser, "Missing", what="list-items") # --------------------------------------------------------------------------- # -# CLI verbs (against the fake backend through a config file) +# CLI verb (against the fake backend through a config file) # --------------------------------------------------------------------------- # -def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_path, args: list[str]) -> int: +def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, args: list[str]) -> int: """Run the CLI with the fake client injected in place of the COM adapter.""" import opcda_diagnozer.main as main_mod from opcda_diagnozer.opc.fake import FakeOpcClient @@ -229,69 +266,117 @@ def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_path, args: list[str]) -> int: return cli_main(args) -def test_cli_list_devices_table(monkeypatch: pytest.MonkeyPatch, tmp_path, capsys) -> None: - code = _run_cli(monkeypatch, tmp_path, ["list-devices", "-c", str(tmp_path / "config.yaml")]) - out = capsys.readouterr().out - assert code == 0 - assert "Simulation Items" in out - assert "Configured Aliases" in out - assert "2 tags" in out - - -def test_cli_list_devices_json(monkeypatch: pytest.MonkeyPatch, tmp_path, capsys) -> None: +def test_cli_list_items_all_tree(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None: code = _run_cli(monkeypatch, tmp_path, [ - "list-devices", "-c", str(tmp_path / "config.yaml"), "--json", + "list-items", "all", "--depth", "0", "-c", str(tmp_path / "config.yaml"), ]) out = capsys.readouterr().out assert code == 0 - import json - - lines = [json.loads(line) for line in out.splitlines() if line.strip()] - assert {d["name"] for d in lines} == {"Simulation Items", "Configured Aliases"} - assert lines[0]["qualified_name"] == "Simulation Items" # walk order = root order + # tree structure: root leaves visible, branches with trailing / + assert "@ClientCount [ItemID: @ClientCount]" in out + assert "Simulation Items/" in out + assert " Bucket Brigade/" in out # nested one level + assert " Int1 [ItemID: Bucket Brigade.Int1]" in out # nested two levels -def test_cli_list_tags_dotted(monkeypatch: pytest.MonkeyPatch, tmp_path, capsys) -> None: +def test_cli_list_items_default_is_all_depth1( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys +) -> None: + code = _run_cli(monkeypatch, tmp_path, ["list-items", "-c", str(tmp_path / "config.yaml")]) + out = capsys.readouterr().out + assert code == 0 + assert "Simulation Items/" in out + assert "Bucket Brigade" not in out # depth 1 stops at the top branches + + +def test_cli_list_items_branches(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None: code = _run_cli(monkeypatch, tmp_path, [ - "list-tags", "-c", str(tmp_path / "config.yaml"), - "Simulation Items.Bucket Brigade", + "list-items", "branches", "--depth", "0", "-c", str(tmp_path / "config.yaml"), + ]) + out = capsys.readouterr().out + assert code == 0 + assert "Simulation Items" in out + assert "Simulation Items.Bucket Brigade" in out + assert "8 sub-branches" in out + assert "[ItemID" not in out # no tags in the branches view + + +def test_cli_list_items_tags_flat(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None: + code = _run_cli(monkeypatch, tmp_path, [ + "list-items", "tags", "--depth", "0", "-c", str(tmp_path / "config.yaml"), ]) out = capsys.readouterr().out assert code == 0 assert "Bucket Brigade.Int1" in out - assert "Bucket Brigade.Real8" in out + assert "Configured Aliases.Alias1" in out + assert "@ClientCount" in out # root-level tags are tags too -def test_cli_list_tags_segments_and_recursive( - monkeypatch: pytest.MonkeyPatch, tmp_path, capsys +def test_cli_list_items_branch_subtree( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys ) -> None: code = _run_cli(monkeypatch, tmp_path, [ - "list-tags", "-c", str(tmp_path / "config.yaml"), - "Simulation Items", "--recursive", "--json", + "list-items", "tags", "--branch", "Simulation Items.Bucket Brigade", + "-c", str(tmp_path / "config.yaml"), ]) out = capsys.readouterr().out assert code == 0 - import json - - item_ids = {json.loads(line)["item_id"] for line in out.splitlines() if line.strip()} - assert "Bucket Brigade.Int1" in item_ids - assert "Random.Money" in item_ids + assert "Bucket Brigade.Int1" in out + assert "Random.Money" not in out # only the selected subtree + assert "Configured Aliases" not in out -def test_cli_list_tags_missing_device_fails_cleanly( - monkeypatch: pytest.MonkeyPatch, tmp_path, capsys +def test_cli_list_items_branch_repeated_segments( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys +) -> None: + """Repeating --branch gives explicit segments (names may contain dots).""" + code = _run_cli(monkeypatch, tmp_path, [ + "list-items", "tags", "--branch", "Simulation Items", "--branch", "Bucket Brigade", + "-c", str(tmp_path / "config.yaml"), + ]) + out = capsys.readouterr().out + assert code == 0 + assert "Bucket Brigade.Int1" in out + + +def test_cli_list_items_json_tree(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None: + code = _run_cli(monkeypatch, tmp_path, [ + "list-items", "--json", "--depth", "0", "-c", str(tmp_path / "config.yaml"), + ]) + out = capsys.readouterr().out + assert code == 0 + lines = [json.loads(line) for line in out.splitlines() if line.strip()] + assert lines[0]["kind"] == "leaf" + sim = next(l for l in lines if l["name"] == "Simulation Items") + assert sim["kind"] == "branch" + assert any(c["name"] == "Bucket Brigade" for c in sim["children"]) + + +def test_cli_list_items_missing_branch_fails_cleanly( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys ) -> None: code = _run_cli(monkeypatch, tmp_path, [ - "list-tags", "-c", str(tmp_path / "config.yaml"), "No.Such.Device", + "list-items", "--branch", "No.Such.Branch", "-c", str(tmp_path / "config.yaml"), ]) captured = capsys.readouterr() assert code == 2 assert "FAILED" in captured.err +def test_cli_old_verbs_are_gone( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys +) -> None: + """list-devices / list-tags were replaced by list-items.""" + for gone in ("list-devices", "list-tags"): + with pytest.raises(SystemExit) as exc_info: + _run_cli(monkeypatch, tmp_path, [gone, "-c", str(tmp_path / "config.yaml")]) + assert exc_info.value.code == 2 # argparse: unknown verb + assert "invalid choice" in capsys.readouterr().err + + @pytest.mark.skipif(sys.platform != "win32", reason="COM browse needs Windows + pywin32") def test_live_automation_browse_smoke() -> None: - """Live smoke test — requires the Matrikon simulation server on this dev box.""" + """Live smoke test — requires a registered Automation Wrapper (dev box).""" pytest.importorskip("pythoncom") from opcda_diagnozer.opc.automation import AutomationOpcClient @@ -299,11 +384,13 @@ def test_live_automation_browse_smoke() -> None: try: client.connect() browser = client.create_browser() - devices = list_devices(browser) - assert devices.organization == "hierarchical" - assert any(d.name == "Simulation Items" for d in devices.devices) - tags = list_tags(browser, device="Simulation Items.Bucket Brigade") + nodes = walk_tree(browser, depth=WHOLE_SUBTREE) + tags = flatten_leaves(nodes) assert any(t.item_id == "Bucket Brigade.Int1" for t in tags) + assert any(t.name == "@ClientCount" for t in tags) # root-level leaf + sim = next(n for n in nodes if n.name == "Simulation Items") + assert sim.kind == "branch" + assert any(c.name == "Bucket Brigade" for c in sim.children) except OpcError as exc: pytest.skip(f"simulation server not usable on this machine: {exc}") finally: