using Akka.Actor; using Akka.TestKit.Xunit2; using NSubstitute; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; 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; /// /// WP2.1b — the alarm subscriber first-segment bucket index must make EXACTLY the routing /// decisions the previous linear scan made. The expectation in each case is computed with /// the original rule (plain StartsWith against every subscribed source), so the /// test pins equivalence rather than restating the new implementation. /// public class DataConnectionActorAlarmIndexTests : TestKit { 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) }; // Sources chosen to exercise every index path: // - "Area1" → NO separator: the residue list (can match other buckets) // - "Area1.Tank" → bucket "Area1", shared prefix with the next one // - "Area1.Tank1.Sub" → bucket "Area1", deeper // - "Area2.Tank" → a different bucket private static readonly string[] Sources = ["Area1", "Area1.Tank", "Area1.Tank1.Sub", "Area2.Tank"]; private static readonly (string SourceRef, string SourceObjectRef)[] Transitions = [ ("Area1.Tank1.Hi", "Area1.Tank1"), // Area1 (residue) + Area1.Tank ("Area1.Tank1.Sub.Hi", "Area1.Tank1.Sub"), // + Area1.Tank1.Sub ("Area1X.Pump.Hi", "Area1X.Pump"), // sub-segment prefix: Area1 ONLY ("Area2.Tank9.Hi", "Area2.Tank9"), // Area2.Tank only ("Zone.A.Hi", "Zone.A"), // nobody ]; private static NativeAlarmTransition Transition(string sourceRef, string sourceObj) => new(sourceRef, sourceObj, "AnalogLimit.Hi", AlarmTransitionKind.Raise, new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, 500), "Process", "hi", "hi", "", "", null, DateTimeOffset.UtcNow, "92", "90"); /// The ORIGINAL linear-scan rule, kept here as the oracle. private static bool LinearScanMatches(string sourceRef, string transitionSourceRef, string transitionSourceObjectRef) => transitionSourceObjectRef.StartsWith(sourceRef, StringComparison.Ordinal) || transitionSourceRef.StartsWith(sourceRef, StringComparison.Ordinal); [Fact] public void BucketIndex_RoutesExactlyLikeTheLinearScan() { 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(ci => Task.FromResult("alarm-" + ci.ArgAt(0))); var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor( "conn", adapter, _options, _health, _factory, "OpcUa")), "alarm-index"); // One probe per source so the routing decision per source is observable. var probes = Sources.ToDictionary(s => s, _ => CreateTestProbe()); foreach (var source in Sources) { actor.Tell(new SubscribeAlarmsRequest("c", "inst-" + source, "conn", source, null, DateTimeOffset.UtcNow), probes[source].Ref); probes[source].ExpectMsg(m => m.Success, TimeSpan.FromSeconds(5)); } Assert.NotNull(cb); foreach (var (sourceRef, sourceObjectRef) in Transitions) { cb!(Transition(sourceRef, sourceObjectRef)); foreach (var source in Sources) { if (LinearScanMatches(source, sourceRef, sourceObjectRef)) { probes[source].ExpectMsg( u => u.Transition.SourceReference == sourceRef, TimeSpan.FromSeconds(5)); } } } // Nothing extra was delivered to anybody. foreach (var source in Sources) probes[source].ExpectNoMsg(TimeSpan.FromMilliseconds(200)); } [Fact] public void SnapshotComplete_StillBroadcastsToEverySubscriber_BypassingTheIndex() { // The framing sentinel carries an EMPTY source reference, so it matches no bucket. // It must still reach every alarm subscriber, or statically-active conditions // delivered only in the snapshot would buffer forever. 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(ci => Task.FromResult("alarm-" + ci.ArgAt(0))); var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor( "conn", adapter, _options, _health, _factory, "OpcUa")), "alarm-index-snapshot"); var probes = Sources.ToDictionary(s => s, _ => CreateTestProbe()); foreach (var source in Sources) { actor.Tell(new SubscribeAlarmsRequest("c", "inst-" + source, "conn", source, null, DateTimeOffset.UtcNow), probes[source].Ref); probes[source].ExpectMsg(m => m.Success, TimeSpan.FromSeconds(5)); } Assert.NotNull(cb); cb!(new NativeAlarmTransition( "", "", "", AlarmTransitionKind.SnapshotComplete, new AlarmConditionState(false, true, null, AlarmShelveState.Unshelved, false, 0), "", "", "", "", "", null, DateTimeOffset.UtcNow, "", "")); foreach (var source in Sources) { probes[source].ExpectMsg( u => u.Transition.Kind == AlarmTransitionKind.SnapshotComplete, TimeSpan.FromSeconds(5)); } } [Fact] public void UnsubscribedSource_IsDroppedFromTheIndex() { 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(ci => Task.FromResult("alarm-" + ci.ArgAt(0))); var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor( "conn", adapter, _options, _health, _factory, "OpcUa")), "alarm-index-unsub"); var probe = CreateTestProbe(); actor.Tell(new SubscribeAlarmsRequest("c", "inst", "conn", "Area1.Tank", null, DateTimeOffset.UtcNow), probe.Ref); probe.ExpectMsg(m => m.Success, TimeSpan.FromSeconds(5)); actor.Tell(new UnsubscribeAlarmsRequest("c2", "inst", "conn", "Area1.Tank", DateTimeOffset.UtcNow), probe.Ref); Thread.Sleep(200); Assert.NotNull(cb); cb!(Transition("Area1.Tank1.Hi", "Area1.Tank1")); probe.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); } }