fix(CLI-47,CLI-48,CLI-49,CLI-50): align alarm feed clients

clients/python/README.md

@@ -132,6 +132,14 @@ sent when the gateway's alarm monitor restarts and drops a truncated verdict.
 Track it if you need set completeness on a live feed without polling
 `query_active_alarms`.
 
+`mxgw-py stream-alarms` writes one JSON row as each feed message arrives. The
+default `--max-messages 0` is unbounded; use a positive value for a finite
+sample, remembering that the provider-status and snapshot-status preamble both
+count toward it. `--timeout` ends the command after an idle wait and the stream
+is closed in every exit path. Protobuf JSON omits false-valued scalar defaults,
+so healthy/clearing frames appear as `{"providerStatus": {}}` and
+`{"snapshotStatus": {}}`; absent means false.
+
 Canceling a Python task cancels the client-side gRPC call or stream wait. It
 does not abort an in-flight MXAccess COM call inside the worker process.
 
@@ -448,7 +456,7 @@ mxgw-py register --session-id <id> --client-name python-client --json
 mxgw-py add-item --session-id <id> --server-handle 1 --item Object.Attribute --json
 mxgw-py advise --session-id <id> --server-handle 1 --item-handle 2 --json
 mxgw-py stream-events --session-id <id> --max-events 1 --json
-mxgw-py stream-alarms --max-messages 1 --json
+mxgw-py stream-alarms --max-messages 4 --json
 mxgw-py acknowledge-alarm --reference "\\Galaxy\Area001.Pump001.PumpFault" --json
 mxgw-py write --session-id <id> --server-handle 1 --item-handle 2 --type int32 --value 123 --json
 ```

clients/python/src/zb_mom_ww_mxgateway_cli/commands.py

@@ -511,16 +511,15 @@ def stream_events(**kwargs: Any) -> None:
 @main.command("stream-alarms")
 @gateway_options
 @click.option("--filter-prefix", default="", help="Alarm-reference prefix filter.")
-@click.option("--max-messages", default=1, type=int, show_default=True)
+@click.option("--max-messages", default=0, type=int, show_default=True, help="Maximum rows to emit; 0 means unbounded.")
 @click.option("--timeout", default=5.0, type=float, show_default=True)
 @click.option("--correlation-id", default="", help="Client correlation id.")
 @click.option("--json", "output_json", is_flag=True, help="Emit JSON output.")
 def stream_alarms(**kwargs: Any) -> None:
-    """Stream a bounded number of messages from the gateway's central alarm feed."""
+    """Stream messages from the gateway's central alarm feed as they arrive."""
 
-    _run(
+    _run_streaming(
         _stream_alarms(**kwargs),
-        output_json=kwargs["output_json"],
         secrets=_secrets(kwargs),
     )
 
@@ -1119,19 +1118,39 @@ async def _stream_events(**kwargs: Any) -> dict[str, Any]:
         return {"events": [_event_row(event) for event in events]}
 
 
-async def _stream_alarms(**kwargs: Any) -> dict[str, Any]:
-    async with await _connect(kwargs) as client:
-        messages = await _collect_alarm_messages(
-            client.stream_alarms(
-                pb.StreamAlarmsRequest(
-                    client_correlation_id=kwargs["correlation_id"],
-                    alarm_filter_prefix=kwargs["filter_prefix"],
-                ),
-            ),
-            max_messages=kwargs["max_messages"],
-            timeout=kwargs["timeout"],
+async def _stream_alarms(**kwargs: Any) -> None:
+    max_messages = int(kwargs["max_messages"])
+    if max_messages > MAX_AGGREGATE_EVENTS:
+        raise click.BadParameter(
+            f"must be less than or equal to {MAX_AGGREGATE_EVENTS}",
+            param_hint="--max-messages",
         )
-        return {"messages": [_message_dict(message) for message in messages]}
+
+    async with await _connect(kwargs) as client:
+        messages = client.stream_alarms(
+            pb.StreamAlarmsRequest(
+                client_correlation_id=kwargs["correlation_id"],
+                alarm_filter_prefix=kwargs["filter_prefix"],
+            ),
+        )
+        iterator = messages.__aiter__()
+        emitted = 0
+        try:
+            while max_messages == 0 or emitted < max_messages:
+                try:
+                    message = await asyncio.wait_for(
+                        iterator.__anext__(),
+                        timeout=kwargs["timeout"],
+                    )
+                except (StopAsyncIteration, asyncio.TimeoutError):
+                    break
+                _emit(_message_dict(message), output_json=kwargs["output_json"])
+                sys.stdout.flush()
+                emitted += 1
+        finally:
+            close = getattr(iterator, "aclose", None)
+            if close is not None:
+                await close()
 
 
 async def _acknowledge_alarm(**kwargs: Any) -> dict[str, Any]:
@@ -1489,6 +1508,17 @@ def _run(
     _emit(payload, output_json=output_json)
 
 
+def _run_streaming(
+    awaitable: Awaitable[None],
+    *,
+    secrets: list[str | None],
+) -> None:
+    try:
+        asyncio.run(awaitable)
+    except MxGatewayError as error:
+        raise click.ClickException(redact_secret(str(error), secrets)) from error
+
+
 def _emit(
     payload: dict[str, Any],
     *,
@@ -1536,34 +1566,6 @@ async def _collect_events(
     return collected
 
 
-async def _collect_alarm_messages(
-    messages: Any,
-    *,
-    max_messages: int,
-    timeout: float,
-) -> list[pb.AlarmFeedMessage]:
-    if max_messages > MAX_AGGREGATE_EVENTS:
-        raise click.BadParameter(
-            f"must be less than or equal to {MAX_AGGREGATE_EVENTS}",
-            param_hint="--max-messages",
-        )
-
-    collected: list[pb.AlarmFeedMessage] = []
-    iterator = messages.__aiter__()
-    try:
-        while len(collected) < max_messages:
-            collected.append(await asyncio.wait_for(iterator.__anext__(), timeout=timeout))
-    except StopAsyncIteration:
-        pass
-    except asyncio.TimeoutError:
-        pass
-    finally:
-        close = getattr(iterator, "aclose", None)
-        if close is not None:
-            await close()
-    return collected
-
-
 async def _collect_deploy_events(
     events: Any,
     *,

clients/python/tests/test_review_findings_022_to_026.py

@@ -7,6 +7,7 @@ fixed source.
 
 from __future__ import annotations
 
+import asyncio
 import json
 import re
 import time as _time_module_ref
@@ -296,6 +297,22 @@ class _AlarmFakeStream:
         self.cancelled = True
 
 
+class _IncrementalAlarmFakeStream(_AlarmFakeStream):
+    def __init__(self, messages: list[pb.AlarmFeedMessage], emitted: list[str]) -> None:
+        super().__init__(messages)
+        self._emitted = emitted
+        self._yielded = 0
+
+    async def __anext__(self) -> pb.AlarmFeedMessage:
+        if self._yielded > 0:
+            assert len(self._emitted) >= self._yielded, "previous alarm row was buffered"
+        if not self._messages:
+            await asyncio.Event().wait()
+            raise AssertionError("unreachable")
+        self._yielded += 1
+        return self._messages.pop(0)
+
+
 class _BulkFakeUnary:
     def __init__(self, replies: list[Any]) -> None:
         self.replies = replies
@@ -667,8 +684,15 @@ def test_cli_write_bulk_happy_path(monkeypatch) -> None:
     assert cmd.kind == pb.MX_COMMAND_KIND_WRITE_BULK
 
 
-def test_cli_stream_alarms_happy_path(monkeypatch) -> None:
-    transitions = [
+def test_cli_stream_alarms_default_streams_status_cached_and_live_rows_incrementally(
+    monkeypatch,
+) -> None:
+    messages = [
+        pb.AlarmFeedMessage(provider_status=pb.AlarmProviderStatus(degraded=False)),
+        pb.AlarmFeedMessage(snapshot_status=pb.AlarmSnapshotStatus(truncated=False)),
+        pb.AlarmFeedMessage(
+            active_alarm=pb.ActiveAlarmSnapshot(alarm_full_reference="Tank01.Level.HiHi"),
+        ),
         pb.AlarmFeedMessage(
             transition=pb.OnAlarmTransitionEvent(
                 alarm_full_reference="Tank01.Level.HiHi",
@@ -676,7 +700,15 @@ def test_cli_stream_alarms_happy_path(monkeypatch) -> None:
             ),
         ),
     ]
-    stream = _AlarmFakeStream(transitions)
+    emitted: list[str] = []
+    real_echo = cli_commands.click.echo
+
+    def recording_echo(message=None, **kwargs) -> None:
+        emitted.append(str(message))
+        real_echo(message, **kwargs)
+
+    monkeypatch.setattr(cli_commands.click, "echo", recording_echo)
+    stream = _IncrementalAlarmFakeStream(messages, emitted)
     stub = _BulkFakeStub()
     stub.set_alarm_stream(stream)
     _install_fake_connect(monkeypatch, stub)
@@ -689,10 +721,8 @@ def test_cli_stream_alarms_happy_path(monkeypatch) -> None:
             "--endpoint",
             "localhost:5000",
             "--plaintext",
-            "--max-messages",
-            "1",
             "--timeout",
-            "5.0",
+            "0.01",
             "--filter-prefix",
             "Tank01.",
             "--json",
@@ -700,8 +730,44 @@ def test_cli_stream_alarms_happy_path(monkeypatch) -> None:
     )
 
     assert result.exit_code == 0, result.output
-    payload = json.loads(result.output)
-    assert payload["messages"][0]["transition"]["alarmFullReference"] == "Tank01.Level.HiHi"
+    rows = [json.loads(line) for line in result.output.splitlines()]
+    assert [next(iter(row)) for row in rows] == [
+        "providerStatus",
+        "snapshotStatus",
+        "activeAlarm",
+        "transition",
+    ]
+    assert stream.cancelled
+
+
+def test_cli_stream_alarms_positive_max_messages_remains_finite(monkeypatch) -> None:
+    messages = [
+        pb.AlarmFeedMessage(provider_status=pb.AlarmProviderStatus()),
+        pb.AlarmFeedMessage(snapshot_status=pb.AlarmSnapshotStatus()),
+        pb.AlarmFeedMessage(active_alarm=pb.ActiveAlarmSnapshot()),
+    ]
+    stream = _AlarmFakeStream(messages)
+    stub = _BulkFakeStub()
+    stub.set_alarm_stream(stream)
+    _install_fake_connect(monkeypatch, stub)
+
+    result = CliRunner().invoke(
+        main,
+        [
+            "stream-alarms",
+            "--endpoint",
+            "localhost:5000",
+            "--plaintext",
+            "--max-messages",
+            "2",
+            "--json",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    rows = [json.loads(line) for line in result.output.splitlines()]
+    assert [next(iter(row)) for row in rows] == ["providerStatus", "snapshotStatus"]
+    assert stream.cancelled
 
 
 def test_cli_acknowledge_alarm_happy_path(monkeypatch) -> None: