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__