test(tst-24): drive the .NET and Python clients against real in-process gRPC servers
TST-24 asked for per-client wire tests against a fake gateway. An audit first corrected the finding's premise: Go, Rust, and Java already had them — bufconn, a loopback tonic server, and InProcessServerBuilder respectively — each already asserting the round trip, the server-observed bearer header, and the ReplayGap sentinel. The two genuine gaps were .NET (every test substituted the transport interface; the test project had no server package at all) and Python (stub monkeypatching everywhere but one opt-in TLS test). Both now serve mxaccess_gateway.v1.MxAccessGateway over a real transport — Kestrel h2c and grpc.aio, each on an ephemeral loopback port — and drive the ordinary public client API against it. Only the gateway's behaviour is canned; the framing, serialization, metadata, and status codes are genuine. Four shapes each: full round trip with every reply field asserted, the authorization header as received by the server (including on the streaming RPC), the ReplayGap sentinel surfaced as the client's typed signal, and a real PERMISSION_DENIED mapping to the typed authorization error. The .NET client was only ever compiled in CI, never tested, so the portable job gains a dotnet test step. Fixes a bug the new tests caught on their first run: Python's connect() built the grpc.aio channel inside asyncio.to_thread, and a grpc.aio channel binds to the event loop current on the constructing thread, so every non-stub connection raised 'There is no current event loop in thread'. No mock-based test could see it, and the test guarding the off-loop behaviour patched create_channel and so asserted the bug. Split resolve_channel_security (blocking TOFU probe, off-loop) from create_channel (on-loop); the guard tests now assert both halves.
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,162 @@
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Drives the public client API against <see cref="WireFakeGatewayServer"/> — a real
|
||||
/// gRPC server on loopback — so the transport, protobuf serialization, call metadata,
|
||||
/// and gRPC status mapping are all exercised. Every other test in this project
|
||||
/// substitutes <see cref="FakeGatewayTransport"/> and therefore proves nothing about
|
||||
/// what actually crosses the wire.
|
||||
/// </summary>
|
||||
public sealed class MxGatewayClientWireTests
|
||||
{
|
||||
private const string ApiKey = "mxgw_wiretest_secret";
|
||||
|
||||
/// <summary>
|
||||
/// Verifies the full session happy path decodes real wire bytes end to end.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task SessionRoundTrip_OverRealTransport_DecodesEveryReplyField()
|
||||
{
|
||||
await using WireFakeGatewayServer server = await WireFakeGatewayServer.StartAsync();
|
||||
await using MxGatewayClient client = server.CreateClient(ApiKey);
|
||||
|
||||
MxGatewaySession session = await client.OpenSessionAsync(
|
||||
new OpenSessionRequest { ClientSessionName = "wire-test" });
|
||||
|
||||
Assert.Equal(WireFakeGatewayServer.FakeGatewayService.SessionId, session.SessionId);
|
||||
Assert.Equal("fake-backend", session.OpenSessionReply.BackendName);
|
||||
Assert.Equal(1234, session.OpenSessionReply.WorkerProcessId);
|
||||
Assert.Equal(3u, session.OpenSessionReply.GatewayProtocolVersion);
|
||||
Assert.Equal(["events", "invoke"], session.OpenSessionReply.Capabilities);
|
||||
|
||||
int serverHandle = await session.RegisterAsync("wire-test-client");
|
||||
Assert.Equal(WireFakeGatewayServer.FakeGatewayService.ServerHandle, serverHandle);
|
||||
|
||||
MxCommandRequest? invoke = server.Service.InvokeRequest;
|
||||
Assert.NotNull(invoke);
|
||||
Assert.Equal(MxCommandKind.Register, invoke.Command.Kind);
|
||||
Assert.Equal("wire-test-client", invoke.Command.Register.ClientName);
|
||||
|
||||
List<MxEvent> events = await CollectAsync(client.StreamEventsAsync(
|
||||
new StreamEventsRequest { SessionId = session.SessionId }));
|
||||
|
||||
MxEvent single = Assert.Single(events);
|
||||
Assert.Equal(MxEventFamily.OnDataChange, single.Family);
|
||||
Assert.Equal(WireFakeGatewayServer.FakeGatewayService.ServerHandle, single.ServerHandle);
|
||||
Assert.Equal(WireFakeGatewayServer.FakeGatewayService.ItemHandle, single.ItemHandle);
|
||||
Assert.Equal(17, single.Value.Int32Value);
|
||||
Assert.Equal(192, single.Quality);
|
||||
Assert.Equal(9ul, single.WorkerSequence);
|
||||
Assert.Equal(MxEvent.BodyOneofCase.OnDataChange, single.BodyCase);
|
||||
|
||||
CloseSessionReply closeReply = await session.CloseAsync();
|
||||
Assert.Equal(SessionState.Closed, closeReply.FinalState);
|
||||
Assert.Equal(
|
||||
WireFakeGatewayServer.FakeGatewayService.SessionId,
|
||||
server.Service.CloseSessionRequest?.SessionId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies the API key reaches the server as a bearer header on unary and
|
||||
/// streaming calls alike. A transport fake can only assert what the client passes;
|
||||
/// this asserts what the server receives.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task ApiKey_ReachesTheServerAsBearerMetadata_OnEveryRpc()
|
||||
{
|
||||
await using WireFakeGatewayServer server = await WireFakeGatewayServer.StartAsync();
|
||||
await using MxGatewayClient client = server.CreateClient(ApiKey);
|
||||
|
||||
MxGatewaySession session = await client.OpenSessionAsync(
|
||||
new OpenSessionRequest { ClientSessionName = "wire-test" });
|
||||
await session.RegisterAsync("wire-test-client");
|
||||
await CollectAsync(client.StreamEventsAsync(
|
||||
new StreamEventsRequest { SessionId = session.SessionId }));
|
||||
await session.CloseAsync();
|
||||
|
||||
string expected = $"Bearer {ApiKey}";
|
||||
Assert.Equal(
|
||||
new Dictionary<string, string>
|
||||
{
|
||||
["OpenSession"] = expected,
|
||||
["Invoke"] = expected,
|
||||
["StreamEvents"] = expected,
|
||||
["CloseSession"] = expected,
|
||||
},
|
||||
server.Service.AuthorizationByMethod);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies the gateway's replay-gap sentinel survives serialization and is
|
||||
/// surfaced as a typed, non-terminal stream item.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task ReplayGapSentinel_SurvivesTheWire_AsTypedStreamItem()
|
||||
{
|
||||
await using WireFakeGatewayServer server = await WireFakeGatewayServer.StartAsync(service =>
|
||||
service.ReplayGap = new ReplayGap
|
||||
{
|
||||
RequestedAfterSequence = 3,
|
||||
OldestAvailableSequence = 8,
|
||||
});
|
||||
await using MxGatewayClient client = server.CreateClient(ApiKey);
|
||||
|
||||
MxGatewaySession session = await client.OpenSessionAsync(
|
||||
new OpenSessionRequest { ClientSessionName = "wire-test" });
|
||||
|
||||
List<MxEventStreamItem> items = [];
|
||||
IAsyncEnumerable<MxEvent> stream = client.StreamEventsAsync(new StreamEventsRequest
|
||||
{
|
||||
SessionId = session.SessionId,
|
||||
AfterWorkerSequence = 3,
|
||||
});
|
||||
|
||||
await foreach (MxEventStreamItem item in stream.AsStreamItemsAsync())
|
||||
{
|
||||
items.Add(item);
|
||||
}
|
||||
|
||||
Assert.Equal(2, items.Count);
|
||||
Assert.True(items[0].IsReplayGap);
|
||||
Assert.Equal(3ul, items[0].ReplayGap!.RequestedAfterSequence);
|
||||
Assert.Equal(8ul, items[0].ReplayGap!.OldestAvailableSequence);
|
||||
Assert.False(items[1].IsReplayGap);
|
||||
Assert.Equal(MxEventFamily.OnDataChange, items[1].Event.Family);
|
||||
Assert.Equal(3ul, server.Service.StreamEventsRequest?.AfterWorkerSequence);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies a genuine <c>PERMISSION_DENIED</c> status maps to the typed client
|
||||
/// exception rather than a bare RpcException.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task PermissionDeniedStatus_MapsToAuthorizationException()
|
||||
{
|
||||
await using WireFakeGatewayServer server = await WireFakeGatewayServer.StartAsync(
|
||||
service => service.DenyInvoke = true);
|
||||
await using MxGatewayClient client = server.CreateClient(ApiKey);
|
||||
|
||||
MxGatewaySession session = await client.OpenSessionAsync(
|
||||
new OpenSessionRequest { ClientSessionName = "wire-test" });
|
||||
|
||||
await Assert.ThrowsAsync<MxGatewayAuthorizationException>(
|
||||
() => session.RegisterAsync("wire-test-client"));
|
||||
}
|
||||
|
||||
private static async Task<List<MxEvent>> CollectAsync(IAsyncEnumerable<MxEvent> stream)
|
||||
{
|
||||
List<MxEvent> events = [];
|
||||
await foreach (MxEvent gatewayEvent in stream)
|
||||
{
|
||||
events.Add(gatewayEvent);
|
||||
}
|
||||
|
||||
return events;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,264 @@
|
||||
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) =>
|
||||
MxGatewayClient.Create(new MxGatewayClientOptions
|
||||
{
|
||||
Endpoint = Endpoint,
|
||||
ApiKey = apiKey,
|
||||
UseTls = false,
|
||||
DefaultCallTimeout = TimeSpan.FromSeconds(30),
|
||||
});
|
||||
|
||||
/// <inheritdoc />
|
||||
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 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 Task<MxCommandReply> Invoke(MxCommandRequest request, ServerCallContext context)
|
||||
{
|
||||
Record(context);
|
||||
InvokeRequest = request;
|
||||
|
||||
if (DenyInvoke)
|
||||
{
|
||||
throw new RpcException(new Status(StatusCode.PermissionDenied, "invoke scope required"));
|
||||
}
|
||||
|
||||
return Task.FromResult(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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+9
@@ -12,6 +12,15 @@
|
||||
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
|
||||
<PackageReference Include="xunit" Version="2.9.3" />
|
||||
<PackageReference Include="xunit.runner.visualstudio" Version="3.1.4" />
|
||||
<!-- Wire tests only (WireFakeGatewayServer): hosts the real MxAccessGateway service
|
||||
on loopback Kestrel so the client is driven over genuine HTTP/2 + protobuf rather
|
||||
than a substituted transport. Version tracks the gateway server's Grpc.AspNetCore
|
||||
(src/ZB.MOM.WW.MxGateway.Server) and the client's Grpc.Net.Client, both 2.76.0. -->
|
||||
<PackageReference Include="Grpc.AspNetCore.Server" Version="2.76.0" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<FrameworkReference Include="Microsoft.AspNetCore.App" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
|
||||
Reference in New Issue
Block a user