From af9c9d78da14c8087d4f7f8b9ef58f9b85736b00 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 23 Jul 2026 16:18:15 -0400 Subject: [PATCH] test(mesh-phase5): 3-port listener reachability + comment fix for the telemetry endpoint Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs | 22 ++- .../TelemetryListenerTests.cs | 176 ++++++++++++++++++ ...OM.WW.OtOpcUa.Host.IntegrationTests.csproj | 5 + 3 files changed, 195 insertions(+), 8 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/TelemetryListenerTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 9fe73366..8853dca4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -417,21 +417,25 @@ builder.Services.AddOtOpcUaSecrets(builder.Configuration); builder.Services.AddOtOpcUaHealth(hasAdmin); builder.Services.AddOtOpcUaObservability(builder.Configuration); -// gRPC server plumbing shared by two endpoints: the LocalDb passive sync endpoint (driver-role, -// mapped below on hasDriver && syncListenPort > 0) and the ConfigServe artifact endpoint -// (admin-role, mapped on hasAdmin && configServeGrpcPort > 0). Each interceptor is scoped strictly -// to its own service path, so both can share one pipeline and each is a harmless pass-through when -// its service is unmapped. Both fail-closed with no ApiKey configured — the interceptor is the ONLY -// inbound auth on either endpoint (the libraries verify nothing). Registered whenever either role is -// present so an admin-only node still serves config and a driver-only node still gates sync. +// gRPC server plumbing shared by three endpoints: the LocalDb passive sync endpoint (driver-role, +// mapped below on hasDriver && syncListenPort > 0), the ConfigServe artifact endpoint (admin-role, +// mapped on hasAdmin && configServeGrpcPort > 0), and the telemetry-stream endpoint (driver-role, +// mapped on hasDriver && telemetryListenPort > 0, per-cluster mesh Phase 5). Each interceptor is +// scoped strictly to its own service path, so all three can share one pipeline and each is a +// harmless pass-through when its service is unmapped. All three fail-closed with no ApiKey +// configured (LocalDbSyncAuthInterceptor, ConfigServeAuthInterceptor, TelemetryStreamAuthInterceptor) +// — the interceptor is the ONLY inbound auth on any endpoint (the libraries verify nothing). +// Registered whenever either role is present so an admin-only node still serves config and a +// driver-only node still gates sync + telemetry. if (hasDriver || hasAdmin) { builder.Services.AddGrpc(o => { if (hasDriver) + { o.Interceptors.Add(); - if (hasDriver) o.Interceptors.Add(); + } if (hasAdmin) o.Interceptors.Add(); }); @@ -596,7 +600,9 @@ if (hasAdmin && configServeGrpcPort > 0) // Driver-node live-telemetry stream (per-cluster mesh Phase 5). Central dials in; gated by // TelemetryStreamAuthInterceptor (fail-closed on empty Telemetry:ApiKey). if (hasDriver && telemetryListenPort > 0) +{ app.MapGrpcService(); +} app.MapOtOpcUaHealth(); app.MapOtOpcUaMetrics(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/TelemetryListenerTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/TelemetryListenerTests.cs new file mode 100644 index 00000000..b133660f --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/TelemetryListenerTests.cs @@ -0,0 +1,176 @@ +using System.Net; +using System.Threading.Channels; +using Grpc.Core; +using Grpc.Net.Client; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Server.Kestrel.Core; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Moq; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1; +using ZB.MOM.WW.OtOpcUa.Host.Configuration; +using ZB.MOM.WW.OtOpcUa.Host.Grpc; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests; + +/// +/// Per-cluster mesh Phase 5 (host-wiring task) — the telemetry-stream h2c listener coexisting with +/// the two earlier dedicated ports (LocalDb sync + ConfigServe) and the main HTTP surface, and the +/// actually being mapped and interceptor-gated on +/// that port. +/// +/// +/// +/// Why this exists. Phase 5 adds a THIRD dedicated h2c port to Program.cs's merged +/// listener block. That block must still re-bind the existing HTTP surface EXACTLY ONCE while +/// adding all three dedicated ports (re-binding it per-port throws "address already in use"; +/// re-binding it zero times silently unbinds the AdminUI behind Traefik). The sibling +/// pins the two-port case; this pins the three-port case +/// AND — the part a port-open check cannot prove — that the telemetry gRPC service is genuinely +/// wired: an un-keyed dial gets (mapped + interceptor +/// reached), not (never mapped), and a correctly-keyed +/// dial opens a real stream. This is the executable regression for the "ships clean but never +/// actually wired" gap. +/// +/// +/// Like , this drives a minimal WebApplication +/// shaped exactly like Program.cs's driver-role wiring (AddGrpc + +/// + Configure<TelemetryOptions> + +/// MapGrpcService<TelemetryStreamGrpcService>) rather than booting the real host +/// (which needs SQL, an Akka mesh and LDAP). What is under test is the Kestrel binding contract +/// and the gRPC map/intercept wiring, both fully reproduced here. The node's own telemetry hub +/// is a stub — this test proves reachability, not fan-out content. +/// +/// +public sealed class TelemetryListenerTests +{ + private const string ApiKey = "telemetry-test-key"; + + [Fact] + public async Task AllThreeDedicatedH2cPorts_AndTheHttpSurface_AnswerSimultaneously_AndTelemetryServiceIsWired() + { + var httpPort = GetFreePort(); + var syncPort = GetFreePort(); + var configServePort = GetFreePort(); + var telemetryPort = GetFreePort(); + + var builder = WebApplication.CreateBuilder(); + builder.Environment.ApplicationName = typeof(TelemetryListenerTests).Assembly.GetName().Name!; + + // Exactly the driver-role wiring Program.cs applies for the telemetry endpoint: the fail-closed + // interceptor on the shared AddGrpc pipeline, the TelemetryOptions the interceptor reads its + // expected key from, and a stub hub so the mapped service can be constructed on the keyed path. + builder.Services.Configure(o => o.ApiKey = ApiKey); + builder.Services.AddSingleton(StubHubYieldingAnEmptyStream()); + builder.Services.AddGrpc(o => o.Interceptors.Add()); + + // The shape Program.cs's merged block applies: compute the existing surface ONCE, apply it ONCE, + // then add each configured dedicated h2c port. Here all THREE dedicated ports are set at once — + // the fused-node case that would double-bind the HTTP surface if the block re-applied it per-port. + var existingBindings = KestrelHttpBinding.Parse($"http://localhost:{httpPort}"); + builder.WebHost.ConfigureKestrel(kestrel => + { + foreach (var binding in existingBindings) + binding.Apply(kestrel); + + kestrel.ListenAnyIP(syncPort, o => o.Protocols = HttpProtocols.Http2); + kestrel.ListenAnyIP(configServePort, o => o.Protocols = HttpProtocols.Http2); + kestrel.ListenAnyIP(telemetryPort, o => o.Protocols = HttpProtocols.Http2); + }); + + await using var app = builder.Build(); + app.MapGet("/healthz", () => "ok"); + app.MapGrpcService(); + await app.StartAsync(TestContext.Current.CancellationToken); + + try + { + var ct = TestContext.Current.CancellationToken; + + // 1) The re-bound HTTP/1.1 surface still answers (the AdminUI/Traefik guarantee) even with + // all three dedicated ports added — i.e. the existing surface was re-bound exactly once. + using var http1 = new HttpClient(); + (await http1.GetStringAsync($"http://localhost:{httpPort}/healthz", ct)).ShouldBe("ok"); + + // 2) The sync + config-serve dedicated ports negotiate prior-knowledge h2c. A 404 is fine — + // it proves the HTTP/2 connection established and routed, which is all that is in question + // for those two (their services are mapped elsewhere; here we only pin the port binding). + using var http2 = new HttpClient + { + DefaultRequestVersion = HttpVersion.Version20, + DefaultVersionPolicy = HttpVersionPolicy.RequestVersionExact, + }; + (await http2.GetAsync($"http://localhost:{syncPort}/", ct)).Version.ShouldBe(HttpVersion.Version20); + (await http2.GetAsync($"http://localhost:{configServePort}/", ct)).Version.ShouldBe(HttpVersion.Version20); + + // 3) The telemetry port serves the MAPPED, interceptor-gated gRPC service. Dial it for real. + using var channel = GrpcChannel.ForAddress( + $"http://localhost:{telemetryPort}", new GrpcChannelOptions { HttpHandler = new SocketsHttpHandler() }); + var client = new TelemetryStreamService.TelemetryStreamServiceClient(channel); + + // 3a) No bearer -> PermissionDenied. This is the load-bearing assertion: PermissionDenied can + // only come from the interceptor, and the interceptor only runs because the method was + // routed to the mapped service. A service that was never mapped would answer Unimplemented. + var noKey = await Should.ThrowAsync(async () => + { + using var call = client.Subscribe(new TelemetryStreamRequest { CorrelationId = "no-key" }, cancellationToken: ct); + await foreach (var _ in call.ResponseStream.ReadAllAsync(ct)) { } + }); + noKey.StatusCode.ShouldBe(StatusCode.PermissionDenied); + noKey.StatusCode.ShouldNotBe(StatusCode.Unimplemented); + + // 3b) Correct bearer -> the mapped handler actually executes end-to-end (interceptor passes, + // the service resolves the stub hub and streams). The stub yields an empty, completed + // stream, so the call completes cleanly with zero envelopes. + var headers = new Metadata { { "authorization", $"Bearer {ApiKey}" } }; + using var keyed = client.Subscribe( + new TelemetryStreamRequest { CorrelationId = "keyed" }, headers, cancellationToken: ct); + var received = 0; + await foreach (var _ in keyed.ResponseStream.ReadAllAsync(ct)) + received++; + received.ShouldBe(0); + } + finally + { + await app.StopAsync(TestContext.Current.CancellationToken); + } + } + + /// + /// A hub whose subscription reader is already completed and empty, so the keyed + /// Subscribe handler runs to completion and closes the stream with zero envelopes — + /// enough to prove the mapped service executed without depending on any live telemetry producer. + /// + private static ITelemetryLocalHub StubHubYieldingAnEmptyStream() + { + var hub = new Mock(); + hub.Setup(h => h.Subscribe(It.IsAny())).Returns(() => new EmptySubscription()); + return hub.Object; + } + + private sealed class EmptySubscription : ITelemetrySubscription + { + private readonly Channel _channel = Channel.CreateBounded(1); + + public EmptySubscription() => _channel.Writer.Complete(); + + public ChannelReader Reader => _channel.Reader; + + public void Dispose() + { + } + } + + private static int GetFreePort() + { + using var listener = new System.Net.Sockets.TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.csproj b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.csproj index 9ae4ff79..64a02c31 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.csproj +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.csproj @@ -9,6 +9,11 @@ + +