"""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, 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_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()