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.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-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, AddressSpaceRealm.Uns), 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, AddressSpaceRealm.Raw), 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, AddressSpaceRealm.Uns), 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, AddressSpaceRealm.Uns), 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, AddressSpaceRealm.Uns), asker.Ref); var result = asker.ExpectMsg(TimeSpan.FromSeconds(2)); result.Success.ShouldBeFalse(); result.Reason.ShouldNotBeNull(); result.Reason!.ShouldContain("stale"); } /// /// Review H1 regression — a raw RawPath and a UNS path that share a BARE s= id but back /// DIFFERENT driver instances must each route to their OWN driver. Seeds raw tag A/B/C/D on /// drv-1 and a UnsTagReference whose UNS NodeId is ALSO A/B/C/D (Area A / Line B / Equip C /// / effective name D) but which backs a raw tag X/Y/Z/W on drv-2. A bare-only routing key /// would collide (last-writer-wins → wrong device); the (realm, bareId) key keeps them distinct. /// [Fact] public void Colliding_raw_and_uns_bare_ids_route_to_their_own_driver_by_realm() { var db = NewInMemoryDbFactory(); var factory = new PerInstanceRecordingDriverFactory("Modbus"); var deploymentId = SeedCollisionDeployment(db, RevA); var actor = SpawnHostAndApply(db, deploymentId, factory); // Write to the RAW node A/B/C/D → drv-1, forwarded by the raw tag's RawPath A/B/C/D. var asker1 = CreateTestProbe(); actor.Tell(new DriverHostActor.RouteNodeWrite("ns=2;s=A/B/C/D", 111.0, AddressSpaceRealm.Raw), asker1.Ref); asker1.ExpectMsg(Timeout).Success.ShouldBeTrue(); // Write to the UNS node A/B/C/D → drv-2, forwarded by the BACKING raw tag's RawPath X/Y/Z/W. var asker2 = CreateTestProbe(); actor.Tell(new DriverHostActor.RouteNodeWrite("ns=3;s=A/B/C/D", 222.0, AddressSpaceRealm.Uns), asker2.Ref); asker2.ExpectMsg(Timeout).Success.ShouldBeTrue(); AwaitAssert(() => { // drv-1 got ONLY the raw write (A/B/C/D, 111); drv-2 got ONLY the UNS-routed write (X/Y/Z/W, 222). var w1 = factory.WritesFor("drv-1"); var w2 = factory.WritesFor("drv-2"); w1.Count.ShouldBe(1); w1[0].FullReference.ShouldBe("A/B/C/D"); w1[0].Value.ShouldBe(111.0); w2.Count.ShouldBe(1); w2[0].FullReference.ShouldBe("X/Y/Z/W"); // the backing RawPath, NOT the shared bare id w2[0].Value.ShouldBe(222.0); }, duration: Timeout); } /// 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; } /// Seeds the review-H1 collision config: raw tag A/B/C/D on drv-1, and a UnsTagReference /// (Area A / Line B / Equip C, effective name D → UNS NodeId A/B/C/D) backing raw tag X/Y/Z/W /// on drv-2. Both drivers ENABLED (Modbus) so both children spawn. private static DeploymentId SeedCollisionDeployment(IDbContextFactory db, RevisionHash rev) { var artifact = JsonSerializer.SerializeToUtf8Bytes(new { RawFolders = new[] { new { RawFolderId = "rf-a", ParentRawFolderId = (string?)null, Name = "A", ClusterId = "c1" }, new { RawFolderId = "rf-x", ParentRawFolderId = (string?)null, Name = "X", ClusterId = "c1" }, }, DriverInstances = new[] { new { DriverInstanceId = "drv-1", RawFolderId = "rf-a", Name = "B", DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = true }, new { DriverInstanceId = "drv-2", RawFolderId = "rf-x", Name = "Y", DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = true }, }, Devices = new[] { new { DeviceId = "dev-c", DriverInstanceId = "drv-1", Name = "C", DeviceConfig = "{}" }, new { DeviceId = "dev-z", DriverInstanceId = "drv-2", Name = "Z", DeviceConfig = "{}" }, }, TagGroups = Array.Empty(), Tags = new[] { new { TagId = "t-d", DeviceId = "dev-c", TagGroupId = (string?)null, Name = "D", DataType = "Double", AccessLevel = 1, TagConfig = "{}" }, new { TagId = "t-w", DeviceId = "dev-z", TagGroupId = (string?)null, Name = "W", DataType = "Double", AccessLevel = 1, TagConfig = "{}" }, }, UnsAreas = new[] { new { UnsAreaId = "a1", Name = "A", ClusterId = "c1" } }, UnsLines = new[] { new { UnsLineId = "l1", UnsAreaId = "a1", Name = "B" } }, Equipment = new[] { new { EquipmentId = "EQ-1", UnsLineId = "l1", Name = "C", MachineCode = "M1" } }, // Effective name "D" (override) → UNS NodeId A/B/C/D, backing raw tag t-w (X/Y/Z/W) on drv-2. UnsTagReferences = new[] { new { UnsTagReferenceId = "r1", EquipmentId = "EQ-1", TagId = "t-w", DisplayNameOverride = "D" } }, }); 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 SEPARATE per driver instance id, so a test /// with two drivers can assert which instance received a given write (used by the H1 collision test). private sealed class PerInstanceRecordingDriverFactory : IDriverFactory { private readonly string _supportedType; private readonly Dictionary _byId = new(StringComparer.Ordinal); private readonly object _lock = new(); public PerInstanceRecordingDriverFactory(string supportedType) { _supportedType = supportedType; } /// The writes the given driver instance received. public IReadOnlyList WritesFor(string driverInstanceId) { lock (_lock) return _byId.TryGetValue(driverInstanceId, out var d) ? d.Writes : Array.Empty(); } /// public IDriver? TryCreate(string driverType, string driverInstanceId, string driverConfigJson) { if (!string.Equals(driverType, _supportedType, StringComparison.Ordinal)) return null; lock (_lock) { if (!_byId.TryGetValue(driverInstanceId, out var d)) { d = new RecordingDriver(); d.Bind(driverInstanceId, driverType); _byId[driverInstanceId] = d; } return d; } } /// public IReadOnlyCollection SupportedTypes => new[] { _supportedType }; } /// 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()); } } }