Joseph Doherty
0e4843612b
Add --parent-gobject-id (integer) to the galaxy-browse CLI command so the Python client matches the Go (-parent) and Rust (--parent-gobject-id) CLIs. When set, drives BrowseChildren paging via browse_children_raw (page size 500, repeated-token guard) and renders the same JSON node shape (flattened object fields + hasChildrenHint + empty children array) and indented-text tree as the root-walk path. --depth is ignored on the parent path with a one-line stderr warning, matching the Go/Rust behaviour. Tests added in TDD order.
648 lines
21 KiB
Python
648 lines
21 KiB
Python
"""Tests for the Python CLI."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
from zb_mom_ww_mxgateway import __version__
|
|
from zb_mom_ww_mxgateway_cli import commands as commands_module
|
|
from zb_mom_ww_mxgateway_cli.commands import main
|
|
|
|
_BATCH_EOR = "__MXGW_BATCH_EOR__"
|
|
|
|
|
|
def test_require_certificate_validation_flag_flows_through_connect(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""The --require-certificate-validation CLI flag must reach ClientOptions (Client.Python-027)."""
|
|
captured: dict[str, object] = {}
|
|
|
|
async def fake_connect(options, **_kwargs):
|
|
captured["options"] = options
|
|
# Return a minimal object that supports the async context-manager protocol
|
|
# used by every CLI command body (async with await _connect(...) as client).
|
|
return _FakeAsyncClient()
|
|
|
|
monkeypatch.setattr(commands_module.GatewayClient, "connect", fake_connect)
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
[
|
|
"open-session",
|
|
"--endpoint",
|
|
"gateway.example:5001",
|
|
"--require-certificate-validation",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["options"].require_certificate_validation is True
|
|
|
|
|
|
def test_require_certificate_validation_defaults_off(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Without the flag the strict-validation posture stays off (TOFU default)."""
|
|
captured: dict[str, object] = {}
|
|
|
|
async def fake_connect(options, **_kwargs):
|
|
captured["options"] = options
|
|
return _FakeAsyncClient()
|
|
|
|
monkeypatch.setattr(commands_module.GatewayClient, "connect", fake_connect)
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["open-session", "--endpoint", "gateway.example:5001", "--plaintext", "--json"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["options"].require_certificate_validation is False
|
|
|
|
|
|
class _FakeAsyncClient:
|
|
"""Minimal async-context-manager fake satisfying the open-session command body."""
|
|
|
|
async def __aenter__(self) -> "_FakeAsyncClient":
|
|
return self
|
|
|
|
async def __aexit__(self, *_exc: object) -> None:
|
|
return None
|
|
|
|
async def open_session_raw(self, *_args, **_kwargs):
|
|
from zb_mom_ww_mxgateway.generated import mxaccess_gateway_pb2 as pb
|
|
|
|
return pb.OpenSessionReply(session_id="cli-test-session")
|
|
|
|
|
|
def test_version_json_is_deterministic() -> None:
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(main, ["version", "--json"])
|
|
|
|
assert result.exit_code == 0
|
|
assert json.loads(result.output) == {
|
|
"client": "mxgw-py",
|
|
"package": "mxaccess-gateway-client",
|
|
"version": __version__,
|
|
}
|
|
|
|
|
|
def test_write_parser_rejects_unknown_value_type() -> None:
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(
|
|
main,
|
|
[
|
|
"write",
|
|
"--session-id",
|
|
"session-1",
|
|
"--server-handle",
|
|
"12",
|
|
"--item-handle",
|
|
"34",
|
|
"--type",
|
|
"unsupported",
|
|
"--value",
|
|
"123",
|
|
"--api-key",
|
|
"mxgw_test_secret",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "unsupported value type" in result.output
|
|
|
|
|
|
def test_stream_alarms_is_registered() -> None:
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(main, ["stream-alarms", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "--filter-prefix" in result.output
|
|
assert "--max-messages" in result.output
|
|
|
|
|
|
def test_acknowledge_alarm_requires_reference() -> None:
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(
|
|
main,
|
|
["acknowledge-alarm", "--api-key", "mxgw_test_secret", "--json"],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "--reference" in result.output
|
|
|
|
|
|
def test_cli_error_output_redacts_api_key() -> None:
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(
|
|
main,
|
|
[
|
|
"open-session",
|
|
"--endpoint",
|
|
"127.0.0.1:1",
|
|
"--api-key",
|
|
"mxgw_test_secret",
|
|
"--plaintext",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "mxgw_test_secret" not in result.output
|
|
|
|
|
|
def test_batch_runs_version_command_and_writes_eor() -> None:
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(main, ["batch"], input="version --json\n")
|
|
|
|
assert result.exit_code == 0
|
|
blocks = [block for block in result.output.split(_BATCH_EOR + "\n") if block]
|
|
assert len(blocks) == 1
|
|
payload = json.loads(blocks[0].strip())
|
|
assert payload == {
|
|
"client": "mxgw-py",
|
|
"package": "mxaccess-gateway-client",
|
|
"version": __version__,
|
|
}
|
|
|
|
|
|
def test_batch_terminates_on_empty_line() -> None:
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(
|
|
main,
|
|
["batch"],
|
|
input="version --json\n\nversion --json\n",
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
# Only the first command runs; the empty line breaks the loop before the second.
|
|
assert result.output.count(_BATCH_EOR) == 1
|
|
|
|
|
|
def test_batch_continues_after_error_line() -> None:
|
|
runner = CliRunner()
|
|
|
|
# First line is invalid (unknown subcommand), second is a valid version call.
|
|
result = runner.invoke(
|
|
main,
|
|
["batch"],
|
|
input="not-a-real-command\nversion --json\n",
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert result.output.count(_BATCH_EOR) == 2
|
|
|
|
blocks = [block for block in result.output.split(_BATCH_EOR + "\n") if block]
|
|
assert len(blocks) == 2
|
|
|
|
# First block: error JSON ({"error": "...", "type": "..."}).
|
|
error_payload = json.loads(blocks[0].strip().splitlines()[-1])
|
|
assert "error" in error_payload
|
|
assert "type" in error_payload
|
|
|
|
# Second block: successful version JSON.
|
|
version_payload = json.loads(blocks[1].strip())
|
|
assert version_payload["version"] == __version__
|
|
|
|
|
|
class _FakeGalaxyClient:
|
|
"""Minimal async-context-manager fake satisfying the galaxy command bodies."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
ok: bool = True,
|
|
objects=None,
|
|
last_deploy=None,
|
|
events=None,
|
|
browse_roots=None,
|
|
browse_children_pages=None,
|
|
) -> None:
|
|
self._ok = ok
|
|
self._objects = objects or []
|
|
self._last_deploy = last_deploy
|
|
self._events = events or []
|
|
self._browse_roots = browse_roots or []
|
|
# List of BrowseChildrenReply-like objects to serve in order (paged).
|
|
self._browse_children_pages = browse_children_pages or []
|
|
self._browse_children_calls: list = []
|
|
self.browse_options = None
|
|
|
|
async def __aenter__(self) -> "_FakeGalaxyClient":
|
|
return self
|
|
|
|
async def __aexit__(self, *_exc: object) -> None:
|
|
return None
|
|
|
|
async def test_connection(self) -> bool:
|
|
return self._ok
|
|
|
|
async def discover_hierarchy(self):
|
|
return self._objects
|
|
|
|
async def browse(self, options=None):
|
|
self.browse_options = options
|
|
return self._browse_roots
|
|
|
|
async def browse_children_raw(self, request):
|
|
"""Return the next queued BrowseChildrenReply page; raises if queue empty."""
|
|
self._browse_children_calls.append(request)
|
|
if not self._browse_children_pages:
|
|
raise AssertionError("browse_children_raw called but no pages queued")
|
|
return self._browse_children_pages.pop(0)
|
|
|
|
async def get_last_deploy_time(self):
|
|
# Mirrors galaxy.py: protobuf ToDatetime() yields a timezone-NAIVE UTC datetime.
|
|
return self._last_deploy
|
|
|
|
def watch_deploy_events(self, _last_seen_deploy_time=None):
|
|
events = self._events
|
|
|
|
async def _iter():
|
|
for event in events:
|
|
yield event
|
|
|
|
return _iter()
|
|
|
|
|
|
def _patch_galaxy_connect(monkeypatch: pytest.MonkeyPatch, fake: _FakeGalaxyClient) -> None:
|
|
async def fake_connect(options, **_kwargs):
|
|
return fake
|
|
|
|
monkeypatch.setattr(commands_module.GalaxyRepositoryClient, "connect", fake_connect)
|
|
|
|
|
|
def test_galaxy_test_connection_emits_ok(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_patch_galaxy_connect(monkeypatch, _FakeGalaxyClient(ok=True))
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-test-connection", "--plaintext", "--json"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
payload = json.loads(result.output)
|
|
assert payload == {"command": "galaxy-test-connection", "ok": True}
|
|
|
|
|
|
def test_galaxy_discover_serializes_objects(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from zb_mom_ww_mxgateway.generated import galaxy_repository_pb2 as galaxy_pb
|
|
|
|
objects = [
|
|
galaxy_pb.GalaxyObject(gobject_id=7, tag_name="Area001", contained_name="Area001"),
|
|
galaxy_pb.GalaxyObject(gobject_id=8, tag_name="Pump001", contained_name="Pump001"),
|
|
]
|
|
_patch_galaxy_connect(monkeypatch, _FakeGalaxyClient(objects=objects))
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-discover", "--plaintext", "--json"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
payload = json.loads(result.output)
|
|
assert payload["command"] == "galaxy-discover"
|
|
assert len(payload["objects"]) == 2
|
|
assert payload["objects"][0]["tagName"] == "Area001"
|
|
assert payload["objects"][1]["gobjectId"] == 8
|
|
|
|
|
|
def test_galaxy_last_deploy_emits_utc_iso(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""The naive-UTC deploy time from the library must be emitted as unambiguous UTC ISO-8601."""
|
|
from datetime import datetime
|
|
|
|
naive_utc = datetime(2025, 6, 15, 12, 0, 0) # noqa: DTZ001 -- mirrors protobuf ToDatetime()
|
|
_patch_galaxy_connect(monkeypatch, _FakeGalaxyClient(last_deploy=naive_utc))
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-last-deploy", "--plaintext", "--json"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
payload = json.loads(result.output)
|
|
assert payload["command"] == "galaxy-last-deploy"
|
|
assert payload["present"] is True
|
|
assert payload["timeOfLastDeploy"].endswith(("+00:00", "Z"))
|
|
|
|
|
|
def test_galaxy_watch_serializes_deploy_events(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from zb_mom_ww_mxgateway.generated import galaxy_repository_pb2 as galaxy_pb
|
|
|
|
events = [galaxy_pb.DeployEvent(sequence=1)]
|
|
_patch_galaxy_connect(monkeypatch, _FakeGalaxyClient(events=events))
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-watch", "--plaintext", "--max-events", "1", "--json"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
payload = json.loads(result.output)
|
|
assert payload["command"] == "galaxy-watch"
|
|
assert len(payload["events"]) == 1
|
|
|
|
|
|
class _FakeBrowseNode:
|
|
"""Minimal stand-in for LazyBrowseNode covering the CLI render path."""
|
|
|
|
def __init__(self, obj, *, has_children_hint=False, children=None) -> None:
|
|
self._object = obj
|
|
self._has_children_hint = has_children_hint
|
|
self._children = list(children or [])
|
|
self._is_expanded = bool(children)
|
|
self.expand_calls = 0
|
|
|
|
@property
|
|
def object(self):
|
|
return self._object
|
|
|
|
@property
|
|
def has_children_hint(self) -> bool:
|
|
return self._has_children_hint
|
|
|
|
@property
|
|
def children(self):
|
|
return list(self._children)
|
|
|
|
@property
|
|
def is_expanded(self) -> bool:
|
|
return self._is_expanded
|
|
|
|
async def expand(self) -> None:
|
|
self.expand_calls += 1
|
|
self._is_expanded = True
|
|
|
|
|
|
def test_galaxy_browse_serializes_nested_nodes(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from zb_mom_ww_mxgateway.generated import galaxy_repository_pb2 as galaxy_pb
|
|
|
|
child = _FakeBrowseNode(
|
|
galaxy_pb.GalaxyObject(gobject_id=8, tag_name="Pump001", contained_name="Pump001"),
|
|
has_children_hint=False,
|
|
)
|
|
root = _FakeBrowseNode(
|
|
galaxy_pb.GalaxyObject(gobject_id=7, tag_name="Area001", contained_name="Area001"),
|
|
has_children_hint=True,
|
|
children=[child],
|
|
)
|
|
_patch_galaxy_connect(monkeypatch, _FakeGalaxyClient(browse_roots=[root]))
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-browse", "--plaintext", "--json"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
payload = json.loads(result.output)
|
|
assert "_text" not in payload
|
|
assert payload["command"] == "galaxy-browse"
|
|
assert len(payload["nodes"]) == 1
|
|
node = payload["nodes"][0]
|
|
assert node["tagName"] == "Area001"
|
|
assert node["hasChildrenHint"] is True
|
|
assert len(node["children"]) == 1
|
|
assert node["children"][0]["gobjectId"] == 8
|
|
assert node["children"][0]["children"] == []
|
|
|
|
|
|
def test_galaxy_browse_renders_indented_text_tree(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from zb_mom_ww_mxgateway.generated import galaxy_repository_pb2 as galaxy_pb
|
|
|
|
child = _FakeBrowseNode(
|
|
galaxy_pb.GalaxyObject(gobject_id=8, tag_name="Pump001", browse_name="Pump001"),
|
|
)
|
|
root = _FakeBrowseNode(
|
|
galaxy_pb.GalaxyObject(gobject_id=7, tag_name="Area001", browse_name="Area001"),
|
|
has_children_hint=True,
|
|
children=[child],
|
|
)
|
|
_patch_galaxy_connect(monkeypatch, _FakeGalaxyClient(browse_roots=[root]))
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-browse", "--plaintext"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
lines = result.output.splitlines()
|
|
assert lines[0] == "1"
|
|
assert lines[1] == "+ Area001 Area001 (gobject 7)"
|
|
assert lines[2] == " - Pump001 Pump001 (gobject 8)"
|
|
|
|
|
|
def test_galaxy_browse_forwards_filter_options(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
fake = _FakeGalaxyClient(browse_roots=[])
|
|
_patch_galaxy_connect(monkeypatch, fake)
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
[
|
|
"galaxy-browse",
|
|
"--plaintext",
|
|
"--category-id",
|
|
"10",
|
|
"--category-id",
|
|
"12",
|
|
"--template-chain-contains",
|
|
"$Pump",
|
|
"--tag-name-glob",
|
|
"Area*",
|
|
"--include-attributes",
|
|
"--alarm-bearing-only",
|
|
"--historized-only",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
options = fake.browse_options
|
|
assert tuple(options.category_ids) == (10, 12)
|
|
assert tuple(options.template_chain_contains) == ("$Pump",)
|
|
assert options.tag_name_glob == "Area*"
|
|
assert options.include_attributes is True
|
|
assert options.alarm_bearing_only is True
|
|
assert options.historized_only is True
|
|
|
|
|
|
def test_galaxy_browse_expands_to_depth(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from zb_mom_ww_mxgateway.generated import galaxy_repository_pb2 as galaxy_pb
|
|
|
|
root = _FakeBrowseNode(
|
|
galaxy_pb.GalaxyObject(gobject_id=7, tag_name="Area001"),
|
|
has_children_hint=True,
|
|
)
|
|
_patch_galaxy_connect(monkeypatch, _FakeGalaxyClient(browse_roots=[root]))
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-browse", "--plaintext", "--depth", "2", "--json"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert root.expand_calls == 1
|
|
|
|
|
|
def test_galaxy_commands_are_registered() -> None:
|
|
runner = CliRunner()
|
|
for command in (
|
|
"galaxy-test-connection",
|
|
"galaxy-last-deploy",
|
|
"galaxy-discover",
|
|
"galaxy-watch",
|
|
"galaxy-browse",
|
|
):
|
|
result = runner.invoke(main, [command, "--help"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "--endpoint" in result.output
|
|
|
|
|
|
@pytest.mark.parametrize("depth_arg", ["99", "-1"])
|
|
def test_galaxy_browse_rejects_out_of_range_depth(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
depth_arg: str,
|
|
) -> None:
|
|
"""--depth values outside [0, 50] must be rejected with a non-zero exit."""
|
|
_patch_galaxy_connect(monkeypatch, _FakeGalaxyClient(browse_roots=[]))
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-browse", "--plaintext", "--depth", depth_arg, "--json"],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "--depth must be between 0 and 50" in result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# --parent-gobject-id drill-down tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _fake_browse_children_reply(children_and_hints, *, next_page_token=""):
|
|
"""Build a minimal fake BrowseChildrenReply-like object."""
|
|
from zb_mom_ww_mxgateway.generated import galaxy_repository_pb2 as galaxy_pb
|
|
|
|
reply = galaxy_pb.BrowseChildrenReply()
|
|
for obj, hint in children_and_hints:
|
|
reply.children.append(obj)
|
|
reply.child_has_children.append(hint)
|
|
reply.next_page_token = next_page_token
|
|
return reply
|
|
|
|
|
|
def test_galaxy_browse_parent_fetches_one_level_json(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""--parent-gobject-id N calls browse_children_raw and renders one-level JSON."""
|
|
from zb_mom_ww_mxgateway.generated import galaxy_repository_pb2 as galaxy_pb
|
|
|
|
child_a = galaxy_pb.GalaxyObject(gobject_id=10, tag_name="PumpA", browse_name="PumpA")
|
|
child_b = galaxy_pb.GalaxyObject(gobject_id=11, tag_name="PumpB", browse_name="PumpB")
|
|
page = _fake_browse_children_reply([(child_a, True), (child_b, False)])
|
|
fake = _FakeGalaxyClient(browse_children_pages=[page])
|
|
_patch_galaxy_connect(monkeypatch, fake)
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-browse", "--plaintext", "--parent-gobject-id", "7", "--json"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
payload = json.loads(result.output)
|
|
|
|
# One BrowseChildren RPC was issued with the correct parent id.
|
|
assert len(fake._browse_children_calls) == 1
|
|
call_req = fake._browse_children_calls[0]
|
|
assert call_req.parent_gobject_id == 7
|
|
|
|
# JSON shape mirrors the lazy-browse node shape.
|
|
assert payload["command"] == "galaxy-browse"
|
|
nodes = payload["nodes"]
|
|
assert len(nodes) == 2
|
|
assert nodes[0]["tagName"] == "PumpA"
|
|
assert nodes[0]["hasChildrenHint"] is True
|
|
assert nodes[0]["children"] == []
|
|
assert nodes[1]["gobjectId"] == 11
|
|
assert nodes[1]["hasChildrenHint"] is False
|
|
assert nodes[1]["children"] == []
|
|
|
|
|
|
def test_galaxy_browse_parent_renders_text_tree(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""--parent-gobject-id N text output: count line then marker lines (no indent)."""
|
|
from zb_mom_ww_mxgateway.generated import galaxy_repository_pb2 as galaxy_pb
|
|
|
|
child = galaxy_pb.GalaxyObject(gobject_id=10, tag_name="PumpA", browse_name="PumpA")
|
|
page = _fake_browse_children_reply([(child, False)])
|
|
fake = _FakeGalaxyClient(browse_children_pages=[page])
|
|
_patch_galaxy_connect(monkeypatch, fake)
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-browse", "--plaintext", "--parent-gobject-id", "7"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
lines = result.output.splitlines()
|
|
assert lines[0] == "1"
|
|
assert lines[1] == "- PumpA PumpA (gobject 10)"
|
|
|
|
|
|
def test_galaxy_browse_parent_pages_correctly(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""--parent-gobject-id loops on next_page_token until exhausted."""
|
|
from zb_mom_ww_mxgateway.generated import galaxy_repository_pb2 as galaxy_pb
|
|
|
|
child_a = galaxy_pb.GalaxyObject(gobject_id=10, tag_name="PumpA", browse_name="PumpA")
|
|
child_b = galaxy_pb.GalaxyObject(gobject_id=11, tag_name="PumpB", browse_name="PumpB")
|
|
page1 = _fake_browse_children_reply([(child_a, False)], next_page_token="tok1")
|
|
page2 = _fake_browse_children_reply([(child_b, True)])
|
|
fake = _FakeGalaxyClient(browse_children_pages=[page1, page2])
|
|
_patch_galaxy_connect(monkeypatch, fake)
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-browse", "--plaintext", "--parent-gobject-id", "7", "--json"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert len(fake._browse_children_calls) == 2
|
|
# Second call must carry the page token from the first reply.
|
|
assert fake._browse_children_calls[1].page_token == "tok1"
|
|
payload = json.loads(result.output)
|
|
assert len(payload["nodes"]) == 2
|
|
|
|
|
|
def test_galaxy_browse_parent_warns_when_depth_also_set(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""When both --parent-gobject-id and --depth>0 are supplied a warning is emitted."""
|
|
from zb_mom_ww_mxgateway.generated import galaxy_repository_pb2 as galaxy_pb
|
|
|
|
child = galaxy_pb.GalaxyObject(gobject_id=10, tag_name="PumpA", browse_name="PumpA")
|
|
page = _fake_browse_children_reply([(child, False)])
|
|
fake = _FakeGalaxyClient(browse_children_pages=[page])
|
|
_patch_galaxy_connect(monkeypatch, fake)
|
|
|
|
# CliRunner mixes stderr into output in this Click version.
|
|
result = CliRunner().invoke(
|
|
main,
|
|
["galaxy-browse", "--plaintext", "--parent-gobject-id", "7", "--depth", "2", "--json"],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert "--depth is ignored" in result.output
|
|
|
|
|
|
def test_galaxy_browse_help_shows_parent_gobject_id() -> None:
|
|
"""--parent-gobject-id appears in the galaxy-browse --help output."""
|
|
result = CliRunner().invoke(main, ["galaxy-browse", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "--parent-gobject-id" in result.output
|