From a8278f65d0779e377981dd94bbd8291a4d123cec Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 10 Jul 2026 12:44:11 -0400 Subject: [PATCH] =?UTF-8?q?test(sitestream):=20end-to-end=20trace=20for=20?= =?UTF-8?q?site-wide=20alarm=20stream=20=E2=86=92=20live=20cache=20(plan?= =?UTF-8?q?=20#10=20T7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds SiteAlarmStreamEndToEndTests wiring the whole feature pipe with only the gRPC HTTP/2 transport mocked: real SiteStreamManager.SubscribeSiteAlarms → real SiteStreamGrpcServer.SubscribeSite → real StreamRelayActor → proto → wire round-trip (ToByteArray/ParseFrom) → real SiteStreamGrpcClient.ConvertToAlarmEvent → real SiteAlarmAggregatorActor cache. Closes the cross-cutting gap the per-task T1–T6 unit tests leave (each mocks its neighbour): proves AlarmKey identity (instance, name, sourceRef) + full native enrichment survive every boundary mapping, attributes and placeholder rows are dropped end-to-end, the single site-wide stream carries alarms for multiple instances (no per-instance filter), and a snapshot-seed row and a live delta for the same native alarm — both mapped through the real pipe — collapse onto one cache row. Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj --- .../Grpc/SiteAlarmStreamEndToEndTests.cs | 298 ++++++++++++++++++ 1 file changed, 298 insertions(+) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs new file mode 100644 index 00000000..9694b975 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs @@ -0,0 +1,298 @@ +using System.Collections.Concurrent; +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Google.Protobuf; +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; +using ZB.MOM.WW.ScadaBridge.SiteRuntime; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming; + +namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Grpc; + +/// +/// End-to-end trace for the aggregated live alarm stream (plan #10 deferred item, Task 7). +/// +/// Every per-task unit test (T1–T6) exercises exactly one layer with its neighbour mocked: +/// T1 the site broadcast filter (subscriber = TestKit probe, domain events), T2/the existing +/// the server+relay+channel with a MOCKED +/// stopping at the proto event, T3 the proto→domain client +/// mapping in isolation, and T4 the aggregator fed hand-built domain events. Nothing wires the +/// whole pipe together. +/// +/// This test assembles the REAL chain end to end — mocking only the gRPC HTTP/2 transport: +/// +/// domain AlarmStateChanged +/// → SiteStreamManager.SubscribeSiteAlarms (real site-wide broadcast + alarm-only filter) +/// → SiteStreamGrpcServer.SubscribeSite (real server handler) +/// → StreamRelayActor (real domain→proto mapping + placeholder drop) +/// → SiteStreamEvent proto → wire round-trip (ToByteArray/ParseFrom, simulates HTTP/2) +/// → SiteStreamGrpcClient.ConvertToAlarmEvent (real proto→domain mapping) +/// → SiteAlarmAggregatorActor cache (real dedup/seed/live) +/// +/// and asserts the invariants the seams can't prove alone: AlarmKey identity +/// (instance, name, sourceRef) + native enrichment survive every boundary; attributes and +/// placeholder rows never reach the live cache; the single site-wide stream carries alarms +/// for MULTIPLE instances (no per-instance filter); and a snapshot-seed row and a live delta +/// for the same native alarm — both mapped through the real pipe — collapse onto ONE cache row. +/// +/// Full gRPC-over-HTTP/2 remains a manual docker-cluster smoke (plan §4 Task 7). +/// +public class SiteAlarmStreamEndToEndTests : TestKit +{ + private const string InstanceA = "SiteA.Pump01"; + private const string InstanceB = "SiteA.Motor07"; + + // ── The full site→proto→client trace ──────────────────────────────────────── + + [Fact] + public async Task SiteWideAlarm_TraversesManagerToRelayToClient_PreservingIdentityAndEnrichment_DroppingAttributesAndPlaceholders() + { + // Site side: a REAL broadcast hub (no mock subscriber) wired straight into the + // REAL SubscribeSite server handler. + var manager = new SiteStreamManager( + new SiteRuntimeOptions { StreamBufferSize = 256 }, + NullLogger.Instance); + manager.Initialize(Sys); + + var server = new SiteStreamGrpcServer(manager, NullLogger.Instance); + server.SetReady(Sys); + + var written = new ConcurrentQueue(); + var writer = Substitute.For>(); + writer.WriteAsync(Arg.Any(), Arg.Any()) + .Returns(Task.CompletedTask) + .AndDoes(ci => written.Enqueue(ci.Arg())); + + using var cts = new CancellationTokenSource(); + var context = CreateMockContext(cts.Token); + + var streamTask = Task.Run(() => server.SubscribeSite( + new SiteStreamRequest { CorrelationId = "e2e-site" }, writer, context)); + + // Wait until the server has actually subscribed to the site-wide hub, otherwise a + // publish can race ahead of the materialized subscription and be missed. + await WaitForConditionAsync(() => manager.SubscriptionCount == 1); + + var raise = new DateTimeOffset(2026, 4, 2, 8, 30, 0, TimeSpan.Zero); + var ts = raise.AddSeconds(5); + + // A fully-enriched NATIVE alarm on instance A (the payload whose fidelity we trace). + var native = new AlarmStateChanged(InstanceA, "Tank01.LevelAlarm", AlarmState.Active, 725, ts) + { + Level = AlarmLevel.HighHigh, + Message = "Tank 01 level critically high", + Kind = AlarmKind.NativeOpcUa, + Condition = new AlarmConditionState( + Active: true, Acknowledged: true, Confirmed: false, + Shelve: AlarmShelveState.OneShotShelved, Suppressed: false, Severity: 725), + SourceReference = "Tank01.Level.HiHi", + AlarmTypeName = "AnalogLimitAlarm.HiHi", + Category = "Process", + OperatorUser = "op.jane", + OperatorComment = "ack — investigating", + OriginalRaiseTime = raise, + CurrentValue = "98.4", + LimitValue = "95.0", + NativeSourceCanonicalName = "Tank01.LevelAlarm", + IsConfiguredPlaceholder = false + }; + + // A COMPUTED alarm on a DIFFERENT instance — must arrive over the SAME site-wide + // stream (proves the per-instance filter is gone). + var computedOtherInstance = new AlarmStateChanged(InstanceB, "OverSpeed", AlarmState.Active, 400, ts); + + // Noise that MUST be dropped somewhere in the pipe: + var attribute = new AttributeValueChanged(InstanceA, "Modules.Flow", "GPM", 12.3, "Good", ts); + var placeholder = new AlarmStateChanged(InstanceA, "Motor1.MotorAlarms", AlarmState.Normal, 0, ts) + { + Kind = AlarmKind.NativeOpcUa, + IsConfiguredPlaceholder = true + }; + + // Interleave so a leak of the dropped rows would change the observed count/order. + manager.PublishAttributeValueChanged(attribute); // filtered at the manager (alarm-only) + manager.PublishAlarmStateChanged(native); // → 1 proto + manager.PublishAlarmStateChanged(placeholder); // dropped by StreamRelayActor + manager.PublishAlarmStateChanged(computedOtherInstance); // → 1 proto + + await WaitForConditionAsync(() => written.Count >= 2); + // Give any leaked (attribute/placeholder) event a chance to show up before asserting. + await Task.Delay(150); + + cts.Cancel(); + await streamTask; + + var protos = written.ToArray(); + Assert.Equal(2, protos.Length); + Assert.All(protos, p => + Assert.Equal(SiteStreamEvent.EventOneofCase.AlarmChanged, p.EventCase)); // no attribute leaked + Assert.DoesNotContain(protos, p => p.AlarmChanged.IsConfiguredPlaceholder); // no placeholder leaked + + // Wire round-trip (serialize→deserialize) then the REAL client mapping back to domain. + var mapped = protos + .Select(p => SiteStreamGrpcClient.ConvertToAlarmEvent(RoundTripOverWire(p))) + .ToList(); + Assert.All(mapped, m => Assert.NotNull(m)); + + // Both instances present — one site-wide stream, no per-instance filter. + var backA = mapped.Single(m => m!.InstanceUniqueName == InstanceA)!; + var backB = mapped.Single(m => m!.InstanceUniqueName == InstanceB)!; + Assert.Equal("OverSpeed", backB.AlarmName); + + // AlarmKey identity survived end to end. + Assert.Equal("Tank01.LevelAlarm", backA.AlarmName); + Assert.Equal("Tank01.Level.HiHi", backA.SourceReference); + + // Native enrichment survived every boundary (manager→relay→proto→wire→client). + Assert.Equal(AlarmKind.NativeOpcUa, backA.Kind); + Assert.Equal(AlarmState.Active, backA.State); + Assert.Equal(725, backA.Priority); + Assert.Equal(AlarmLevel.HighHigh, backA.Level); + Assert.Equal("Tank 01 level critically high", backA.Message); + Assert.True(backA.Condition.Active); + Assert.True(backA.Condition.Acknowledged); + Assert.Equal(AlarmShelveState.OneShotShelved, backA.Condition.Shelve); + Assert.Equal(725, backA.Condition.Severity); + Assert.Equal("AnalogLimitAlarm.HiHi", backA.AlarmTypeName); + Assert.Equal("Process", backA.Category); + Assert.Equal("op.jane", backA.OperatorUser); + Assert.Equal("ack — investigating", backA.OperatorComment); + Assert.Equal(raise, backA.OriginalRaiseTime); + Assert.Equal("98.4", backA.CurrentValue); + Assert.Equal("95.0", backA.LimitValue); + Assert.Equal("Tank01.LevelAlarm", backA.NativeSourceCanonicalName); + Assert.Equal(ts, backA.Timestamp); + Assert.False(backA.IsConfiguredPlaceholder); + } + + // ── Snapshot-seed vs live-delta identity parity, through the real pipe ──────── + + [Fact] + public async Task RealPipeMapped_SeedRow_And_LiveDelta_ForSameNativeAlarm_CollapseOntoOneCacheRow() + { + // Produce TWO proto events for the SAME native alarm identity through the real + // manager→relay→proto pipe: an older one (used as the snapshot-seed row) and a + // newer one (fed as the live delta). Because both are mapped by the SAME real + // ConvertToAlarmEvent, their AlarmKeys are identical by construction — so the real + // aggregator must collapse them onto ONE row (no seed/live ghost duplicate). + var manager = new SiteStreamManager( + new SiteRuntimeOptions { StreamBufferSize = 256 }, + NullLogger.Instance); + manager.Initialize(Sys); + var server = new SiteStreamGrpcServer(manager, NullLogger.Instance); + server.SetReady(Sys); + + var written = new ConcurrentQueue(); + var writer = Substitute.For>(); + writer.WriteAsync(Arg.Any(), Arg.Any()) + .Returns(Task.CompletedTask) + .AndDoes(ci => written.Enqueue(ci.Arg())); + using var cts = new CancellationTokenSource(); + var streamTask = Task.Run(() => server.SubscribeSite( + new SiteStreamRequest { CorrelationId = "e2e-parity" }, writer, CreateMockContext(cts.Token))); + await WaitForConditionAsync(() => manager.SubscriptionCount == 1); + + AlarmStateChanged Native(int priority, DateTimeOffset t) => + new(InstanceA, "Tank01.LevelAlarm", AlarmState.Active, priority, t) + { + Kind = AlarmKind.NativeOpcUa, + SourceReference = "Tank01.Level.HiHi", + NativeSourceCanonicalName = "Tank01.LevelAlarm" + }; + + var t0 = new DateTimeOffset(2026, 4, 2, 9, 0, 0, TimeSpan.Zero); + manager.PublishAlarmStateChanged(Native(300, t0)); // → seed row (older) + manager.PublishAlarmStateChanged(Native(800, t0.AddSeconds(5))); // → live delta (newer) + await WaitForConditionAsync(() => written.Count >= 2); + cts.Cancel(); + await streamTask; + + var protos = written.ToArray(); + var seedRow = SiteStreamGrpcClient.ConvertToAlarmEvent(RoundTripOverWire(protos[0]))!; + var liveRow = SiteStreamGrpcClient.ConvertToAlarmEvent(RoundTripOverWire(protos[1]))!; + + // Drive a REAL aggregator: seed returns the older mapped row; then Tell it the newer + // mapped delta on the live path. + var factory = new NoopSiteStreamClientFactory(); + var sink = new PublishSink(); + var aggregator = Sys.ActorOf(Props.Create(() => new SiteAlarmAggregatorActor( + "site-alpha", "e2e-parity", _ => Task.FromResult>(new[] { seedRow }), + sink.Publish, factory, "http://a:5100", "http://b:5100", TimeSpan.FromMinutes(10)))); + + await WaitForConditionAsync(() => sink.Latest is { Count: 1 }); // seed applied + aggregator.Tell(liveRow); + await WaitForConditionAsync(() => + sink.Latest is { Count: 1 } l && l[0].Priority == 800); // live delta collapsed in place + + var final = sink.Latest!; + Assert.Single(final); // exactly ONE row for the (instance, name, sourceRef) identity + Assert.Equal("Tank01.Level.HiHi", final[0].SourceReference); + Assert.Equal(AlarmKind.NativeOpcUa, final[0].Kind); + } + + // ── helpers ────────────────────────────────────────────────────────────────── + + /// + /// Serialize→deserialize a proto event to mimic the gRPC HTTP/2 hop, proving the + /// enriched AlarmStateUpdate is fully wire-representable (no field lost in codec). + /// + private static SiteStreamEvent RoundTripOverWire(SiteStreamEvent evt) => + SiteStreamEvent.Parser.ParseFrom(evt.ToByteArray()); + + private static ServerCallContext CreateMockContext(CancellationToken cancellationToken) + { + var context = Substitute.For(); + context.CancellationToken.Returns(cancellationToken); + return context; + } + + private static async Task WaitForConditionAsync(Func condition, int timeoutMs = 5000) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); + while (!condition() && DateTime.UtcNow < deadline) + await Task.Delay(25); + Assert.True(condition(), $"Condition not met within {timeoutMs}ms"); + } + + private sealed class PublishSink + { + private readonly object _lock = new(); + private IReadOnlyList? _latest; + public void Publish(IReadOnlyList snapshot) + { + lock (_lock) { _latest = snapshot; } + } + public IReadOnlyList? Latest + { + get { lock (_lock) { return _latest; } } + } + } + + /// Aggregator transport double: its live stream simply stays open until cancelled. + private sealed class NoopSiteStreamClient : SiteStreamGrpcClient + { + public override Task SubscribeSiteAsync( + string correlationId, Action onAlarmEvent, Action onError, CancellationToken ct) + { + var tcs = new TaskCompletionSource(); + ct.Register(() => tcs.TrySetResult()); + return tcs.Task; + } + + public override void Unsubscribe(string correlationId) { } + } + + private sealed class NoopSiteStreamClientFactory : SiteStreamGrpcClientFactory + { + private readonly NoopSiteStreamClient _client = new(); + public NoopSiteStreamClientFactory() : base(NullLoggerFactory.Instance) { } + public override SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint) => _client; + public override SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint) => _client; + } +}