Joseph Doherty
828e3e6cf6
Adds the session-less alarm CLI subcommands to mxgw-py. stream-alarms reads a
bounded slice of the gateway's central alarm feed (--filter-prefix,
--max-messages, --timeout, --json; aggregate `{messages: [...]}`);
acknowledge-alarm is a unary ack (--reference required, --comment, --operator).
GatewayClient.stream_alarms joins query_active_alarms via a
_canceling_alarm_feed_iterator helper mirroring the existing
_canceling_active_alarms_iterator pattern.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
149 lines
3.7 KiB
Python
149 lines
3.7 KiB
Python
"""Tests for the Python CLI."""
|
|
|
|
import json
|
|
|
|
from click.testing import CliRunner
|
|
|
|
from zb_mom_ww_mxgateway import __version__
|
|
from zb_mom_ww_mxgateway_cli.commands import main
|
|
|
|
_BATCH_EOR = "__MXGW_BATCH_EOR__"
|
|
|
|
|
|
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__
|