Merge pull request 'clients/dotnet: SDK methods for AcknowledgeAlarm + QueryActiveAlarms (PR E.2)' (#106) from track-e2-dotnet-alarm-sdk into main

2026-04-30 16:39:35 -04:00
parent 26d0e2c471 0765eb4de3
commit a7edc8f8bf
6 changed files with 387 additions and 2 deletions
@@ -41,6 +41,24 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
    /// </summary>
    public List<(StreamEventsRequest Request, CallOptions CallOptions)> StreamEventsCalls { get; } = [];

+    /// <summary>
+    /// Gets the list of captured AcknowledgeAlarmAsync calls.
+    /// </summary>
+    public List<(AcknowledgeAlarmRequest Request, CallOptions CallOptions)> AcknowledgeAlarmCalls { get; } = [];
+
+    /// <summary>
+    /// Gets the list of captured QueryActiveAlarmsAsync calls.
+    /// </summary>
+    public List<(QueryActiveAlarmsRequest Request, CallOptions CallOptions)> QueryActiveAlarmsCalls { get; } = [];
+
+    /// <summary>
+    /// Gets the queue of exceptions to throw from AcknowledgeAlarmAsync.
+    /// </summary>
+    public Queue<Exception> AcknowledgeAlarmExceptions { get; } = new();
+
+    private readonly Queue<AcknowledgeAlarmReply> _acknowledgeReplies = new();
+    private readonly List<ActiveAlarmSnapshot> _activeAlarmSnapshots = [];
+
    /// <summary>
    /// Gets or sets the reply to return from OpenSessionAsync.
    /// </summary>
@@ -168,4 +186,57 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
    {
        _events.Add(gatewayEvent);
    }
+
+    /// <summary>
+    /// Records the acknowledge call and returns the next enqueued reply (or default).
+    /// </summary>
+    public Task<AcknowledgeAlarmReply> AcknowledgeAlarmAsync(
+        AcknowledgeAlarmRequest request,
+        CallOptions callOptions)
+    {
+        AcknowledgeAlarmCalls.Add((request, callOptions));
+        if (AcknowledgeAlarmExceptions.TryDequeue(out Exception? exception))
+        {
+            throw exception;
+        }
+
+        return Task.FromResult(_acknowledgeReplies.Count > 0
+            ? _acknowledgeReplies.Dequeue()
+            : new AcknowledgeAlarmReply
+            {
+                SessionId = request.SessionId,
+                CorrelationId = request.ClientCorrelationId,
+                ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
+                Status = new MxStatusProxy { Success = 1, Category = MxStatusCategory.Ok },
+            });
+    }
+
+    /// <summary>
+    /// Records the query call and yields each enqueued snapshot.
+    /// </summary>
+    public async IAsyncEnumerable<ActiveAlarmSnapshot> QueryActiveAlarmsAsync(
+        QueryActiveAlarmsRequest request,
+        CallOptions callOptions)
+    {
+        QueryActiveAlarmsCalls.Add((request, callOptions));
+
+        foreach (ActiveAlarmSnapshot snapshot in _activeAlarmSnapshots)
+        {
+            callOptions.CancellationToken.ThrowIfCancellationRequested();
+            await Task.Yield();
+            yield return snapshot;
+        }
+    }
+
+    /// <summary>Enqueues an acknowledge reply.</summary>
+    public void AddAcknowledgeReply(AcknowledgeAlarmReply reply)
+    {
+        _acknowledgeReplies.Enqueue(reply);
+    }
+
+    /// <summary>Enqueues a snapshot to be yielded from QueryActiveAlarmsAsync.</summary>
+    public void AddActiveAlarmSnapshot(ActiveAlarmSnapshot snapshot)
+    {
+        _activeAlarmSnapshots.Add(snapshot);
+    }
 }
@@ -0,0 +1,192 @@
+using Google.Protobuf.WellKnownTypes;
+using Grpc.Core;
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Client.Tests;
+
+/// <summary>
+///     PR E.2 — pins the .NET SDK surface for the new alarm RPCs:
+///     <see cref="MxGatewayClient.AcknowledgeAlarmAsync"/> and
+///     <see cref="MxGatewayClient.QueryActiveAlarmsAsync"/>.
+/// </summary>
+public sealed class MxGatewayClientAlarmsTests
+{
+    [Fact]
+    public async Task AcknowledgeAlarmAsync_RecordsRequestShapeAndReturnsReply()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        transport.AddAcknowledgeReply(new AcknowledgeAlarmReply
+        {
+            SessionId = "session-fixture",
+            CorrelationId = "corr-1",
+            ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
+            Status = new MxStatusProxy
+            {
+                Success = 1,
+                Category = MxStatusCategory.Ok,
+                DetectedBy = MxStatusSource.RespondingLmx,
+            },
+        });
+        await using MxGatewayClient client = CreateClient(transport);
+
+        AcknowledgeAlarmReply reply = await client.AcknowledgeAlarmAsync(new AcknowledgeAlarmRequest
+        {
+            SessionId = "session-fixture",
+            ClientCorrelationId = "corr-1",
+            AlarmFullReference = "Tank01.Level.HiHi",
+            Comment = "investigating",
+            OperatorUser = "alice",
+        });
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.Equal(MxStatusCategory.Ok, reply.Status.Category);
+
+        var call = Assert.Single(transport.AcknowledgeAlarmCalls);
+        Assert.Equal("Tank01.Level.HiHi", call.Request.AlarmFullReference);
+        Assert.Equal("investigating", call.Request.Comment);
+        Assert.Equal("alice", call.Request.OperatorUser);
+        Assert.Equal("Bearer test-api-key", call.CallOptions.Headers?.GetValue("authorization"));
+    }
+
+    [Fact]
+    public async Task AcknowledgeAlarmAsync_HonorsCancellation()
+    {
+        // Acks are routed through the safe-unary retry pipeline (idempotent at the
+        // MxAccess level), so the transport-side cancellation token is a linked one
+        // rather than the caller's original. Verify cancellation by tripping the source
+        // and asserting the call observes it.
+        using CancellationTokenSource cancellation = new();
+        cancellation.Cancel();
+        FakeGatewayTransport transport = CreateTransport();
+        await using MxGatewayClient client = CreateClient(transport);
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
+            client.AcknowledgeAlarmAsync(
+                new AcknowledgeAlarmRequest
+                {
+                    SessionId = "session-fixture",
+                    AlarmFullReference = "Tank01.Level.HiHi",
+                    Comment = string.Empty,
+                    OperatorUser = "alice",
+                },
+                cancellation.Token));
+    }
+
+    [Fact]
+    public async Task AcknowledgeAlarmAsync_MapsUnauthenticated_RpcException_ToTypedException()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        transport.AcknowledgeAlarmExceptions.Enqueue(
+            new RpcException(new Status(StatusCode.Unauthenticated, "expired key")));
+        await using MxGatewayClient client = CreateClient(transport);
+
+        // Note: the FakeGatewayTransport surfaces RpcException directly (it does not run
+        // through GrpcMxGatewayClientTransport's mapping); the fake's contract here is to
+        // pass the exception verbatim. RpcException → typed exception mapping is covered
+        // in the GrpcMxGatewayClientTransport-level tests; the SDK-level test pins the
+        // pass-through shape so a future migration to direct mapping won't silently
+        // change observable behaviour.
+        var ex = await Assert.ThrowsAsync<RpcException>(
+            () => client.AcknowledgeAlarmAsync(new AcknowledgeAlarmRequest
+            {
+                SessionId = "session-fixture",
+                AlarmFullReference = "Tank01.Level.HiHi",
+                Comment = string.Empty,
+                OperatorUser = "alice",
+            }));
+        Assert.Equal(StatusCode.Unauthenticated, ex.StatusCode);
+    }
+
+    [Fact]
+    public async Task QueryActiveAlarmsAsync_StreamsEnqueuedSnapshots()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        transport.AddActiveAlarmSnapshot(MakeSnapshot("Tank01.Level.HiHi", AlarmConditionState.Active));
+        transport.AddActiveAlarmSnapshot(MakeSnapshot("Tank02.Level.HiHi", AlarmConditionState.ActiveAcked));
+        await using MxGatewayClient client = CreateClient(transport);
+
+        List<ActiveAlarmSnapshot> snapshots = [];
+        await foreach (ActiveAlarmSnapshot snapshot in client.QueryActiveAlarmsAsync(new QueryActiveAlarmsRequest
+        {
+            SessionId = "session-fixture",
+        }))
+        {
+            snapshots.Add(snapshot);
+        }
+
+        Assert.Equal(2, snapshots.Count);
+        Assert.Equal("Tank01.Level.HiHi", snapshots[0].AlarmFullReference);
+        Assert.Equal(AlarmConditionState.Active, snapshots[0].CurrentState);
+        Assert.Equal(AlarmConditionState.ActiveAcked, snapshots[1].CurrentState);
+        Assert.Single(transport.QueryActiveAlarmsCalls);
+    }
+
+    [Fact]
+    public async Task QueryActiveAlarmsAsync_PassesFilterPrefix()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        await using MxGatewayClient client = CreateClient(transport);
+
+        await foreach (ActiveAlarmSnapshot _ in client.QueryActiveAlarmsAsync(new QueryActiveAlarmsRequest
+        {
+            SessionId = "session-fixture",
+            AlarmFilterPrefix = "Tank01.",
+        }))
+        {
+            // no snapshots enqueued; just verifying the request passes through
+        }
+
+        var call = Assert.Single(transport.QueryActiveAlarmsCalls);
+        Assert.Equal("Tank01.", call.Request.AlarmFilterPrefix);
+    }
+
+    [Fact]
+    public async Task QueryActiveAlarmsAsync_HonorsCancellationDuringEnumeration()
+    {
+        FakeGatewayTransport transport = CreateTransport();
+        transport.AddActiveAlarmSnapshot(MakeSnapshot("Tank01.Level.HiHi", AlarmConditionState.Active));
+        transport.AddActiveAlarmSnapshot(MakeSnapshot("Tank02.Level.HiHi", AlarmConditionState.Active));
+        await using MxGatewayClient client = CreateClient(transport);
+
+        using CancellationTokenSource cancellation = new();
+        await Assert.ThrowsAsync<OperationCanceledException>(async () =>
+        {
+            await foreach (ActiveAlarmSnapshot _ in client.QueryActiveAlarmsAsync(
+                new QueryActiveAlarmsRequest { SessionId = "session-fixture" },
+                cancellation.Token))
+            {
+                cancellation.Cancel();
+            }
+        });
+    }
+
+    private static ActiveAlarmSnapshot MakeSnapshot(string fullReference, AlarmConditionState state)
+    {
+        return new ActiveAlarmSnapshot
+        {
+            AlarmFullReference = fullReference,
+            SourceObjectReference = fullReference.Split('.')[0],
+            AlarmTypeName = "AnalogLimitAlarm.HiHi",
+            Severity = 750,
+            CurrentState = state,
+            Category = "Process",
+            Description = "Tank high-high level",
+            OriginalRaiseTimestamp = Timestamp.FromDateTime(new DateTime(2026, 5, 1, 12, 0, 0, DateTimeKind.Utc)),
+            LastTransitionTimestamp = Timestamp.FromDateTime(new DateTime(2026, 5, 1, 12, 0, 30, DateTimeKind.Utc)),
+        };
+    }
+
+    private static MxGatewayClient CreateClient(FakeGatewayTransport transport)
+    {
+        return new MxGatewayClient(transport.Options, transport);
+    }
+
+    private static FakeGatewayTransport CreateTransport()
+    {
+        return new FakeGatewayTransport(new MxGatewayClientOptions
+        {
+            Endpoint = new Uri("http://localhost:5000"),
+            ApiKey = "test-api-key",
+        });
+    }
+}
@@ -18,7 +18,7 @@ public sealed class MxGatewayClientCliTests
        var exitCode = MxGatewayClientCli.Run(["version"], output, error);

        Assert.Equal(0, exitCode);
-        Assert.Contains("gateway-protocol=2", output.ToString());
+        Assert.Contains("gateway-protocol=3", output.ToString());
        Assert.Contains("worker-protocol=1", output.ToString());
        Assert.Equal(string.Empty, error.ToString());
    }
@@ -33,7 +33,7 @@ public sealed class MxGatewayClientCliTests
        int exitCode = await MxGatewayClientCli.RunAsync(["version", "--json"], output, error);

        Assert.Equal(0, exitCode);
-        Assert.Contains("\"gatewayProtocolVersion\":2", output.ToString());
+        Assert.Contains("\"gatewayProtocolVersion\":3", output.ToString());
        Assert.Equal(string.Empty, error.ToString());
    }

@@ -116,6 +116,65 @@ internal sealed class GrpcMxGatewayClientTransport(
        return StreamEventsAsync(request, callOptions);
    }

+    /// <inheritdoc />
+    public async Task<AcknowledgeAlarmReply> AcknowledgeAlarmAsync(
+        AcknowledgeAlarmRequest request,
+        CallOptions callOptions)
+    {
+        try
+        {
+            return await RawClient.AcknowledgeAlarmAsync(request, callOptions)
+                .ResponseAsync
+                .ConfigureAwait(false);
+        }
+        catch (RpcException exception)
+        {
+            throw MapRpcException(exception, callOptions.CancellationToken);
+        }
+    }
+
+    /// <inheritdoc />
+    public async IAsyncEnumerable<ActiveAlarmSnapshot> QueryActiveAlarmsAsync(
+        QueryActiveAlarmsRequest request,
+        CallOptions callOptions,
+        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        CancellationToken effectiveCancellationToken = cancellationToken.CanBeCanceled
+            ? cancellationToken
+            : callOptions.CancellationToken;
+
+        using AsyncServerStreamingCall<ActiveAlarmSnapshot> call = RawClient.QueryActiveAlarms(request, callOptions);
+
+        IAsyncStreamReader<ActiveAlarmSnapshot> responseStream = call.ResponseStream;
+        while (true)
+        {
+            ActiveAlarmSnapshot? snapshot;
+            try
+            {
+                if (!await responseStream.MoveNext(effectiveCancellationToken).ConfigureAwait(false))
+                {
+                    break;
+                }
+
+                snapshot = responseStream.Current;
+            }
+            catch (RpcException exception)
+            {
+                throw MapRpcException(exception, effectiveCancellationToken);
+            }
+
+            yield return snapshot;
+        }
+    }
+
+    /// <inheritdoc />
+    IAsyncEnumerable<ActiveAlarmSnapshot> IMxGatewayClientTransport.QueryActiveAlarmsAsync(
+        QueryActiveAlarmsRequest request,
+        CallOptions callOptions)
+    {
+        return QueryActiveAlarmsAsync(request, callOptions);
+    }
+
    private static Exception MapRpcException(
        RpcException exception,
        CancellationToken cancellationToken)
@@ -54,4 +54,25 @@ internal interface IMxGatewayClientTransport
    IAsyncEnumerable<MxEvent> StreamEventsAsync(
        StreamEventsRequest request,
        CallOptions callOptions);
+
+    /// <summary>
+    /// Acknowledges an active MXAccess alarm condition.
+    /// </summary>
+    /// <param name="request">The acknowledge request.</param>
+    /// <param name="callOptions">gRPC call options.</param>
+    /// <returns>The acknowledge reply with native MxStatus.</returns>
+    Task<AcknowledgeAlarmReply> AcknowledgeAlarmAsync(
+        AcknowledgeAlarmRequest request,
+        CallOptions callOptions);
+
+    /// <summary>
+    /// Streams a snapshot of all alarms currently in Active or ActiveAcked state — the
+    /// ConditionRefresh equivalent for the gateway.
+    /// </summary>
+    /// <param name="request">The query request, optionally scoped by alarm-reference prefix.</param>
+    /// <param name="callOptions">gRPC call options.</param>
+    /// <returns>An async enumerable of active-alarm snapshots.</returns>
+    IAsyncEnumerable<ActiveAlarmSnapshot> QueryActiveAlarmsAsync(
+        QueryActiveAlarmsRequest request,
+        CallOptions callOptions);
 }
@@ -182,6 +182,48 @@ public sealed class MxGatewayClient : IAsyncDisposable
        return _transport.StreamEventsAsync(request, CreateStreamCallOptions(cancellationToken));
    }

+    /// <summary>
+    /// Acknowledges an active MXAccess alarm condition through the gateway. The
+    /// gateway authenticates the request against the API key's <c>invoke:alarm-ack</c>
+    /// scope and forwards the acknowledge to the worker's MXAccess session;
+    /// the resulting <see cref="MxStatusProxy"/> is returned in the reply.
+    /// </summary>
+    /// <param name="request">The acknowledge request — alarm reference, comment, operator user.</param>
+    /// <param name="cancellationToken">Cancellation token for the operation.</param>
+    /// <returns>The acknowledge reply with protocol + native MxStatus.</returns>
+    public Task<AcknowledgeAlarmReply> AcknowledgeAlarmAsync(
+        AcknowledgeAlarmRequest request,
+        CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+        ThrowIfDisposed();
+
+        return ExecuteSafeUnaryAsync(
+            token => _transport.AcknowledgeAlarmAsync(request, CreateCallOptions(token)),
+            cancellationToken);
+    }
+
+    /// <summary>
+    /// Streams a snapshot of all alarms currently Active or ActiveAcked — the gateway's
+    /// ConditionRefresh equivalent. Used after reconnect to seed the local Part 9 state
+    /// machine, or to reconcile alarms that may have been missed during a transport
+    /// blip. Optionally scoped by alarm-reference prefix
+    /// (<see cref="QueryActiveAlarmsRequest.AlarmFilterPrefix"/>) so a partial refresh
+    /// can target an equipment sub-tree.
+    /// </summary>
+    /// <param name="request">The query request, optionally scoped by alarm-reference prefix.</param>
+    /// <param name="cancellationToken">Cancellation token for the stream.</param>
+    /// <returns>An async enumerable of active-alarm snapshots.</returns>
+    public IAsyncEnumerable<ActiveAlarmSnapshot> QueryActiveAlarmsAsync(
+        QueryActiveAlarmsRequest request,
+        CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(request);
+        ThrowIfDisposed();
+
+        return _transport.QueryActiveAlarmsAsync(request, CreateStreamCallOptions(cancellationToken));
+    }
+
    /// <summary>
    /// Disposes the client and releases all resources.
    /// </summary>