Joseph Doherty
37cb3b0df8
All five client CLIs now share one credential contract for `authenticate-user`: flags `--password` / `--password-env` (Go: `-password` / `-password-env`) with default env `MXGATEWAY_VERIFY_PASSWORD`, resolution flag-then-env, and a resolved credential that is missing *or empty* is a usage error naming the flag and the variable. The value is never echoed and never reaches the wire. Go and Java previously sent an empty credential when the variable was unset, turning a misconfigured environment into a real MXAccess authentication attempt. Go now returns the guard error before dialing; Java throws a picocli ParameterException instead of falling back to "". Python's `--password-env` gained the canonical default and its UsageError names the resolved variable. Rust treats an empty flag or env value as missing, with the resolution extracted into a testable `resolve_verify_user_password`. .NET adopts the canonical flags and keeps `--verify-user-password`, `--verify-user-password-env`, and MXGATEWAY_VERIFY_USER_PASSWORD as deprecated aliases for one release. Docs same commit: CrossLanguageSmokeMatrix.md gains the credential contract and the per-CLI subcommand-coverage table (the documented-not-fixed half of the finding); all five READMEs name the canonical variable and the fail-fast rule, and the .NET README carries the deprecation note. Tracking flipped to Done in both remediation registers with a change-log row. No .proto changed; no generated code regenerated.
959 lines
30 KiB
Python
959 lines
30 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
|
|
|
|
|
|
class _FakeInvokeClient:
|
|
"""Async-context-manager fake whose invoke_raw returns a scripted reply.
|
|
|
|
Satisfies the session-backed CLI command bodies (register / write-secured /
|
|
authenticate-user) which build a Session over this client and call through to
|
|
``invoke_raw``. Records the last command so tests can assert credentials are
|
|
carried on the wire but never echoed to stdout.
|
|
"""
|
|
|
|
def __init__(self, reply) -> None:
|
|
self._reply = reply
|
|
self.last_request = None
|
|
|
|
async def __aenter__(self) -> "_FakeInvokeClient":
|
|
return self
|
|
|
|
async def __aexit__(self, *_exc: object) -> None:
|
|
return None
|
|
|
|
async def invoke_raw(self, request):
|
|
self.last_request = request
|
|
return self._reply
|
|
|
|
|
|
def test_authenticate_user_command_returns_user_id_without_echoing_password(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from zb_mom_ww_mxgateway.generated import mxaccess_gateway_pb2 as pb
|
|
|
|
reply = pb.MxCommandReply(
|
|
session_id="s1",
|
|
kind=pb.MX_COMMAND_KIND_AUTHENTICATE_USER,
|
|
protocol_status=pb.ProtocolStatus(code=pb.PROTOCOL_STATUS_CODE_OK),
|
|
authenticate_user=pb.AuthenticateUserReply(user_id=42),
|
|
)
|
|
fake = _FakeInvokeClient(reply)
|
|
|
|
async def fake_connect(options, **_kwargs):
|
|
return fake
|
|
|
|
monkeypatch.setattr(commands_module.GatewayClient, "connect", fake_connect)
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
[
|
|
"authenticate-user",
|
|
"--plaintext",
|
|
"--session-id",
|
|
"s1",
|
|
"--server-handle",
|
|
"3",
|
|
"--verify-user",
|
|
"operator",
|
|
"--password",
|
|
"cli-secret-pw",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert json.loads(result.output)["userId"] == 42
|
|
assert "cli-secret-pw" not in result.output
|
|
# Credential is carried on the wire, not echoed.
|
|
assert fake.last_request.command.authenticate_user.verify_user_password == "cli-secret-pw"
|
|
|
|
|
|
def test_authenticate_user_reads_password_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from zb_mom_ww_mxgateway.generated import mxaccess_gateway_pb2 as pb
|
|
|
|
reply = pb.MxCommandReply(
|
|
session_id="s1",
|
|
kind=pb.MX_COMMAND_KIND_AUTHENTICATE_USER,
|
|
protocol_status=pb.ProtocolStatus(code=pb.PROTOCOL_STATUS_CODE_OK),
|
|
authenticate_user=pb.AuthenticateUserReply(user_id=7),
|
|
)
|
|
fake = _FakeInvokeClient(reply)
|
|
|
|
async def fake_connect(options, **_kwargs):
|
|
return fake
|
|
|
|
monkeypatch.setattr(commands_module.GatewayClient, "connect", fake_connect)
|
|
monkeypatch.setenv("MXGW_TEST_PW", "env-secret-pw")
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
[
|
|
"authenticate-user",
|
|
"--plaintext",
|
|
"--session-id",
|
|
"s1",
|
|
"--server-handle",
|
|
"3",
|
|
"--verify-user",
|
|
"operator",
|
|
"--password-env",
|
|
"MXGW_TEST_PW",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert "env-secret-pw" not in result.output
|
|
assert fake.last_request.command.authenticate_user.verify_user_password == "env-secret-pw"
|
|
|
|
|
|
def test_authenticate_user_requires_a_password(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("MXGATEWAY_VERIFY_PASSWORD", raising=False)
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
[
|
|
"authenticate-user",
|
|
"--plaintext",
|
|
"--session-id",
|
|
"s1",
|
|
"--server-handle",
|
|
"3",
|
|
"--verify-user",
|
|
"operator",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "password is required" in result.output
|
|
# CLI-45: the usage error names the option and the canonical env var.
|
|
assert "--password" in result.output
|
|
assert "MXGATEWAY_VERIFY_PASSWORD" in result.output
|
|
|
|
|
|
def test_authenticate_user_reads_password_from_canonical_default_env(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""CLI-45: --password-env defaults to MXGATEWAY_VERIFY_PASSWORD.
|
|
|
|
Exporting the canonical variable alone must satisfy the credential, with no
|
|
explicit --password-env flag — the same operator workflow as the other CLIs.
|
|
"""
|
|
from zb_mom_ww_mxgateway.generated import mxaccess_gateway_pb2 as pb
|
|
|
|
reply = pb.MxCommandReply(
|
|
session_id="s1",
|
|
kind=pb.MX_COMMAND_KIND_AUTHENTICATE_USER,
|
|
protocol_status=pb.ProtocolStatus(code=pb.PROTOCOL_STATUS_CODE_OK),
|
|
authenticate_user=pb.AuthenticateUserReply(user_id=11),
|
|
)
|
|
fake = _FakeInvokeClient(reply)
|
|
|
|
async def fake_connect(options, **_kwargs):
|
|
return fake
|
|
|
|
monkeypatch.setattr(commands_module.GatewayClient, "connect", fake_connect)
|
|
monkeypatch.setenv("MXGATEWAY_VERIFY_PASSWORD", "canonical-env-pw")
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
[
|
|
"authenticate-user",
|
|
"--plaintext",
|
|
"--session-id",
|
|
"s1",
|
|
"--server-handle",
|
|
"3",
|
|
"--verify-user",
|
|
"operator",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert json.loads(result.output)["userId"] == 11
|
|
assert "canonical-env-pw" not in result.output
|
|
assert fake.last_request.command.authenticate_user.verify_user_password == "canonical-env-pw"
|
|
|
|
|
|
def test_authenticate_user_rejects_empty_password_value(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""CLI-45: an empty resolved credential fails fast, never reaching the wire."""
|
|
monkeypatch.setenv("MXGATEWAY_VERIFY_PASSWORD", "")
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
[
|
|
"authenticate-user",
|
|
"--plaintext",
|
|
"--session-id",
|
|
"s1",
|
|
"--server-handle",
|
|
"3",
|
|
"--verify-user",
|
|
"operator",
|
|
"--password",
|
|
"",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "password is required" in result.output
|
|
|
|
|
|
def test_write_secured_command_does_not_echo_value_on_failure(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from zb_mom_ww_mxgateway.generated import mxaccess_gateway_pb2 as pb
|
|
|
|
reply = pb.MxCommandReply(
|
|
session_id="s1",
|
|
kind=pb.MX_COMMAND_KIND_WRITE_SECURED,
|
|
protocol_status=pb.ProtocolStatus(
|
|
code=pb.PROTOCOL_STATUS_CODE_MXACCESS_FAILURE,
|
|
message="WriteSecured rejected value cli-secret-value",
|
|
),
|
|
hresult=-1,
|
|
)
|
|
fake = _FakeInvokeClient(reply)
|
|
|
|
async def fake_connect(options, **_kwargs):
|
|
return fake
|
|
|
|
monkeypatch.setattr(commands_module.GatewayClient, "connect", fake_connect)
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
[
|
|
"write-secured",
|
|
"--plaintext",
|
|
"--session-id",
|
|
"s1",
|
|
"--server-handle",
|
|
"3",
|
|
"--item-handle",
|
|
"4",
|
|
"--value",
|
|
"cli-secret-value",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert "cli-secret-value" not in result.output
|
|
|
|
|
|
def test_write_secured_and_authenticate_user_commands_are_registered() -> None:
|
|
names = set(main.commands)
|
|
assert {"write-secured", "authenticate-user"} <= names
|
|
|
|
|
|
class _FakeReplayGapSession:
|
|
"""Session stand-in whose event stream starts with a ReplayGap sentinel.
|
|
|
|
Mirrors what ``Session.stream_events`` yields on a resume that predates the
|
|
gateway's retained replay ring: the typed gap first, then normal events.
|
|
"""
|
|
|
|
def __init__(self, gap, event) -> None:
|
|
self._gap = gap
|
|
self._event = event
|
|
|
|
def stream_events(self, **_kwargs):
|
|
async def _iterate():
|
|
yield self._gap
|
|
yield self._event
|
|
|
|
return _iterate()
|
|
|
|
|
|
def test_stream_events_renders_replay_gap(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""CLI-35: a ReplayGap renders as its own JSON row instead of crashing the command."""
|
|
from zb_mom_ww_mxgateway.events import ReplayGap
|
|
from zb_mom_ww_mxgateway.generated import mxaccess_gateway_pb2 as pb
|
|
|
|
gap = ReplayGap(requested_after_sequence=7, oldest_available_sequence=42)
|
|
event = pb.MxEvent(session_id="cli-test-session", worker_sequence=43)
|
|
|
|
async def fake_connect(options, **_kwargs):
|
|
return _FakeAsyncClient()
|
|
|
|
monkeypatch.setattr(commands_module.GatewayClient, "connect", fake_connect)
|
|
monkeypatch.setattr(
|
|
commands_module,
|
|
"_session",
|
|
lambda _client, _session_id: _FakeReplayGapSession(gap, event),
|
|
)
|
|
|
|
result = CliRunner().invoke(
|
|
main,
|
|
[
|
|
"stream-events",
|
|
"--plaintext",
|
|
"--session-id",
|
|
"cli-test-session",
|
|
"--after-worker-sequence",
|
|
"7",
|
|
"--max-events",
|
|
"2",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
rows = json.loads(result.output)["events"]
|
|
assert rows[0] == {
|
|
"replayGap": {"requestedAfterSequence": 7, "oldestAvailableSequence": 42},
|
|
}
|
|
# The gap is rendered, never swallowed, and the normal event still follows it.
|
|
assert "replayGap" not in rows[1]
|
|
assert rows[1]["workerSequence"] == "43"
|