mxaccessgw/clients/dotnet/MxGateway.Client.Tests/FakeGatewayTransport.cs

using Grpc.Core;
using MxGateway.Contracts.Proto;

namespace MxGateway.Client.Tests;

/// <summary>
/// Fake implementation of IMxGatewayClientTransport for testing.
/// </summary>
internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMxGatewayClientTransport
{
    private readonly Queue<MxCommandReply> _invokeReplies = new();
    private readonly List<MxEvent> _events = [];

    /// <summary>
    /// Gets the gateway client options.
    /// </summary>
    public MxGatewayClientOptions Options { get; } = options;

    /// <summary>
    /// Gets null, since this is a test fake without a real gRPC client.
    /// </summary>
    public MxAccessGateway.MxAccessGatewayClient? RawClient => null;

    /// <summary>
    /// Gets the list of captured OpenSessionAsync calls.
    /// </summary>
    public List<(OpenSessionRequest Request, CallOptions CallOptions)> OpenSessionCalls { get; } = [];

    /// <summary>
    /// Gets the list of captured CloseSessionAsync calls.
    /// </summary>
    public List<(CloseSessionRequest Request, CallOptions CallOptions)> CloseSessionCalls { get; } = [];

    /// <summary>
    /// Gets the list of captured InvokeAsync calls.
    /// </summary>
    public List<(MxCommandRequest Request, CallOptions CallOptions)> InvokeCalls { get; } = [];

    /// <summary>
    /// Gets the list of captured StreamEventsAsync calls.
    /// </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 StreamAlarmsAsync calls.
    /// </summary>
    public List<(StreamAlarmsRequest Request, CallOptions CallOptions)> StreamAlarmsCalls { 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>
    public OpenSessionReply OpenSessionReply { get; set; } = new()
    {
        SessionId = "session-fixture",
        BackendName = "mxaccess-worker",
        GatewayProtocolVersion = 1,
        WorkerProtocolVersion = 1,
        ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
    };

    /// <summary>
    /// Gets or sets the reply to return from CloseSessionAsync.
    /// </summary>
    public CloseSessionReply CloseSessionReply { get; set; } = new()
    {
        SessionId = "session-fixture",
        FinalState = SessionState.Closed,
        ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
    };

    /// <summary>
    /// Gets the queue of exceptions to throw from OpenSessionAsync.
    /// </summary>
    public Queue<Exception> OpenSessionExceptions { get; } = new();

    /// <summary>
    /// Gets the queue of exceptions to throw from CloseSessionAsync.
    /// </summary>
    public Queue<Exception> CloseSessionExceptions { get; } = new();

    /// <summary>
    /// Gets or sets a value indicating whether thrown <see cref="RpcException"/>s are mapped
    /// to <see cref="MxGatewayException"/> the way the production gRPC transport does. Lets
    /// retry tests exercise the wrapped-exception predicate branch that runs in production.
    /// </summary>
    public bool MapTransportExceptions { get; set; }

    /// <summary>
    /// Gets or sets an optional hook awaited inside CloseSessionAsync after the call is
    /// recorded; lets tests pause a close mid-flight to observe concurrent dispose.
    /// </summary>
    public Func<Task>? CloseSessionHook { get; set; }

    /// <summary>
    /// Gets the queue of exceptions to throw from InvokeAsync.
    /// </summary>
    public Queue<Exception> InvokeExceptions { get; } = new();

    /// <summary>
    /// Verifies that the OpenSessionAsync call is recorded and returns the configured reply.
    /// </summary>
    /// <param name="request">The OpenSessionRequest to process.</param>
    /// <param name="callOptions">Call options specifying RPC behavior.</param>
    public Task<OpenSessionReply> OpenSessionAsync(
        OpenSessionRequest request,
        CallOptions callOptions)
    {
        OpenSessionCalls.Add((request, callOptions));
        if (OpenSessionExceptions.TryDequeue(out Exception? exception))
        {
            throw Translate(exception, callOptions);
        }

        return Task.FromResult(OpenSessionReply);
    }

    /// <summary>
    /// Verifies that the CloseSessionAsync call is recorded and returns the configured reply.
    /// </summary>
    /// <param name="request">The CloseSessionRequest to process.</param>
    /// <param name="callOptions">Call options specifying RPC behavior.</param>
    public async Task<CloseSessionReply> CloseSessionAsync(
        CloseSessionRequest request,
        CallOptions callOptions)
    {
        CloseSessionCalls.Add((request, callOptions));

        if (CloseSessionHook is not null)
        {
            await CloseSessionHook().ConfigureAwait(false);
        }

        if (CloseSessionExceptions.TryDequeue(out Exception? exception))
        {
            throw Translate(exception, callOptions);
        }

        return CloseSessionReply;
    }

    /// <summary>
    /// Verifies that the InvokeAsync call is recorded and returns the next enqueued reply.
    /// </summary>
    /// <param name="request">The MxCommandRequest to process.</param>
    /// <param name="callOptions">Call options specifying RPC behavior.</param>
    public Task<MxCommandReply> InvokeAsync(
        MxCommandRequest request,
        CallOptions callOptions)
    {
        InvokeCalls.Add((request, callOptions));
        if (InvokeExceptions.TryDequeue(out Exception? exception))
        {
            throw Translate(exception, callOptions);
        }

        return Task.FromResult(_invokeReplies.Dequeue());
    }

    /// <summary>
    /// Verifies that the StreamEventsAsync call is recorded and yields all enqueued events.
    /// </summary>
    /// <param name="request">The StreamEventsRequest to process.</param>
    /// <param name="callOptions">Call options specifying RPC behavior.</param>
    public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
        StreamEventsRequest request,
        CallOptions callOptions)
    {
        StreamEventsCalls.Add((request, callOptions));

        foreach (MxEvent gatewayEvent in _events)
        {
            callOptions.CancellationToken.ThrowIfCancellationRequested();
            await Task.Yield();
            yield return gatewayEvent;
        }
    }

    /// <summary>
    /// Enqueues a reply to be returned from the next InvokeAsync call.
    /// </summary>
    /// <param name="reply">The reply to enqueue.</param>
    public void AddInvokeReply(MxCommandReply reply)
    {
        _invokeReplies.Enqueue(reply);
    }

    /// <summary>
    /// Enqueues an event to be yielded from StreamEventsAsync.
    /// </summary>
    /// <param name="gatewayEvent">The event to enqueue.</param>
    public void AddEvent(MxEvent gatewayEvent)
    {
        _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 Translate(exception, callOptions);
        }

        return Task.FromResult(_acknowledgeReplies.Count > 0
            ? _acknowledgeReplies.Dequeue()
            : new AcknowledgeAlarmReply
            {
                CorrelationId = request.ClientCorrelationId,
                ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
                Status = new MxStatusProxy { Success = 1, Category = MxStatusCategory.Ok },
            });
    }

    /// <summary>
    /// Records the call and yields each enqueued snapshot as an active-alarm
    /// feed message, then a snapshot-complete sentinel.
    /// </summary>
    public async IAsyncEnumerable<AlarmFeedMessage> StreamAlarmsAsync(
        StreamAlarmsRequest request,
        CallOptions callOptions)
    {
        StreamAlarmsCalls.Add((request, callOptions));

        foreach (ActiveAlarmSnapshot snapshot in _activeAlarmSnapshots)
        {
            callOptions.CancellationToken.ThrowIfCancellationRequested();
            await Task.Yield();
            yield return new AlarmFeedMessage { ActiveAlarm = snapshot };
        }

        yield return new AlarmFeedMessage { SnapshotComplete = true };
    }

    /// <summary>Enqueues an acknowledge reply.</summary>
    public void AddAcknowledgeReply(AcknowledgeAlarmReply reply)
    {
        _acknowledgeReplies.Enqueue(reply);
    }

    /// <summary>Enqueues a snapshot yielded from StreamAlarmsAsync as an active-alarm message.</summary>
    public void AddActiveAlarmSnapshot(ActiveAlarmSnapshot snapshot)
    {
        _activeAlarmSnapshots.Add(snapshot);
    }

    /// <summary>
    /// Maps a queued exception the way the production gRPC transport does when
    /// <see cref="MapTransportExceptions"/> is set; otherwise returns it unchanged.
    /// </summary>
    private Exception Translate(Exception exception, CallOptions callOptions)
    {
        if (MapTransportExceptions && exception is RpcException rpcException)
        {
            return RpcExceptionMapper.Map(rpcException, callOptions.CancellationToken);
        }

        return exception;
    }
}