test(sitestream): end-to-end trace for site-wide alarm stream → live cache (plan #10 T7)
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
This commit is contained in:
Joseph Doherty
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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
|
||||
/// <see cref="GrpcStreamIntegrationTests"/> the server+relay+channel with a MOCKED
|
||||
/// <see cref="ISiteStreamSubscriber"/> 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).
|
||||
/// </summary>
|
||||
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<SiteStreamManager>.Instance);
|
||||
manager.Initialize(Sys);
|
||||
|
||||
var server = new SiteStreamGrpcServer(manager, NullLogger<SiteStreamGrpcServer>.Instance);
|
||||
server.SetReady(Sys);
|
||||
|
||||
var written = new ConcurrentQueue<SiteStreamEvent>();
|
||||
var writer = Substitute.For<IServerStreamWriter<SiteStreamEvent>>();
|
||||
writer.WriteAsync(Arg.Any<SiteStreamEvent>(), Arg.Any<CancellationToken>())
|
||||
.Returns(Task.CompletedTask)
|
||||
.AndDoes(ci => written.Enqueue(ci.Arg<SiteStreamEvent>()));
|
||||
|
||||
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<SiteStreamManager>.Instance);
|
||||
manager.Initialize(Sys);
|
||||
var server = new SiteStreamGrpcServer(manager, NullLogger<SiteStreamGrpcServer>.Instance);
|
||||
server.SetReady(Sys);
|
||||
|
||||
var written = new ConcurrentQueue<SiteStreamEvent>();
|
||||
var writer = Substitute.For<IServerStreamWriter<SiteStreamEvent>>();
|
||||
writer.WriteAsync(Arg.Any<SiteStreamEvent>(), Arg.Any<CancellationToken>())
|
||||
.Returns(Task.CompletedTask)
|
||||
.AndDoes(ci => written.Enqueue(ci.Arg<SiteStreamEvent>()));
|
||||
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<IReadOnlyList<AlarmStateChanged>>(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 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// Serialize→deserialize a proto event to mimic the gRPC HTTP/2 hop, proving the
|
||||
/// enriched <c>AlarmStateUpdate</c> is fully wire-representable (no field lost in codec).
|
||||
/// </summary>
|
||||
private static SiteStreamEvent RoundTripOverWire(SiteStreamEvent evt) =>
|
||||
SiteStreamEvent.Parser.ParseFrom(evt.ToByteArray());
|
||||
|
||||
private static ServerCallContext CreateMockContext(CancellationToken cancellationToken)
|
||||
{
|
||||
var context = Substitute.For<ServerCallContext>();
|
||||
context.CancellationToken.Returns(cancellationToken);
|
||||
return context;
|
||||
}
|
||||
|
||||
private static async Task WaitForConditionAsync(Func<bool> 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<AlarmStateChanged>? _latest;
|
||||
public void Publish(IReadOnlyList<AlarmStateChanged> snapshot)
|
||||
{
|
||||
lock (_lock) { _latest = snapshot; }
|
||||
}
|
||||
public IReadOnlyList<AlarmStateChanged>? Latest
|
||||
{
|
||||
get { lock (_lock) { return _latest; } }
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Aggregator transport double: its live stream simply stays open until cancelled.</summary>
|
||||
private sealed class NoopSiteStreamClient : SiteStreamGrpcClient
|
||||
{
|
||||
public override Task SubscribeSiteAsync(
|
||||
string correlationId, Action<AlarmStateChanged> onAlarmEvent, Action<Exception> 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;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user