diff --git a/clients/dotnet/README.md b/clients/dotnet/README.md index 6a2864d..3c711f5 100644 --- a/clients/dotnet/README.md +++ b/clients/dotnet/README.md @@ -166,6 +166,11 @@ 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 `QueryActiveAlarmsAsync`. +The CLI's text rows always spell out `degraded=false` and `truncated=false` so +healthy and clearing frames remain visible. Its protobuf-JSON modes retain the +standard proto3 omission rule, so those same false-valued payloads appear as +`{"providerStatus": {}}` and `{"snapshotStatus": {}}`; absent means false. + `MxGatewaySession.CloseAsync` is explicit and idempotent. Repeated calls return the first `CloseSessionReply` instead of sending another close request. @@ -285,7 +290,7 @@ dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- advise --s dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- write --session-id --server-handle 1 --item-handle 1 --type int32 --value 123 --json dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- write2 --session-id --server-handle 1 --item-handle 1 --type int32 --value 123 --timestamp 2026-01-01T00:00:00Z --json dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- stream-events --session-id --max-events 1 --json -dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- stream-alarms --filter-prefix Area001 --max-events 1 --json +dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- stream-alarms --filter-prefix Area001 --max-events 4 --json dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- acknowledge-alarm --reference "\\Galaxy\Area001.Pump001.PumpFault" --comment "ack from cli" --operator operator1 --json dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- smoke --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --item Area001.Pump001.Speed --json ``` diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli/MxGatewayClientCli.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli/MxGatewayClientCli.cs index 353c21a..a84a324 100644 --- a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli/MxGatewayClientCli.cs +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli/MxGatewayClientCli.cs @@ -1561,9 +1561,9 @@ public static class MxGatewayClientCli AlarmFeedMessage.PayloadOneofCase.Transition => $"transition {ProtobufJsonFormatter.Format(feedMessage.Transition)}", AlarmFeedMessage.PayloadOneofCase.ProviderStatus => - $"provider-status {ProtobufJsonFormatter.Format(feedMessage.ProviderStatus)}", + $"provider-status mode={feedMessage.ProviderStatus.Mode} degraded={feedMessage.ProviderStatus.Degraded.ToString().ToLowerInvariant()} reason=\"{feedMessage.ProviderStatus.Reason}\"", AlarmFeedMessage.PayloadOneofCase.SnapshotStatus => - $"snapshot-status {ProtobufJsonFormatter.Format(feedMessage.SnapshotStatus)}", + $"snapshot-status truncated={feedMessage.SnapshotStatus.Truncated.ToString().ToLowerInvariant()}", _ => $"unknown-payload {feedMessage.PayloadCase}", }; } diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientCliTests.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientCliTests.cs index 726f591..b0b4f3e 100644 --- a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientCliTests.cs +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientCliTests.cs @@ -705,6 +705,50 @@ public sealed class MxGatewayClientCliTests Assert.Equal(string.Empty, error.ToString()); } + /// Verifies that false alarm-feed status flags remain explicit in text while protobuf JSON omits defaults. + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAsync_StreamAlarms_StatusFramesDistinguishTextFromProtobufJson() + { + FakeCliClient fakeClient = new(); + fakeClient.AlarmFeedMessages.Add(new AlarmFeedMessage + { + ProviderStatus = new AlarmProviderStatus { Degraded = false }, + }); + fakeClient.AlarmFeedMessages.Add(new AlarmFeedMessage + { + SnapshotStatus = new AlarmSnapshotStatus { Truncated = false }, + }); + + using var textOutput = new StringWriter(); + using var textError = new StringWriter(); + int textExit = await MxGatewayClientCli.RunAsync( + ["stream-alarms", "--endpoint", "http://localhost:5000", "--api-key", "test-api-key", "--max-events", "2"], + textOutput, + textError, + _ => fakeClient); + + Assert.Equal(0, textExit); + Assert.Equal( + ["provider-status mode=Unspecified degraded=false reason=\"\"", "snapshot-status truncated=false"], + textOutput.ToString().Split(Environment.NewLine, StringSplitOptions.RemoveEmptyEntries)); + Assert.Equal(string.Empty, textError.ToString()); + + using var jsonOutput = new StringWriter(); + using var jsonError = new StringWriter(); + int jsonExit = await MxGatewayClientCli.RunAsync( + ["stream-alarms", "--endpoint", "http://localhost:5000", "--api-key", "test-api-key", "--max-events", "2", "--jsonl"], + jsonOutput, + jsonError, + _ => fakeClient); + + Assert.Equal(0, jsonExit); + Assert.Equal( + ["{ \"providerStatus\": { } }", "{ \"snapshotStatus\": { } }"], + jsonOutput.ToString().Split(Environment.NewLine, StringSplitOptions.RemoveEmptyEntries)); + Assert.Equal(string.Empty, jsonError.ToString()); + } + /// Verifies that acknowledge-alarm builds a request and prints the JSON reply. /// A task that represents the asynchronous operation. [Fact] diff --git a/clients/go/README.md b/clients/go/README.md index 84ea96a..577fd57 100644 --- a/clients/go/README.md +++ b/clients/go/README.md @@ -162,6 +162,11 @@ on every verdict change — including the clearing frame 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 `QueryActiveAlarms`. +The CLI's text rows always spell out `degraded=false` and `truncated=false`. +Its protobuf-JSON mode retains the standard proto3 omission rule, so those +false-valued payloads appear as `{"providerStatus":{}}` and +`{"snapshotStatus":{}}`; absent means false. + ## Write Semantics And Common Pitfalls These are MXAccess parity behaviors that surprise new callers. The gateway @@ -439,7 +444,7 @@ go run ./cmd/mxgw-go write -session-id -server-handle 1 -item-handle 1 -typ go run ./cmd/mxgw-go write-bulk -session-id -server-handle 1 -item-handles 1,2 -values 10,20 -type int32 -plaintext -json go run ./cmd/mxgw-go read-bulk -session-id -item-handles 1,2 -plaintext -json go run ./cmd/mxgw-go stream-events -session-id -plaintext -json -go run ./cmd/mxgw-go stream-alarms -plaintext -json +go run ./cmd/mxgw-go stream-alarms -plaintext -limit 4 -json go run ./cmd/mxgw-go smoke -item Area001.Tag.Value -plaintext -json go run ./cmd/mxgw-go galaxy-test-connection -plaintext -json go run ./cmd/mxgw-go galaxy-last-deploy -plaintext -json diff --git a/clients/go/cmd/mxgw-go/main_test.go b/clients/go/cmd/mxgw-go/main_test.go index f69c0b3..b515df7 100644 --- a/clients/go/cmd/mxgw-go/main_test.go +++ b/clients/go/cmd/mxgw-go/main_test.go @@ -283,6 +283,68 @@ type pingFakeGateway struct { pb.UnimplementedMxAccessGatewayServer } +// alarmStatusFakeGateway emits both status arms with their proto3 boolean +// defaults so the CLI contract covers values that protobuf JSON omits. +type alarmStatusFakeGateway struct { + pb.UnimplementedMxAccessGatewayServer +} + +func (g *alarmStatusFakeGateway) StreamAlarms( + _ *pb.StreamAlarmsRequest, + stream grpc.ServerStreamingServer[pb.AlarmFeedMessage], +) error { + if err := stream.Send(&pb.AlarmFeedMessage{ + Payload: &pb.AlarmFeedMessage_ProviderStatus{ + ProviderStatus: &pb.AlarmProviderStatus{Degraded: false}, + }, + }); err != nil { + return err + } + return stream.Send(&pb.AlarmFeedMessage{ + Payload: &pb.AlarmFeedMessage_SnapshotStatus{ + SnapshotStatus: &pb.AlarmSnapshotStatus{Truncated: false}, + }, + }) +} + +func startAlarmStatusGateway(t *testing.T) string { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + server := grpc.NewServer() + pb.RegisterMxAccessGatewayServer(server, &alarmStatusFakeGateway{}) + go func() { _ = server.Serve(listener) }() + t.Cleanup(func() { + server.Stop() + _ = listener.Close() + }) + return listener.Addr().String() +} + +func TestRunStreamAlarmsPrintsStatusArmsInTextAndProtobufJSON(t *testing.T) { + endpoint := startAlarmStatusGateway(t) + baseArgs := []string{"stream-alarms", "-endpoint", endpoint, "-plaintext", "-api-key", "test", "-limit", "2"} + + var textOut, textErr bytes.Buffer + if err := runWithIO(t.Context(), baseArgs, &textOut, &textErr); err != nil { + t.Fatalf("text stream-alarms error = %v; stderr = %s", err, textErr.String()) + } + if got, want := strings.TrimSpace(textOut.String()), "provider-status mode=ALARM_PROVIDER_MODE_UNSPECIFIED degraded=false reason=\"\"\nsnapshot-status truncated=false"; got != want { + t.Fatalf("text stream-alarms output = %q, want %q", got, want) + } + + var jsonOut, jsonErr bytes.Buffer + jsonArgs := append(append([]string{}, baseArgs...), "-json") + if err := runWithIO(t.Context(), jsonArgs, &jsonOut, &jsonErr); err != nil { + t.Fatalf("JSON stream-alarms error = %v; stderr = %s", err, jsonErr.String()) + } + if got, want := strings.TrimSpace(jsonOut.String()), "{\"providerStatus\":{}}\n{\"snapshotStatus\":{}}"; got != want { + t.Fatalf("JSON stream-alarms output = %q, want protobuf JSON %q", got, want) + } +} + func (g *pingFakeGateway) Invoke(_ context.Context, req *pb.MxCommandRequest) (*pb.MxCommandReply, error) { echo := "pong:" + req.GetCommand().GetPing().GetMessage() return &pb.MxCommandReply{ diff --git a/clients/go/mxgateway/types.go b/clients/go/mxgateway/types.go index 8b9f402..e5a6f1a 100644 --- a/clients/go/mxgateway/types.go +++ b/clients/go/mxgateway/types.go @@ -127,9 +127,14 @@ type ( QueryActiveAlarmsRequest = pb.QueryActiveAlarmsRequest // StreamAlarmsRequest is the gateway StreamAlarms request message. StreamAlarmsRequest = pb.StreamAlarmsRequest - // AlarmFeedMessage is one message on the StreamAlarms feed — an - // active-alarm snapshot row, a snapshot-complete sentinel, or a transition. + // AlarmFeedMessage is one message on the StreamAlarms feed — an active-alarm + // snapshot row, snapshot-complete sentinel, transition, provider status, or + // snapshot-completeness status. AlarmFeedMessage = pb.AlarmFeedMessage + // AlarmProviderStatus reports the alarm provider mode and degradation state. + AlarmProviderStatus = pb.AlarmProviderStatus + // AlarmSnapshotStatus reports whether the cached alarm snapshot is truncated. + AlarmSnapshotStatus = pb.AlarmSnapshotStatus // ActiveAlarmSnapshot is one row in a ConditionRefresh stream. ActiveAlarmSnapshot = pb.ActiveAlarmSnapshot // OnAlarmTransitionEvent is the body carried by alarm-transition MxEvents. diff --git a/clients/go/mxgateway/types_alias_test.go b/clients/go/mxgateway/types_alias_test.go new file mode 100644 index 0000000..a783ffd --- /dev/null +++ b/clients/go/mxgateway/types_alias_test.go @@ -0,0 +1,17 @@ +package mxgateway_test + +import ( + "testing" + + "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/mxgateway" +) + +// TestAlarmFeedStatusAliasesArePublic is a compile-time consumer contract for +// the two status payloads carried by AlarmFeedMessage. +func TestAlarmFeedStatusAliasesArePublic(t *testing.T) { + t.Helper() + var provider mxgateway.AlarmProviderStatus + var snapshot mxgateway.AlarmSnapshotStatus + _ = provider + _ = snapshot +} diff --git a/clients/java/README.md b/clients/java/README.md index e811ec7..f530f2e 100644 --- a/clients/java/README.md +++ b/clients/java/README.md @@ -135,6 +135,11 @@ the clearing frame 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 `queryActiveAlarms`. +The CLI's text rows always spell out `degraded=false` and `truncated=false`. +Its protobuf-JSON mode retains the standard proto3 omission rule, so those +false-valued payloads appear as `{"providerStatus":{}}` and +`{"snapshotStatus":{}}`; absent means false. + ## Write Semantics And Common Pitfalls These are MXAccess parity behaviors that surprise new callers. The gateway @@ -411,7 +416,7 @@ gradle :zb-mom-ww-mxgateway-cli:run --args="advise-supervisory --endpoint localh gradle :zb-mom-ww-mxgateway-cli:run --args="authenticate-user --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --session-id --server-handle 1 --verify-user operator --password-env MXGATEWAY_VERIFY_PASSWORD --json" gradle :zb-mom-ww-mxgateway-cli:run --args="write-secured --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --session-id --server-handle 1 --item-handle 1 --type int32 --value 123 --current-user-id 100 --verifier-user-id 100 --json" gradle :zb-mom-ww-mxgateway-cli:run --args="stream-events --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --session-id --limit 1 --json" -gradle :zb-mom-ww-mxgateway-cli:run --args="stream-alarms --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --filter-prefix Galaxy --limit 1 --json" +gradle :zb-mom-ww-mxgateway-cli:run --args="stream-alarms --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --filter-prefix Galaxy --limit 4 --json" gradle :zb-mom-ww-mxgateway-cli:run --args="acknowledge-alarm --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --reference \"\\Galaxy\Area001.Pump001.PumpFault\" --json" gradle :zb-mom-ww-mxgateway-cli:run --args="smoke --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --item TestObject.TestInt --json" ``` diff --git a/clients/java/zb-mom-ww-mxgateway-cli/src/test/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCliTests.java b/clients/java/zb-mom-ww-mxgateway-cli/src/test/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCliTests.java index 69609c3..c8e0bb8 100644 --- a/clients/java/zb-mom-ww-mxgateway-cli/src/test/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCliTests.java +++ b/clients/java/zb-mom-ww-mxgateway-cli/src/test/java/com/zb/mom/ww/mxgateway/cli/MxGatewayCliTests.java @@ -27,6 +27,8 @@ import mxaccess_gateway.v1.MxaccessGateway.ActiveAlarmSnapshot; import mxaccess_gateway.v1.MxaccessGateway.AddItemReply; import mxaccess_gateway.v1.MxaccessGateway.AlarmConditionState; import mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage; +import mxaccess_gateway.v1.MxaccessGateway.AlarmProviderStatus; +import mxaccess_gateway.v1.MxaccessGateway.AlarmSnapshotStatus; import mxaccess_gateway.v1.MxaccessGateway.AlarmTransitionKind; import mxaccess_gateway.v1.MxaccessGateway.BulkReadResult; import mxaccess_gateway.v1.MxaccessGateway.BulkWriteResult; @@ -685,6 +687,21 @@ final class MxGatewayCliTests { assertTrue(run.output().contains("\"snapshotComplete\""), run.output()); } + @Test + void streamAlarmsCommandPrintsStatusArmsInTextAndProtobufJson() { + FakeClientFactory factory = new FakeClientFactory(); + + CliRun text = execute(factory, "stream-alarms"); + List textRows = text.output().lines().filter(line -> !line.isBlank()).toList(); + assertEquals("provider-status mode=ALARM_PROVIDER_MODE_UNSPECIFIED degraded=false reason=", textRows.get(3)); + assertEquals("snapshot-status truncated=false", textRows.get(4)); + + CliRun json = execute(factory, "stream-alarms", "--json"); + List jsonRows = json.output().lines().filter(line -> !line.isBlank()).toList(); + assertEquals("{\"providerStatus\":{}}", jsonRows.get(3)); + assertEquals("{\"snapshotStatus\":{}}", jsonRows.get(4)); + } + @Test void acknowledgeAlarmCommandForwardsOptionsAndPrintsReply() { FakeClientFactory factory = new FakeClientFactory(); @@ -1409,6 +1426,12 @@ final class MxGatewayCliTests { .setTransitionKind(AlarmTransitionKind.ALARM_TRANSITION_KIND_ACKNOWLEDGE) .setSeverity(700)) .build()); + observer.onNext(AlarmFeedMessage.newBuilder() + .setProviderStatus(AlarmProviderStatus.newBuilder().setDegraded(false)) + .build()); + observer.onNext(AlarmFeedMessage.newBuilder() + .setSnapshotStatus(AlarmSnapshotStatus.newBuilder().setTruncated(false)) + .build()); observer.onCompleted(); return new MxGatewayAlarmFeedSubscription(); } diff --git a/clients/python/README.md b/clients/python/README.md index 339c700..c9904be 100644 --- a/clients/python/README.md +++ b/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 --client-name python-client --json mxgw-py add-item --session-id --server-handle 1 --item Object.Attribute --json mxgw-py advise --session-id --server-handle 1 --item-handle 2 --json mxgw-py stream-events --session-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 --server-handle 1 --item-handle 2 --type int32 --value 123 --json ``` diff --git a/clients/python/src/zb_mom_ww_mxgateway_cli/commands.py b/clients/python/src/zb_mom_ww_mxgateway_cli/commands.py index 005949c..6fbe2aa 100644 --- a/clients/python/src/zb_mom_ww_mxgateway_cli/commands.py +++ b/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, *, diff --git a/clients/python/tests/test_review_findings_022_to_026.py b/clients/python/tests/test_review_findings_022_to_026.py index b9a046e..81a1477 100644 --- a/clients/python/tests/test_review_findings_022_to_026.py +++ b/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: diff --git a/clients/rust/README.md b/clients/rust/README.md index f8853d9..119b36f 100644 --- a/clients/rust/README.md +++ b/clients/rust/README.md @@ -77,7 +77,7 @@ cargo run -p mxgw-cli -- register --session-id --client-name mxgw-r cargo run -p mxgw-cli -- add-item --session-id --server-handle 1 --item TestChildObject.TestInt --json cargo run -p mxgw-cli -- advise --session-id --server-handle 1 --item-handle 1 --json cargo run -p mxgw-cli -- stream-events --session-id --max-events 1 --json -cargo run -p mxgw-cli -- stream-alarms --session-id --max-messages 1 --json +cargo run -p mxgw-cli -- stream-alarms --session-id --max-messages 4 --json cargo run -p mxgw-cli -- acknowledge-alarm --session-id --alarm-reference "\\Galaxy\Area001.Pump001.PumpFault" --json cargo run -p mxgw-cli -- write --session-id --server-handle 1 --item-handle 1 --value-type int32 --value 123 --json ``` @@ -141,6 +141,11 @@ every verdict change — including the clearing frame 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`. +Unlike the protobuf-JSON formatters used by the other CLIs, the Rust CLI's +hand-built JSON status rows always include `"degraded": false` and +`"truncated": false`. This deliberate representational difference is +documented in the cross-language smoke matrix; both shapes mean the same thing. + The session helpers keep MXAccess handles visible: ```rust diff --git a/docs/CrossLanguageSmokeMatrix.md b/docs/CrossLanguageSmokeMatrix.md index 4a7e0bb..fc0a0fc 100644 --- a/docs/CrossLanguageSmokeMatrix.md +++ b/docs/CrossLanguageSmokeMatrix.md @@ -66,6 +66,23 @@ declaration-ordered JSON; and the row sits alone on its own line for Go and for Rust's `--jsonl`, but inside an `events` array for Python and for Rust's aggregate `--json`. +Alarm-feed status rows deliberately preserve each CLI's established JSON +mapping while making false values explicit in every text formatter: + +| CLI | Healthy provider text | Clearing snapshot text | JSON false-value shape | +|-----|-----------------------|------------------------|------------------------| +| `mxgw-dotnet` | `provider-status mode=Unspecified degraded=false reason=""` | `snapshot-status truncated=false` | `{"providerStatus": {}}` / `{"snapshotStatus": {}}` | +| `mxgw-go` | `provider-status mode=ALARM_PROVIDER_MODE_UNSPECIFIED degraded=false reason=""` | `snapshot-status truncated=false` | `{"providerStatus":{}}` / `{"snapshotStatus":{}}` | +| `mxgw-py` | same JSON row as `--json` | same JSON row as `--json` | `{"providerStatus": {}}` / `{"snapshotStatus": {}}`, emitted one row at a time | +| `mxgw-java` | `provider-status mode=ALARM_PROVIDER_MODE_UNSPECIFIED degraded=false reason=` | `snapshot-status truncated=false` | `{"providerStatus":{}}` / `{"snapshotStatus":{}}` | +| `mxgw-rs` | explicit `degraded=false` summary | explicit `truncated=false` summary | always includes `"degraded":false` / `"truncated":false` | + +For the protobuf-JSON rows, an absent proto3 boolean means `false`; Rust's +always-explicit booleans are semantically equivalent. Python `stream-alarms` +defaults to an unbounded feed and emits each row immediately; a positive +`--max-messages` value provides a bounded sample and includes both status +preamble rows in the count. + ## Integration Gate Cross-language smoke execution is opt-in. Runners should skip the matrix unless