feat(python): add galaxy-* CLI commands (§4.2)

clients/python/README.md

@@ -214,9 +214,22 @@ The method returns an async iterator yielding the generated `DeployEvent`
 proto. Breaking out of the loop, calling `aclose()` on the iterator, or
 cancelling the surrounding task closes the underlying gRPC stream
 cleanly. The streaming RPC requires the same `metadata:read` scope as
-the other Galaxy methods. The CLI does not currently expose a
-streaming `watch-deploy-events` subcommand — use the library API
-directly when subscribing to deploy events from Python.
+the other Galaxy methods.
+
+The CLI exposes the Galaxy Repository RPCs through four subcommands that
+mirror the other clients:
+
+```bash
+mxgw-py galaxy-test-connection --plaintext --json
+mxgw-py galaxy-last-deploy --plaintext --json
+mxgw-py galaxy-discover --plaintext --json
+mxgw-py galaxy-watch --plaintext --json
+```
+
+`galaxy-watch` is bounded by `--max-events` (default `1`) and `--timeout`
+(seconds) so it always terminates; pass `--last-seen-deploy-time` (an
+ISO-8601 timestamp) to suppress the bootstrap event when it matches the
+current cached deploy time.
 
 ## Authentication And TLS
 

clients/python/src/zb_mom_ww_mxgateway_cli/commands.py

@@ -21,6 +21,7 @@ from zb_mom_ww_mxgateway import __version__
 from zb_mom_ww_mxgateway.auth import redact_secret
 from zb_mom_ww_mxgateway.client import GatewayClient
 from zb_mom_ww_mxgateway.errors import MxGatewayError
+from zb_mom_ww_mxgateway.galaxy import GalaxyRepositoryClient
 from zb_mom_ww_mxgateway.generated import mxaccess_gateway_pb2 as pb
 from zb_mom_ww_mxgateway.options import ClientOptions
 from zb_mom_ww_mxgateway.values import MxValueInput, to_mx_value
@@ -512,6 +513,67 @@ def smoke(**kwargs: Any) -> None:
     _run(_smoke(**kwargs), output_json=kwargs["output_json"], secrets=_secrets(kwargs))
 
 
+@main.command("galaxy-test-connection")
+@gateway_options
+@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
+def galaxy_test_connection(**kwargs: Any) -> None:
+    """Test whether the gateway can reach the Galaxy Repository DB."""
+
+    _run(
+        _galaxy_test_connection(**kwargs),
+        output_json=kwargs["output_json"],
+        secrets=_secrets(kwargs),
+    )
+
+
+@main.command("galaxy-last-deploy")
+@gateway_options
+@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
+def galaxy_last_deploy(**kwargs: Any) -> None:
+    """Read the last Galaxy deploy timestamp."""
+
+    _run(
+        _galaxy_last_deploy(**kwargs),
+        output_json=kwargs["output_json"],
+        secrets=_secrets(kwargs),
+    )
+
+
+@main.command("galaxy-discover")
+@gateway_options
+@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
+def galaxy_discover(**kwargs: Any) -> None:
+    """Enumerate the deployed Galaxy object hierarchy."""
+
+    _run(
+        _galaxy_discover(**kwargs),
+        output_json=kwargs["output_json"],
+        secrets=_secrets(kwargs),
+    )
+
+
+@main.command("galaxy-watch")
+@gateway_options
+@click.option(
+    "--last-seen-deploy-time",
+    "last_seen_deploy_time",
+    default=None,
+    help="ISO-8601 timestamp; when it matches the current cached deploy time the "
+    "bootstrap event is suppressed.",
+)
+@click.option("--max-events", default=1, type=int, show_default=True)
+@click.option("--timeout", default=5.0, type=float, show_default=True)
+@click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
+def galaxy_watch(**kwargs: Any) -> None:
+    """Stream a bounded number of Galaxy deploy events."""
+
+    _run(
+        _galaxy_watch(**kwargs),
+        output_json=kwargs["output_json"],
+        secrets=_secrets(kwargs),
+    )
+
+
 async def _open_session(**kwargs: Any) -> dict[str, Any]:
     async with await _connect(kwargs) as client:
         reply = await client.open_session_raw(
@@ -922,6 +984,48 @@ async def _smoke(**kwargs: Any) -> dict[str, Any]:
                 await session.close()
 
 
+async def _galaxy_test_connection(**kwargs: Any) -> dict[str, Any]:
+    async with await _connect_galaxy(kwargs) as galaxy:
+        ok = await galaxy.test_connection()
+        return {"command": "galaxy-test-connection", "ok": ok}
+
+
+async def _galaxy_last_deploy(**kwargs: Any) -> dict[str, Any]:
+    async with await _connect_galaxy(kwargs) as galaxy:
+        last_deploy = await galaxy.get_last_deploy_time()
+        payload: dict[str, Any] = {
+            "command": "galaxy-last-deploy",
+            "present": last_deploy is not None,
+        }
+        if last_deploy is not None:
+            payload["timeOfLastDeploy"] = last_deploy.isoformat()
+        return payload
+
+
+async def _galaxy_discover(**kwargs: Any) -> dict[str, Any]:
+    async with await _connect_galaxy(kwargs) as galaxy:
+        objects = await galaxy.discover_hierarchy()
+        return {
+            "command": "galaxy-discover",
+            "objects": [_message_dict(obj) for obj in objects],
+        }
+
+
+async def _galaxy_watch(**kwargs: Any) -> dict[str, Any]:
+    last_seen = kwargs.get("last_seen_deploy_time")
+    last_seen_dt = _parse_datetime(last_seen) if last_seen else None
+    async with await _connect_galaxy(kwargs) as galaxy:
+        events = await _collect_events(
+            galaxy.watch_deploy_events(last_seen_dt),
+            max_events=kwargs["max_events"],
+            timeout=kwargs["timeout"],
+        )
+        return {
+            "command": "galaxy-watch",
+            "events": [_message_dict(event) for event in events],
+        }
+
+
 async def _connect(kwargs: dict[str, Any]) -> GatewayClient:
     api_key = kwargs.get("api_key") or _api_key_from_env(kwargs.get("api_key_env"))
     return await GatewayClient.connect(
@@ -938,6 +1042,22 @@ async def _connect(kwargs: dict[str, Any]) -> GatewayClient:
     )
 
 
+async def _connect_galaxy(kwargs: dict[str, Any]) -> GalaxyRepositoryClient:
+    api_key = kwargs.get("api_key") or _api_key_from_env(kwargs.get("api_key_env"))
+    return await GalaxyRepositoryClient.connect(
+        ClientOptions(
+            endpoint=kwargs["endpoint"],
+            api_key=api_key,
+            plaintext=_use_plaintext(kwargs),
+            ca_file=kwargs.get("ca_file"),
+            require_certificate_validation=bool(kwargs.get("require_certificate_validation")),
+            server_name_override=kwargs.get("server_name_override"),
+            call_timeout=kwargs.get("call_timeout"),
+            stream_timeout=kwargs.get("stream_timeout"),
+        ),
+    )
+
+
 def _session(client: GatewayClient, session_id: str):
     from zb_mom_ww_mxgateway.session import Session
 

clients/python/tests/test_cli.py

@@ -211,3 +211,78 @@ def test_batch_continues_after_error_line() -> None:
     # 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) -> None:
+        self._ok = ok
+        self._objects = objects or []
+
+    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
+
+
+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_commands_are_registered() -> None:
+    runner = CliRunner()
+    for command in (
+        "galaxy-test-connection",
+        "galaxy-last-deploy",
+        "galaxy-discover",
+        "galaxy-watch",
+    ):
+        result = runner.invoke(main, [command, "--help"])
+        assert result.exit_code == 0, result.output
+        assert "--endpoint" in result.output