mxaccessgw/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientRemediationTests.cs

using Grpc.Core;
using Grpc.Net.Client;
using ZB.MOM.WW.MxGateway.Contracts.Proto;

namespace ZB.MOM.WW.MxGateway.Client.Tests;

/// <summary>Tests for ack no-retry, caller-owned channels, safe dispose, typed diagnostics, and AttachSession.</summary>
public sealed class MxGatewayClientRemediationTests
{
    /// <summary>An acknowledge is a non-idempotent operator write and is never retried.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AcknowledgeAlarmAsync_DoesNotRetryTransientFailure()
    {
        FakeGatewayTransport transport = CreateTransport();
        transport.AcknowledgeAlarmExceptions.Enqueue(new RpcException(new Status(StatusCode.Unavailable, "down")));
        await using MxGatewayClient client = new(transport.Options, transport);

        await Assert.ThrowsAsync<RpcException>(() => client.AcknowledgeAlarmAsync(
            new AcknowledgeAlarmRequest { OperatorUser = "op", OperatorPassword = "pw" }));

        Assert.Single(transport.AcknowledgeAlarmCalls);
    }

    /// <summary>A client created over a caller-owned channel does not dispose it.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Create_WithExistingChannel_DoesNotDisposeChannel()
    {
        using GrpcChannel channel = GrpcChannel.ForAddress("http://localhost:1");
        MxGatewayClient gateway = MxGatewayClient.Create(channel, Options());
        GalaxyRepositoryClient galaxy = GalaxyRepositoryClient.Create(channel, Options());

        await gateway.DisposeAsync();
        await galaxy.DisposeAsync();

        // A disposed channel throws ObjectDisposedException when a call is created.
        var invoker = channel.CreateCallInvoker();
        Exception? ex = Record.Exception(() => invoker.AsyncUnaryCall(
            new Method<byte[], byte[]>(
                MethodType.Unary,
                "s",
                "m",
                Marshallers.Create(b => b, b => b),
                Marshallers.Create(b => b, b => b)),
            null,
            new CallOptions(deadline: DateTime.UtcNow.AddMilliseconds(1)),
            []).Dispose());
        Assert.IsNotType<ObjectDisposedException>(ex);
    }

    /// <summary>Create over an existing channel rejects a null channel.</summary>
    [Fact]
    public void Create_WithNullChannel_Throws()
    {
        Assert.Throws<ArgumentNullException>(() => MxGatewayClient.Create(null!, Options()));
        Assert.Throws<ArgumentNullException>(() => GalaxyRepositoryClient.Create(null!, Options()));
    }

    /// <summary>DisposeAsync swallows a gateway failure during close and still disposes.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Session_DisposeAsync_SwallowsCloseFailure()
    {
        FakeGatewayTransport transport = CreateTransport();
        transport.CloseSessionExceptions.Enqueue(new MxGatewayException("unreachable"));
        await using MxGatewayClient client = new(transport.Options, transport);
        MxGatewaySession session = await client.OpenSessionAsync();

        await session.DisposeAsync();

        Assert.Single(transport.CloseSessionCalls);
        await Assert.ThrowsAsync<ObjectDisposedException>(() => session.CloseAsync());
    }

    /// <summary>DisposeAsync after the client is disposed does not throw.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Session_DisposeAsync_AfterClientDisposed_DoesNotThrow()
    {
        FakeGatewayTransport transport = CreateTransport();
        MxGatewayClient client = new(transport.Options, transport);
        MxGatewaySession session = await client.OpenSessionAsync();
        await client.DisposeAsync();

        await session.DisposeAsync();

        await Assert.ThrowsAsync<ObjectDisposedException>(() => session.CloseAsync());
    }

    /// <summary>PingAsync sends a Ping command carrying the message.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task PingAsync_SendsPingCommand()
    {
        FakeGatewayTransport transport = CreateTransport();
        transport.AddInvokeReply(OkReply(MxCommandKind.Ping));
        await using MxGatewayClient client = new(transport.Options, transport);
        MxGatewaySession session = await client.OpenSessionAsync();

        await session.PingAsync("hello");

        MxCommandRequest request = Assert.Single(transport.InvokeCalls).Request;
        Assert.Equal(MxCommandKind.Ping, request.Command.Kind);
        Assert.Equal("hello", request.Command.Ping.Message);
        Assert.Equal("session-fixture", request.SessionId);
    }

    /// <summary>GetSessionStateAsync returns the typed state.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task GetSessionStateAsync_ReturnsState()
    {
        FakeGatewayTransport transport = CreateTransport();
        MxCommandReply reply = OkReply(MxCommandKind.GetSessionState);
        reply.SessionState = new SessionStateReply { State = SessionState.Ready };
        transport.AddInvokeReply(reply);
        await using MxGatewayClient client = new(transport.Options, transport);
        MxGatewaySession session = await client.OpenSessionAsync();

        SessionState state = await session.GetSessionStateAsync();

        Assert.Equal(SessionState.Ready, state);
        Assert.Equal(MxCommandKind.GetSessionState, Assert.Single(transport.InvokeCalls).Request.Command.Kind);
    }

    /// <summary>GetSessionStateAsync throws on a reply without payload.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task GetSessionStateAsync_MissingPayload_ThrowsMalformed()
    {
        FakeGatewayTransport transport = CreateTransport();
        transport.AddInvokeReply(OkReply(MxCommandKind.GetSessionState));
        await using MxGatewayClient client = new(transport.Options, transport);
        MxGatewaySession session = await client.OpenSessionAsync();

        await Assert.ThrowsAsync<MxGatewayMalformedReplyException>(() => session.GetSessionStateAsync());
    }

    /// <summary>GetWorkerInfoAsync returns the worker info payload.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task GetWorkerInfoAsync_ReturnsWorkerInfo()
    {
        FakeGatewayTransport transport = CreateTransport();
        MxCommandReply reply = OkReply(MxCommandKind.GetWorkerInfo);
        reply.WorkerInfo = new WorkerInfoReply { WorkerProcessId = 42, WorkerVersion = "1.2.3" };
        transport.AddInvokeReply(reply);
        await using MxGatewayClient client = new(transport.Options, transport);
        MxGatewaySession session = await client.OpenSessionAsync();

        WorkerInfoReply info = await session.GetWorkerInfoAsync();

        Assert.Equal(42, info.WorkerProcessId);
        Assert.Equal("1.2.3", info.WorkerVersion);
    }

    /// <summary>DrainEventsAsync sends max_events and returns the events.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DrainEventsAsync_ReturnsEventsAndSendsMax()
    {
        FakeGatewayTransport transport = CreateTransport();
        MxCommandReply reply = OkReply(MxCommandKind.DrainEvents);
        reply.DrainEvents = new DrainEventsReply();
        reply.DrainEvents.Events.Add(new MxEvent { SessionId = "session-fixture", WorkerSequence = 5 });
        transport.AddInvokeReply(reply);
        await using MxGatewayClient client = new(transport.Options, transport);
        MxGatewaySession session = await client.OpenSessionAsync();

        IReadOnlyList<MxEvent> events = await session.DrainEventsAsync(10);

        Assert.Equal(5UL, Assert.Single(events).WorkerSequence);
        Assert.Equal(10U, Assert.Single(transport.InvokeCalls).Request.Command.DrainEvents.MaxEvents);
    }

    /// <summary>DrainEventsAsync treats a missing payload as an empty drain.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DrainEventsAsync_MissingPayload_ReturnsEmpty()
    {
        FakeGatewayTransport transport = CreateTransport();
        transport.AddInvokeReply(OkReply(MxCommandKind.DrainEvents));
        await using MxGatewayClient client = new(transport.Options, transport);
        MxGatewaySession session = await client.OpenSessionAsync();

        Assert.Empty(await session.DrainEventsAsync());
    }

    /// <summary>AttachSession binds to an existing id without opening a session and resumes the stream.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AttachSession_BindsExistingIdWithoutOpening_AndResumesStream()
    {
        FakeGatewayTransport transport = CreateTransport();
        transport.AddEvent(new MxEvent { SessionId = "existing", WorkerSequence = 8 });
        await using MxGatewayClient client = new(transport.Options, transport);

        MxGatewaySession session = client.AttachSession("existing");
        List<MxEvent> received = [];
        await foreach (MxEvent e in session.StreamEventsAsync(afterWorkerSequence: 7))
        {
            received.Add(e);
        }

        Assert.Empty(transport.OpenSessionCalls);
        Assert.Equal("existing", session.SessionId);
        StreamEventsRequest request = Assert.Single(transport.StreamEventsCalls).Request;
        Assert.Equal("existing", request.SessionId);
        Assert.Equal(7UL, request.AfterWorkerSequence);
        Assert.Single(received);
    }

    /// <summary>AttachSession rejects blank ids and a disposed client.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task AttachSession_InvalidArguments_Throw()
    {
        FakeGatewayTransport transport = CreateTransport();
        MxGatewayClient client = new(transport.Options, transport);

        Assert.Throws<ArgumentException>(() => client.AttachSession(" "));
        await client.DisposeAsync();
        Assert.Throws<ObjectDisposedException>(() => client.AttachSession("x"));
    }

    private static MxCommandReply OkReply(MxCommandKind kind) => new()
    {
        SessionId = "session-fixture",
        Kind = kind,
        ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
    };

    private static MxGatewayClientOptions Options() => new()
    {
        Endpoint = new Uri("http://localhost:5000"),
        ApiKey = "test-api-key",
    };

    private static FakeGatewayTransport CreateTransport() => new(Options());
}