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