mxaccessgw/clients/python/tests/test_cli.py

"""Tests for the Python CLI."""

import io
import json
from typing import Any

import click
import pytest
from click.testing import CliRunner

from mxgateway import __version__
from mxgateway_cli.commands import _BATCH_EOR, _use_plaintext, main


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_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


# Regression tests for Client.Python-013: ``_use_plaintext`` must not silently
# downgrade ``localhost:`` / ``127.0.0.1:`` endpoints to plaintext. TLS is the
# default; users must pass ``--plaintext`` to opt in.


def test_use_plaintext_requires_explicit_flag_for_localhost_endpoint() -> None:
    """A ``localhost:`` endpoint with no flags must resolve to TLS."""

    assert (
        _use_plaintext(
            {"endpoint": "localhost:5000", "plaintext": False, "use_tls": False}
        )
        is False
    )


def test_use_plaintext_requires_explicit_flag_for_loopback_ip_endpoint() -> None:
    """A ``127.0.0.1:`` endpoint with no flags must resolve to TLS."""

    assert (
        _use_plaintext(
            {"endpoint": "127.0.0.1:5000", "plaintext": False, "use_tls": False}
        )
        is False
    )


def test_use_plaintext_explicit_plaintext_flag_opts_in() -> None:
    """``--plaintext`` must select plaintext regardless of endpoint host."""

    assert (
        _use_plaintext(
            {"endpoint": "localhost:5000", "plaintext": True, "use_tls": False}
        )
        is True
    )
    assert (
        _use_plaintext(
            {
                "endpoint": "mxgateway.example.local:5001",
                "plaintext": True,
                "use_tls": False,
            }
        )
        is True
    )


def test_use_plaintext_explicit_tls_flag_is_accepted_and_idempotent() -> None:
    """``--tls`` is accepted as a redundant affirmation of the default."""

    assert (
        _use_plaintext(
            {
                "endpoint": "mxgateway.example.local:5001",
                "plaintext": False,
                "use_tls": True,
            }
        )
        is False
    )
    # Even for a localhost endpoint, ``--tls`` (the default) must yield TLS.
    assert (
        _use_plaintext(
            {"endpoint": "localhost:5000", "plaintext": False, "use_tls": True}
        )
        is False
    )


def test_use_plaintext_rejects_conflicting_flags() -> None:
    """``--plaintext`` combined with ``--tls`` is a usage error."""

    with pytest.raises(click.UsageError):
        _use_plaintext(
            {"endpoint": "localhost:5000", "plaintext": True, "use_tls": True}
        )


def test_cli_localhost_endpoint_defaults_to_tls_via_open_session(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """End-to-end: ``open-session`` against ``localhost:`` with no flags
    must build a TLS ``ClientOptions`` (plaintext=False)."""

    captured: dict[str, object] = {}

    async def _fake_connect(options):  # type: ignore[no-untyped-def]
        captured["plaintext"] = options.plaintext
        raise RuntimeError("stop-before-network")

    monkeypatch.setattr(
        "mxgateway_cli.commands.GatewayClient.connect", _fake_connect
    )

    runner = CliRunner()
    result = runner.invoke(
        main,
        [
            "open-session",
            "--endpoint",
            "localhost:5000",
            "--api-key",
            "mxgw_test_secret",
            "--json",
        ],
    )

    assert result.exit_code != 0  # connect was stubbed to raise
    assert captured.get("plaintext") is False, (
        "localhost endpoint must default to TLS without an explicit --plaintext "
        "flag (Client.Python-013 regression)."
    )


def test_cli_localhost_endpoint_with_plaintext_flag_uses_plaintext(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """End-to-end: ``--plaintext`` opts in to plaintext as expected."""

    captured: dict[str, object] = {}

    async def _fake_connect(options):  # type: ignore[no-untyped-def]
        captured["plaintext"] = options.plaintext
        raise RuntimeError("stop-before-network")

    monkeypatch.setattr(
        "mxgateway_cli.commands.GatewayClient.connect", _fake_connect
    )

    runner = CliRunner()
    result = runner.invoke(
        main,
        [
            "open-session",
            "--endpoint",
            "localhost:5000",
            "--api-key",
            "mxgw_test_secret",
            "--plaintext",
            "--json",
        ],
    )

    assert result.exit_code != 0
    assert captured.get("plaintext") is True


# ---------------------------------------------------------------------------
# batch subcommand tests
# ---------------------------------------------------------------------------


def _run_batch(lines: list[str]) -> tuple[int, list[str]]:
    """Invoke ``batch`` with the given stdin lines; return (exit_code, stdout_lines)."""
    runner = CliRunner()
    stdin_text = "\n".join(lines) + "\n"
    result = runner.invoke(main, ["batch"], input=stdin_text)
    stdout_lines = result.output.splitlines()
    return result.exit_code, stdout_lines


def _split_records(stdout_lines: list[str]) -> list[list[str]]:
    """Split stdout lines on ``__MXGW_BATCH_EOR__`` sentinels into per-command records."""
    records: list[list[str]] = []
    current: list[str] = []
    for line in stdout_lines:
        if line == _BATCH_EOR:
            records.append(current)
            current = []
        else:
            current.append(line)
    # Any trailing lines without a sentinel are ignored (shouldn't occur).
    return records


def test_batch_version_json_produces_eor_sentinel() -> None:
    """A single ``version --json`` line produces the version JSON followed by the EOR sentinel."""
    exit_code, lines = _run_batch(["version --json"])

    assert exit_code == 0
    records = _split_records(lines)
    assert len(records) == 1
    payload = json.loads(records[0][0])
    assert payload == {
        "client": "mxgw-py",
        "package": "mxaccess-gateway-client",
        "version": __version__,
    }


def test_batch_two_commands_produce_two_delimited_records() -> None:
    """Two input lines produce exactly two EOR-delimited records."""
    exit_code, lines = _run_batch(["version --json", "version --json"])

    assert exit_code == 0
    records = _split_records(lines)
    assert len(records) == 2
    for record in records:
        payload = json.loads(record[0])
        assert payload["client"] == "mxgw-py"


def test_batch_eof_exits_zero() -> None:
    """EOF on stdin exits with code 0."""
    runner = CliRunner()
    result = runner.invoke(main, ["batch"], input="")
    assert result.exit_code == 0


def test_batch_empty_line_exits_zero() -> None:
    """An empty line signals a clean exit with code 0."""
    exit_code, lines = _run_batch([""])
    assert exit_code == 0
    # No EOR sentinels should have been emitted.
    assert _BATCH_EOR not in lines


def test_batch_empty_line_stops_processing_subsequent_commands() -> None:
    """Commands after the first empty line must not be executed."""
    exit_code, lines = _run_batch(["", "version --json"])

    assert exit_code == 0
    # No records should appear because the empty line stopped the loop.
    records = _split_records(lines)
    assert records == []


def test_batch_failure_does_not_terminate_loop() -> None:
    """A failing command (bad parse) must not terminate the batch loop."""
    exit_code, lines = _run_batch([
        "open-session --unknown-flag",
        "version --json",
    ])

    assert exit_code == 0
    records = _split_records(lines)
    # Two records: one error + one success.
    assert len(records) == 2
    # First record must be a JSON error object.
    error_payload = json.loads(records[0][0])
    assert "error" in error_payload
    assert "type" in error_payload
    # Second record must be the version JSON.
    version_payload = json.loads(records[1][0])
    assert version_payload["client"] == "mxgw-py"


def test_batch_error_record_has_required_json_shape() -> None:
    """A failing command must produce ``{"error": "...", "type": "..."}`` JSON."""
    exit_code, lines = _run_batch(["open-session --unknown-flag"])

    assert exit_code == 0
    records = _split_records(lines)
    assert len(records) == 1
    payload = json.loads(records[0][0])
    assert isinstance(payload.get("error"), str)
    assert isinstance(payload.get("type"), str)


def test_batch_network_error_produces_error_json_not_terminates(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A network-level failure (MxGatewayError) on one command must not stop the loop."""

    async def _fake_connect(kwargs: dict[str, Any]) -> Any:
        raise RuntimeError("injected-network-failure")

    monkeypatch.setattr("mxgateway_cli.commands.GatewayClient.connect", _fake_connect)

    exit_code, lines = _run_batch([
        "open-session --endpoint localhost:5000 --api-key mxgw_test --plaintext --json",
        "version --json",
    ])

    assert exit_code == 0
    records = _split_records(lines)
    assert len(records) == 2
    # First record is an error.
    error_payload = json.loads(records[0][0])
    assert "error" in error_payload
    assert "type" in error_payload
    # Second record is success.
    version_payload = json.loads(records[1][0])
    assert version_payload["client"] == "mxgw-py"