From c78034f0b9e683664ca7de5001966a618e2bf158 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 23 Jul 2026 15:55:06 -0400 Subject: [PATCH] feat(mesh-phase5): central telemetry dialer client + proto->domain converters Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- .../Telemetry/TelemetryProtoMapCentral.cs | 132 ++++++++ .../Telemetry/TelemetryStreamClient.cs | 123 +++++++ .../ZB.MOM.WW.OtOpcUa.ControlPlane.csproj | 3 + .../TelemetryProtoMapCentralTests.cs | 307 ++++++++++++++++++ .../Telemetry/TelemetryStreamClientTests.cs | 125 +++++++ 5 files changed, 690 insertions(+) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryStreamClient.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryProtoMapCentralTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryStreamClientTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs new file mode 100644 index 00000000..462994b0 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs @@ -0,0 +1,132 @@ +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry; + +/// +/// Central-side projection of a wire envelope back onto its domain +/// record (per-cluster mesh Phase 5). The exact reverse of TelemetryProtoMapNode: it +/// transcribes every proto field onto the matching domain field and selects the domain record from +/// the envelope's oneof arm. +/// +/// +/// +/// Timestamp presence is honoured. A proto google.protobuf.Timestamp field is a +/// message: when the node left it unset (a null nullable-DateTime on the domain side) the +/// generated property is , so the nullable domain fields map via +/// ?.ToDateTime() and an absent Timestamp becomes . The +/// non-nullable domain fields (published / sampled / transition timestamps) are always set by +/// the node mapper, so they map via a direct .ToDateTime() (which yields a +/// value). +/// +/// +/// optional-string presence is honoured. A proto3 optional string that the node +/// left unset reads back as (via the generated Has… presence), +/// distinguishing null from the empty string; likewise the optional bool +/// historize_to_aveva round-trips its three states (null / true / false). +/// +/// +public static class TelemetryProtoMapCentral +{ + /// + /// Projects a wire onto its domain record, selecting the type from + /// the populated oneof arm. This switch is the coverage guard: every + /// value has an arm here, and an unset (or a + /// future unmapped) case throws so adding a fifth oneof case + /// without a converter fails the coverage test. + /// + /// The wire envelope to project. + /// The domain record for the populated arm. + /// is null. + /// The envelope carries no handled oneof arm. + public static object MapEvent(TelemetryEvent evt) + { + ArgumentNullException.ThrowIfNull(evt); + + return evt.EventCase switch + { + TelemetryEvent.EventOneofCase.AlarmTransition => ToAlarm(evt.AlarmTransition), + TelemetryEvent.EventOneofCase.ScriptLog => ToScript(evt.ScriptLog), + TelemetryEvent.EventOneofCase.DriverHealth => ToHealth(evt.DriverHealth), + TelemetryEvent.EventOneofCase.DriverResilience => ToResilience(evt.DriverResilience), + _ => throw new NotSupportedException( + $"TelemetryEvent carries no handled oneof arm (EventCase = {evt.EventCase})."), + }; + } + + /// Projects an onto an . + /// The proto sub-message. + /// The domain record. + public static AlarmTransitionEvent ToAlarm(AlarmTransition msg) + { + ArgumentNullException.ThrowIfNull(msg); + + return new AlarmTransitionEvent( + AlarmId: msg.AlarmId, + EquipmentPath: msg.EquipmentPath, + AlarmName: msg.AlarmName, + TransitionKind: msg.TransitionKind, + Severity: msg.Severity, + Message: msg.Message, + User: msg.User, + TimestampUtc: msg.TimestampUtc.ToDateTime(), + AlarmTypeName: msg.AlarmTypeName, + Comment: msg.HasComment ? msg.Comment : null, + HistorizeToAveva: msg.HasHistorizeToAveva ? msg.HistorizeToAveva : null, + ReferencingEquipmentPaths: msg.ReferencingEquipmentPaths.ToList()); + } + + /// Projects a onto a . + /// The proto sub-message. + /// The domain record. + public static ScriptLogEntry ToScript(ScriptLog msg) + { + ArgumentNullException.ThrowIfNull(msg); + + return new ScriptLogEntry( + ScriptId: msg.ScriptId, + Level: msg.Level, + Message: msg.Message, + TimestampUtc: msg.TimestampUtc.ToDateTime(), + VirtualTagId: msg.HasVirtualTagId ? msg.VirtualTagId : null, + AlarmId: msg.HasAlarmId ? msg.AlarmId : null, + EquipmentId: msg.HasEquipmentId ? msg.EquipmentId : null); + } + + /// Projects a onto a . + /// The proto sub-message. + /// The domain record. + public static DriverHealthChanged ToHealth(DriverHealth msg) + { + ArgumentNullException.ThrowIfNull(msg); + + return new DriverHealthChanged( + ClusterId: msg.ClusterId, + DriverInstanceId: msg.DriverInstanceId, + State: msg.State, + LastSuccessfulReadUtc: msg.LastSuccessfulReadUtc?.ToDateTime(), + LastError: msg.HasLastError ? msg.LastError : null, + ErrorCount5Min: msg.ErrorCount5Min, + PublishedUtc: msg.PublishedUtc.ToDateTime()); + } + + /// Projects a onto a . + /// The proto sub-message. + /// The domain record. + public static DriverResilienceStatusChanged ToResilience(DriverResilienceStatus msg) + { + ArgumentNullException.ThrowIfNull(msg); + + return new DriverResilienceStatusChanged( + DriverInstanceId: msg.DriverInstanceId, + HostName: msg.HostName, + BreakerOpen: msg.BreakerOpen, + ConsecutiveFailures: msg.ConsecutiveFailures, + CurrentInFlight: msg.CurrentInFlight, + LastBreakerOpenUtc: msg.LastBreakerOpenUtc?.ToDateTime(), + LastSampledUtc: msg.LastSampledUtc.ToDateTime(), + PublishedUtc: msg.PublishedUtc.ToDateTime()); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryStreamClient.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryStreamClient.cs new file mode 100644 index 00000000..fb3acaa8 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryStreamClient.cs @@ -0,0 +1,123 @@ +using System.Net; +using Grpc.Core; +using Grpc.Net.Client; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry; + +/// +/// Central-side per-node dialer for the Phase 5 telemetry stream. Owns one h2c +/// to a single driver node's TelemetryStreamService, opens the +/// server-streaming Subscribe call, and pumps every to a +/// caller-supplied sink. The raw proto envelope is delivered as-is — the caller maps it via +/// . +/// +/// +/// +/// Deliberately dumb. Connect, pump, surface errors — no reconnect, no backoff, no +/// buffering. A normal shutdown (the caller's firing, or the +/// server ending the stream with ) completes +/// without calling onError; any other failure is routed to +/// onError exactly once as the reconnect trigger the supervisor (a later task) owns. +/// +/// +/// h2c + Bearer. The endpoint is a prior-knowledge http://host:port address (no +/// TLS); the shared node bearer key rides the authorization header on every call. The +/// channel sets HTTP/2 keepalive pings so a long-idle telemetry stream is not silently dropped +/// by an intermediary. +/// +/// +public sealed class TelemetryStreamClient : IDisposable +{ + private readonly string _apiKey; + private readonly GrpcChannel? _ownedChannel; + private readonly TelemetryStreamService.TelemetryStreamServiceClient _client; + + /// + /// Initializes a new instance of the class dialing a live + /// driver node over h2c. + /// + /// Prior-knowledge http://host:port address of the node's telemetry server. + /// Shared node bearer key sent as authorization: Bearer <key>. + public TelemetryStreamClient(string endpoint, string apiKey) + { + ArgumentException.ThrowIfNullOrEmpty(endpoint); + ArgumentNullException.ThrowIfNull(apiKey); + + _apiKey = apiKey; + _ownedChannel = GrpcChannel.ForAddress(endpoint, new GrpcChannelOptions + { + HttpHandler = new SocketsHttpHandler + { + KeepAlivePingDelay = TimeSpan.FromSeconds(15), + KeepAlivePingTimeout = TimeSpan.FromSeconds(10), + KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always, + EnableMultipleHttp2Connections = true, + }, + }); + _client = new TelemetryStreamService.TelemetryStreamServiceClient(_ownedChannel); + } + + /// + /// Initializes a new instance of the class over a + /// caller-supplied client — the test seam. No channel is owned or disposed. + /// + /// The generated client to drive (typically a fake in tests). + /// Shared node bearer key sent as authorization: Bearer <key>. + public TelemetryStreamClient(TelemetryStreamService.TelemetryStreamServiceClient client, string apiKey) + { + ArgumentNullException.ThrowIfNull(client); + ArgumentNullException.ThrowIfNull(apiKey); + + _client = client; + _apiKey = apiKey; + _ownedChannel = null; + } + + /// + /// Opens the telemetry stream and pumps every envelope to until the + /// server ends the stream or fires. A normal shutdown returns without + /// invoking ; any other failure invokes + /// exactly once and returns. + /// + /// Stream correlation id echoed on every envelope. + /// Sink for each received raw . + /// Invoked once on an abnormal stream failure (the reconnect trigger). + /// Cancels the stream (normal shutdown). + public async Task RunAsync( + string correlationId, + Action onEvent, + Action onError, + CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(onEvent); + ArgumentNullException.ThrowIfNull(onError); + + var headers = new Metadata { { "authorization", $"Bearer {_apiKey}" } }; + + try + { + using var call = _client.Subscribe( + new TelemetryStreamRequest { CorrelationId = correlationId }, headers, cancellationToken: ct); + + await foreach (var evt in call.ResponseStream.ReadAllAsync(ct)) + onEvent(evt); + } + catch (OperationCanceledException) + { + // Normal shutdown: the caller's token fired. + } + catch (RpcException rex) when (rex.StatusCode == StatusCode.Cancelled) + { + // Normal shutdown: the server (or a token-driven cancel) ended the stream. + } + catch (Exception ex) + { + // The reconnect trigger — a later supervisor task decides retry/backoff. + onError(ex); + } + } + + /// + public void Dispose() => _ownedChannel?.Dispose(); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj index 4c0ed37c..d2e52480 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj @@ -15,6 +15,9 @@ + + diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryProtoMapCentralTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryProtoMapCentralTests.cs new file mode 100644 index 00000000..d51ab2e5 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryProtoMapCentralTests.cs @@ -0,0 +1,307 @@ +using Google.Protobuf.WellKnownTypes; +using Shouldly; +using ZB.MOM.WW.OtOpcUa.Commons.Protos; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Telemetry; + +/// +/// Unit tests for — the central-side proto→domain +/// projection. Each kind asserts every field round-trips, including the null-vs-absent presence +/// handling for nullable Timestamps, optional strings, and the tri-state optional bool. A coverage +/// guard proves every value has a converter. +/// +public sealed class TelemetryProtoMapCentralTests +{ + private static readonly DateTime SampleUtc = + new(2026, 7, 23, 10, 30, 45, DateTimeKind.Utc); + + private static readonly DateTime OtherUtc = + new(2026, 7, 23, 9, 15, 0, DateTimeKind.Utc); + + [Fact] + public void ToAlarm_transcribes_every_field_with_nullables_present() + { + var proto = new AlarmTransition + { + AlarmId = "Plant/Modbus/dev1/Speed", + EquipmentPath = "Area/Line/Equip", + AlarmName = "HighSpeed", + TransitionKind = "Activated", + Severity = 700, + Message = "Speed high", + User = "operator1", + TimestampUtc = Timestamp.FromDateTime(SampleUtc), + AlarmTypeName = "LimitAlarm", + Comment = "ack comment", + HistorizeToAveva = false, + }; + proto.ReferencingEquipmentPaths.AddRange(["Area/Line/E1", "Area/Line/E2"]); + + var e = TelemetryProtoMapCentral.ToAlarm(proto); + + e.AlarmId.ShouldBe("Plant/Modbus/dev1/Speed"); + e.EquipmentPath.ShouldBe("Area/Line/Equip"); + e.AlarmName.ShouldBe("HighSpeed"); + e.TransitionKind.ShouldBe("Activated"); + e.Severity.ShouldBe(700); + e.Message.ShouldBe("Speed high"); + e.User.ShouldBe("operator1"); + e.TimestampUtc.ShouldBe(SampleUtc); + e.TimestampUtc.Kind.ShouldBe(DateTimeKind.Utc); + e.AlarmTypeName.ShouldBe("LimitAlarm"); + e.Comment.ShouldBe("ack comment"); + e.HistorizeToAveva.ShouldBe(false); + e.ReferencingEquipmentPaths.ShouldBe(["Area/Line/E1", "Area/Line/E2"]); + } + + [Fact] + public void ToAlarm_absent_optionals_map_to_null_and_empty_repeated_to_empty_list() + { + var proto = new AlarmTransition + { + AlarmId = "a", + EquipmentPath = "p", + AlarmName = "n", + TransitionKind = "Cleared", + Severity = 1, + Message = "m", + User = "system", + TimestampUtc = Timestamp.FromDateTime(SampleUtc), + AlarmTypeName = "AlarmCondition", + // Comment, HistorizeToAveva unset; no referencing paths added. + }; + + var e = TelemetryProtoMapCentral.ToAlarm(proto); + + e.Comment.ShouldBeNull(); + e.HistorizeToAveva.ShouldBeNull(); + e.ReferencingEquipmentPaths.ShouldNotBeNull(); + e.ReferencingEquipmentPaths.ShouldBeEmpty(); + } + + [Fact] + public void ToAlarm_historize_present_true_round_trips() + { + var proto = new AlarmTransition + { + AlarmId = "a", EquipmentPath = "p", AlarmName = "n", TransitionKind = "Activated", + Severity = 1, Message = "m", User = "system", + TimestampUtc = Timestamp.FromDateTime(SampleUtc), AlarmTypeName = "AlarmCondition", + HistorizeToAveva = true, + }; + + TelemetryProtoMapCentral.ToAlarm(proto).HistorizeToAveva.ShouldBe(true); + } + + [Fact] + public void ToScript_transcribes_every_field_with_optionals_present() + { + var proto = new ScriptLog + { + ScriptId = "script-1", + Level = "Information", + Message = "hello", + TimestampUtc = Timestamp.FromDateTime(SampleUtc), + VirtualTagId = "vt-1", + AlarmId = "al-1", + EquipmentId = "eq-1", + }; + + var e = TelemetryProtoMapCentral.ToScript(proto); + + e.ScriptId.ShouldBe("script-1"); + e.Level.ShouldBe("Information"); + e.Message.ShouldBe("hello"); + e.TimestampUtc.ShouldBe(SampleUtc); + e.TimestampUtc.Kind.ShouldBe(DateTimeKind.Utc); + e.VirtualTagId.ShouldBe("vt-1"); + e.AlarmId.ShouldBe("al-1"); + e.EquipmentId.ShouldBe("eq-1"); + } + + [Fact] + public void ToScript_absent_optionals_map_to_null() + { + var proto = new ScriptLog + { + ScriptId = "script-1", + Level = "Error", + Message = "boom", + TimestampUtc = Timestamp.FromDateTime(SampleUtc), + // VirtualTagId, AlarmId, EquipmentId unset. + }; + + var e = TelemetryProtoMapCentral.ToScript(proto); + + e.VirtualTagId.ShouldBeNull(); + e.AlarmId.ShouldBeNull(); + e.EquipmentId.ShouldBeNull(); + } + + [Fact] + public void ToHealth_transcribes_every_field_with_nullables_present() + { + var proto = new DriverHealth + { + ClusterId = "c1", + DriverInstanceId = "d1", + State = "Faulted", + LastSuccessfulReadUtc = Timestamp.FromDateTime(OtherUtc), + LastError = "timeout", + ErrorCount5Min = 3, + PublishedUtc = Timestamp.FromDateTime(SampleUtc), + }; + + var e = TelemetryProtoMapCentral.ToHealth(proto); + + e.ClusterId.ShouldBe("c1"); + e.DriverInstanceId.ShouldBe("d1"); + e.State.ShouldBe("Faulted"); + e.LastSuccessfulReadUtc.ShouldBe(OtherUtc); + e.LastSuccessfulReadUtc!.Value.Kind.ShouldBe(DateTimeKind.Utc); + e.LastError.ShouldBe("timeout"); + e.ErrorCount5Min.ShouldBe(3); + e.PublishedUtc.ShouldBe(SampleUtc); + e.PublishedUtc.Kind.ShouldBe(DateTimeKind.Utc); + } + + [Fact] + public void ToHealth_absent_nullable_timestamp_and_optional_string_map_to_null() + { + var proto = new DriverHealth + { + ClusterId = "c1", + DriverInstanceId = "d1", + State = "Healthy", + ErrorCount5Min = 0, + PublishedUtc = Timestamp.FromDateTime(SampleUtc), + // LastSuccessfulReadUtc, LastError unset. + }; + + var e = TelemetryProtoMapCentral.ToHealth(proto); + + e.LastSuccessfulReadUtc.ShouldBeNull(); + e.LastError.ShouldBeNull(); + } + + [Fact] + public void ToResilience_transcribes_every_field_with_nullable_present() + { + var proto = new DriverResilienceStatus + { + DriverInstanceId = "d1", + HostName = "host-a", + BreakerOpen = true, + ConsecutiveFailures = 5, + CurrentInFlight = 2, + LastBreakerOpenUtc = Timestamp.FromDateTime(OtherUtc), + LastSampledUtc = Timestamp.FromDateTime(SampleUtc), + PublishedUtc = Timestamp.FromDateTime(SampleUtc), + }; + + var e = TelemetryProtoMapCentral.ToResilience(proto); + + e.DriverInstanceId.ShouldBe("d1"); + e.HostName.ShouldBe("host-a"); + e.BreakerOpen.ShouldBeTrue(); + e.ConsecutiveFailures.ShouldBe(5); + e.CurrentInFlight.ShouldBe(2); + e.LastBreakerOpenUtc.ShouldBe(OtherUtc); + e.LastBreakerOpenUtc!.Value.Kind.ShouldBe(DateTimeKind.Utc); + e.LastSampledUtc.ShouldBe(SampleUtc); + e.PublishedUtc.ShouldBe(SampleUtc); + } + + [Fact] + public void ToResilience_absent_nullable_timestamp_maps_to_null() + { + var proto = new DriverResilienceStatus + { + DriverInstanceId = "d1", + HostName = "host-a", + BreakerOpen = false, + ConsecutiveFailures = 0, + CurrentInFlight = 0, + LastSampledUtc = Timestamp.FromDateTime(SampleUtc), + PublishedUtc = Timestamp.FromDateTime(SampleUtc), + // LastBreakerOpenUtc unset. + }; + + TelemetryProtoMapCentral.ToResilience(proto).LastBreakerOpenUtc.ShouldBeNull(); + } + + [Fact] + public void MapEvent_routes_each_case_to_its_record_type() + { + var alarm = new TelemetryEvent { AlarmTransition = MinimalAlarm() }; + var script = new TelemetryEvent { ScriptLog = MinimalScript() }; + var health = new TelemetryEvent { DriverHealth = MinimalHealth() }; + var resilience = new TelemetryEvent { DriverResilience = MinimalResilience() }; + + TelemetryProtoMapCentral.MapEvent(alarm).ShouldBeOfType(); + TelemetryProtoMapCentral.MapEvent(script).ShouldBeOfType(); + TelemetryProtoMapCentral.MapEvent(health).ShouldBeOfType(); + TelemetryProtoMapCentral.MapEvent(resilience).ShouldBeOfType(); + } + + [Fact] + public void MapEvent_unset_case_throws_NotSupported() + { + Should.Throw(() => TelemetryProtoMapCentral.MapEvent(new TelemetryEvent())); + } + + /// + /// Coverage guard: every value must be routed + /// by to a non-null domain record. Adding a + /// fifth oneof case (and thus a fifth HandledCases entry) without a converter arm fails here. + /// + [Fact] + public void MapEvent_covers_every_handled_case() + { + foreach (var handled in TelemetryProtoContract.HandledCases) + { + var evt = BuildFor(handled); + evt.EventCase.ShouldBe(handled); + TelemetryProtoMapCentral.MapEvent(evt).ShouldNotBeNull(); + } + } + + private static TelemetryEvent BuildFor(TelemetryEvent.EventOneofCase handled) => handled switch + { + TelemetryEvent.EventOneofCase.AlarmTransition => new TelemetryEvent { AlarmTransition = MinimalAlarm() }, + TelemetryEvent.EventOneofCase.ScriptLog => new TelemetryEvent { ScriptLog = MinimalScript() }, + TelemetryEvent.EventOneofCase.DriverHealth => new TelemetryEvent { DriverHealth = MinimalHealth() }, + TelemetryEvent.EventOneofCase.DriverResilience => new TelemetryEvent { DriverResilience = MinimalResilience() }, + _ => throw new InvalidOperationException($"Test does not know how to build case {handled}."), + }; + + private static AlarmTransition MinimalAlarm() => new() + { + AlarmId = "a", EquipmentPath = "p", AlarmName = "n", TransitionKind = "Activated", + Severity = 1, Message = "m", User = "system", + TimestampUtc = Timestamp.FromDateTime(SampleUtc), AlarmTypeName = "AlarmCondition", + }; + + private static ScriptLog MinimalScript() => new() + { + ScriptId = "s", Level = "Information", Message = "m", + TimestampUtc = Timestamp.FromDateTime(SampleUtc), + }; + + private static DriverHealth MinimalHealth() => new() + { + ClusterId = "c", DriverInstanceId = "d", State = "Healthy", + ErrorCount5Min = 0, PublishedUtc = Timestamp.FromDateTime(SampleUtc), + }; + + private static DriverResilienceStatus MinimalResilience() => new() + { + DriverInstanceId = "d", HostName = "h", BreakerOpen = false, + ConsecutiveFailures = 0, CurrentInFlight = 0, + LastSampledUtc = Timestamp.FromDateTime(SampleUtc), + PublishedUtc = Timestamp.FromDateTime(SampleUtc), + }; +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryStreamClientTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryStreamClientTests.cs new file mode 100644 index 00000000..8335f6b4 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryStreamClientTests.cs @@ -0,0 +1,125 @@ +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Shouldly; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Telemetry; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Telemetry; + +/// +/// Unit tests for driving a fake generated client (the ctor +/// seam), so the pump/normal-shutdown/error-routing behaviour is provable without a live gRPC +/// server. The wire itself is covered by the Phase 5 live gate. +/// +public sealed class TelemetryStreamClientTests +{ + [Fact] + public async Task RunAsync_pumps_every_event_to_onEvent() + { + var events = new[] + { + new TelemetryEvent { ScriptLog = Script("one") }, + new TelemetryEvent { ScriptLog = Script("two") }, + }; + var fake = new FakeClient(new FakeStreamReader(events)); + using var sut = new TelemetryStreamClient(fake, "key"); + + var received = new List(); + Exception? errored = null; + + await sut.RunAsync("corr-1", received.Add, ex => errored = ex, CancellationToken.None); + + received.Count.ShouldBe(2); + received[0].ScriptLog.Message.ShouldBe("one"); + received[1].ScriptLog.Message.ShouldBe("two"); + errored.ShouldBeNull(); + } + + [Fact] + public async Task RunAsync_swallows_rpc_cancelled_without_calling_onError() + { + var fake = new FakeClient(new FakeStreamReader( + [new TelemetryEvent { ScriptLog = Script("one") }], + throwAtEnd: new RpcException(new Status(StatusCode.Cancelled, "cancelled")))); + using var sut = new TelemetryStreamClient(fake, "key"); + + var received = new List(); + Exception? errored = null; + + await sut.RunAsync("corr-1", received.Add, ex => errored = ex, CancellationToken.None); + + received.Count.ShouldBe(1); + errored.ShouldBeNull(); + } + + [Fact] + public async Task RunAsync_routes_real_rpc_failure_to_onError() + { + var boom = new RpcException(new Status(StatusCode.Unavailable, "node down")); + var fake = new FakeClient(new FakeStreamReader([], throwAtEnd: boom)); + using var sut = new TelemetryStreamClient(fake, "key"); + + Exception? errored = null; + + await sut.RunAsync("corr-1", _ => { }, ex => errored = ex, CancellationToken.None); + + errored.ShouldBeSameAs(boom); + } + + private static ScriptLog Script(string message) => new() + { + ScriptId = "s", + Level = "Information", + Message = message, + TimestampUtc = Timestamp.FromDateTime(new DateTime(2026, 7, 23, 0, 0, 0, DateTimeKind.Utc)), + }; + + /// Fake generated client whose Subscribe returns a canned server-streaming call. + private sealed class FakeClient : TelemetryStreamService.TelemetryStreamServiceClient + { + private readonly IAsyncStreamReader _reader; + + public FakeClient(IAsyncStreamReader reader) => _reader = reader; + + public override AsyncServerStreamingCall Subscribe( + TelemetryStreamRequest request, + Metadata? headers = null, + DateTime? deadline = null, + CancellationToken cancellationToken = default) => + new( + _reader, + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => [], + () => { }); + } + + /// Replays a fixed sequence of events, optionally throwing at the end. + private sealed class FakeStreamReader : IAsyncStreamReader + { + private readonly IReadOnlyList _items; + private readonly Exception? _throwAtEnd; + private int _index = -1; + + public FakeStreamReader(IReadOnlyList items, Exception? throwAtEnd = null) + { + _items = items; + _throwAtEnd = throwAtEnd; + } + + public TelemetryEvent Current => _items[_index]; + + public Task MoveNext(CancellationToken cancellationToken) + { + _index++; + if (_index < _items.Count) + return Task.FromResult(true); + + if (_throwAtEnd is not null) + throw _throwAtEnd; + + return Task.FromResult(false); + } + } +}