dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(runtime): delete dead per-value host ingress; batch fan-out review fixes

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 13:01:17 -04:00
parent 4f177bc3e8
commit ec6f8bb0ea
7 changed files with 305 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBatchFanOutTests.cs
+28 -106
View File
@@ -1,18 +1,12 @@
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.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;
using static ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness.V3ArtifactSeeder;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
@@ -29,7 +23,8 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// </para>
/// <para>
/// Arrangement mirrors <see cref="DriverHostActorLiveValueTests"/> — a real apply through the seeded
/// v3 artifact builds the map; sink + mux are <see cref="Akka.TestKit.TestProbe"/>s.
/// v3 artifact (<see cref="V3ArtifactSeeder"/>) builds the map; sink + mux are
/// <see cref="Akka.TestKit.TestProbe"/>s.
/// </para>
/// </summary>
public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
@@ -54,7 +49,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
},
unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) });
var (actor, publish, _) = SpawnHostAndApply(db, deploymentId);
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[]
{
@@ -62,7 +57,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/temp", 7.5, OpcUaStatus.BadNoCommunication, TsB),
}));
var batch = publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(TimeSpan.FromSeconds(5));
var batch = publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget);
batch.Items.Count.ShouldBe(3);
var byId = batch.Items.ToDictionary(w => w.NodeId);
@@ -103,7 +98,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
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);
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[]
{
@@ -111,7 +106,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA),
}));
var batch = publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(TimeSpan.FromSeconds(5));
var batch = publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget);
batch.Items.Count.ShouldBe(1);
batch.Items[0].NodeId.ShouldBe("Plant/Modbus/dev1/speed");
batch.Items[0].Value.ShouldBe(42.0);
@@ -121,7 +116,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
{
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 43.0, OpcUaStatus.Good, TsB),
}));
publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(TimeSpan.FromSeconds(5))
publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget)
.Items[0].Value.ShouldBe(43.0);
}
@@ -139,7 +134,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
},
unsRefs: Array.Empty<(string, string, string, string, string?)>());
var (actor, _, mux) = SpawnHostAndApply(db, deploymentId);
var (actor, _, mux) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[]
{
@@ -147,7 +142,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/temp", 7.5, OpcUaStatus.Good, TsB),
}));
var fwd = mux.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(TimeSpan.FromSeconds(5));
var fwd = mux.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget);
fwd.DriverInstanceId.ShouldBe("drv-1");
fwd.Items.Select(i => i.FullReference).ShouldBe(
new[] { "Plant/Modbus/dev1/speed", "Plant/Modbus/dev1/temp" });
@@ -165,7 +160,7 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
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);
var (actor, publish, mux) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[]
{
@@ -173,104 +168,31 @@ public sealed class DriverHostActorBatchFanOutTests : RuntimeActorTestBase
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/alsonope", 2.0, OpcUaStatus.Good, TsB),
}));
mux.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(TimeSpan.FromSeconds(5))
mux.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget)
.Items.Count.ShouldBe(2);
publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
}
/// <summary>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.</summary>
private (IActorRef Actor, Akka.TestKit.TestProbe Publish, Akka.TestKit.TestProbe Mux) SpawnHostAndApply(
IDbContextFactory<OtOpcUaConfigDbContext> db, DeploymentId deploymentId)
/// <summary>A host with NO OPC UA publish actor wired (the ctor default — the dev/Mac shape) still
/// makes the mux hop: the batch fan-out returns early on a null sink, and returning early BEFORE the
/// mux Tell would silently strand virtual-tag and continuous-historization inputs.</summary>
[Fact]
public void Mux_still_receives_batch_when_no_publish_actor_is_wired()
{
var coordinator = CreateTestProbe();
var publish = CreateTestProbe();
var mux = CreateTestProbe();
var vtHost = CreateTestProbe();
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 = Sys.ActorOf(DriverHostActor.Props(
db, TestNode, coordinator.Ref,
localRoles: new HashSet<string> { "driver" },
dependencyMux: mux.Ref,
opcUaPublishActor: publish.Ref,
virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
virtualTagHostOverride: vtHost.Ref));
var (actor, mux) = SpawnHostAndApplyWithoutPublishActor(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(TimeSpan.FromSeconds(5));
return (actor, publish, mux);
}
/// <summary>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 — the same shape <see cref="DriverHostActorLiveValueTests"/> seeds.</summary>
private static DeploymentId SeedV3Deployment(
IDbContextFactory<OtOpcUaConfigDbContext> 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
actor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("drv-1", new[]
{
TagId = t.Tag,
DeviceId = $"{t.Driver}:{t.Device}",
TagGroupId = (string?)null,
Name = t.Tag,
DataType = t.DataType,
AccessLevel = t.Writable ? 1 : 0,
TagConfig = "{}",
}).ToArray();
new DriverInstanceActor.AttributeValueItem("Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, TsA),
}));
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<object>(),
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;
var fwd = mux.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget);
fwd.DriverInstanceId.ShouldBe("drv-1");
fwd.Items.Single().FullReference.ShouldBe("Plant/Modbus/dev1/speed");
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorLiveValueTests.cs
+46 -144
View File
@@ -1,19 +1,12 @@
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;
using static ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness.V3ArtifactSeeder;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
@@ -22,14 +15,22 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
/// <see cref="DriverHostActor"/>. The driver publishes a value keyed by its wire-ref (== the tag's
/// <c>RawPath</c>); the host's <c>_nodeIdByDriverRef</c> map — rebuilt each apply from the composition's
/// <c>RawTags</c> ∪ <c>UnsReferenceVariables</c> — resolves <c>(DriverInstanceId, RawPath)</c> to the RAW
/// NodeId AND every referencing UNS NodeId, so <c>ForwardToMux</c> lands ONE publish on the raw node
/// NodeId AND every referencing UNS NodeId, so <c>ForwardBatchToMux</c> lands ONE write on the raw node
/// (<see cref="AddressSpaceRealm.Raw"/>) plus every UNS node (<see cref="AddressSpaceRealm.Uns"/>) 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.
/// <para>
/// 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. <c>DispatchDeployment</c> triggers the
/// Driven over the BATCH ingress, because that is the only one a driver child has: since the
/// value-pipeline batching work <c>DriverInstanceActor.OnDataChangeForward</c> wraps even a single
/// data change in an <see cref="DriverInstanceActor.AttributeValueBatchPublished"/>, and the host
/// no longer receives the per-value message at all. These fan-out cases are therefore stated as
/// single-item bursts; what they pin is the ROUTING, while
/// <see cref="DriverHostActorBatchFanOutTests"/> pins the message counts of a multi-item burst.
/// </para>
/// <para>
/// Drives a real apply through <see cref="V3ArtifactSeeder"/>: the seeded artifact carries the v3
/// raw-tag chain (RawFolder → DriverInstance(RawFolderId) → Device → Tag) + an optional
/// UnsTagReference projecting each raw tag into an equipment. <c>DispatchDeployment</c> triggers the
/// <c>ApplyAndAck → PushDesiredSubscriptions</c> pass that builds the map; the sink and mux are
/// injected as <see cref="Akka.TestKit.TestProbe"/>s.
/// </para>
@@ -40,6 +41,12 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
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);
/// <summary>The single-item burst a driver child emits for one data change — every value ingress into
/// the host is an AttributeValueBatchPublished, so the fan-out cases below state their input that
/// way.</summary>
private static DriverInstanceActor.AttributeValueBatchPublished Burst((string Ref, object Value) item) =>
new("drv-1", new[] { new DriverInstanceActor.AttributeValueItem(item.Ref, item.Value, OpcUaStatus.Good, Ts) });
/// <summary>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.</summary>
[Fact]
@@ -50,15 +57,14 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
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);
var (actor, publish, mux) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
// The driver publishes by its wire-ref RawPath.
actor.Tell(new DriverInstanceActor.AttributeValuePublished(
"drv-1", "Plant/Modbus/dev1/speed", 42.0, OpcUaStatus.Good, Ts));
// The driver publishes by its wire-ref RawPath, in the single-item burst DriverInstanceActor
// emits for one data change.
actor.Tell(Burst(("Plant/Modbus/dev1/speed", 42.0)));
var a = publish.ExpectMsg<OpcUaPublishActor.AttributeValueUpdate>(TimeSpan.FromSeconds(5));
var b = publish.ExpectMsg<OpcUaPublishActor.AttributeValueUpdate>(TimeSpan.FromSeconds(5));
var byId = new[] { a, b }.ToDictionary(u => u.NodeId);
var byId = publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget)
.Items.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.
@@ -69,12 +75,12 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
{
u.Value.ShouldBe(42.0);
u.StatusCode.ShouldBe(OpcUaStatus.Good);
u.TimestampUtc.ShouldBe(Ts);
u.SourceTimestampUtc.ShouldBe(Ts);
}
// The raw publish still reaches the dependency mux (VirtualTag inputs), keyed by RawPath.
mux.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(TimeSpan.FromSeconds(5))
.FullReference.ShouldBe("Plant/Modbus/dev1/speed");
mux.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget)
.Items.Single().FullReference.ShouldBe("Plant/Modbus/dev1/speed");
}
/// <summary>A raw tag with NO UNS reference fans to ONLY its raw NodeId (Raw realm).</summary>
@@ -86,12 +92,12 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
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);
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.AttributeValuePublished(
"drv-1", "Plant/Modbus/dev1/speed", 5.0, OpcUaStatus.Good, Ts));
actor.Tell(Burst(("Plant/Modbus/dev1/speed", 5.0)));
var only = publish.ExpectMsg<OpcUaPublishActor.AttributeValueUpdate>(TimeSpan.FromSeconds(5));
var only = publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget)
.Items.ShouldHaveSingleItem();
only.NodeId.ShouldBe("Plant/Modbus/dev1/speed");
only.Realm.ShouldBe(AddressSpaceRealm.Raw);
publish.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
@@ -111,17 +117,13 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
(Area: "filling", Line: "line1", Equip: "station2", BackingTag: "speed", Effective: (string?)null),
});
var (actor, publish, _) = SpawnHostAndApply(db, deploymentId);
var (actor, publish, _) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.AttributeValuePublished(
"drv-1", "Plant/Modbus/dev1/speed", 7.0, OpcUaStatus.Good, Ts));
actor.Tell(Burst(("Plant/Modbus/dev1/speed", 7.0)));
var updates = new[]
{
publish.ExpectMsg<OpcUaPublishActor.AttributeValueUpdate>(TimeSpan.FromSeconds(5)),
publish.ExpectMsg<OpcUaPublishActor.AttributeValueUpdate>(TimeSpan.FromSeconds(5)),
publish.ExpectMsg<OpcUaPublishActor.AttributeValueUpdate>(TimeSpan.FromSeconds(5)),
};
// 3-wide fan-out inside ONE batch message — the batch path's own 1:N coverage.
var updates = publish.ExpectMsg<OpcUaPublishActor.AttributeValueBatchUpdate>(PresenceBudget).Items;
updates.Count.ShouldBe(3);
updates.Select(u => u.NodeId).ShouldBe(new[]
{
"Plant/Modbus/dev1/speed",
@@ -141,14 +143,13 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
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);
var (actor, publish, mux) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.AttributeValuePublished(
"drv-1", "Plant/Modbus/dev1/nope", 99.0, OpcUaStatus.Good, Ts));
actor.Tell(Burst(("Plant/Modbus/dev1/nope", 99.0)));
publish.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
mux.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(TimeSpan.FromSeconds(5))
.FullReference.ShouldBe("Plant/Modbus/dev1/nope");
mux.ExpectMsg<DriverInstanceActor.AttributeValueBatchPublished>(PresenceBudget)
.Items.Single().FullReference.ShouldBe("Plant/Modbus/dev1/nope");
}
// ---- #481 (Layer 4): comms loss reaches the dependency mux ------------------------------------
@@ -172,14 +173,14 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
},
unsRefs: Array.Empty<(string, string, string, string, string?)>());
var (actor, _, mux) = SpawnHostAndApply(db, deploymentId);
var (actor, _, mux) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-1", Connected: false));
var pushes = new[]
{
mux.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(TimeSpan.FromSeconds(5)),
mux.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(TimeSpan.FromSeconds(5)),
mux.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(PresenceBudget),
mux.ExpectMsg<DriverInstanceActor.AttributeValuePublished>(PresenceBudget),
};
pushes.Select(p => p.FullReference).ShouldBe(
@@ -215,7 +216,7 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
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, _, mux) = SpawnHostAndApply(db, deploymentId);
var (actor, _, mux) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-1", Connected: true));
@@ -232,110 +233,11 @@ public sealed class DriverHostActorLiveValueTests : RuntimeActorTestBase
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, _, mux) = SpawnHostAndApply(db, deploymentId);
var (actor, _, mux) = SpawnHostAndApply(this, db, TestNode, deploymentId, RevA);
actor.Tell(new DriverInstanceActor.ConnectivityChanged("drv-OTHER", Connected: false));
mux.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
}
/// <summary>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.</summary>
private (IActorRef Actor, Akka.TestKit.TestProbe Publish, Akka.TestKit.TestProbe Mux) SpawnHostAndApply(
IDbContextFactory<OtOpcUaConfigDbContext> 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<string> { "driver" },
dependencyMux: mux.Ref,
opcUaPublishActor: publish.Ref,
virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
virtualTagHostOverride: vtHost.Ref));
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(TimeSpan.FromSeconds(5));
return (actor, publish, mux);
}
/// <summary>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 <c>Plant/{DriverName}/{Device}/{Tag}</c>; each UNS
/// reference's NodeId is <c>{Area}/{Line}/{Equip}/{Effective ?? backing tag Name}</c>. Enums serialize
/// numerically (AccessLevel: ReadWrite = 1).</summary>
private static DeploymentId SeedV3Deployment(
IDbContextFactory<OtOpcUaConfigDbContext> 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<object>(),
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;
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/RuntimeActorTestBase.cs
+4 -1
View File
@@ -27,8 +27,11 @@ public abstract class RuntimeActorTestBase : TestKit
/// idle machine are comfortable in isolation and marginal under that contention: three consecutive
/// 30-run verification rounds each surfaced a <em>different</em> test failing on one, in three
/// different files. Fixing them one at a time was chasing a distribution rather than a defect.</para>
/// <para>Public rather than protected so Harness helpers that are not themselves
/// <see cref="TestKit"/> subclasses — <see cref="V3ArtifactSeeder"/> — budget their waits from the
/// same constant instead of re-deriving one.</para>
/// </remarks>
protected static readonly TimeSpan PresenceBudget = TimeSpan.FromSeconds(15);
public static readonly TimeSpan PresenceBudget = TimeSpan.FromSeconds(15);
/// <summary>Gets the Akka test HOCON configuration string for single-node cluster setup.</summary>
protected static string AkkaTestHocon => @"
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs
+173
View File
@@ -0,0 +1,173 @@
using System.Text.Json;
using Akka.Actor;
using Akka.TestKit;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
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;
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
/// <summary>
/// Shared v3 deployment fixture for the <see cref="DriverHostActor"/> value-routing suites: seeds a
/// Sealed artifact carrying the raw-tag chain (+ optional UNS references) and spawns a host that has
/// actually applied it, so <c>_nodeIdByDriverRef</c> is populated before the test publishes anything.
/// <para>
/// Kept here rather than in either test class because the per-value and batched fan-out suites
/// assert different deliveries over the SAME arrangement — duplicating the seeder let the two
/// copies drift. Deliberately narrow: the assembly's other artifact fixtures seed their own
/// shapes and are not folded in.
/// </para>
/// </summary>
public static class V3ArtifactSeeder
{
/// <summary>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 <c>Plant/{DriverName}/{Device}/{Tag}</c>; each UNS
/// reference's NodeId is <c>{Area}/{Line}/{Equip}/{Effective ?? backing tag Name}</c>. Enums serialize
/// numerically (AccessLevel: ReadWrite = 1).</summary>
/// <param name="db">The in-memory config-db factory to seed.</param>
/// <param name="rev">The revision hash the deployment is sealed under.</param>
/// <param name="rawTags">The raw tags to materialise, one per (driver, device, tag).</param>
/// <param name="unsRefs">UNS references projecting a raw tag (by <c>BackingTag</c>) into an equipment.</param>
/// <returns>The seeded deployment's id.</returns>
public static DeploymentId SeedV3Deployment(
IDbContextFactory<OtOpcUaConfigDbContext> 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<object>(),
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;
}
/// <summary>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.</summary>
/// <param name="kit">The calling TestKit, supplying the ActorSystem + probes.</param>
/// <param name="db">The factory whose deployment was seeded by <see cref="SeedV3Deployment"/>.</param>
/// <param name="localNode">The host's local node id.</param>
/// <param name="deploymentId">The deployment to dispatch.</param>
/// <param name="rev">The revision hash the deployment was sealed under.</param>
/// <returns>The host actor plus the publish and mux probes wired into it.</returns>
public static (IActorRef Actor, TestProbe Publish, TestProbe Mux) SpawnHostAndApply(
TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db,
NodeId localNode, DeploymentId deploymentId, RevisionHash rev)
{
var publish = kit.CreateTestProbe();
var (actor, mux, coordinator) = Spawn(kit, db, localNode, publish.Ref);
Dispatch(actor, coordinator, deploymentId, rev);
publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(RuntimeActorTestBase.PresenceBudget);
return (actor, publish, mux);
}
/// <summary>The same arrangement with NO OPC UA publish actor wired (the ctor default) — the dev/Mac
/// shape, where the mux hop must still happen. Returns only the mux probe, since there is no sink.</summary>
/// <param name="kit">The calling TestKit, supplying the ActorSystem + probes.</param>
/// <param name="db">The factory whose deployment was seeded by <see cref="SeedV3Deployment"/>.</param>
/// <param name="localNode">The host's local node id.</param>
/// <param name="deploymentId">The deployment to dispatch.</param>
/// <param name="rev">The revision hash the deployment was sealed under.</param>
/// <returns>The host actor plus the mux probe wired into it.</returns>
public static (IActorRef Actor, TestProbe Mux) SpawnHostAndApplyWithoutPublishActor(
TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db,
NodeId localNode, DeploymentId deploymentId, RevisionHash rev)
{
var (actor, mux, coordinator) = Spawn(kit, db, localNode, opcUaPublishActor: null);
Dispatch(actor, coordinator, deploymentId, rev);
return (actor, mux);
}
private static (IActorRef Actor, TestProbe Mux, TestProbe Coordinator) Spawn(
TestKitBase kit, IDbContextFactory<OtOpcUaConfigDbContext> db, NodeId localNode,
IActorRef? opcUaPublishActor)
{
var coordinator = kit.CreateTestProbe();
var mux = kit.CreateTestProbe();
var vtHost = kit.CreateTestProbe();
var actor = kit.Sys.ActorOf(DriverHostActor.Props(
db, localNode, coordinator.Ref,
localRoles: new HashSet<string> { "driver" },
dependencyMux: mux.Ref,
opcUaPublishActor: opcUaPublishActor,
virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
virtualTagHostOverride: vtHost.Ref));
return (actor, mux, coordinator);
}
private static void Dispatch(IActorRef actor, TestProbe coordinator, DeploymentId deploymentId, RevisionHash rev)
{
actor.Tell(new DispatchDeployment(deploymentId, rev, CorrelationId.NewId()));
coordinator.ExpectMsg<ApplyAck>(RuntimeActorTestBase.PresenceBudget)
.Outcome.ShouldBe(ApplyAckOutcome.Applied);
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/VirtualTags/DependencyMuxActorTests.cs
+11 -9
View File
@@ -124,13 +124,14 @@ public sealed class DependencyMuxActorTests : RuntimeActorTestBase
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
}
/// <summary>Verifies that the DriverHostActor forwards AttributeValuePublished messages through the dependency mux.</summary>
/// <summary>Verifies that the DriverHostActor forwards a driver burst through the dependency mux to the
/// interested subscriber.</summary>
[Fact]
public void DriverHostActor_forwards_AttributeValuePublished_through_to_mux()
public void DriverHostActor_forwards_AttributeValueBatchPublished_through_to_mux()
{
// Spin a mux + a stand-in for DriverHostActor that wraps the real DriverHostActor's
// forward path. We feed it AttributeValuePublished directly to verify the routing —
// exercising the actual DriverHost spawn would require a deployment artifact + DI.
// Spin a mux + a real DriverHostActor and feed it the batch a driver child would send —
// exercising a full DriverHost apply would require a deployment artifact + DI, and the mux hop
// does not depend on one (it precedes the NodeId resolution).
var mux = Sys.ActorOf(DependencyMuxActor.Props());
var subscriber = CreateTestProbe();
mux.Tell(new DependencyMuxActor.RegisterInterest(new[] { "ref-1" }, subscriber.Ref));
@@ -142,9 +143,10 @@ public sealed class DependencyMuxActorTests : RuntimeActorTestBase
ackRouter: hostProbe.Ref,
dependencyMux: mux));
// Tell the host an AttributeValuePublished — it should fan out to the mux + subscriber.
hostActor.Tell(new DriverInstanceActor.AttributeValuePublished(
"driver-1", "ref-1", 42, OpcUaStatus.Good, DateTime.UtcNow));
hostActor.Tell(new DriverInstanceActor.AttributeValueBatchPublished("driver-1", new[]
{
new DriverInstanceActor.AttributeValueItem("ref-1", 42, OpcUaStatus.Good, DateTime.UtcNow),
}));
subscriber.ExpectMsg<VirtualTagActor.DependencyValueChanged>().TagId.ShouldBe("ref-1");
}
@@ -168,7 +170,7 @@ public sealed class DependencyMuxActorTests : RuntimeActorTestBase
new DriverInstanceActor.AttributeValueItem("tag-x", 10, OpcUaStatus.BadNoCommunication, tsX),
}));
var dep = sub.ExpectMsg<VirtualTagActor.DependencyValueChanged>(TimeSpan.FromSeconds(3));
var dep = sub.ExpectMsg<VirtualTagActor.DependencyValueChanged>(PresenceBudget);
dep.TagId.ShouldBe("tag-x");
dep.Value.ShouldBe(10);
dep.TimestampUtc.ShouldBe(tsX);