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.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.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) — the driver single-source fan-out wired into /// . The driver publishes a value keyed by its wire-ref (== the tag's /// RawPath); the host's _nodeIdByDriverRef map — rebuilt each apply from the composition's /// RawTagsUnsReferenceVariables — resolves (DriverInstanceId, RawPath) to the RAW /// NodeId AND every referencing UNS NodeId, so ForwardToMux lands ONE publish on the raw node /// () plus every UNS node () with /// IDENTICAL value/quality/timestamp (no independent buffer ⇒ no drift). The raw value is still forwarded /// to the dependency mux (VirtualTag inputs), keyed by the RawPath. /// /// Drives a real apply through the existing harness: the seeded artifact carries the v3 raw-tag chain /// (RawFolder → DriverInstance(RawFolderId) → Device → Tag) + an optional UnsTagReference projecting /// each raw tag into an equipment. DispatchDeployment triggers the /// ApplyAndAck → PushDesiredSubscriptions pass that builds the map; the sink and mux are /// injected as s. /// /// public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase { private static readonly NodeId TestNode = NodeId.Parse("driver-lv-test"); private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); private static readonly DateTime Ts = new(2026, 6, 13, 10, 0, 0, DateTimeKind.Utc); /// A driver value published by RawPath fans to the RAW node AND the referencing UNS node, each /// carrying identical value/quality/timestamp — the fan-out drift guard. [Fact] public void Driver_value_fans_out_to_raw_and_uns_NodeIds_with_identical_value() { var db = NewInMemoryDbFactory(); var deploymentId = SeedV3Deployment(db, RevA, rawTags: new[] { (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false) }, unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) }); var (actor, publish, mux) = SpawnHostAndApply(db, deploymentId); // The driver publishes by its wire-ref RawPath. actor.Tell(new DriverInstanceActor.AttributeValuePublished( "drv-1", "Plant/Modbus/dev1/speed", 42.0, OpcUaQuality.Good, Ts)); var a = publish.ExpectMsg(TimeSpan.FromSeconds(5)); var b = publish.ExpectMsg(TimeSpan.FromSeconds(5)); var byId = new[] { a, b }.ToDictionary(u => u.NodeId); byId.Keys.ShouldBe(new[] { "Plant/Modbus/dev1/speed", "filling/line1/station1/speed" }, ignoreOrder: true); // Realm travels with each update so the sink resolves the right namespace. byId["Plant/Modbus/dev1/speed"].Realm.ShouldBe(AddressSpaceRealm.Raw); byId["filling/line1/station1/speed"].Realm.ShouldBe(AddressSpaceRealm.Uns); // Fan-out drift guard: identical value + quality + timestamp on BOTH node states. foreach (var u in byId.Values) { u.Value.ShouldBe(42.0); u.Quality.ShouldBe(OpcUaQuality.Good); u.TimestampUtc.ShouldBe(Ts); } // The raw publish still reaches the dependency mux (VirtualTag inputs), keyed by RawPath. mux.ExpectMsg(TimeSpan.FromSeconds(5)) .FullReference.ShouldBe("Plant/Modbus/dev1/speed"); } /// A raw tag with NO UNS reference fans to ONLY its raw NodeId (Raw realm). [Fact] public void Unreferenced_raw_tag_fans_to_raw_node_only() { var db = NewInMemoryDbFactory(); var deploymentId = SeedV3Deployment(db, RevA, rawTags: new[] { (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false) }, unsRefs: Array.Empty<(string, string, string, string, string?)>()); var (actor, publish, _) = SpawnHostAndApply(db, deploymentId); actor.Tell(new DriverInstanceActor.AttributeValuePublished( "drv-1", "Plant/Modbus/dev1/speed", 5.0, OpcUaQuality.Good, Ts)); var only = publish.ExpectMsg(TimeSpan.FromSeconds(5)); only.NodeId.ShouldBe("Plant/Modbus/dev1/speed"); only.Realm.ShouldBe(AddressSpaceRealm.Raw); publish.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); } /// A raw tag referenced by TWO equipments fans to the raw node PLUS both UNS nodes (3 updates, /// identical value) — the 1:N fan-out. [Fact] public void Raw_tag_referenced_by_two_equipments_fans_to_raw_plus_both_uns() { var db = NewInMemoryDbFactory(); var deploymentId = SeedV3Deployment(db, RevA, rawTags: new[] { (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false) }, unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null), (Area: "filling", Line: "line1", Equip: "station2", BackingTag: "speed", Effective: (string?)null), }); var (actor, publish, _) = SpawnHostAndApply(db, deploymentId); actor.Tell(new DriverInstanceActor.AttributeValuePublished( "drv-1", "Plant/Modbus/dev1/speed", 7.0, OpcUaQuality.Good, Ts)); var updates = new[] { publish.ExpectMsg(TimeSpan.FromSeconds(5)), publish.ExpectMsg(TimeSpan.FromSeconds(5)), publish.ExpectMsg(TimeSpan.FromSeconds(5)), }; updates.Select(u => u.NodeId).ShouldBe(new[] { "Plant/Modbus/dev1/speed", "filling/line1/station1/speed", "filling/line1/station2/speed", }, ignoreOrder: true); updates.ShouldAllBe(u => (double)u.Value! == 7.0); } /// A value for a ref not in the composition is NOT pushed to the OPC UA sink (no matching /// NodeId), but is still forwarded to the dependency mux. [Fact] public void Unmatched_ref_is_dropped_from_sink_but_still_forwarded_to_mux() { var db = NewInMemoryDbFactory(); var deploymentId = SeedV3Deployment(db, RevA, rawTags: new[] { (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false) }, unsRefs: Array.Empty<(string, string, string, string, string?)>()); var (actor, publish, mux) = SpawnHostAndApply(db, deploymentId); actor.Tell(new DriverInstanceActor.AttributeValuePublished( "drv-1", "Plant/Modbus/dev1/nope", 99.0, OpcUaQuality.Good, Ts)); publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); mux.ExpectMsg(TimeSpan.FromSeconds(5)) .FullReference.ShouldBe("Plant/Modbus/dev1/nope"); } /// Spawns the host with publish + mux probes, dispatches the deployment, and waits for the /// Applied ACK so the map build in PushDesiredSubscriptions has completed before the test publishes a /// value. A VirtualTag-host probe is injected so the real host isn't spawned. private (IActorRef Actor, Akka.TestKit.TestProbe Publish, Akka.TestKit.TestProbe Mux) SpawnHostAndApply( IDbContextFactory db, DeploymentId deploymentId) { var coordinator = CreateTestProbe(); var publish = CreateTestProbe(); var mux = CreateTestProbe(); var vtHost = CreateTestProbe(); var actor = Sys.ActorOf(DriverHostActor.Props( db, TestNode, coordinator.Ref, localRoles: new HashSet { "driver" }, dependencyMux: mux.Ref, opcUaPublishActor: publish.Ref, virtualTagEvaluator: NullVirtualTagEvaluator.Instance, virtualTagHostOverride: vtHost.Ref)); actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); publish.ExpectMsg(TimeSpan.FromSeconds(5)); return (actor, publish, mux); } /// Seeds a Sealed deployment whose artifact carries the v3 raw-tag chain (RawFolder "Plant" → /// DriverInstance(RawFolderId) → Device → Tag) plus optional UnsTagReferences projecting a raw tag into an /// Area/Line/Equipment. Each raw tag's RawPath is Plant/{DriverName}/{Device}/{Tag}; each UNS /// reference's NodeId is {Area}/{Line}/{Equip}/{Effective ?? backing tag Name}. Enums serialize /// numerically (AccessLevel: ReadWrite = 1). private static DeploymentId SeedV3Deployment( IDbContextFactory db, RevisionHash rev, (string Driver, string DriverName, string Device, string Tag, string DataType, bool Writable)[] rawTags, (string Area, string Line, string Equip, string BackingTag, string? Effective)[] unsRefs) { var drivers = rawTags .Select(t => (t.Driver, t.DriverName)) .Distinct() .Select(d => new { DriverInstanceId = d.Driver, RawFolderId = "rf-plant", Name = d.DriverName, DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = false }) .ToArray(); var devices = rawTags .Select(t => (t.Driver, t.Device)) .Distinct() .Select(d => new { DeviceId = $"{d.Driver}:{d.Device}", DriverInstanceId = d.Driver, Name = d.Device, DeviceConfig = "{}" }) .ToArray(); var tags = rawTags.Select(t => new { TagId = t.Tag, // tag id == name for test simplicity (unique per driver here) DeviceId = $"{t.Driver}:{t.Device}", TagGroupId = (string?)null, Name = t.Tag, DataType = t.DataType, AccessLevel = t.Writable ? 1 : 0, // TagAccessLevel.ReadWrite = 1 TagConfig = "{}", }).ToArray(); // UNS topology: one area/line per distinct (Area, Line); one equipment per (Area, Line, Equip). var areas = unsRefs.Select(r => r.Area).Distinct(StringComparer.Ordinal) .Select(a => new { UnsAreaId = $"a-{a}", Name = a, ClusterId = "c1" }).ToArray(); var lines = unsRefs.Select(r => (r.Area, r.Line)).Distinct() .Select(l => new { UnsLineId = $"l-{l.Area}-{l.Line}", UnsAreaId = $"a-{l.Area}", Name = l.Line }).ToArray(); var equipment = unsRefs.Select(r => (r.Area, r.Line, r.Equip)).Distinct() .Select(e => new { EquipmentId = $"EQ-{e.Area}-{e.Line}-{e.Equip}", UnsLineId = $"l-{e.Area}-{e.Line}", Name = e.Equip, MachineCode = e.Equip }).ToArray(); var unsTagReferences = unsRefs.Select((r, i) => new { UnsTagReferenceId = $"ref-{i}", EquipmentId = $"EQ-{r.Area}-{r.Line}-{r.Equip}", TagId = r.BackingTag, DisplayNameOverride = r.Effective, }).ToArray(); var artifact = JsonSerializer.SerializeToUtf8Bytes(new { RawFolders = new[] { new { RawFolderId = "rf-plant", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } }, DriverInstances = drivers, Devices = devices, TagGroups = Array.Empty(), Tags = tags, UnsAreas = areas, UnsLines = lines, Equipment = equipment, UnsTagReferences = unsTagReferences, }); 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; } }