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

using System.Collections.Concurrent;
using System.Net;
using Grpc.Core;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Hosting.Server;
using Microsoft.AspNetCore.Hosting.Server.Features;
using Microsoft.AspNetCore.Server.Kestrel.Core;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.MxGateway.Contracts.Proto;

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

/// <summary>
/// Hosts the real <c>mxaccess_gateway.v1.MxAccessGateway</c> service on a loopback
/// Kestrel endpoint so client tests exercise genuine HTTP/2 framing, protobuf
/// serialization, call metadata, and gRPC status propagation.
/// </summary>
/// <remarks>
/// <para>
/// This is the counterpart of <see cref="FakeGatewayTransport"/>: that fake replaces
/// <c>IMxGatewayClientTransport</c>, so nothing below the client wrapper runs. This one
/// replaces only the gateway's <em>behaviour</em> — every byte between the client and
/// the service is the real wire format. Contract breaks that a transport fake cannot
/// see (a field the client never decodes, metadata it does not actually send, a status
/// code it maps differently once it arrives as a real <see cref="RpcException"/>) fail
/// here.
/// </para>
/// <para>
/// Plaintext h2c is used deliberately: TLS is covered by
/// <c>MxGatewayClientTlsHandlerTests</c>, and h2c keeps the harness certificate-free so
/// it runs identically on every CI host. See <c>docs/GatewayTesting.md</c>
/// (Client Wire Tests) for the shared pattern and its Python counterpart.
/// </para>
/// </remarks>
internal sealed class WireFakeGatewayServer : IAsyncDisposable
{
    private readonly WebApplication _app;

    private WireFakeGatewayServer(WebApplication app, FakeGatewayService service, int port)
    {
        _app = app;
        Service = service;
        Endpoint = new Uri($"http://127.0.0.1:{port}");
    }

    /// <summary>
    /// Gets the canned service backing the endpoint; tests read its recorded requests.
    /// </summary>
    public FakeGatewayService Service { get; }

    /// <summary>
    /// Gets the h2c endpoint to point <see cref="MxGatewayClientOptions.Endpoint"/> at.
    /// </summary>
    public Uri Endpoint { get; }

    /// <summary>
    /// Starts a server on an ephemeral loopback port.
    /// </summary>
    /// <param name="configure">Optional configuration of the canned service.</param>
    /// <returns>The started server.</returns>
    public static async Task<WireFakeGatewayServer> StartAsync(Action<FakeGatewayService>? configure = null)
    {
        FakeGatewayService service = new();
        configure?.Invoke(service);

        WebApplicationBuilder builder = WebApplication.CreateBuilder();
        builder.Logging.ClearProviders();
        builder.WebHost.ConfigureKestrel(options =>
            // Port 0 lets the OS pick; HTTP/2 without TLS (h2c) is what the client's
            // plain http:// endpoint negotiates via RequestVersionExact.
            options.Listen(IPAddress.Loopback, 0, listen => listen.Protocols = HttpProtocols.Http2));
        builder.Services.AddGrpc();
        builder.Services.AddSingleton(service);

        WebApplication app = builder.Build();
        app.MapGrpcService<FakeGatewayService>();
        await app.StartAsync().ConfigureAwait(false);

        return new WireFakeGatewayServer(app, service, ResolvePort(app));
    }

    /// <summary>
    /// Creates a client bound to this server's endpoint.
    /// </summary>
    /// <param name="apiKey">API key the client should present.</param>
    /// <returns>A client that talks to this server over h2c.</returns>
    public MxGatewayClient CreateClient(string apiKey) =>
        CreateClient(apiKey, TimeSpan.FromSeconds(30), new MxGatewayClientRetryOptions());

    /// <summary>
    /// Creates a client bound to this server's endpoint with an explicit call timeout and retry policy.
    /// </summary>
    /// <param name="apiKey">API key the client should present.</param>
    /// <param name="defaultCallTimeout">Per-attempt unary deadline.</param>
    /// <param name="retry">Retry options for safe unary calls.</param>
    /// <returns>A client that talks to this server over h2c.</returns>
    public MxGatewayClient CreateClient(
        string apiKey,
        TimeSpan defaultCallTimeout,
        MxGatewayClientRetryOptions retry) =>
        MxGatewayClient.Create(new MxGatewayClientOptions
        {
            Endpoint = Endpoint,
            ApiKey = apiKey,
            UseTls = false,
            DefaultCallTimeout = defaultCallTimeout,
            Retry = retry,
        });

    /// <summary>Stops and disposes the underlying web application.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async ValueTask DisposeAsync()
    {
        await _app.StopAsync().ConfigureAwait(false);
        await _app.DisposeAsync().ConfigureAwait(false);
    }

    private static int ResolvePort(WebApplication app)
    {
        IServerAddressesFeature? addresses = app.Services
            .GetRequiredService<IServer>()
            .Features
            .Get<IServerAddressesFeature>();
        string address = addresses?.Addresses.FirstOrDefault()
            ?? throw new InvalidOperationException("Kestrel did not report a bound address.");
        return new Uri(address).Port;
    }

    /// <summary>
    /// Canned gateway answering the four session RPCs with gateway-shaped replies.
    /// </summary>
    internal sealed class FakeGatewayService : MxAccessGateway.MxAccessGatewayBase
    {
        /// <summary>The session id every reply carries.</summary>
        public const string SessionId = "wire-session-1";

        /// <summary>The server handle the canned Register reply returns.</summary>
        public const int ServerHandle = 4242;

        /// <summary>The item handle the canned data-change event carries.</summary>
        public const int ItemHandle = 77;

        /// <summary>
        /// Gets the <c>authorization</c> header value observed per RPC name.
        /// </summary>
        public ConcurrentDictionary<string, string> AuthorizationByMethod { get; } = new();

        /// <summary>
        /// Gets or sets a value indicating whether <c>Invoke</c> fails with
        /// <see cref="StatusCode.PermissionDenied"/> instead of replying.
        /// </summary>
        public bool DenyInvoke { get; set; }

        /// <summary>
        /// Gets or sets the replay-gap sentinel emitted at the head of the event stream.
        /// </summary>
        public ReplayGap? ReplayGap { get; set; }

        /// <summary>
        /// Gets or sets how many leading <c>Invoke</c> calls stall until the client's deadline
        /// expires, so the client sees <see cref="StatusCode.DeadlineExceeded"/> for them.
        /// </summary>
        public int StallFirstInvokes { get; set; }

        /// <summary>Gets the number of <c>Invoke</c> calls the server received.</summary>
        public int InvokeCount => Volatile.Read(ref _invokeCount);

        private int _invokeCount;

        /// <summary>
        /// Gets the last <c>Invoke</c> request the client sent, as decoded from the wire.
        /// </summary>
        public MxCommandRequest? InvokeRequest { get; private set; }

        /// <summary>
        /// Gets the last <c>StreamEvents</c> request the client sent.
        /// </summary>
        public StreamEventsRequest? StreamEventsRequest { get; private set; }

        /// <summary>
        /// Gets the last <c>CloseSession</c> request the client sent.
        /// </summary>
        public CloseSessionRequest? CloseSessionRequest { get; private set; }

        /// <inheritdoc />
        public override Task<OpenSessionReply> OpenSession(
            OpenSessionRequest request,
            ServerCallContext context)
        {
            Record(context);
            return Task.FromResult(new OpenSessionReply
            {
                SessionId = SessionId,
                BackendName = "fake-backend",
                WorkerProcessId = 1234,
                WorkerProtocolVersion = 1,
                GatewayProtocolVersion = 3,
                Capabilities = { "events", "invoke" },
                ProtocolStatus = Ok(),
            });
        }

        /// <inheritdoc />
        public override async Task<MxCommandReply> Invoke(MxCommandRequest request, ServerCallContext context)
        {
            Record(context);
            InvokeRequest = request;

            if (Interlocked.Increment(ref _invokeCount) <= StallFirstInvokes)
            {
                // Hold the call past the client's deadline; the cancellation is the deadline
                // firing (or the client giving up), either way the caller has stopped waiting.
                try
                {
                    await Task.Delay(Timeout.InfiniteTimeSpan, context.CancellationToken).ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                }

                throw new RpcException(new Status(StatusCode.DeadlineExceeded, "stalled attempt"));
            }

            if (DenyInvoke)
            {
                throw new RpcException(new Status(StatusCode.PermissionDenied, "invoke scope required"));
            }

            return new MxCommandReply
            {
                SessionId = request.SessionId,
                CorrelationId = request.ClientCorrelationId,
                Kind = request.Command.Kind,
                ProtocolStatus = Ok(),
                Hresult = 0,
                Register = new RegisterReply { ServerHandle = ServerHandle },
            };
        }

        /// <inheritdoc />
        public override async Task StreamEvents(
            StreamEventsRequest request,
            IServerStreamWriter<MxEvent> responseStream,
            ServerCallContext context)
        {
            Record(context);
            StreamEventsRequest = request;

            if (ReplayGap is not null)
            {
                // The sentinel shape the gateway emits: family unspecified, body unset,
                // only replay_gap populated.
                await responseStream.WriteAsync(new MxEvent
                {
                    SessionId = request.SessionId,
                    ReplayGap = ReplayGap,
                }).ConfigureAwait(false);
            }

            await responseStream.WriteAsync(new MxEvent
            {
                SessionId = request.SessionId,
                Family = MxEventFamily.OnDataChange,
                ServerHandle = ServerHandle,
                ItemHandle = ItemHandle,
                Value = new MxValue { Int32Value = 17 },
                Quality = 192,
                WorkerSequence = 9,
                OnDataChange = new OnDataChangeEvent(),
            }).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public override Task<CloseSessionReply> CloseSession(
            CloseSessionRequest request,
            ServerCallContext context)
        {
            Record(context);
            CloseSessionRequest = request;
            return Task.FromResult(new CloseSessionReply
            {
                SessionId = request.SessionId,
                FinalState = SessionState.Closed,
                ProtocolStatus = Ok(),
            });
        }

        private static ProtocolStatus Ok() => new() { Code = ProtocolStatusCode.Ok };

        private void Record(ServerCallContext context)
        {
            string? authorization = context.RequestHeaders.GetValue("authorization");
            if (authorization is not null)
            {
                // context.Method is the fully-qualified "/package.Service/Method";
                // key on the bare method name so assertions stay readable.
                AuthorizationByMethod[context.Method[(context.Method.LastIndexOf('/') + 1)..]] =
                    authorization;
            }
        }
    }
}