mxaccessgw/clients/python/tests/test_cli.py

"""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,
    ) -> None:
        self._ok = ok
        self._objects = objects or []
        self._last_deploy = last_deploy
        self._events = events or []
        self._browse_roots = browse_roots or []
        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 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 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