🎉 Initial commit

2026-10-07 10:38:21 +02:00
commit 1c64aeb071
38 changed files with 3391 additions and 0 deletions
@@ -0,0 +1,310 @@
+"""Browse tests (opc/browse.py + fake browser + CLI verbs).
+
+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
+§9: COM adapters are import-guarded).
+"""
+
+import sys
+
+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.browse import (
+    MAX_DEPTH,
+    list_devices,
+    list_tags,
+    split_device_path,
+    tags_for_device,
+)
+from opcda_diagnozer.opc.fake import FAKE_SPACE, FakeOpcClient
+
+
+@pytest.fixture
+def browser() -> Browser:
+    client = FakeOpcClient()
+    client.connect()
+    return client.create_browser()
+
+
+# --------------------------------------------------------------------------- #
+# split_device_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_device_path(raw, expected: tuple[str, ...]) -> None:
+    assert split_device_path(raw) == expected
+
+
+# --------------------------------------------------------------------------- #
+# list_devices
+# --------------------------------------------------------------------------- #
+
+
+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_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_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] == (
+        "Bucket Brigade", "Random", "Read Error", "Saw-toothed Waves",
+    )
+
+
+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)
+
+
+@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_list_devices_cycle_guard(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A pathological self-nesting branch must not hang the recursive walk."""
+
+    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"):
+        list_tags(_LoopBrowser(), recursive=True)  # type: ignore[arg-type]
+
+
+# --------------------------------------------------------------------------- #
+# list_tags
+# --------------------------------------------------------------------------- #
+
+
+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_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_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()
+    )
+
+
+# --------------------------------------------------------------------------- #
+# fake browser & client protocol conformance
+# --------------------------------------------------------------------------- #
+
+
+def test_fake_browser_matches_protocol(browser: Browser) -> None:
+    # runtime_checkable protocol: structural check
+    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"
+    space = list_devices(browser)
+    assert space.devices == ()
+    assert space.root.tag_names == ("Tag1", "Tag2")
+
+
+# --------------------------------------------------------------------------- #
+# CLI verbs (against the fake backend through a config file)
+# --------------------------------------------------------------------------- #
+
+
+def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_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):
+        client.connect()
+        return client
+
+    monkeypatch.setattr(main_mod, "_connect_from_config", _fake_connect)
+    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:
+    code = _run_cli(monkeypatch, tmp_path, [
+        "list-devices", "-c", str(tmp_path / "config.yaml"), "--json",
+    ])
+    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
+
+
+def test_cli_list_tags_dotted(monkeypatch: pytest.MonkeyPatch, tmp_path, capsys) -> None:
+    code = _run_cli(monkeypatch, tmp_path, [
+        "list-tags", "-c", str(tmp_path / "config.yaml"),
+        "Simulation Items.Bucket Brigade",
+    ])
+    out = capsys.readouterr().out
+    assert code == 0
+    assert "Bucket Brigade.Int1" in out
+    assert "Bucket Brigade.Real8" in out
+
+
+def test_cli_list_tags_segments_and_recursive(
+    monkeypatch: pytest.MonkeyPatch, tmp_path, capsys
+) -> None:
+    code = _run_cli(monkeypatch, tmp_path, [
+        "list-tags", "-c", str(tmp_path / "config.yaml"),
+        "Simulation Items", "--recursive", "--json",
+    ])
+    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
+
+
+def test_cli_list_tags_missing_device_fails_cleanly(
+    monkeypatch: pytest.MonkeyPatch, tmp_path, capsys
+) -> None:
+    code = _run_cli(monkeypatch, tmp_path, [
+        "list-tags", "-c", str(tmp_path / "config.yaml"), "No.Such.Device",
+    ])
+    captured = capsys.readouterr()
+    assert code == 2
+    assert "FAILED" in captured.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."""
+    pytest.importorskip("pythoncom")
+    from opcda_diagnozer.opc.automation import AutomationOpcClient
+
+    client = AutomationOpcClient(server="Matrikon.OPC.Simulation.1")
+    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")
+        assert any(t.item_id == "Bucket Brigade.Int1" for t in tags)
+    except OpcError as exc:
+        pytest.skip(f"simulation server not usable on this machine: {exc}")
+    finally:
+        client.close()