using System.Collections.Concurrent; using System.Text.Json; using Akka.Actor; 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.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-WP3) — inbound operator-write routing through EITHER NodeId. A /// carries the FULL ns-qualified NodeId string (the node /// manager's write hook passes node.NodeId.ToString()); the host normalises it to the bare id and /// resolves the NodeId → (DriverInstanceId, RawPath) reverse map — populated in /// PushDesiredSubscriptions for BOTH the raw NodeId AND every referencing UNS NodeId — gates on the /// driver PRIMARY, and forwards a carrying the tag's /// RawPath to the driver child. So a write to the raw node and a write to the referencing UNS node /// both reach the SAME driver point (they share the driver ref). /// /// Drives a real apply (Enabled Modbus driver ⇒ a real child backed /// by a recording driver), then routes writes and asserts the forwarded wire-ref + the primary gate. /// /// public sealed class DriverHostActorWriteRoutingTests : RuntimeActorTestBase { private static readonly NodeId TestNode = NodeId.Parse("driver-wr-test"); private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5); // The single seeded raw tag + its UNS reference. private const string RawPath = "Plant/Modbus/dev1/speed"; private const string UnsNodeId = "filling/line1/station1/speed"; /// On the PRIMARY, a RouteNodeWrite for the ns-qualified UNS NodeId resolves to the raw tag's /// driver ref and forwards the write keyed by the tag's RawPath (dual-NodeId write routing). [Fact] public void Primary_routes_write_via_uns_nodeid_to_driver_by_rawpath() { var db = NewInMemoryDbFactory(); var recorder = new RecordingDriverFactory("Modbus"); var deploymentId = SeedV3Deployment(db, RevA); var actor = SpawnHostAndApply(db, deploymentId, recorder); var asker = CreateTestProbe(); // The node manager passes the FULL ns-qualified id; the host must normalise + resolve it. actor.Tell(new DriverHostActor.RouteNodeWrite($"ns=3;s={UnsNodeId}", 123.0), asker.Ref); asker.ExpectMsg(Timeout).Success.ShouldBeTrue(); AwaitAssert(() => { recorder.Writes.Count.ShouldBe(1); recorder.Writes[0].FullReference.ShouldBe(RawPath); // the driver writes by RawPath, not the UNS id recorder.Writes[0].Value.ShouldBe(123.0); }, duration: Timeout); } /// On the PRIMARY, a RouteNodeWrite for the ns-qualified RAW NodeId resolves to the same driver /// ref and forwards the write keyed by the RawPath. [Fact] public void Primary_routes_write_via_raw_nodeid_to_driver_by_rawpath() { var db = NewInMemoryDbFactory(); var recorder = new RecordingDriverFactory("Modbus"); var deploymentId = SeedV3Deployment(db, RevA); var actor = SpawnHostAndApply(db, deploymentId, recorder); var asker = CreateTestProbe(); actor.Tell(new DriverHostActor.RouteNodeWrite($"ns=2;s={RawPath}", 456.0), asker.Ref); asker.ExpectMsg(Timeout).Success.ShouldBeTrue(); AwaitAssert(() => { recorder.Writes.Count.ShouldBe(1); recorder.Writes[0].FullReference.ShouldBe(RawPath); recorder.Writes[0].Value.ShouldBe(456.0); }, duration: Timeout); } /// On a SECONDARY, RouteNodeWrite (via the UNS NodeId) replies "not primary" and the driver /// receives NO write — the primary gate fires before the reverse-map lookup. [Fact] public void Secondary_rejects_write_and_does_not_forward_to_driver() { var db = NewInMemoryDbFactory(); var recorder = new RecordingDriverFactory("Modbus"); var deploymentId = SeedV3Deployment(db, RevA); var actor = SpawnHostAndApply(db, deploymentId, recorder); actor.Tell(new RedundancyStateChanged( new[] { new NodeRedundancyState(TestNode, RedundancyRole.Secondary, IsClusterLeader: false, IsRoleLeaderForDriver: false, AsOfUtc: DateTime.UtcNow), }, CorrelationId.NewId())); var asker = CreateTestProbe(); actor.Tell(new DriverHostActor.RouteNodeWrite($"ns=3;s={UnsNodeId}", 123.0), asker.Ref); var result = asker.ExpectMsg(Timeout); result.Success.ShouldBeFalse(); result.Reason.ShouldBe("not primary"); recorder.Writes.ShouldBeEmpty(); } /// An unknown NodeId (no reverse-map entry) replies failure and writes nothing. [Fact] public void Unknown_node_id_replies_failure() { var db = NewInMemoryDbFactory(); var recorder = new RecordingDriverFactory("Modbus"); var deploymentId = SeedV3Deployment(db, RevA); var actor = SpawnHostAndApply(db, deploymentId, recorder); var asker = CreateTestProbe(); actor.Tell(new DriverHostActor.RouteNodeWrite("ns=3;s=filling/line1/station1/does-not-exist", 1.0), asker.Ref); var result = asker.ExpectMsg(Timeout); result.Success.ShouldBeFalse(); result.Reason.ShouldNotBeNull(); recorder.Writes.ShouldBeEmpty(); } /// A RouteNodeWrite arriving while the host is Stale (config DB unreachable) fast-fails with an /// immediate negative reply mentioning "stale". [Fact] public void Stale_host_fast_fails_route_node_write() { var db = new ThrowingDbFactory(); var coordinator = CreateTestProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" })); var asker = CreateTestProbe(); actor.Tell(new DriverHostActor.RouteNodeWrite($"ns=3;s={UnsNodeId}", 123.0), asker.Ref); var result = asker.ExpectMsg(TimeSpan.FromSeconds(2)); result.Success.ShouldBeFalse(); result.Reason.ShouldNotBeNull(); result.Reason!.ShouldContain("stale"); } /// Spawns the host with the recording driver factory, dispatches the deployment, and waits for /// the Applied ACK so the reverse-map build in PushDesiredSubscriptions has completed before routing a /// write. No OPC UA / mux probes are wired — the write path doesn't depend on the publish actor. private IActorRef SpawnHostAndApply( IDbContextFactory db, DeploymentId deploymentId, IDriverFactory factory) { var coordinator = CreateTestProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( db, TestNode, coordinator.Ref, driverFactory: factory, localRoles: new HashSet { "driver" })); actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); coordinator.ExpectMsg(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied); return actor; } /// Seeds a Sealed v3 deployment: RawFolder "Plant" → DriverInstance "drv-1" (Modbus, ENABLED so a /// real child spawns) → Device "dev1" → Tag "speed" (RawPath Plant/Modbus/dev1/speed, ReadWrite), /// projected into equipment filling/line1/station1 by a UnsTagReference (UNS NodeId /// filling/line1/station1/speed). private static DeploymentId SeedV3Deployment(IDbContextFactory db, RevisionHash rev) { var artifact = JsonSerializer.SerializeToUtf8Bytes(new { RawFolders = new[] { new { RawFolderId = "rf-plant", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } }, DriverInstances = new[] { new { DriverInstanceId = "drv-1", RawFolderId = "rf-plant", Name = "Modbus", DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = true }, }, Devices = new[] { new { DeviceId = "dev-1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" } }, TagGroups = Array.Empty(), Tags = new[] { new { TagId = "t-speed", DeviceId = "dev-1", TagGroupId = (string?)null, Name = "speed", DataType = "Double", AccessLevel = 1, TagConfig = "{}" }, }, UnsAreas = new[] { new { UnsAreaId = "a1", Name = "filling", ClusterId = "c1" } }, UnsLines = new[] { new { UnsLineId = "l1", UnsAreaId = "a1", Name = "line1" } }, Equipment = new[] { new { EquipmentId = "EQ-1", UnsLineId = "l1", Name = "station1", MachineCode = "M1" } }, UnsTagReferences = new[] { new { UnsTagReferenceId = "r1", EquipmentId = "EQ-1", TagId = "t-speed", DisplayNameOverride = (string?)null } }, }); 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; } /// An whose CreateDbContext always throws — drives the host /// into the Stale path. private sealed class ThrowingDbFactory : IDbContextFactory { /// public OtOpcUaConfigDbContext CreateDbContext() => throw new InvalidOperationException("config DB unreachable (test stub)"); } /// Factory producing a single for the supported type. private sealed class RecordingDriverFactory : IDriverFactory { private readonly string _supportedType; private readonly RecordingDriver _driver = new(); public RecordingDriverFactory(string supportedType) { _supportedType = supportedType; } /// The writes the spawned driver received. public IReadOnlyList Writes => _driver.Writes; /// 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 write and returns Good. private sealed class RecordingDriver : IDriver, IWritable { private readonly ConcurrentQueue _writes = new(); /// public string DriverInstanceId { get; private set; } = string.Empty; /// public string DriverType { get; private set; } = string.Empty; /// The writes received so far. public IReadOnlyList Writes => _writes.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 long GetMemoryFootprint() => 0; /// public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask; /// public Task> WriteAsync( IReadOnlyList writes, CancellationToken cancellationToken) { foreach (var w in writes) _writes.Enqueue(w); return Task.FromResult>( writes.Select(_ => new WriteResult(0u)).ToArray()); } } }