397 lines
15 KiB
Python
397 lines
15 KiB
Python
"""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 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, OpcError, SpaceNode
|
|
from opcda_diagnozer.opc.browse import (
|
|
DEFAULT_MAX_DEPTH,
|
|
WHOLE_SUBTREE,
|
|
flatten_branches,
|
|
flatten_leaves,
|
|
require_branch,
|
|
split_branch_path,
|
|
walk_tree,
|
|
)
|
|
from opcda_diagnozer.opc.fake import FakeOpcClient
|
|
|
|
|
|
@pytest.fixture
|
|
def browser() -> Browser:
|
|
client = FakeOpcClient()
|
|
client.connect()
|
|
return client.create_browser()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# split_branch_path
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("raw", "expected"),
|
|
[
|
|
(None, ()),
|
|
((), ()),
|
|
("", ()),
|
|
(" ", ()),
|
|
("Simulation Items.Bucket Brigade", ("Simulation Items", "Bucket Brigade")),
|
|
("Simulation Items", ("Simulation Items",)),
|
|
("a..b", ("a", "b")), # empty segments dropped
|
|
(("Simulation Items", "Bucket Brigade"), ("Simulation Items", "Bucket Brigade")),
|
|
(("Simulation Items", " Bucket Brigade "), ("Simulation Items", "Bucket Brigade")),
|
|
(("Simulation Items", "", "Bucket Brigade"), ("Simulation Items", "Bucket Brigade")),
|
|
# explicit segments keep embedded dots intact (no split)
|
|
(("Bucket. Inc", "Device 1"), ("Bucket. Inc", "Device 1")),
|
|
],
|
|
)
|
|
def test_split_branch_path(raw, expected: tuple[str, ...]) -> None:
|
|
assert split_branch_path(raw) == expected
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# walk_tree
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
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_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_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_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)
|
|
|
|
|
|
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_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
|
|
def organization(self) -> str:
|
|
return "hierarchical"
|
|
|
|
def list_branches(self, path=()):
|
|
return ["Loop"] # "Loop" always contains another "Loop" — cycle
|
|
|
|
def list_leaves(self, path=()):
|
|
return []
|
|
|
|
def move_to(self, path):
|
|
return None
|
|
|
|
def close(self) -> None:
|
|
return None
|
|
|
|
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
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# flatten helpers
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
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_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_spacenode_is_frozen() -> None:
|
|
node = SpaceNode(kind="leaf", name="X", path=(), item_id="X")
|
|
with pytest.raises(AttributeError):
|
|
node.name = "Y" # type: ignore[misc]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# fake browser & client protocol conformance
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_fake_browser_matches_protocol(browser: Browser) -> None:
|
|
assert isinstance(browser, Browser)
|
|
|
|
|
|
def test_fake_client_browser_requires_connect() -> None:
|
|
client = FakeOpcClient()
|
|
with pytest.raises(OpcError, match="before connect"):
|
|
client.create_browser()
|
|
|
|
|
|
def test_fake_browser_unknown_branch_raises() -> None:
|
|
client = FakeOpcClient()
|
|
client.connect()
|
|
browser = client.create_browser()
|
|
with pytest.raises(OpcError, match="no such branch"):
|
|
browser.list_branches(["Simulation Items", "Missing"])
|
|
with pytest.raises(OpcError, match="no such branch"):
|
|
browser.list_leaves(["Missing"])
|
|
|
|
|
|
def test_fake_flat_space_reports_flat_organization() -> None:
|
|
client = FakeOpcClient(space={(): ((), ("Tag1", "Tag2"))})
|
|
client.connect()
|
|
browser = client.create_browser()
|
|
assert browser.organization == "flat"
|
|
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 verb (against the fake backend through a config file)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
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
|
|
|
|
config = tmp_path / "config.yaml"
|
|
config.write_text(
|
|
"opc:\n server: Fake.Server.1\ntags:\n - name: Bucket Brigade.Int1\n",
|
|
encoding="utf-8",
|
|
)
|
|
client = FakeOpcClient()
|
|
|
|
def _fake_connect(config_path: str, server=None, host=None):
|
|
client.connect()
|
|
return client
|
|
|
|
monkeypatch.setattr(main_mod, "_connect_from_config", _fake_connect)
|
|
return cli_main(args)
|
|
|
|
|
|
def test_cli_list_items_all_tree(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
|
|
code = _run_cli(monkeypatch, tmp_path, [
|
|
"list-items", "all", "--depth", "0", "-c", str(tmp_path / "config.yaml"),
|
|
])
|
|
out = capsys.readouterr().out
|
|
assert code == 0
|
|
# 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_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-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 "Configured Aliases.Alias1" in out
|
|
assert "@ClientCount" in out # root-level tags are tags too
|
|
|
|
|
|
def test_cli_list_items_branch_subtree(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
|
|
) -> None:
|
|
code = _run_cli(monkeypatch, tmp_path, [
|
|
"list-items", "tags", "--branch", "Simulation Items.Bucket Brigade",
|
|
"-c", str(tmp_path / "config.yaml"),
|
|
])
|
|
out = capsys.readouterr().out
|
|
assert code == 0
|
|
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_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-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 a registered Automation Wrapper (dev box)."""
|
|
pytest.importorskip("pythoncom")
|
|
from opcda_diagnozer.opc.automation import AutomationOpcClient
|
|
|
|
client = AutomationOpcClient(server="Matrikon.OPC.Simulation.1")
|
|
try:
|
|
client.connect()
|
|
browser = client.create_browser()
|
|
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:
|
|
client.close() |