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.Enums; using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; using ZB.MOM.WW.ScadaBridge.HealthMonitoring; using ZB.MOM.WW.ScadaBridge.SiteEventLogging; namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors; /// /// Low-severity housekeeping findings from review 2026-10-21 (02/L2, 02/L4, 02/U4): bookkeeping /// that outlived the subscription or the session it belonged to, and the connection-transition /// event-log contract. /// [Collection("DataConnectionManagerActor")] public class DataConnectionHousekeepingTests : TestKit { private const string Tag = "sensor/temp"; private readonly ISiteHealthCollector _health = Substitute.For(); private readonly IDataConnectionFactory _factory = Substitute.For(); private static DataConnectionOptions Options() => new() { ReconnectInterval = TimeSpan.FromMilliseconds(100), TagResolutionRetryInterval = TimeSpan.FromSeconds(30), WriteTimeout = TimeSpan.FromSeconds(5), SeedReadMaxAttempts = 1, SeedReadRetryDelay = TimeSpan.FromMilliseconds(10), SeedOverallTimeout = TimeSpan.FromSeconds(60), SeedReadTimeout = TimeSpan.FromSeconds(60), QualityFlushInterval = TimeSpan.FromMilliseconds(100) }; private IActorRef CreateActor(FakeBatchDataConnection adapter, string name) => Sys.ActorOf(Props.Create(() => new DataConnectionActor( name, adapter, Options(), _health, _factory, "OpcUa")), name); private int[]? LastQualityPush() => _health.ReceivedCalls() .Where(c => c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateTagQuality)) .Select(c => c.GetArguments()) .Select(a => new[] { (int)a[1]!, (int)a[2]!, (int)a[3]! }) .LastOrDefault(); private int ConnectedTransitions(string name) => _health.ReceivedCalls().Count(c => c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionHealth) && c.GetArguments() is [string n, ConnectionHealth h, ..] && n == name && h == ConnectionHealth.Connected); /// /// 02/L2: a value for a tag nobody subscribes to any more (delivered after the last /// unsubscribe, before the device unadvise lands) must not re-enter the quality bookkeeping — /// that leaked a _lastTagQuality entry and a phantom Good count above the subscribed /// total until the next reconnect. /// [Fact] public void ValueForUnsubscribedTag_DoesNotReEnterQualityBookkeeping() { const string name = "late-value-after-unsubscribe"; var adapter = new FakeBatchDataConnection(); var actor = CreateActor(adapter, name); actor.Tell(new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); ExpectMsg(m => m.Success, TimeSpan.FromSeconds(5)); AwaitCondition(() => LastQualityPush() is [1, 0, 0], TimeSpan.FromSeconds(5)); actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", name, DateTimeOffset.UtcNow)); AwaitCondition(() => !adapter.UnsubscribeBatches.IsEmpty, TimeSpan.FromSeconds(5)); AwaitCondition(() => LastQualityPush() is [0, 0, 0], TimeSpan.FromSeconds(5)); // The late callback. adapter.ValueCallback!(Tag, new TagValue(7, QualityCode.Good, DateTimeOffset.UtcNow)); Thread.Sleep(500); // > QualityFlushInterval: a dirty counter would have been pushed Assert.Equal([0, 0, 0], LastQualityPush()!); } /// /// 02/L4: the reconnect re-seed is bound to its session. A drop while it is still reading must /// cancel it instead of leaving a full-tag-set read running against a struggling device for /// up to SeedOverallTimeout — one more per flap. /// [Fact] public void ReconnectReseed_IsCancelled_WhenTheSessionDropsAgain() { const string name = "reseed-cancelled-on-drop"; var adapter = new FakeBatchDataConnection(); var actor = CreateActor(adapter, name); actor.Tell(new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow)); ExpectMsg(m => m.Success, TimeSpan.FromSeconds(5)); adapter.HangReads = true; var readsBefore = adapter.ReadTokens.Count; adapter.RaiseDisconnected(); AwaitCondition(() => ConnectedTransitions(name) >= 2, TimeSpan.FromSeconds(5)); AwaitCondition(() => adapter.ReadTokens.Count > readsBefore, TimeSpan.FromSeconds(5)); var reseedToken = adapter.ReadTokens.ElementAt(readsBefore); Assert.False(reseedToken.IsCancellationRequested); adapter.RaiseDisconnected(); AwaitCondition(() => reseedToken.IsCancellationRequested, TimeSpan.FromSeconds(5)); } /// /// 02/U4(b): the spec says every connection transition is logged to Site Event Logging — /// the INITIAL connect included, not just reconnect-restored / lost / failover. /// [Fact] public void InitialConnect_IsLoggedToSiteEventLog() { const string name = "initial-connect-logged"; var eventLog = Substitute.For(); var adapter = new FakeBatchDataConnection(); Sys.ActorOf(Props.Create(() => new DataConnectionActor( name, adapter, Options(), _health, _factory, "OpcUa", null, null, 3, eventLog)), name); AwaitCondition(() => eventLog.ReceivedCalls().Any(c => c.GetMethodInfo().Name == nameof(ISiteEventLogger.LogEventAsync) && c.GetArguments() is ["connection", "Info", null, name, string message, ..] && message.Contains("established", StringComparison.OrdinalIgnoreCase)), TimeSpan.FromSeconds(5)); } }