using System.Collections.Concurrent; using System.Text.Json; using Akka.Actor; using Akka.TestKit; using Microsoft.EntityFrameworkCore; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.Engines; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Commons.Types; using ZB.MOM.WW.OtOpcUa.Configuration; using ZB.MOM.WW.OtOpcUa.Configuration.Entities; using ZB.MOM.WW.OtOpcUa.Configuration.Enums; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; /// /// v3 Batch 4 (B4-WP4) — the inbound native-condition acknowledge routing wired into /// re-expressed for the v3 raw-condition model. An OPC UA client Acknowledges /// a NATIVE condition (materialised at the raw tag, ConditionId = its RawPath); the node manager invokes /// NativeAlarmAckRouter and the host receives a . /// The host resolves the condition NodeId (== the RawPath) → owning (DriverInstanceId, RawPath) via /// the _driverRefByAlarmNodeId inverse map (built alongside the alarm forward map in /// PushDesiredSubscriptions from the alarm-bearing composition.RawTags), applies the SAME /// primary gate the inbound write path uses, and routes to the owning driver child's /// carrying the principal. /// /// /// Mirrors DriverHostActorLiveValueTests: a real apply through the harness spawns a real /// (non-stubbed) child backed by a recording /// driver (DriverType "GalaxyMxGateway", Enabled), so the inverse map is populated authentically and the /// forwarded acknowledge request can be observed. The seeded raw tag carries an alarm object so it /// materialises as a Part 9 condition at its RawPath, not a value variable. /// /// public sealed class DriverHostActorNativeAlarmAckRoutingTests : RuntimeActorTestBase { private static readonly NodeId TestNode = NodeId.Parse("driver-ack-test"); private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); private static readonly TimeSpan Timeout = PresenceBudget; // The v3 RawPath for the seeded alarm tag: RawFolder "Plant" / DriverName "gw" / Device "dev1" / Tag. private const string AlarmRawPath = "Plant/gw/dev1/temp_hi"; /// On the PRIMARY (role unknown ⇒ Primary), a RouteNativeAlarmAck for a mapped raw condition NodeId /// (== the RawPath) forwards exactly one to the owning driver's /// , correlated on the RawPath, with the operator principal + the /// comment. [Fact] public void RouteNativeAlarmAck_routes_to_driver_AcknowledgeAsync_with_principal() { var db = NewInMemoryDbFactory(); var recorder = new RecordingAlarmDriverFactory("GalaxyMxGateway"); var deploymentId = SeedV3AlarmDeployment(db, RevA, Driver: "drv-1", Tag: "temp_hi"); var actor = SpawnHostAndApply(db, deploymentId, recorder); // Local role unknown ⇒ treated as Primary ⇒ ack allowed (default-allow semantics). actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "cmt", "alice", ApiKeyId: "0a1b2c")); // The driver received exactly one acknowledge, correlated on its wire-ref RawPath, with principal // and the session's API key id (lmxopcua#749 — the Galaxy driver resolves the key's ack identity). AwaitAssert(() => { recorder.Acks.Count.ShouldBe(1); recorder.Acks[0].ConditionId.ShouldBe(AlarmRawPath); recorder.Acks[0].SourceNodeId.ShouldBe(AlarmRawPath); recorder.Acks[0].Comment.ShouldBe("cmt"); recorder.Acks[0].OperatorUser.ShouldBe("alice"); recorder.Acks[0].ApiKeyId.ShouldBe("0a1b2c"); }, duration: Timeout, cancellationToken: TestContext.Current.CancellationToken); } /// An ack for an unmapped condition NodeId is dropped: no throw, and the driver's /// is never called. [Fact] public void RouteNativeAlarmAck_unknown_node_is_dropped() { var db = NewInMemoryDbFactory(); var recorder = new RecordingAlarmDriverFactory("GalaxyMxGateway"); var deploymentId = SeedV3AlarmDeployment(db, RevA, Driver: "drv-1", Tag: "temp_hi"); var actor = SpawnHostAndApply(db, deploymentId, recorder); actor.Tell(new DriverHostActor.RouteNativeAlarmAck("Plant/gw/dev1/does-not-exist", "cmt", "alice")); // POSITIVE CONTROL, not a sleep (#501). `Acks` is empty at t=0, so any assertion made before the // host has processed the message above passes without testing anything — and AwaitAssert returns on // its FIRST success, so it spends none of its duration and is the wrong tool for an absence. // // Instead send a MAPPED ack afterwards. The host processes its mailbox in order and forwards to the // child via Tell, and the child handles RouteAlarmAck with ReceiveAsync (which suspends its own // mailbox), so by the time this one reaches the driver the unmapped one has provably already been // handled. Whatever `Acks` holds then is the complete answer. actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "control", "alice")); AwaitAssert(() => recorder.Acks.ShouldContain(a => a.Comment == "control"), duration: Timeout, cancellationToken: TestContext.Current.CancellationToken); recorder.Acks.Count.ShouldBe(1, "only the control ack may reach the driver — the unmapped condition NodeId must have been dropped"); } /// On a SECONDARY node the ack is gated off (same primary gate as the inbound write path): the /// driver's is NOT called. [Fact] public void RouteNativeAlarmAck_on_non_primary_is_dropped() { var db = NewInMemoryDbFactory(); var recorder = new RecordingAlarmDriverFactory("GalaxyMxGateway"); var deploymentId = SeedV3AlarmDeployment(db, RevA, Driver: "drv-1", Tag: "temp_hi"); var actor = SpawnHostAndApply(db, deploymentId, recorder); // Force this node Secondary so the primary gate rejects the ack. actor.Tell(new RedundancyStateChanged( new[] { new NodeRedundancyState(TestNode, RedundancyRole.Secondary, IsClusterLeader: false, IsDriverPrimary: false, AsOfUtc: DateTime.UtcNow), }, CorrelationId.NewId())); actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "gated", "alice")); // POSITIVE CONTROL (#501) — same reasoning as the unmapped-node test, with the role itself as the // control: return the node to PRIMARY and re-send the SAME mapped ack. Mailbox ordering means that // once this one reaches the driver, the Secondary-gated one has already been processed. The // discriminator is the comment, so a leaked ack is visible rather than merged into the count. actor.Tell(new RedundancyStateChanged( new[] { new NodeRedundancyState(TestNode, RedundancyRole.Primary, IsClusterLeader: true, IsDriverPrimary: true, AsOfUtc: DateTime.UtcNow), }, CorrelationId.NewId())); actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "control", "alice")); AwaitAssert(() => recorder.Acks.ShouldContain(a => a.Comment == "control"), duration: Timeout, cancellationToken: TestContext.Current.CancellationToken); recorder.Acks.ShouldNotContain(a => a.Comment == "gated", "the Primary gate must have dropped the ack sent while this node was Secondary"); recorder.Acks.Count.ShouldBe(1); } /// /// lmxopcua#760 — the driver REFUSES the acknowledge (no Galaxy ack credential): the host reverts the /// condition the node manager optimistically acknowledged, re-projecting it Active + UNacknowledged with /// its message and severity intact, so the address space stops claiming an ack Galaxy never received. /// [Fact] public void A_refused_ack_reverts_the_condition_to_unacknowledged() { var db = NewInMemoryDbFactory(); var recorder = new RecordingAlarmDriverFactory("GalaxyMxGateway") { AckFault = new DriverOperationRefusedException("no Galaxy ack credential"), }; var deploymentId = SeedV3AlarmDeployment(db, RevA, Driver: "drv-1", Tag: "temp_hi"); var publish = this.CreateCleanOutcomePublishProbe(); var actor = SpawnHostAndApply(db, deploymentId, recorder, publish.Ref); RaiseAndExpectUnacked(actor, publish); actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "cmt", "alice")); var reverted = NextAlarmUpdate(publish); reverted.AlarmNodeId.ShouldBe(AlarmRawPath); reverted.Realm.ShouldBe(AddressSpaceRealm.Raw); reverted.State.Active.ShouldBeTrue(); reverted.State.Acknowledged.ShouldBeFalse(); reverted.State.Message.ShouldBe("temperature high"); reverted.State.Severity.ShouldBe((ushort)700); recorder.Acks.Count.ShouldBe(1, "the ack did reach the driver, which refused it"); } /// /// lmxopcua#760 — any backend failure (here: the gateway rejecting the ack) reverts the same way, not just /// a deliberate refusal. /// [Fact] public void A_failed_ack_reverts_the_condition_to_unacknowledged() { var db = NewInMemoryDbFactory(); var recorder = new RecordingAlarmDriverFactory("GalaxyMxGateway") { AckFault = new InvalidOperationException("gateway hresult 0x80070005"), }; var deploymentId = SeedV3AlarmDeployment(db, RevA, Driver: "drv-1", Tag: "temp_hi"); var publish = this.CreateCleanOutcomePublishProbe(); var actor = SpawnHostAndApply(db, deploymentId, recorder, publish.Ref); RaiseAndExpectUnacked(actor, publish); actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "cmt", "alice")); NextAlarmUpdate(publish).State.Acknowledged.ShouldBeFalse(); } /// /// lmxopcua#760 — a SUCCESSFUL ack reverts nothing. Positive control: the next driver Clear projects /// inactive + ACKED, which it only can when the local ack stayed in the projector (a revert would have /// left it unacked) — and it must be the very next condition update, so no revert slipped in between. /// [Fact] public void A_successful_ack_is_not_reverted() { var db = NewInMemoryDbFactory(); var recorder = new RecordingAlarmDriverFactory("GalaxyMxGateway"); var deploymentId = SeedV3AlarmDeployment(db, RevA, Driver: "drv-1", Tag: "temp_hi"); var publish = this.CreateCleanOutcomePublishProbe(); var actor = SpawnHostAndApply(db, deploymentId, recorder, publish.Ref); RaiseAndExpectUnacked(actor, publish); actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "cmt", "alice")); AwaitAssert(() => recorder.Acks.Count.ShouldBe(1), duration: Timeout, cancellationToken: TestContext.Current.CancellationToken); actor.Tell(new DriverInstanceActor.AttributeAlarmPublished("drv-1", Evt(AlarmTransitionKind.Clear))); var next = NextAlarmUpdate(publish); next.State.Active.ShouldBeFalse(); next.State.Acknowledged.ShouldBeTrue("the successful ack must have survived — no revert"); } /// /// lmxopcua#760 — the host publishes the condition → owning-driver pre-check snapshot the OPC UA node /// manager uses to refuse an acknowledge synchronously: the driver's own verdict for the condition it /// owns, nothing for an unknown condition. /// [Fact] public void The_ack_preflight_view_answers_with_the_owning_drivers_verdict() { var db = NewInMemoryDbFactory(); var recorder = new RecordingAlarmDriverFactory("GalaxyMxGateway") { PreflightRefusal = key => key is null ? "anonymous sessions cannot acknowledge" : null, }; var deploymentId = SeedV3AlarmDeployment(db, RevA, Driver: "drv-1", Tag: "temp_hi"); var view = new NativeAlarmAckPreflightView(); SpawnHostAndApply(db, deploymentId, recorder, ackPreflightView: view); AwaitAssert(() => view.Check(AlarmRawPath, apiKeyId: null).ShouldBe("anonymous sessions cannot acknowledge"), duration: Timeout, cancellationToken: TestContext.Current.CancellationToken); view.Check(AlarmRawPath, apiKeyId: "0a1b2c").ShouldBeNull(); view.Check("Plant/gw/dev1/does-not-exist", apiKeyId: null).ShouldBeNull(); } /// /// lmxopcua#765 — an ack routed to a driver that is not running (disabled here) reaches no alarm system, so /// the host reverts the condition the node manager optimistically acknowledged, as for a refused ack. /// [Fact] public void An_ack_to_a_driver_that_is_not_running_reverts_the_condition() { var db = NewInMemoryDbFactory(); var recorder = new RecordingAlarmDriverFactory("GalaxyMxGateway"); var deploymentId = SeedV3AlarmDeployment(db, RevA, Driver: "drv-1", Tag: "temp_hi", enabled: false); var publish = this.CreateCleanOutcomePublishProbe(); var actor = SpawnHostAndApply(db, deploymentId, recorder, publish.Ref, awaitConnected: false); RaiseAndExpectUnacked(actor, publish); actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "cmt", "alice")); var reverted = NextAlarmUpdate(publish); reverted.AlarmNodeId.ShouldBe(AlarmRawPath); reverted.State.Active.ShouldBeTrue(); reverted.State.Acknowledged.ShouldBeFalse(); recorder.Acks.ShouldBeEmpty("no driver was running to take the ack"); } /// /// lmxopcua#765 — likewise a STUBBED child (here: no factory for the driver type, so the host falls back to /// a stub that has no alarm backend) silently swallows the ack; the host reverts it rather than routing it. /// [Fact] public void An_ack_to_a_stubbed_driver_reverts_the_condition() { var db = NewInMemoryDbFactory(); var noFactoryForType = new RecordingAlarmDriverFactory("SomeOtherType"); var deploymentId = SeedV3AlarmDeployment(db, RevA, Driver: "drv-1", Tag: "temp_hi"); var publish = this.CreateCleanOutcomePublishProbe(); var actor = SpawnHostAndApply(db, deploymentId, noFactoryForType, publish.Ref, awaitConnected: false); RaiseAndExpectUnacked(actor, publish); actor.Tell(new DriverHostActor.RouteNativeAlarmAck(AlarmRawPath, "cmt", "alice")); NextAlarmUpdate(publish).State.Acknowledged.ShouldBeFalse(); } /// /// lmxopcua#765 — a host that came up Stale (config DB unreachable) drops every native ack before the /// projector records it, so it could not be reverted. The pre-check view refuses every native ack while the /// host is Stale, so the node manager answers synchronously instead of applying an ack nothing will route. /// [Fact] public void A_stale_host_refuses_every_native_ack_through_the_preflight_view() { var view = new NativeAlarmAckPreflightView(); view.Check(AlarmRawPath, apiKeyId: "0a1b2c").ShouldBeNull("positive control: nothing refuses before the host starts"); var host = Sys.ActorOf(DriverHostActor.Props( new ThrowingDbFactory(), TestNode, CreateTestProbe().Ref, localRoles: new HashSet { "driver" }, ackPreflightView: view)); // Barrier: Stale answers RouteNodeWrite synchronously, so a reply proves the host is in Stale. var asker = CreateTestProbe(); host.Tell(new DriverHostActor.RouteNodeWrite("eq-x/speed", 0.0, AddressSpaceRealm.Uns), asker.Ref); asker.ExpectMsg(Timeout, cancellationToken: TestContext.Current.CancellationToken).Reason!.ShouldContain("stale"); view.Check(AlarmRawPath, apiKeyId: "0a1b2c").ShouldNotBeNull().ShouldContain("stale"); view.Check("any/other/condition", apiKeyId: null).ShouldNotBeNull(); } private sealed class ThrowingDbFactory : IDbContextFactory { public OtOpcUaConfigDbContext CreateDbContext() => throw new InvalidOperationException("db unreachable"); } private static AlarmEventArgs Evt(AlarmTransitionKind kind) => new( new StubAlarmHandleForTests(), SourceNodeId: "Temp", ConditionId: AlarmRawPath, AlarmType: "OffNormalAlarm", Message: "temperature high", Severity: AlarmSeverity.High, SourceTimestampUtc: DateTime.UtcNow, Kind: kind); private void RaiseAndExpectUnacked(IActorRef actor, TestProbe publish) { actor.Tell(new DriverInstanceActor.AttributeAlarmPublished("drv-1", Evt(AlarmTransitionKind.Raise))); var raised = NextAlarmUpdate(publish); raised.State.Active.ShouldBeTrue(); raised.State.Acknowledged.ShouldBeFalse(); } private static OpcUaPublishActor.AlarmStateUpdate NextAlarmUpdate(TestProbe publish) => publish.FishForMessage(_ => true, Timeout); private sealed class StubAlarmHandleForTests : IAlarmSubscriptionHandle { public string DiagnosticId => "test-alarm-sub"; } /// Spawns the host with the recording alarm-driver factory, dispatches the deployment, and waits /// for the Applied ACK so the apply (and thus the inverse-map build in PushDesiredSubscriptions) has /// completed before the test routes an ack. private IActorRef SpawnHostAndApply( IDbContextFactory db, DeploymentId deploymentId, IDriverFactory factory, IActorRef? opcUaPublishActor = null, NativeAlarmAckPreflightView? ackPreflightView = null, bool awaitConnected = true) { var coordinator = CreateTestProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( db, TestNode, coordinator.Ref, driverFactory: factory, localRoles: new HashSet { "driver" }, opcUaPublishActor: opcUaPublishActor, ackPreflightView: ackPreflightView)); actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); // Applied does NOT mean the driver child is Connected: the apply acks once the child is spawned, // and a RouteAlarmAck reaching a still-Connecting child is dropped by design (fire-and-forget, no // upstream to reach). Each test sends its positive-control ack exactly once, so on a loaded runner // the control itself was dropped and the wait read [] — 4 of 20 runs under thread-pool starvation, // and on Gitea (#606). SubscribeAlarmsAsync is only ever called from the Connected state (a // pre-connect request is swallowed and re-issued on the Connected entry), so wait for it. if (awaitConnected && factory is RecordingAlarmDriverFactory recording) AwaitCondition(() => recording.AlarmsSubscribed, Timeout); return actor; } /// /// Seeds a Sealed deployment whose artifact carries the v3 raw-tag chain (RawFolder "Plant" → /// DriverInstance(RawFolderId, Name "gw", DriverType "GalaxyMxGateway", Enabled) → Device "dev1" → Tag) /// with the tag's TagConfig carrying an alarm object so the composer projects a non-null /// EquipmentTagAlarmInfo — making the raw tag a Part 9 condition at its RawPath. The non-Windows /// DriverType + Enabled flag spawn a REAL child. /// private static DeploymentId SeedV3AlarmDeployment( IDbContextFactory db, RevisionHash rev, string Driver, string Tag, bool enabled = true) { var artifact = JsonSerializer.SerializeToUtf8Bytes(new { RawFolders = new[] { new { RawFolderId = "rf-plant", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } }, DriverInstances = new[] { new { DriverInstanceRowId = Guid.NewGuid(), DriverInstanceId = Driver, RawFolderId = "rf-plant", Name = "gw", DriverType = "GalaxyMxGateway", // not Windows-only ⇒ a real child is spawned (not stubbed) Enabled = enabled, DriverConfig = "{}", ClusterId = "c1", }, }, Devices = new[] { new { DeviceId = $"{Driver}:dev1", DriverInstanceId = Driver, Name = "dev1", DeviceConfig = "{}" }, }, TagGroups = Array.Empty(), Tags = new[] { new { TagId = "tag-0", DeviceId = $"{Driver}:dev1", TagGroupId = (string?)null, Name = Tag, DataType = "Boolean", AccessLevel = 0, // TagAccessLevel.Read TagConfig = JsonSerializer.Serialize(new { alarm = new { alarmType = "OffNormalAlarm", severity = 700 } }), }, }, }); var id = DeploymentId.NewId(); using var ctx = db.CreateDbContext(); ctx.Deployments.Add(new Deployment { DeploymentId = id.Value, RevisionHash = rev.Value, Status = DeploymentStatus.Sealed, CreatedBy = "test", SealedAtUtc = DateTime.UtcNow, ArtifactBlob = artifact, }); ctx.SaveChanges(); return id; } /// Factory producing a single for the supported type, whose /// recorded acknowledge list is exposed for assertions. private sealed class RecordingAlarmDriverFactory : IDriverFactory { private readonly string _supportedType; private readonly RecordingAlarmDriver _driver = new(); public RecordingAlarmDriverFactory(string supportedType) { _supportedType = supportedType; } /// The acknowledge requests the spawned driver received (thread-safe — AcknowledgeAsync runs /// off the actor thread). public IReadOnlyList Acks => _driver.Acks; /// True once the driver child has reached Connected and opened its alarm subscription. public bool AlarmsSubscribed => _driver.AlarmsSubscribed; /// When set, every acknowledge is recorded and then fails with this exception. public Exception? AckFault { init => _driver.AckFault = value; } /// The driver's synchronous pre-check verdict (lmxopcua#760); null ⇒ always proceed. public Func? PreflightRefusal { init => _driver.PreflightRefusal = value; } /// public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson) { if (!string.Equals(driverType, _supportedType, StringComparison.Ordinal)) return null; _driver.Bind(driverInstanceId, driverType); return _driver; } /// public IReadOnlyCollection SupportedTypes => new[] { _supportedType }; } /// An + that records every acknowledge. private sealed class RecordingAlarmDriver : IDriver, IAlarmSource, IAlarmAcknowledgePreflight { /// When set, records the request and then throws this. public Exception? AckFault { get; set; } /// The verdict; null ⇒ always proceed. public Func? PreflightRefusal { get; set; } /// public string? GetAcknowledgeRefusal(string? apiKeyId) => PreflightRefusal?.Invoke(apiKeyId); private readonly ConcurrentQueue _acks = new(); /// public string DriverInstanceId { get; private set; } = string.Empty; /// public string DriverType { get; private set; } = string.Empty; /// The acknowledge requests received so far. public IReadOnlyList Acks => _acks.ToArray(); /// Sets the identity once the factory is asked to create it. public void Bind(string id, string type) { DriverInstanceId = id; DriverType = type; } /// public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; /// public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; /// public Task ShutdownAsync(CancellationToken cancellationToken) => Task.CompletedTask; /// public DriverHealth GetHealth() => new(DriverState.Healthy, DateTime.UtcNow, LastError: null); /// public Task SubscribeAlarmsAsync( IReadOnlyList sourceNodeIds, CancellationToken cancellationToken) { Volatile.Write(ref _alarmsSubscribed, true); return Task.FromResult(new StubAlarmHandle()); } private bool _alarmsSubscribed; /// True once has been called (only ever from Connected). public bool AlarmsSubscribed => Volatile.Read(ref _alarmsSubscribed); /// public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask; /// public Task AcknowledgeAsync( IReadOnlyList acknowledgements, CancellationToken cancellationToken) { foreach (var a in acknowledgements) _acks.Enqueue(a); return AckFault is { } fault ? Task.FromException(fault) : Task.CompletedTask; } /// public event EventHandler? OnAlarmEvent { add { } remove { } } private sealed class StubAlarmHandle : IAlarmSubscriptionHandle { public string DiagnosticId => "stub-alarm-sub"; } } }