using Akka.Actor; using Akka.Event; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; /// /// Covers WS-4b: subscribes a driver's native /// (when the driver is an ) /// and forwards every transition to its parent as /// — mirroring the OnDataChange → /// forward. The driver fires /// OnAlarmEvent on its OWN thread; the actor marshals it onto the actor thread via Self. /// public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase { /// /// Driving an driver to Connected then raising a native alarm forwards /// it to the parent as carrying the /// driver-instance id + the original (SourceNodeId preserved). /// [Fact] public async Task Native_alarm_is_forwarded_to_parent_as_AttributeAlarmPublished() { var driver = new AlarmSourceStubDriver(); var parent = CreateTestProbe(); var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver)); actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); AwaitCondition(() => driver.InitializeCount > 0, TimeSpan.FromSeconds(2)); // The alarm handler is attached after Become(Connected); wait for it to be wired before raising. AwaitCondition(() => driver.AlarmSubscriberCount == 1, TimeSpan.FromSeconds(2)); // Also establish a data-change subscription so the OnDataChange path is wired (the alarm attach is // independent of Subscribe — this proves both event paths coexist on one driver). await actor.Ask( new DriverInstanceActor.Subscribe(new[] { "tag-z" }, TimeSpan.FromMilliseconds(100)), TimeSpan.FromSeconds(3)); driver.RaiseAlarm(new AlarmEventArgs( new StubAlarmHandle(), SourceNodeId: "src-node-7", ConditionId: "cond-1", AlarmType: "AnalogLimitAlarm.HiHi", Message: "level too high", Severity: AlarmSeverity.High, SourceTimestampUtc: DateTime.UtcNow, Kind: AlarmTransitionKind.Raise)); var published = parent.ExpectMsg(TimeSpan.FromSeconds(2)); published.DriverInstanceId.ShouldBe(driver.DriverInstanceId); published.Args.SourceNodeId.ShouldBe("src-node-7"); published.Args.ConditionId.ShouldBe("cond-1"); published.Args.Kind.ShouldBe(AlarmTransitionKind.Raise); // The same driver's OnDataChange still flows independently — alarm + value events coexist. driver.FireDataChange("tag-z", value: 9.0, statusCode: 0u); parent.ExpectMsg(TimeSpan.FromSeconds(2)) .FullReference.ShouldBe("tag-z"); } /// /// The native-alarm path is gated at the driver: an suppresses /// OnAlarmEvent until at least one alarm subscription exists (e.g. GalaxyDriver gates its /// central feed). When the host pushes a /// carrying native-alarm refs to a Connected driver — the live deploy path — the actor must /// call with those refs to un-gate the feed, and must /// NOT re-subscribe when the same set is redeployed (the feed is already un-gated). /// [Fact] public async Task Alarm_subscription_is_established_when_alarm_refs_are_pushed_while_connected() { var driver = new AlarmSourceStubDriver(); var parent = CreateTestProbe(); var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver)); actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); AwaitCondition(() => driver.AlarmSubscriberCount == 1, TimeSpan.FromSeconds(2)); // reached Connected driver.SubscribeAlarmsCallCount.ShouldBe(0, "no alarm subscription before any alarm refs are pushed"); // A deploy delivers the desired set with a native-alarm ref while the driver is already Connected. actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions( Array.Empty(), TimeSpan.FromMilliseconds(100), new[] { "Temp.HiHi" })); AwaitCondition(() => driver.SubscribeAlarmsCallCount == 1, TimeSpan.FromSeconds(2)); driver.LastAlarmRefs.ShouldBe(new[] { "Temp.HiHi" }); // Redeploying the same alarm set must NOT re-subscribe (idempotent — already un-gated). Round-trip a // data Subscribe to flush the mailbox so the second SetDesiredSubscriptions is fully processed first. actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions( Array.Empty(), TimeSpan.FromMilliseconds(100), new[] { "Temp.HiHi" })); await actor.Ask( new DriverInstanceActor.Subscribe(new[] { "tag-z" }, TimeSpan.FromMilliseconds(100)), TimeSpan.FromSeconds(3)); driver.SubscribeAlarmsCallCount.ShouldBe(1, "an already-established alarm subscription is not re-issued"); } /// /// A driver that is NOT an (only ) connects /// and serves data changes normally — AttachAlarmSource is a safe no-op (no crash) and no /// is ever produced. /// [Fact] public async Task Non_alarm_source_driver_serves_data_changes_and_never_publishes_alarms() { var driver = new SubscribableOnlyStubDriver(); var parent = CreateTestProbe(); var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver)); actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); AwaitCondition(() => driver.InitializeCount > 0, TimeSpan.FromSeconds(2)); await actor.Ask( new DriverInstanceActor.Subscribe(new[] { "tag-a" }, TimeSpan.FromMilliseconds(100)), TimeSpan.FromSeconds(3)); driver.FireDataChange("tag-a", value: 1.5, statusCode: 0u); // Data-change forwarding still works (no crash on AttachAlarmSource for a non-IAlarmSource driver). var dc = parent.ExpectMsg(TimeSpan.FromSeconds(2)); dc.FullReference.ShouldBe("tag-a"); // An AttributeAlarmPublished can never be produced for a non-IAlarmSource driver. parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); } /// /// A native alarm transition that races in while the actor is in Reconnecting is silently /// dropped (debug log only) — it NEVER reaches the parent as /// AND is NOT dead-lettered (the /// Reconnecting state's explicit Receive<NativeAlarmRaised> handler consumes and /// discards it, so it never becomes unhandled). The feed re-delivers active alarms once the actor /// re-enters Connected, so dropping here is safe. /// ( line ~345: the Reconnecting state's /// Receive<NativeAlarmRaised> logs a debug message and discards.) /// /// /// To reproduce the race deterministically: a ForceReconnect is enqueued first, then /// the driver fires an alarm while its OnAlarmEvent handler is still attached (the actor /// hasn't yet processed ForceReconnect). The handler's self.Tell(NativeAlarmRaised) /// lands second in the mailbox. The actor then processes ForceReconnect → Reconnecting /// (detaches handler), then processes the queued NativeAlarmRaised in Reconnecting /// → drops it. The default 10 s reconnect interval ensures no retry fires during the check. /// /// /// /// Both properties are actively asserted: (a) the parent probe receives no /// ; (b) a dead-letter probe subscribed /// to on the also receives /// nothing — proving the drop handler is present (removing it would cause a dead-letter and fail /// this assertion). NativeAlarmRaised is private to the actor, so the dead-letter probe /// subscribes to the unfiltered channel; this is safe because /// exactly one message is injected and the reconnect timer (10 s) cannot fire in the window. /// /// [Fact] public void Native_alarm_during_reconnect_is_dropped_not_forwarded() { // Subscribe a dead-letter probe BEFORE injecting the alarm so we don't miss any early publish. // NativeAlarmRaised is private, so we subscribe to the unfiltered AllDeadLetters channel. // Only one message is injected and the 10 s reconnect timer can't fire in this window, so // a plain "no dead letters at all" assertion is safe and non-flaky. var deadLetters = CreateTestProbe(); Sys.EventStream.Subscribe(deadLetters.Ref, typeof(AllDeadLetters)); // Long reconnect interval (default 10 s) so the retry doesn't fire during the assertion window. var driver = new AlarmSourceStubDriver(); var parent = CreateTestProbe(); var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver)); // Drive to Connected; confirm the alarm handler is attached. actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); AwaitCondition(() => driver.AlarmSubscriberCount == 1, TimeSpan.FromSeconds(2)); // Enqueue ForceReconnect FIRST — the actor hasn't processed it yet, so the handler is // still wired. The test thread then immediately fires the alarm on the driver; the handler's // self.Tell(NativeAlarmRaised) lands SECOND in the mailbox. actor.Tell(new DriverInstanceActor.ForceReconnect()); driver.RaiseAlarm(new AlarmEventArgs( new StubAlarmHandle(), SourceNodeId: "src-node-reconnect", ConditionId: "cond-reconnect", AlarmType: "T", Message: "during-reconnect raise", Severity: AlarmSeverity.High, SourceTimestampUtc: DateTime.UtcNow, Kind: AlarmTransitionKind.Raise)); // (a) The parent must NOT receive AttributeAlarmPublished from that alarm. // The actor processes: (1) ForceReconnect → Reconnecting (handler detached); // (2) NativeAlarmRaised → dropped (debug log, no forward). // Wait generously — the default reconnect interval of 10 s means no retry fires here. parent.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); // (b) The alarm must also NOT have dead-lettered — the Reconnecting state's explicit // Receive handler must have consumed it. If that handler were removed the // message would become unhandled → AllDeadLetters, and this assertion would catch the regression. deadLetters.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); // The actor must still be alive — Watch + no Terminated. Watch(actor); parent.ExpectNoMsg(TimeSpan.FromMilliseconds(100)); } /// /// After a full reconnect cycle (Connected → Reconnecting → Connected), a single raised alarm still /// yields EXACTLY ONE — the /// _alarmEventHandler is not null guard in AttachAlarmSource plus the detach on the /// Connected → Reconnecting transition prevent a double-attach (which would publish twice). /// [Fact] public void Reconnect_does_not_double_attach_alarm_handler() { var driver = new AlarmSourceStubDriver(); var parent = CreateTestProbe(); var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver, reconnectInterval: TimeSpan.FromMilliseconds(50))); actor.Tell(new DriverInstanceActor.InitializeRequested("{}")); AwaitCondition(() => driver.AlarmSubscriberCount == 1, TimeSpan.FromSeconds(2)); // Force Connected → Reconnecting (detaches) → Connected (re-attaches). InitializeCount climbs on // the retry; the alarm handler count must settle back to exactly one (no leaked extra handler). var initBefore = driver.InitializeCount; actor.Tell(new DriverInstanceActor.DisconnectObserved("backend blip")); AwaitCondition(() => driver.InitializeCount > initBefore, TimeSpan.FromSeconds(3)); AwaitCondition(() => driver.AlarmSubscriberCount == 1, TimeSpan.FromSeconds(3)); driver.RaiseAlarm(new AlarmEventArgs( new StubAlarmHandle(), SourceNodeId: "src-node-1", ConditionId: "cond-x", AlarmType: "T", Message: "m", Severity: AlarmSeverity.Low, SourceTimestampUtc: DateTime.UtcNow, Kind: AlarmTransitionKind.Raise)); // Exactly one forward — a double-attach would surface as a second AttributeAlarmPublished. parent.ExpectMsg(TimeSpan.FromSeconds(2)) .Args.SourceNodeId.ShouldBe("src-node-1"); parent.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); } // --- stub drivers ---------------------------------------------------------------------------- private class StubDriver : IDriver { /// Gets the number of times initialization was called. public int InitializeCount; /// Gets the driver instance ID. public string DriverInstanceId => "alarm-stub-driver-1"; /// Gets the driver type. public string DriverType => "Stub"; /// Initializes the driver. public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { Interlocked.Increment(ref InitializeCount); return Task.CompletedTask; } /// Reinitializes the driver. public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; /// Shuts down the driver. public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; /// Gets the health status of the driver. public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, null); /// Gets the memory footprint of the driver. public long GetMemoryFootprint() => 0; /// Flushes optional caches in the driver. public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask; } /// A driver that implements + and lets the /// test raise on demand (the driver fires it on its own thread in /// production; here the test thread stands in for that). private sealed class AlarmSourceStubDriver : StubDriver, ISubscribable, IAlarmSource { private readonly StubHandle _subHandle = new(); /// Occurs when data changes. public event EventHandler? OnDataChange; /// Server-pushed alarm transition. public event EventHandler? OnAlarmEvent; /// Number of live subscribers on — proves attach/detach. public int AlarmSubscriberCount => OnAlarmEvent?.GetInvocationList().Length ?? 0; /// Subscribes to the specified full references. public Task SubscribeAsync( IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) => Task.FromResult(_subHandle); /// Unsubscribes from the specified subscription handle. public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask; private int _subscribeAlarmsCallCount; /// Number of calls — proves the actor established the /// native-alarm subscription that un-gates the feed (incremented on the actor dispatcher thread). public int SubscribeAlarmsCallCount => Volatile.Read(ref _subscribeAlarmsCallCount); /// The node set handed to the most recent call. public IReadOnlyList? LastAlarmRefs { get; private set; } /// Subscribes to alarm events for the specified node set. public Task SubscribeAlarmsAsync( IReadOnlyList sourceNodeIds, CancellationToken cancellationToken) { LastAlarmRefs = sourceNodeIds; Interlocked.Increment(ref _subscribeAlarmsCallCount); return Task.FromResult(new StubAlarmHandle()); } /// Cancels an alarm subscription. public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask; /// Acknowledges a batch of alarms. public Task AcknowledgeAsync( IReadOnlyList acknowledgements, CancellationToken cancellationToken) => Task.CompletedTask; /// Fires — stands in for the driver's native feed. public void RaiseAlarm(AlarmEventArgs args) => OnAlarmEvent?.Invoke(this, args); /// Fires a data change event (keeps OnDataChange exercised; the actor wires both events). public void FireDataChange(string fullRef, object? value, uint statusCode) { var snapshot = new DataValueSnapshot(value, statusCode, DateTime.UtcNow, DateTime.UtcNow); OnDataChange?.Invoke(this, new DataChangeEventArgs(_subHandle, fullRef, snapshot)); } private sealed class StubHandle : ISubscriptionHandle { /// Gets the diagnostic ID of the subscription. public string DiagnosticId => "stub-sub"; } } /// A driver that is but NOT — proves /// AttachAlarmSource is a no-op (no crash) and no alarm forward ever happens. private sealed class SubscribableOnlyStubDriver : StubDriver, ISubscribable { private readonly StubHandle _subHandle = new(); /// Occurs when data changes. public event EventHandler? OnDataChange; /// Subscribes to the specified full references. public Task SubscribeAsync( IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) => Task.FromResult(_subHandle); /// Unsubscribes from the specified subscription handle. public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask; /// Fires a data change event with the specified parameters. public void FireDataChange(string fullRef, object? value, uint statusCode) { var snapshot = new DataValueSnapshot(value, statusCode, DateTime.UtcNow, DateTime.UtcNow); OnDataChange?.Invoke(this, new DataChangeEventArgs(_subHandle, fullRef, snapshot)); } private sealed class StubHandle : ISubscriptionHandle { /// Gets the diagnostic ID of the subscription. public string DiagnosticId => "stub-sub"; } } /// Minimal for building . private sealed class StubAlarmHandle : IAlarmSubscriptionHandle { /// Gets the diagnostic ID of the alarm subscription. public string DiagnosticId => "stub-alarm-sub"; } }