using Akka.Actor; using Akka.TestKit.Xunit2; using NSubstitute; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; using ZB.MOM.WW.ScadaBridge.HealthMonitoring; namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors; /// /// ScadaBridge#138 / OtOpcUa Warm design §5 "Alarms are compatible", at the actor: the promotion /// refresh OtOpcUa sends on its own (an unsolicited RefreshStart / retained conditions / RefreshEnd /// on an EXISTING alarm feed, already framed by RealOpcUaClient as SnapshotStart / Snapshot / /// SnapshotComplete under a new epoch — see RealOpcUaPromotionRefreshTests) goes through the /// existing snapshot handling: it is forwarded in order to the subscriber, and it RECONCILES the /// connection's alarm cache — a condition the promoted node no longer retains is gone from what a /// later subscriber is replayed. /// public class DataConnectionPromotionRefreshTests : TestKit { private const string Source = "Tank01"; private readonly ISiteHealthCollector _health = Substitute.For(); private readonly IDataConnectionFactory _factory = Substitute.For(); private readonly DataConnectionOptions _options = new() { ReconnectInterval = TimeSpan.FromMilliseconds(100), TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200), WriteTimeout = TimeSpan.FromSeconds(5), }; private static NativeAlarmTransition Boundary(AlarmTransitionKind kind, Guid epoch) => new( "", Source, "", kind, new AlarmConditionState(false, true, null, AlarmShelveState.Unshelved, false, 0), "", "", "", "", "", null, DateTimeOffset.UtcNow, "", "", SnapshotSourceReference: Source, SnapshotId: epoch); private static NativeAlarmTransition Condition( string condition, AlarmTransitionKind kind, Guid? epoch, bool active = true, bool acked = false) => new( $"{Source}.{condition}", Source, "AlarmConditionType", kind, new AlarmConditionState(active, acked, null, AlarmShelveState.Unshelved, false, 500), "", "", "msg", "", "", null, DateTimeOffset.UtcNow, "", "", SnapshotSourceReference: epoch is null ? null : Source, SnapshotId: epoch); /// /// Subscribe-time snapshot {Hi, Lo}; then the unsolicited promotion snapshot {Hi} (Lo cleared /// while the node was Standby). The subscriber sees the promotion frame in order, and a second /// subscriber that joins afterwards is replayed Hi only. /// [Fact] public void UnsolicitedPromotionSnapshot_IsForwardedInOrder_AndReconcilesTheCache() { AlarmTransitionCallback? cb = null; var adapter = Substitute.For(); adapter.ConnectAsync(Arg.Any>(), Arg.Any()) .Returns(Task.CompletedTask); ((IAlarmSubscribableConnection)adapter) .SubscribeAlarmsAsync(Arg.Any(), Arg.Any(), Arg.Do(c => cb = c), Arg.Any()) .Returns(Task.FromResult("alarm-sub-1")); var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor( "conn", adapter, _options, _health, _factory, "OpcUa"))); var first = CreateTestProbe(); actor.Tell(new SubscribeAlarmsRequest("c1", "inst1", "conn", Source, null, DateTimeOffset.UtcNow), first); first.ExpectMsg(m => m.Success); Assert.NotNull(cb); // Subscribe-time refresh on the then-Active node. var subscribeEpoch = Guid.NewGuid(); cb!(Boundary(AlarmTransitionKind.SnapshotStart, subscribeEpoch)); cb(Condition("Hi", AlarmTransitionKind.Snapshot, subscribeEpoch)); cb(Condition("Lo", AlarmTransitionKind.Snapshot, subscribeEpoch)); cb(Boundary(AlarmTransitionKind.SnapshotComplete, subscribeEpoch)); first.ReceiveN(4, TimeSpan.FromSeconds(3)); // Promotion on the other node: an unsolicited refresh on this existing feed. var promotionEpoch = Guid.NewGuid(); cb(Boundary(AlarmTransitionKind.SnapshotStart, promotionEpoch)); cb(Condition("Hi", AlarmTransitionKind.Snapshot, promotionEpoch)); cb(Boundary(AlarmTransitionKind.SnapshotComplete, promotionEpoch)); var frame = first.ReceiveN(3, TimeSpan.FromSeconds(3)).Cast().ToList(); Assert.Equal( [AlarmTransitionKind.SnapshotStart, AlarmTransitionKind.Snapshot, AlarmTransitionKind.SnapshotComplete], frame.Select(u => u.Transition.Kind)); Assert.All(frame, u => Assert.Equal(promotionEpoch, u.Transition.SnapshotId)); Assert.Equal($"{Source}.Hi", frame[1].Transition.SourceReference); // A live event after the promotion is still routed live. cb(Condition("Hi", AlarmTransitionKind.Acknowledge, epoch: null, acked: true)); first.ExpectMsg(u => u.Transition.Kind == AlarmTransitionKind.Acknowledge); // A subscriber joining now is replayed the reconciled state: Hi only, Lo is gone. var second = CreateTestProbe(); actor.Tell(new SubscribeAlarmsRequest("c2", "inst2", "conn", Source, null, DateTimeOffset.UtcNow), second); var replay = second.ReceiveN(4, TimeSpan.FromSeconds(3)); var transitions = replay.OfType().Select(u => u.Transition).ToList(); Assert.Single(replay.OfType(), r => r.Success); Assert.Equal( [AlarmTransitionKind.SnapshotStart, AlarmTransitionKind.Snapshot, AlarmTransitionKind.SnapshotComplete], transitions.Select(t => t.Kind)); Assert.Equal([$"{Source}.Hi"], transitions.Where(t => t.Kind == AlarmTransitionKind.Snapshot).Select(t => t.SourceReference)); second.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); } }