Phase 7 Stream B — Core.VirtualTags project (engine + dep graph + timer + source)
Ships the evaluation engine that consumes compiled scripts from Stream A, subscribes to upstream driver tags, runs on change + on timer, cascades evaluations through dependent virtual tags in topological order, and emits changes through a driver-capability-shaped adapter the DriverNodeManager can dispatch to per ADR-002. DependencyGraph owns the directed dep-graph where nodes are tag paths (driver tags implicit leaves, virtual tags registered internal nodes) and edges run from a virtual tag to each tag it reads. Kahn algorithm produces the topological sort. Tarjan iterative SCC detects every cycle in one pass so publish-time rejection surfaces all offending cycles together. Both iterative so 10k-deep chains do not StackOverflow. Re-adding a node overwrites prior dependency set cleanly (supports config-publish reloads). VirtualTagDefinition is the operator-authored config row (Path, DataType, ScriptSource, ChangeTriggered, TimerInterval, Historize). Stream E config DB materializes these on publish. ITagUpstreamSource is the abstraction the engine pulls driver tag values from. Stream G bridges this to IReadable + ISubscribable on live drivers; tests use FakeUpstream that tracks subscription count for leak-test assertions. IHistoryWriter is the per-tag Historize sink. NullHistoryWriter default when caller does not pass one. VirtualTagContext is the per-evaluation ScriptContext. Reads from engine last-known-value cache, writes route through SetVirtualTag callback so cross-tag side effects participate in change cascades. Injectable Now clock for deterministic tests. VirtualTagEngine orchestrates. Load compiles every script via ScriptSandbox, builds the dep graph via DependencyExtractor, checks for cycles, reports every compile failure in one error, subscribes to each referenced upstream path, seeds the value cache. EvaluateAllAsync runs topological order. EvaluateOneAsync is timer path. Read returns cached value. Subscribe registers observer. OnUpstreamChange updates cache, fans out, schedules transitive dependents (change-driven=false tags skipped). EvaluateInternalAsync holds a SemaphoreSlim so cascades do not interleave. Script exceptions and timeouts map per-tag to BadInternalError. Coercion from script double to config Int32 uses Convert.ToInt32. TimerTriggerScheduler groups tags by interval into shared Timers. Tags without TimerInterval not scheduled. VirtualTagSource implements IReadable + ISubscribable per ADR-002. ReadAsync returns cache. SubscribeAsync fires initial-data callback per OPC UA convention. IWritable deliberately not implemented — OPC UA writes to virtual tags rejected in DriverNodeManager per Phase 7 decision 6. 36 unit tests across 4 files: DependencyGraphTests 12, VirtualTagEngineTests 13, VirtualTagSourceTests 6, TimerTriggerSchedulerTests 4. Coverage includes cycle detection (self-loop, 2-node, 3-node, multiple disjoint), 2-level change cascade, per-tag error isolation (one tag throws, others keep working), timeout isolation, Historize toggle, ChangeTriggered=false ignore, reload cleans subscriptions, Dispose releases resources, SetVirtualTag fires observers, type coercion, 10k deep graph no stack overflow, initial-data callback, Unsubscribe stops events. Fixed two bugs during implementation. Monitor.Enter/Exit cannot be held across await (Monitor ownership is thread-local and lost across suspension) — switched to SemaphoreSlim. Kahn edge-direction was inverted — for dependency ordering (X depends on Y means Y comes before X) in-degree should be count of a node own deps, not count of nodes pointing to it; was incrementing inDegree[dep] instead of inDegree[nodeId], causing false cycle detection on valid DAGs. Full Phase 7 test count after Stream B: 99 green (63 Scripting + 36 VirtualTags). Streams C and G will plug engine + source into live OPC UA dispatch path. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,166 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Verifies cycle detection + topological sort on the virtual-tag dependency
|
||||
/// graph. Publish-time correctness depends on these being right — a missed cycle
|
||||
/// would deadlock cascade evaluation; a wrong topological order would miscompute
|
||||
/// chained virtual tags.
|
||||
/// </summary>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class DependencyGraphTests
|
||||
{
|
||||
private static IReadOnlySet<string> Set(params string[] items) =>
|
||||
new HashSet<string>(items, StringComparer.Ordinal);
|
||||
|
||||
[Fact]
|
||||
public void Empty_graph_produces_empty_sort_and_no_cycles()
|
||||
{
|
||||
var g = new DependencyGraph();
|
||||
g.TopologicalSort().ShouldBeEmpty();
|
||||
g.DetectCycles().ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Single_node_with_no_deps()
|
||||
{
|
||||
var g = new DependencyGraph();
|
||||
g.Add("A", Set());
|
||||
g.TopologicalSort().ShouldBe(new[] { "A" });
|
||||
g.DetectCycles().ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Topological_order_places_dependencies_before_dependents()
|
||||
{
|
||||
var g = new DependencyGraph();
|
||||
g.Add("B", Set("A")); // B depends on A
|
||||
g.Add("C", Set("B", "A")); // C depends on B + A
|
||||
g.Add("A", Set()); // A is a leaf
|
||||
|
||||
var order = g.TopologicalSort();
|
||||
var idx = order.Select((x, i) => (x, i)).ToDictionary(p => p.x, p => p.i);
|
||||
idx["A"].ShouldBeLessThan(idx["B"]);
|
||||
idx["B"].ShouldBeLessThan(idx["C"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Self_loop_detected_as_cycle()
|
||||
{
|
||||
var g = new DependencyGraph();
|
||||
g.Add("A", Set("A"));
|
||||
var cycles = g.DetectCycles();
|
||||
cycles.Count.ShouldBe(1);
|
||||
cycles[0].ShouldContain("A");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Two_node_cycle_detected()
|
||||
{
|
||||
var g = new DependencyGraph();
|
||||
g.Add("A", Set("B"));
|
||||
g.Add("B", Set("A"));
|
||||
var cycles = g.DetectCycles();
|
||||
cycles.Count.ShouldBe(1);
|
||||
cycles[0].Count.ShouldBe(2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Three_node_cycle_detected()
|
||||
{
|
||||
var g = new DependencyGraph();
|
||||
g.Add("A", Set("B"));
|
||||
g.Add("B", Set("C"));
|
||||
g.Add("C", Set("A"));
|
||||
var cycles = g.DetectCycles();
|
||||
cycles.Count.ShouldBe(1);
|
||||
cycles[0].Count.ShouldBe(3);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Multiple_disjoint_cycles_all_reported()
|
||||
{
|
||||
var g = new DependencyGraph();
|
||||
// Cycle 1: A -> B -> A
|
||||
g.Add("A", Set("B"));
|
||||
g.Add("B", Set("A"));
|
||||
// Cycle 2: X -> Y -> Z -> X
|
||||
g.Add("X", Set("Y"));
|
||||
g.Add("Y", Set("Z"));
|
||||
g.Add("Z", Set("X"));
|
||||
// Clean leaf: M
|
||||
g.Add("M", Set());
|
||||
|
||||
var cycles = g.DetectCycles();
|
||||
cycles.Count.ShouldBe(2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Topological_sort_throws_DependencyCycleException_on_cycle()
|
||||
{
|
||||
var g = new DependencyGraph();
|
||||
g.Add("A", Set("B"));
|
||||
g.Add("B", Set("A"));
|
||||
Should.Throw<DependencyCycleException>(() => g.TopologicalSort())
|
||||
.Cycles.ShouldNotBeEmpty();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DirectDependents_returns_direct_only()
|
||||
{
|
||||
var g = new DependencyGraph();
|
||||
g.Add("B", Set("A"));
|
||||
g.Add("C", Set("B"));
|
||||
g.DirectDependents("A").ShouldBe(new[] { "B" });
|
||||
g.DirectDependents("B").ShouldBe(new[] { "C" });
|
||||
g.DirectDependents("C").ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void TransitiveDependentsInOrder_returns_topological_closure()
|
||||
{
|
||||
var g = new DependencyGraph();
|
||||
g.Add("B", Set("A"));
|
||||
g.Add("C", Set("B"));
|
||||
g.Add("D", Set("C"));
|
||||
var closure = g.TransitiveDependentsInOrder("A");
|
||||
closure.ShouldBe(new[] { "B", "C", "D" });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Readding_a_node_overwrites_prior_dependencies()
|
||||
{
|
||||
var g = new DependencyGraph();
|
||||
g.Add("X", Set("A"));
|
||||
g.DirectDependencies("X").ShouldBe(new[] { "A" });
|
||||
// Re-add with different deps (simulates script edit + republish).
|
||||
g.Add("X", Set("B", "C"));
|
||||
g.DirectDependencies("X").OrderBy(s => s).ShouldBe(new[] { "B", "C" });
|
||||
// A should no longer list X as a dependent.
|
||||
g.DirectDependents("A").ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Leaf_dependencies_not_registered_as_nodes_are_treated_as_implicit()
|
||||
{
|
||||
// A is referenced but never Add'd as a node — it's an upstream driver tag.
|
||||
var g = new DependencyGraph();
|
||||
g.Add("B", Set("A"));
|
||||
g.TopologicalSort().ShouldBe(new[] { "B" });
|
||||
g.DirectDependents("A").ShouldBe(new[] { "B" });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Deep_graph_no_stack_overflow()
|
||||
{
|
||||
// Iterative Tarjan's + Kahn's — 10k deep chain must complete without blowing the stack.
|
||||
var g = new DependencyGraph();
|
||||
for (var i = 1; i < 10_000; i++)
|
||||
g.Add($"N{i}", Set($"N{i - 1}"));
|
||||
var order = g.TopologicalSort();
|
||||
order.Count.ShouldBe(9_999);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
using System.Collections.Concurrent;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// In-memory <see cref="ITagUpstreamSource"/> for tests. Seed tag values via
|
||||
/// <see cref="Set"/>, push changes via <see cref="Push"/>. Tracks subscriptions so
|
||||
/// tests can assert the engine disposes them on reload / shutdown.
|
||||
/// </summary>
|
||||
public sealed class FakeUpstream : ITagUpstreamSource
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, DataValueSnapshot> _values = new(StringComparer.Ordinal);
|
||||
private readonly ConcurrentDictionary<string, List<Action<string, DataValueSnapshot>>> _subs = new(StringComparer.Ordinal);
|
||||
|
||||
public int ActiveSubscriptionCount { get; private set; }
|
||||
|
||||
public void Set(string path, object value, uint statusCode = 0u)
|
||||
{
|
||||
var now = DateTime.UtcNow;
|
||||
_values[path] = new DataValueSnapshot(value, statusCode, now, now);
|
||||
}
|
||||
|
||||
public void Push(string path, object value, uint statusCode = 0u)
|
||||
{
|
||||
Set(path, value, statusCode);
|
||||
if (_subs.TryGetValue(path, out var list))
|
||||
{
|
||||
Action<string, DataValueSnapshot>[] snap;
|
||||
lock (list) { snap = list.ToArray(); }
|
||||
foreach (var obs in snap) obs(path, _values[path]);
|
||||
}
|
||||
}
|
||||
|
||||
public DataValueSnapshot ReadTag(string path)
|
||||
=> _values.TryGetValue(path, out var v)
|
||||
? v
|
||||
: new DataValueSnapshot(null, 0x80340000u, null, DateTime.UtcNow);
|
||||
|
||||
public IDisposable SubscribeTag(string path, Action<string, DataValueSnapshot> observer)
|
||||
{
|
||||
var list = _subs.GetOrAdd(path, _ => []);
|
||||
lock (list) { list.Add(observer); }
|
||||
ActiveSubscriptionCount++;
|
||||
return new Unsub(this, path, observer);
|
||||
}
|
||||
|
||||
private sealed class Unsub : IDisposable
|
||||
{
|
||||
private readonly FakeUpstream _up;
|
||||
private readonly string _path;
|
||||
private readonly Action<string, DataValueSnapshot> _observer;
|
||||
public Unsub(FakeUpstream up, string path, Action<string, DataValueSnapshot> observer)
|
||||
{
|
||||
_up = up; _path = path; _observer = observer;
|
||||
}
|
||||
public void Dispose()
|
||||
{
|
||||
if (_up._subs.TryGetValue(_path, out var list))
|
||||
{
|
||||
lock (list)
|
||||
{
|
||||
if (list.Remove(_observer))
|
||||
_up.ActiveSubscriptionCount--;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
using Serilog;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests;
|
||||
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class TimerTriggerSchedulerTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Timer_interval_causes_periodic_reevaluation()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
// Counter source — re-eval should pick up new value each tick.
|
||||
var counter = 0;
|
||||
var logger = new LoggerConfiguration().CreateLogger();
|
||||
|
||||
using var engine = new VirtualTagEngine(up,
|
||||
new ScriptLoggerFactory(logger),
|
||||
logger);
|
||||
|
||||
engine.Load([new VirtualTagDefinition(
|
||||
"Counter", DriverDataType.Int32,
|
||||
"""return ctx.Now.Millisecond;""", // changes on every evaluation
|
||||
ChangeTriggered: false,
|
||||
TimerInterval: TimeSpan.FromMilliseconds(100))]);
|
||||
|
||||
using var sched = new TimerTriggerScheduler(engine, logger);
|
||||
sched.Start([new VirtualTagDefinition(
|
||||
"Counter", DriverDataType.Int32,
|
||||
"""return ctx.Now.Millisecond;""",
|
||||
ChangeTriggered: false,
|
||||
TimerInterval: TimeSpan.FromMilliseconds(100))]);
|
||||
|
||||
// Watch the value change across ticks.
|
||||
var snapshots = new List<object?>();
|
||||
using var sub = engine.Subscribe("Counter", (_, v) => snapshots.Add(v.Value));
|
||||
|
||||
await Task.Delay(500);
|
||||
|
||||
snapshots.Count.ShouldBeGreaterThanOrEqualTo(3, "At least 3 ticks in 500ms at 100ms cadence");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Tags_without_TimerInterval_not_scheduled()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
var logger = new LoggerConfiguration().CreateLogger();
|
||||
using var engine = new VirtualTagEngine(up,
|
||||
new ScriptLoggerFactory(logger), logger);
|
||||
engine.Load([new VirtualTagDefinition(
|
||||
"NoTimer", DriverDataType.Int32, """return 1;""")]);
|
||||
|
||||
using var sched = new TimerTriggerScheduler(engine, logger);
|
||||
sched.Start([new VirtualTagDefinition(
|
||||
"NoTimer", DriverDataType.Int32, """return 1;""")]);
|
||||
|
||||
var events = new List<int>();
|
||||
using var sub = engine.Subscribe("NoTimer", (_, v) => events.Add((int)(v.Value ?? 0)));
|
||||
|
||||
await Task.Delay(300);
|
||||
events.Count.ShouldBe(0, "No TimerInterval = no timer ticks");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Start_groups_tags_by_interval_into_shared_timers()
|
||||
{
|
||||
// Smoke test — Start on a definition list with two distinct intervals must not
|
||||
// throw. Group count matches unique intervals.
|
||||
var up = new FakeUpstream();
|
||||
var logger = new LoggerConfiguration().CreateLogger();
|
||||
using var engine = new VirtualTagEngine(up,
|
||||
new ScriptLoggerFactory(logger), logger);
|
||||
engine.Load([
|
||||
new VirtualTagDefinition("Fast", DriverDataType.Int32, """return 1;""",
|
||||
TimerInterval: TimeSpan.FromSeconds(1)),
|
||||
new VirtualTagDefinition("Slow", DriverDataType.Int32, """return 2;""",
|
||||
TimerInterval: TimeSpan.FromSeconds(5)),
|
||||
new VirtualTagDefinition("AlsoFast", DriverDataType.Int32, """return 3;""",
|
||||
TimerInterval: TimeSpan.FromSeconds(1)),
|
||||
]);
|
||||
|
||||
using var sched = new TimerTriggerScheduler(engine, logger);
|
||||
Should.NotThrow(() => sched.Start(new[]
|
||||
{
|
||||
new VirtualTagDefinition("Fast", DriverDataType.Int32, """return 1;""", TimerInterval: TimeSpan.FromSeconds(1)),
|
||||
new VirtualTagDefinition("Slow", DriverDataType.Int32, """return 2;""", TimerInterval: TimeSpan.FromSeconds(5)),
|
||||
new VirtualTagDefinition("AlsoFast", DriverDataType.Int32, """return 3;""", TimerInterval: TimeSpan.FromSeconds(1)),
|
||||
}));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Disposed_scheduler_stops_firing()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
var logger = new LoggerConfiguration().CreateLogger();
|
||||
using var engine = new VirtualTagEngine(up,
|
||||
new ScriptLoggerFactory(logger), logger);
|
||||
engine.Load([new VirtualTagDefinition(
|
||||
"T", DriverDataType.Int32, """return 1;""",
|
||||
TimerInterval: TimeSpan.FromMilliseconds(50))]);
|
||||
|
||||
var sched = new TimerTriggerScheduler(engine, logger);
|
||||
sched.Start([new VirtualTagDefinition(
|
||||
"T", DriverDataType.Int32, """return 1;""",
|
||||
TimerInterval: TimeSpan.FromMilliseconds(50))]);
|
||||
sched.Dispose();
|
||||
|
||||
// After dispose, second Start throws ObjectDisposedException.
|
||||
Should.Throw<ObjectDisposedException>(() =>
|
||||
sched.Start([new VirtualTagDefinition(
|
||||
"T", DriverDataType.Int32, """return 1;""",
|
||||
TimerInterval: TimeSpan.FromMilliseconds(50))]));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,307 @@
|
||||
using Serilog;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// End-to-end VirtualTagEngine behavior: load config, subscribe to upstream,
|
||||
/// evaluate on change, cascade through dependent virtual tags, timer-driven
|
||||
/// re-evaluation, error isolation, historize flag, cycle rejection.
|
||||
/// </summary>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class VirtualTagEngineTests
|
||||
{
|
||||
private static VirtualTagEngine Build(
|
||||
FakeUpstream upstream,
|
||||
IHistoryWriter? history = null,
|
||||
TimeSpan? scriptTimeout = null,
|
||||
Func<DateTime>? clock = null)
|
||||
{
|
||||
var rootLogger = new LoggerConfiguration().CreateLogger();
|
||||
return new VirtualTagEngine(
|
||||
upstream,
|
||||
new ScriptLoggerFactory(rootLogger),
|
||||
rootLogger,
|
||||
history,
|
||||
clock,
|
||||
scriptTimeout);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Simple_script_reads_upstream_and_returns_coerced_value()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("InTag", 10.0);
|
||||
using var engine = Build(up);
|
||||
|
||||
engine.Load([new VirtualTagDefinition(
|
||||
Path: "LineRate",
|
||||
DataType: DriverDataType.Float64,
|
||||
ScriptSource: """return (double)ctx.GetTag("InTag").Value * 2.0;""")]);
|
||||
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
var result = engine.Read("LineRate");
|
||||
result.StatusCode.ShouldBe(0u);
|
||||
result.Value.ShouldBe(20.0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Upstream_change_triggers_cascade_through_two_levels()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("A", 1.0);
|
||||
using var engine = Build(up);
|
||||
|
||||
engine.Load([
|
||||
new VirtualTagDefinition("B", DriverDataType.Float64,
|
||||
"""return (double)ctx.GetTag("A").Value + 10.0;"""),
|
||||
new VirtualTagDefinition("C", DriverDataType.Float64,
|
||||
"""return (double)ctx.GetTag("B").Value * 2.0;"""),
|
||||
]);
|
||||
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
engine.Read("B").Value.ShouldBe(11.0);
|
||||
engine.Read("C").Value.ShouldBe(22.0);
|
||||
|
||||
// Change upstream — cascade should recompute B (11→15.0) then C (30.0)
|
||||
up.Push("A", 5.0);
|
||||
await WaitForConditionAsync(() => Equals(engine.Read("B").Value, 15.0));
|
||||
engine.Read("B").Value.ShouldBe(15.0);
|
||||
engine.Read("C").Value.ShouldBe(30.0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Cycle_in_virtual_tags_rejected_at_Load()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
using var engine = Build(up);
|
||||
|
||||
Should.Throw<DependencyCycleException>(() => engine.Load([
|
||||
new VirtualTagDefinition("A", DriverDataType.Int32, """return (int)ctx.GetTag("B").Value + 1;"""),
|
||||
new VirtualTagDefinition("B", DriverDataType.Int32, """return (int)ctx.GetTag("A").Value + 1;"""),
|
||||
]));
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Script_compile_error_surfaces_at_Load_with_all_failures()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
using var engine = Build(up);
|
||||
|
||||
var ex = Should.Throw<InvalidOperationException>(() => engine.Load([
|
||||
new VirtualTagDefinition("A", DriverDataType.Int32, """return undefinedIdentifier;"""),
|
||||
new VirtualTagDefinition("B", DriverDataType.Int32, """return 42;"""),
|
||||
new VirtualTagDefinition("C", DriverDataType.Int32, """var x = anotherUndefined; return x;"""),
|
||||
]));
|
||||
ex.Message.ShouldContain("2 script(s) did not compile");
|
||||
ex.Message.ShouldContain("A");
|
||||
ex.Message.ShouldContain("C");
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Script_runtime_exception_isolates_to_owning_tag()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("OK", 10);
|
||||
using var engine = Build(up);
|
||||
|
||||
engine.Load([
|
||||
new VirtualTagDefinition("GoodTag", DriverDataType.Int32,
|
||||
"""return (int)ctx.GetTag("OK").Value * 2;"""),
|
||||
new VirtualTagDefinition("BadTag", DriverDataType.Int32,
|
||||
"""throw new InvalidOperationException("boom");"""),
|
||||
]);
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
engine.Read("GoodTag").StatusCode.ShouldBe(0u);
|
||||
engine.Read("GoodTag").Value.ShouldBe(20);
|
||||
engine.Read("BadTag").StatusCode.ShouldBe(0x80020000u, "BadInternalError for thrown script");
|
||||
engine.Read("BadTag").Value.ShouldBeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Timeout_maps_to_BadInternalError_without_killing_the_engine()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
using var engine = Build(up, scriptTimeout: TimeSpan.FromMilliseconds(30));
|
||||
|
||||
engine.Load([
|
||||
new VirtualTagDefinition("Hang", DriverDataType.Int32, """
|
||||
var end = Environment.TickCount64 + 5000;
|
||||
while (Environment.TickCount64 < end) { }
|
||||
return 1;
|
||||
"""),
|
||||
new VirtualTagDefinition("Ok", DriverDataType.Int32, """return 42;"""),
|
||||
]);
|
||||
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
engine.Read("Hang").StatusCode.ShouldBe(0x80020000u);
|
||||
engine.Read("Ok").Value.ShouldBe(42);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Subscribers_receive_engine_emitted_changes()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("In", 1);
|
||||
using var engine = Build(up);
|
||||
|
||||
engine.Load([new VirtualTagDefinition(
|
||||
"Out", DriverDataType.Int32, """return (int)ctx.GetTag("In").Value + 100;""")]);
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
var received = new List<DataValueSnapshot>();
|
||||
using var sub = engine.Subscribe("Out", (p, v) => received.Add(v));
|
||||
|
||||
up.Push("In", 5);
|
||||
await WaitForConditionAsync(() => received.Count >= 1);
|
||||
|
||||
received[^1].Value.ShouldBe(105);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Historize_flag_routes_to_history_writer()
|
||||
{
|
||||
var recorded = new List<(string, DataValueSnapshot)>();
|
||||
var history = new TestHistory(recorded);
|
||||
var up = new FakeUpstream();
|
||||
up.Set("In", 1);
|
||||
using var engine = Build(up, history);
|
||||
|
||||
engine.Load([
|
||||
new VirtualTagDefinition("H", DriverDataType.Int32,
|
||||
"""return (int)ctx.GetTag("In").Value + 1;""", Historize: true),
|
||||
new VirtualTagDefinition("NoH", DriverDataType.Int32,
|
||||
"""return (int)ctx.GetTag("In").Value - 1;""", Historize: false),
|
||||
]);
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
recorded.Select(p => p.Item1).ShouldContain("H");
|
||||
recorded.Select(p => p.Item1).ShouldNotContain("NoH");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Change_driven_false_ignores_upstream_push()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("In", 1);
|
||||
using var engine = Build(up);
|
||||
engine.Load([new VirtualTagDefinition(
|
||||
"Manual", DriverDataType.Int32,
|
||||
"""return (int)ctx.GetTag("In").Value * 10;""",
|
||||
ChangeTriggered: false)]);
|
||||
|
||||
// Initial eval seeds the value.
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
engine.Read("Manual").Value.ShouldBe(10);
|
||||
|
||||
// Upstream change fires but change-driven is off — no recompute.
|
||||
up.Push("In", 99);
|
||||
await Task.Delay(100);
|
||||
engine.Read("Manual").Value.ShouldBe(10, "change-driven=false ignores upstream deltas");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Reload_replaces_existing_tags_and_resubscribes_cleanly()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("A", 1);
|
||||
up.Set("B", 2);
|
||||
using var engine = Build(up);
|
||||
|
||||
engine.Load([new VirtualTagDefinition(
|
||||
"T", DriverDataType.Int32, """return (int)ctx.GetTag("A").Value * 2;""")]);
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
engine.Read("T").Value.ShouldBe(2);
|
||||
up.ActiveSubscriptionCount.ShouldBe(1);
|
||||
|
||||
// Reload — T now depends on B instead of A.
|
||||
engine.Load([new VirtualTagDefinition(
|
||||
"T", DriverDataType.Int32, """return (int)ctx.GetTag("B").Value * 3;""")]);
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
engine.Read("T").Value.ShouldBe(6);
|
||||
up.ActiveSubscriptionCount.ShouldBe(1, "previous subscription on A must be disposed");
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Dispose_releases_upstream_subscriptions()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("A", 1);
|
||||
var engine = Build(up);
|
||||
engine.Load([new VirtualTagDefinition(
|
||||
"T", DriverDataType.Int32, """return (int)ctx.GetTag("A").Value;""")]);
|
||||
up.ActiveSubscriptionCount.ShouldBe(1);
|
||||
|
||||
engine.Dispose();
|
||||
up.ActiveSubscriptionCount.ShouldBe(0);
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SetVirtualTag_within_script_updates_target_and_triggers_observers()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("In", 5);
|
||||
using var engine = Build(up);
|
||||
|
||||
engine.Load([
|
||||
new VirtualTagDefinition("Target", DriverDataType.Int32,
|
||||
"""return 0;""", ChangeTriggered: false), // placeholder value, operator-written via SetVirtualTag
|
||||
new VirtualTagDefinition("Driver", DriverDataType.Int32,
|
||||
"""
|
||||
var v = (int)ctx.GetTag("In").Value;
|
||||
ctx.SetVirtualTag("Target", v * 100);
|
||||
return v;
|
||||
"""),
|
||||
]);
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
engine.Read("Target").Value.ShouldBe(500);
|
||||
engine.Read("Driver").Value.ShouldBe(5);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Type_coercion_from_script_double_to_config_int32()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("In", 3.7);
|
||||
using var engine = Build(up);
|
||||
|
||||
engine.Load([new VirtualTagDefinition(
|
||||
"Rounded", DriverDataType.Int32,
|
||||
"""return (double)ctx.GetTag("In").Value;""")]);
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
engine.Read("Rounded").Value.ShouldBe(4, "Convert.ToInt32 rounds 3.7 to 4");
|
||||
}
|
||||
|
||||
private static async Task WaitForConditionAsync(Func<bool> cond, int timeoutMs = 2000)
|
||||
{
|
||||
var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs);
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
if (cond()) return;
|
||||
await Task.Delay(25);
|
||||
}
|
||||
throw new TimeoutException("Condition did not become true in time");
|
||||
}
|
||||
|
||||
private sealed class TestHistory : IHistoryWriter
|
||||
{
|
||||
private readonly List<(string, DataValueSnapshot)> _buf;
|
||||
public TestHistory(List<(string, DataValueSnapshot)> buf) => _buf = buf;
|
||||
public void Record(string path, DataValueSnapshot value)
|
||||
{
|
||||
lock (_buf) { _buf.Add((path, value)); }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
using Serilog;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Verifies the IReadable + ISubscribable adapter that DriverNodeManager dispatches
|
||||
/// to for NodeSource.Virtual per ADR-002. Key contract: OPC UA clients see virtual
|
||||
/// tags via the same capability interfaces as driver tags, so dispatch stays
|
||||
/// source-agnostic.
|
||||
/// </summary>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class VirtualTagSourceTests
|
||||
{
|
||||
private static (VirtualTagEngine engine, VirtualTagSource source, FakeUpstream up) Build()
|
||||
{
|
||||
var up = new FakeUpstream();
|
||||
up.Set("In", 10);
|
||||
var logger = new LoggerConfiguration().CreateLogger();
|
||||
var engine = new VirtualTagEngine(up, new ScriptLoggerFactory(logger), logger);
|
||||
engine.Load([new VirtualTagDefinition(
|
||||
"Out", DriverDataType.Int32, """return (int)ctx.GetTag("In").Value * 2;""")]);
|
||||
return (engine, new VirtualTagSource(engine), up);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ReadAsync_returns_engine_cached_values()
|
||||
{
|
||||
var (engine, source, _) = Build();
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
var results = await source.ReadAsync(["Out"], TestContext.Current.CancellationToken);
|
||||
results.Count.ShouldBe(1);
|
||||
results[0].Value.ShouldBe(20);
|
||||
results[0].StatusCode.ShouldBe(0u);
|
||||
engine.Dispose();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ReadAsync_unknown_path_returns_Bad_quality()
|
||||
{
|
||||
var (engine, source, _) = Build();
|
||||
var results = await source.ReadAsync(["NoSuchTag"], TestContext.Current.CancellationToken);
|
||||
results[0].StatusCode.ShouldBe(0x80340000u);
|
||||
engine.Dispose();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SubscribeAsync_fires_initial_data_callback()
|
||||
{
|
||||
var (engine, source, _) = Build();
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
var events = new List<DataChangeEventArgs>();
|
||||
source.OnDataChange += (_, e) => events.Add(e);
|
||||
|
||||
var handle = await source.SubscribeAsync(["Out"], TimeSpan.FromMilliseconds(100),
|
||||
TestContext.Current.CancellationToken);
|
||||
handle.ShouldNotBeNull();
|
||||
|
||||
// Per OPC UA convention, initial-data callback fires on subscribe.
|
||||
events.Count.ShouldBeGreaterThanOrEqualTo(1);
|
||||
events[0].FullReference.ShouldBe("Out");
|
||||
events[0].Snapshot.Value.ShouldBe(20);
|
||||
|
||||
await source.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
|
||||
engine.Dispose();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SubscribeAsync_fires_on_upstream_change_via_engine_cascade()
|
||||
{
|
||||
var (engine, source, up) = Build();
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
var events = new List<DataChangeEventArgs>();
|
||||
source.OnDataChange += (_, e) => events.Add(e);
|
||||
var handle = await source.SubscribeAsync(["Out"], TimeSpan.Zero,
|
||||
TestContext.Current.CancellationToken);
|
||||
|
||||
var initialCount = events.Count;
|
||||
up.Push("In", 50);
|
||||
|
||||
// Wait for the cascade.
|
||||
var deadline = DateTime.UtcNow.AddSeconds(2);
|
||||
while (DateTime.UtcNow < deadline && events.Count <= initialCount) await Task.Delay(25);
|
||||
|
||||
events.Count.ShouldBeGreaterThan(initialCount);
|
||||
events[^1].Snapshot.Value.ShouldBe(100);
|
||||
|
||||
await source.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
|
||||
engine.Dispose();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task UnsubscribeAsync_stops_further_events()
|
||||
{
|
||||
var (engine, source, up) = Build();
|
||||
await engine.EvaluateAllAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
var events = new List<DataChangeEventArgs>();
|
||||
source.OnDataChange += (_, e) => events.Add(e);
|
||||
var handle = await source.SubscribeAsync(["Out"], TimeSpan.Zero,
|
||||
TestContext.Current.CancellationToken);
|
||||
|
||||
await source.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
|
||||
var countAfterUnsub = events.Count;
|
||||
|
||||
up.Push("In", 99);
|
||||
await Task.Delay(200);
|
||||
|
||||
events.Count.ShouldBe(countAfterUnsub, "Unsubscribe must stop OnDataChange emissions");
|
||||
engine.Dispose();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Null_arguments_rejected()
|
||||
{
|
||||
var (engine, source, _) = Build();
|
||||
await Should.ThrowAsync<ArgumentNullException>(async () =>
|
||||
await source.ReadAsync(null!, TestContext.Current.CancellationToken));
|
||||
await Should.ThrowAsync<ArgumentNullException>(async () =>
|
||||
await source.SubscribeAsync(null!, TimeSpan.Zero, TestContext.Current.CancellationToken));
|
||||
await Should.ThrowAsync<ArgumentNullException>(async () =>
|
||||
await source.UnsubscribeAsync(null!, TestContext.Current.CancellationToken));
|
||||
engine.Dispose();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net10.0</TargetFramework>
|
||||
<Nullable>enable</Nullable>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<IsPackable>false</IsPackable>
|
||||
<IsTestProject>true</IsTestProject>
|
||||
<RootNamespace>ZB.MOM.WW.OtOpcUa.Core.VirtualTags.Tests</RootNamespace>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="xunit.v3" Version="1.1.0"/>
|
||||
<PackageReference Include="Shouldly" Version="4.3.0"/>
|
||||
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.12.0"/>
|
||||
<PackageReference Include="xunit.runner.visualstudio" Version="3.0.2">
|
||||
<PrivateAssets>all</PrivateAssets>
|
||||
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
|
||||
</PackageReference>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\src\ZB.MOM.WW.OtOpcUa.Core.VirtualTags\ZB.MOM.WW.OtOpcUa.Core.VirtualTags.csproj"/>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<NuGetAuditSuppress Include="https://github.com/advisories/GHSA-37gx-xxp4-5rgx"/>
|
||||
<NuGetAuditSuppress Include="https://github.com/advisories/GHSA-w3x6-4m5h-cxqf"/>
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
Reference in New Issue
Block a user