using System.Text.Json; using Moq; using Opc.Ua; using Opc.Ua.Client; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests; /// /// Gitea #670 — upstream session loss must reach the driver's health and its subscribers. Before the fix a bad /// keep-alive only flipped the host status: the subscribed values stayed Good forever, health stayed Healthy (so the /// #527 comms-loss watchdog could never trip — its first conjunct is "non-Healthy"), and LastSuccessfulRead /// only moved on init / ReadAsync, which a subscription-only driver never calls. /// [Trait("Category", "Unit")] public sealed class OpcUaClientSessionLossTests { private const uint BadNoCommunication = 0x80310000u; private const string SpeedRef = "Plant/dev/Speed"; private const string CounterRef = "Plant/dev/Counter"; private sealed class ManualClock(DateTimeOffset start) : TimeProvider { public DateTimeOffset Now { get; set; } = start; public override DateTimeOffset GetUtcNow() => Now; public void Advance(TimeSpan by) => Now += by; } private static RawTagEntry Tag(string rawPath, string nodeId) => new(rawPath, JsonSerializer.Serialize(new { nodeId }), WriteIdempotent: true); private static readonly IReadOnlyList Tags = [ Tag(SpeedRef, "ns=0;s=Speed"), Tag(CounterRef, "ns=0;s=Counter"), ]; private static Mock NewSession() { var mock = new Mock(MockBehavior.Loose); mock.SetupGet(s => s.MessageContext) .Returns(new ServiceMessageContext(SdkTelemetry.Context)); mock.SetupGet(s => s.NamespaceUris).Returns(new NamespaceTable()); return mock; } private static (OpcUaClientDriver Driver, Mock Session, ManualClock Clock) NewConnectedDriver() { var drv = new OpcUaClientDriver(new OpcUaClientDriverOptions { RawTags = Tags }, "opcua-670"); var session = NewSession(); var clock = new ManualClock(new DateTimeOffset(2026, 10, 11, 12, 0, 0, TimeSpan.Zero)); drv.Clock = clock; drv.SetSessionForTest(session.Object); drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags)); drv.SetHealthForTest(new DriverHealth(DriverState.Healthy, clock.Now.UtcDateTime, null)); return (drv, session, clock); } private static DataValue Good(object value, DateTime ts) => new(new Variant(value), StatusCodes.Good, ts, ts); private static void SetupRead(Mock session, params object[] values) => session.Setup(s => s.ReadAsync( It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) .ReturnsAsync((RequestHeader _, double _, TimestampsToReturn _, ReadValueIdCollection nodes, CancellationToken _) => new ReadResponse { ResponseHeader = new ResponseHeader(), Results = [.. nodes.Select((_, i) => new DataValue(new Variant(values[i]), StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow))], DiagnosticInfos = [], }); private static async Task WaitForAsync(Func condition, string because) { var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); while (DateTime.UtcNow < deadline) { if (condition()) return; await Task.Delay(10); } condition().ShouldBeTrue(because); } [Fact] public void A_bad_keep_alive_degrades_health_and_preserves_the_last_successful_read() { var (drv, _, clock) = NewConnectedDriver(); using var _ = drv; var before = drv.GetHealth().LastSuccessfulRead; clock.Advance(TimeSpan.FromSeconds(5)); drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); var h = drv.GetHealth(); h.State.ShouldNotBe(DriverState.Healthy, "a lost session must be visible to the #527 watchdog's first conjunct"); h.State.ShouldBe(DriverState.Degraded); h.LastSuccessfulRead.ShouldBe(before, "the watchdog measures silence from this stamp, so loss must not move it"); h.LastError.ShouldNotBeNull().ShouldContain("keep-alive"); } [Fact] public void A_bad_keep_alive_publishes_BadNoCommunication_once_per_ref_retaining_the_value() { var (drv, _, clock) = NewConnectedDriver(); using var _ = drv; var handle = drv.AddSubscriptionForTest([SpeedRef, CounterRef]); var t0 = clock.Now.UtcDateTime; drv.DeliverNotificationForTest(handle, SpeedRef, Good(42.5f, t0)); drv.DeliverNotificationForTest(handle, CounterRef, Good(7, t0)); var seen = new List(); drv.OnDataChange += (_, e) => seen.Add(e); drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); seen.Count.ShouldBe(2, "exactly one Bad per subscribed ref"); var speed = seen.Single(e => e.FullReference == SpeedRef); speed.Snapshot.StatusCode.ShouldBe(BadNoCommunication); speed.Snapshot.Value.ShouldBe(42.5f, "the value is retained — only the quality changes (as the host does)"); speed.SubscriptionHandle.ShouldBe(handle); seen.Single(e => e.FullReference == CounterRef).Snapshot.Value.ShouldBe(7); // The SDK re-raises a bad keep-alive every interval while the channel stays down: no further publishes. drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); drv.RaiseKeepAliveForTest(StatusCodes.BadConnectionClosed); seen.Count.ShouldBe(2, "a second bad keep-alive in the same outage must publish nothing"); } [Fact] public void Good_keep_alives_and_notifications_advance_the_last_successful_read() { var (drv, _, clock) = NewConnectedDriver(); using var _ = drv; var handle = drv.AddSubscriptionForTest([SpeedRef]); clock.Advance(TimeSpan.FromSeconds(30)); drv.RaiseKeepAliveForTest(StatusCodes.Good); drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime); clock.Advance(TimeSpan.FromSeconds(30)); drv.DeliverNotificationForTest(handle, SpeedRef, Good(1.0f, clock.Now.UtcDateTime)); drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime); drv.GetHealth().State.ShouldBe(DriverState.Healthy); } /// /// Falsification control for the R2 half of the fix: a long-running, idle, subscription-only driver whose /// server is perfectly alive must never look dead to the #527 watchdog (non-Healthy AND silent ≥ 90 s). /// Without good keep-alives advancing LastSuccessfulRead, fixing R1 alone would trip it on the first /// bad tick of any connection older than 90 s. /// [Fact] public void An_idle_healthy_subscription_driver_never_looks_dead_to_the_watchdog() { var (drv, _, clock) = NewConnectedDriver(); using var _ = drv; drv.AddSubscriptionForTest([SpeedRef]); // static value — no notifications ever arrive for (var i = 0; i < 60; i++) // 5 minutes of 5 s keep-alives { clock.Advance(TimeSpan.FromSeconds(5)); drv.RaiseKeepAliveForTest(StatusCodes.Good); var h = drv.GetHealth(); h.State.ShouldBe(DriverState.Healthy); (clock.Now.UtcDateTime - h.LastSuccessfulRead!.Value).ShouldBeLessThan(TimeSpan.FromSeconds(90)); } } [Fact] public async Task Recovery_on_the_same_session_republishes_the_identical_value_Good() { var (drv, session, clock) = NewConnectedDriver(); using var _ = drv; var handle = drv.AddSubscriptionForTest([SpeedRef]); drv.DeliverNotificationForTest(handle, SpeedRef, Good(42.5f, clock.Now.UtcDateTime)); SetupRead(session, 42.5f); // the server still holds the SAME value — it will not resend it var seen = new List(); drv.OnDataChange += (_, e) => { lock (seen) seen.Add(e); }; drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); clock.Advance(TimeSpan.FromSeconds(20)); drv.RaiseKeepAliveForTest(StatusCodes.Good); drv.GetHealth().State.ShouldBe(DriverState.Healthy); drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime); await WaitForAsync(() => { lock (seen) return seen.Count >= 2; }, "the recovery read must publish"); lock (seen) { seen.Count.ShouldBe(2); seen[0].Snapshot.StatusCode.ShouldBe(BadNoCommunication); seen[1].FullReference.ShouldBe(SpeedRef); seen[1].SubscriptionHandle.ShouldBe(handle); seen[1].Snapshot.StatusCode.ShouldBe(0u); seen[1].Snapshot.Value.ShouldBe(42.5f, "the IDENTICAL value must come back Good — a (value,status) dedup lets it through"); } } [Fact] public async Task A_notification_during_recovery_wins_over_the_recovery_read() { var (drv, session, clock) = NewConnectedDriver(); using var _ = drv; var handle = drv.AddSubscriptionForTest([SpeedRef]); drv.DeliverNotificationForTest(handle, SpeedRef, Good(1.0f, clock.Now.UtcDateTime)); drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); // A fresh notification arrives before the keep-alive recovers: the read must not overwrite it. drv.DeliverNotificationForTest(handle, SpeedRef, Good(2.0f, clock.Now.UtcDateTime)); SetupRead(session, 1.0f); var seen = new List(); drv.OnDataChange += (_, e) => { lock (seen) seen.Add(e); }; drv.RaiseKeepAliveForTest(StatusCodes.Good); await Task.Delay(200, TestContext.Current.CancellationToken); lock (seen) seen.ShouldBeEmpty(); } [Fact] public void A_bad_keep_alive_annotates_subscribed_alarm_conditions_quality_only_once() { var (drv, _, _) = NewConnectedDriver(); using var _ = drv; drv.AddAlarmSubscriptionForTest([SpeedRef]); var seen = new List(); drv.OnAlarmEvent += (_, e) => seen.Add(e); drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); var e = seen.ShouldHaveSingleItem(); e.ConditionId.ShouldBe(SpeedRef); e.IsQualityOnly.ShouldBeTrue("an outage says nothing about the alarm state — quality only, no transition"); e.QualityStatusCode.ShouldBe(BadNoCommunication); } [Fact] public async Task Recovery_restores_alarm_condition_quality_Good() { var (drv, session, _) = NewConnectedDriver(); using var _ = drv; SetupRead(session); drv.AddAlarmSubscriptionForTest([SpeedRef]); var seen = new List(); drv.OnAlarmEvent += (_, e) => { lock (seen) seen.Add(e); }; drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); drv.RaiseKeepAliveForTest(StatusCodes.Good); await WaitForAsync(() => { lock (seen) return seen.Count >= 2; }, "recovery must re-annotate Good"); lock (seen) { seen[1].IsQualityOnly.ShouldBeTrue(); seen[1].QualityStatusCode.ShouldBe(0u); } } [Fact] public void A_keep_alive_from_a_superseded_session_is_ignored() { var (drv, _, _) = NewConnectedDriver(); using var _ = drv; var stale = NewSession(); drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication, stale.Object); drv.GetHealth().State.ShouldBe(DriverState.Healthy); } [Fact] public async Task Shutdown_clears_the_outage_so_a_fresh_session_reports_its_own_loss() { var (drv, _, _) = NewConnectedDriver(); using var _ = drv; drv.AddSubscriptionForTest([SpeedRef]); drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); await drv.ShutdownAsync(TestContext.Current.CancellationToken); // As after a watchdog reinit: a new session + subscription, and a new outage must be reported again. var fresh = NewSession(); drv.SetSessionForTest(fresh.Object); drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags)); drv.AddSubscriptionForTest([SpeedRef]); var seen = new List(); drv.OnDataChange += (_, e) => seen.Add(e); drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); seen.ShouldHaveSingleItem().Snapshot.StatusCode.ShouldBe(BadNoCommunication); } /// /// Review M2: a recovery overtaken by a re-initialise (watchdog Shutdown + Initialize) must not leak its /// outcome onto the NEW session — previously its failed read re-armed the outage latch, so the new session's /// first real outage was swallowed. /// [Fact] public async Task A_recovery_overtaken_by_a_reinit_does_not_latch_the_new_session() { var ct = TestContext.Current.CancellationToken; var (drv, _, clock) = NewConnectedDriver(); using var _ = drv; var handle = drv.AddSubscriptionForTest([SpeedRef]); drv.DeliverNotificationForTest(handle, SpeedRef, Good(1.0f, clock.Now.UtcDateTime)); drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); await drv.Gate.WaitAsync(ct); // park the recovery read on the gate drv.RaiseKeepAliveForTest(StatusCodes.Good); await drv.ShutdownAsync(ct); // the watchdog's re-initialise overtakes it … var fresh = NewSession(); // … and a new session is up (its reads would fail) drv.SetSessionForTest(fresh.Object); drv.SetNamespaceMapForTest(NamespaceMap.FromTable(new NamespaceTable(), Tags)); drv.SetHealthForTest(new DriverHealth(DriverState.Healthy, clock.Now.UtcDateTime, null)); drv.AddSubscriptionForTest([SpeedRef]); drv.Gate.Release(); await drv.RecoveryTaskForTest; drv.GetHealth().State.ShouldBe(DriverState.Healthy, "the stale recovery must not degrade the new session"); var seen = new List(); drv.OnDataChange += (_, e) => seen.Add(e); drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); seen.ShouldHaveSingleItem("the new session's first outage must be published, not swallowed by a stale latch") .Snapshot.StatusCode.ShouldBe(BadNoCommunication); } /// /// Review M1: the subscription dies while the session lives (an outage longer than the subscription lifetime /// but shorter than the session timeout). Keep-alives stay Good, so without detection the driver ends Healthy /// with fresh stamps and frozen values forever. It must go Degraded, publish the loss once, and STOP advancing /// the read stamp — so the watchdog's reinit recreates the subscription. /// [Fact] public void A_subscription_that_stops_publishing_while_the_session_lives_freezes_the_stamp_and_goes_Bad() { var (drv, _, clock) = NewConnectedDriver(); using var _ = drv; var handle = drv.AddSubscriptionForTest([SpeedRef]); var otherHandle = drv.AddSubscriptionForTest([CounterRef]); drv.DeliverNotificationForTest(handle, SpeedRef, Good(3.0f, clock.Now.UtcDateTime)); var stalled = false; drv.IsPublishingStopped = sub => stalled && sub.DisplayName == "opcua-sub-1"; drv.PublishingStallGrace = _ => TimeSpan.FromSeconds(10); var seen = new List(); drv.OnDataChange += (_, e) => seen.Add(e); stalled = true; clock.Advance(TimeSpan.FromSeconds(5)); drv.RaiseKeepAliveForTest(StatusCodes.Good); // first observation — within grace drv.GetHealth().State.ShouldBe(DriverState.Healthy); clock.Advance(TimeSpan.FromSeconds(10)); drv.RaiseKeepAliveForTest(StatusCodes.Good); // grace elapsed var h = drv.GetHealth(); h.State.ShouldBe(DriverState.Degraded); h.LastError.ShouldNotBeNull().ShouldContain("subscription lost"); var frozen = h.LastSuccessfulRead; seen.Count(e => e.Snapshot.StatusCode == BadNoCommunication).ShouldBe(2, "every subscribed ref published Bad once"); seen.Single(e => e.FullReference == SpeedRef).Snapshot.Value.ShouldBe(3.0f); // Nothing may revive the stamp: not good keep-alives, not a bad→good keep-alive cycle, not notifications // from a still-healthy sibling subscription. for (var i = 0; i < 30; i++) { clock.Advance(TimeSpan.FromSeconds(5)); drv.RaiseKeepAliveForTest(StatusCodes.Good); drv.DeliverNotificationForTest(otherHandle, CounterRef, Good(i, clock.Now.UtcDateTime)); } drv.RaiseKeepAliveForTest(StatusCodes.BadNoCommunication); drv.RaiseKeepAliveForTest(StatusCodes.Good); h = drv.GetHealth(); h.State.ShouldBe(DriverState.Degraded); h.LastSuccessfulRead.ShouldBe(frozen, "the #527 watchdog needs a frozen stamp to re-initialise the driver"); seen.Count(e => e.Snapshot.StatusCode == BadNoCommunication).ShouldBe(2, "the loss is published once per connection"); } /// Control for M1: a subscription that is briefly silent (e.g. right after an outage) and then resumes /// within its grace is NOT declared dead. [Fact] public void A_subscription_silent_for_less_than_its_grace_is_not_declared_lost() { var (drv, _, clock) = NewConnectedDriver(); using var _ = drv; drv.AddSubscriptionForTest([SpeedRef]); var stalled = true; drv.IsPublishingStopped = _ => stalled; drv.PublishingStallGrace = _ => TimeSpan.FromSeconds(10); drv.RaiseKeepAliveForTest(StatusCodes.Good); clock.Advance(TimeSpan.FromSeconds(9)); drv.RaiseKeepAliveForTest(StatusCodes.Good); stalled = false; for (var i = 0; i < 10; i++) { clock.Advance(TimeSpan.FromSeconds(5)); drv.RaiseKeepAliveForTest(StatusCodes.Good); } stalled = true; // a NEW stall restarts the grace rather than inheriting the old one clock.Advance(TimeSpan.FromSeconds(5)); drv.RaiseKeepAliveForTest(StatusCodes.Good); drv.GetHealth().State.ShouldBe(DriverState.Healthy); drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime); } [Fact] public void A_BadNoSubscription_publish_error_with_tracked_subscriptions_declares_the_subscription_lost() { var (drv, _, _) = NewConnectedDriver(); using var _ = drv; drv.AddSubscriptionForTest([SpeedRef]); var seen = new List(); drv.OnDataChange += (_, e) => seen.Add(e); drv.RaisePublishErrorForTest(StatusCodes.BadSubscriptionIdInvalid, subscriptionId: 4242); // not ours drv.GetHealth().State.ShouldBe(DriverState.Healthy); drv.RaisePublishErrorForTest(StatusCodes.BadNoSubscription); drv.GetHealth().State.ShouldBe(DriverState.Degraded); seen.ShouldHaveSingleItem().Snapshot.StatusCode.ShouldBe(BadNoCommunication); } [Fact] public void A_BadNoSubscription_with_nothing_tracked_is_not_a_loss() { var (drv, _, _) = NewConnectedDriver(); using var _ = drv; drv.RaisePublishErrorForTest(StatusCodes.BadNoSubscription); drv.GetHealth().State.ShouldBe(DriverState.Healthy); } [Fact] public async Task Shutdown_clears_a_subscription_loss() { var (drv, _, clock) = NewConnectedDriver(); using var _ = drv; drv.AddSubscriptionForTest([SpeedRef]); drv.RaisePublishErrorForTest(StatusCodes.BadNoSubscription); await drv.ShutdownAsync(TestContext.Current.CancellationToken); drv.SetSessionForTest(NewSession().Object); drv.SetHealthForTest(new DriverHealth(DriverState.Healthy, clock.Now.UtcDateTime, null)); clock.Advance(TimeSpan.FromSeconds(5)); drv.RaiseKeepAliveForTest(StatusCodes.Good); drv.GetHealth().LastSuccessfulRead.ShouldBe(clock.Now.UtcDateTime, "the re-subscribed connection starts clean"); } [Fact] public void Sdk_log_forwarding_is_rate_limited() { var target = new CountingLogger(); var limited = new SdkTelemetry.RateLimitedLogger(target, max: 3, window: TimeSpan.FromHours(1)); for (var i = 0; i < 50; i++) Microsoft.Extensions.Logging.LoggerExtensions.LogWarning(limited, "sdk warning {I}", i); target.Count.ShouldBe(3); } private sealed class CountingLogger : Microsoft.Extensions.Logging.ILogger { public int Count; public IDisposable? BeginScope(TState state) where TState : notnull => null; public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true; public void Log(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId, TState state, Exception? exception, Func formatter) => Count++; } }