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

Phase 7 follow-up #243 — CachedTagUpstreamSource + Phase7EngineComposer

Browse Source 
Ships the composition kernel that maps Config DB rows (Script / VirtualTag /
ScriptedAlarm) to the runtime definitions VirtualTagEngine + ScriptedAlarmEngine
consume, builds the engine instances, and wires OnEvent → historian-sink routing.

## src/ZB.MOM.WW.OtOpcUa.Server/Phase7/

- CachedTagUpstreamSource — implements both Core.VirtualTags.ITagUpstreamSource and
  Core.ScriptedAlarms.ITagUpstreamSource (identical shape, distinct namespaces) on one
  concrete type so the composer can hand one instance to both engines. Thread-safe
  ConcurrentDictionary value cache with synchronous ReadTag + fire-on-write
  Push(path, snapshot) that fans out to every observer registered via SubscribeTag.
  Unknown-path reads return a BadNodeIdUnknown-quality snapshot (status 0x80340000)
  so scripts branch on quality naturally.
- Phase7EngineComposer.Compose(scripts, virtualTags, scriptedAlarms, upstream,
  alarmStateStore, historianSink, rootScriptLogger, loggerFactory) — single static
  entry point that:
  * Indexes scripts by ScriptId, resolves VirtualTag.ScriptId + ScriptedAlarm.PredicateScriptId
    to full SourceCode
  * Projects DB rows to VirtualTagDefinition + ScriptedAlarmDefinition (mapping
    DataType string → DriverDataType enum, AlarmType string → AlarmKind enum,
    Severity 1..1000 → AlarmSeverity bucket matching the OPC UA Part 9 bands
    that AbCipAlarmProjection + OpcUaClient MapSeverity already use)
  * Constructs VirtualTagEngine + loads definitions (throws InvalidOperationException
    with the list of scripts that failed to compile — aggregated like Streams B+C)
  * Constructs ScriptedAlarmEngine + loads definitions + wires OnEvent →
    IAlarmHistorianSink.EnqueueAsync using ScriptedAlarmEvent.Emission as the event
    kind + Condition.LastAckUser/LastAckComment for audit fields
  * Returns Phase7ComposedSources with Disposables list the caller owns

Empty Phase 7 config returns Phase7ComposedSources.Empty so deployments without
scripts / alarms behave exactly as pre-Phase-7. Non-null sources flow into
OpcUaApplicationHost's virtualReadable / scriptedAlarmReadable plumbing landed by
task #239 — DriverNodeManager then dispatches reads by NodeSourceKind per PR #186.

## Tests — 12/12

CachedTagUpstreamSourceTests (6):
- Unknown-path read returns BadNodeIdUnknown-quality snapshot
- Push-then-Read returns cached value
- Push fans out to subscribers in registration order
- Push to one path doesn't fire another path's observer
- Dispose of subscription handle stops fan-out
- Satisfies both Core.VirtualTags + Core.ScriptedAlarms ITagUpstreamSource interfaces

Phase7EngineComposerTests (6):
- Empty rows → Phase7ComposedSources.Empty (both sources null)
- VirtualTag rows → VirtualReadable non-null + Disposables populated
- Missing script reference throws InvalidOperationException with the missing ScriptId
  in the message
- Disabled VirtualTag row skipped by projection
- TimerIntervalMs → TimeSpan.FromMilliseconds round-trip
- Severity 1..1000 maps to Low/Medium/High/Critical at 250/500/750 boundaries
  (matches AbCipAlarmProjection + OpcUaClient.MapSeverity banding)

## Scope — what this PR does NOT do

The composition kernel is the tricky part; the remaining wiring is three narrower
follow-ups that each build on this PR:

- task #244 — driver-bridge feed that populates CachedTagUpstreamSource from live
  driver subscriptions. Without this, ctx.GetTag returns BadNodeIdUnknown even when
  the driver has a fresh value.
- task #245 — ScriptedAlarmReadable adapter exposing each alarm's current Active
  state as IReadable. Phase7EngineComposer.Compose currently returns
  ScriptedAlarmReadable=null so reads on Source=ScriptedAlarm variables return
  BadNotFound per the ADR-002 "misconfiguration not silent fallback" signal.
- task #246 — Program.cs call to Phase7EngineComposer.Compose with config rows
  loaded from the sealed-cache DB read, plus SqliteStoreAndForwardSink lifecycle
  wiring at %ProgramData%/OtOpcUa/alarm-historian-queue.db with the Galaxy.Host
  IPC writer from Stream D.

Task #240 (live OPC UA E2E smoke) depends on all three follow-ups landing.
This commit is contained in:
Joseph Doherty
2026-04-20 21:23:31 -04:00
parent c7f0855427
commit f64a8049d8
5 changed files with 522 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
tests/ZB.MOM.WW.OtOpcUa.Server.Tests/Phase7/CachedTagUpstreamSourceTests.cs Normal file
View File

@@ -0,0 +1,83 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Server.Phase7;
namespace ZB.MOM.WW.OtOpcUa.Server.Tests.Phase7;
/// <summary>
/// Covers the Phase 7 driver-to-engine bridge cache (task #243). Verifies the
/// cache serves last-known values synchronously, fans out Push updates to
/// subscribers, and cleans up on Dispose.
/// </summary>
[Trait("Category", "Unit")]
public sealed class CachedTagUpstreamSourceTests
{
private static DataValueSnapshot Snap(object? v) =>
new(v, 0u, DateTime.UtcNow, DateTime.UtcNow);
[Fact]
public void ReadTag_unknown_path_returns_BadNodeIdUnknown_snapshot()
{
var c = new CachedTagUpstreamSource();
var snap = c.ReadTag("/nowhere");
snap.Value.ShouldBeNull();
snap.StatusCode.ShouldBe(CachedTagUpstreamSource.UpstreamNotConfigured);
}
[Fact]
public void Push_then_Read_returns_cached_value()
{
var c = new CachedTagUpstreamSource();
c.Push("/Line1/Temp", Snap(42));
c.ReadTag("/Line1/Temp").Value.ShouldBe(42);
}
[Fact]
public void Push_fans_out_to_subscribers_in_registration_order()
{
var c = new CachedTagUpstreamSource();
var events = new List<string>();
c.SubscribeTag("/X", (p, s) => events.Add($"A:{p}:{s.Value}"));
c.SubscribeTag("/X", (p, s) => events.Add($"B:{p}:{s.Value}"));
c.Push("/X", Snap(7));
events.ShouldBe(["A:/X:7", "B:/X:7"]);
}
[Fact]
public void Push_to_different_path_does_not_fire_foreign_observer()
{
var c = new CachedTagUpstreamSource();
var fired = 0;
c.SubscribeTag("/X", (_, _) => fired++);
c.Push("/Y", Snap(1));
fired.ShouldBe(0);
}
[Fact]
public void Dispose_of_subscription_stops_fan_out()
{
var c = new CachedTagUpstreamSource();
var fired = 0;
var sub = c.SubscribeTag("/X", (_, _) => fired++);
c.Push("/X", Snap(1));
sub.Dispose();
c.Push("/X", Snap(2));
fired.ShouldBe(1);
}
[Fact]
public void Satisfies_both_VirtualTag_and_ScriptedAlarm_upstream_interfaces()
{
var c = new CachedTagUpstreamSource();
// Single instance is assignable to both — the composer passes it through for
// both engine constructors per the task #243 wiring.
((Core.VirtualTags.ITagUpstreamSource)c).ShouldNotBeNull();
((Core.ScriptedAlarms.ITagUpstreamSource)c).ShouldNotBeNull();
}
}

143
tests/ZB.MOM.WW.OtOpcUa.Server.Tests/Phase7/Phase7EngineComposerTests.cs Normal file
View File

@@ -0,0 +1,143 @@
using Microsoft.Extensions.Logging.Abstractions;
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
using ZB.MOM.WW.OtOpcUa.Server.Phase7;
namespace ZB.MOM.WW.OtOpcUa.Server.Tests.Phase7;
/// <summary>
/// Phase 7 follow-up (task #243) — verifies the composer that maps Config DB
/// rows to runtime engine definitions + wires up VirtualTagEngine +
/// ScriptedAlarmEngine + historian routing.
/// </summary>
[Trait("Category", "Unit")]
public sealed class Phase7EngineComposerTests
{
private static Script ScriptRow(string id, string source) => new()
{
ScriptRowId = Guid.NewGuid(), GenerationId = 1,
ScriptId = id, Name = id, SourceCode = source, SourceHash = "h",
};
private static VirtualTag VtRow(string id, string scriptId) => new()
{
VirtualTagRowId = Guid.NewGuid(), GenerationId = 1,
VirtualTagId = id, EquipmentId = "eq-1", Name = id,
DataType = "Float32", ScriptId = scriptId,
};
private static ScriptedAlarm AlarmRow(string id, string scriptId) => new()
{
ScriptedAlarmRowId = Guid.NewGuid(), GenerationId = 1,
ScriptedAlarmId = id, EquipmentId = "eq-1", Name = id,
AlarmType = "LimitAlarm", Severity = 500,
MessageTemplate = "x", PredicateScriptId = scriptId,
};
[Fact]
public void Compose_empty_rows_returns_Empty_sentinel()
{
var result = Phase7EngineComposer.Compose(
scripts: [],
virtualTags: [],
scriptedAlarms: [],
upstream: new CachedTagUpstreamSource(),
alarmStateStore: new InMemoryAlarmStateStore(),
historianSink: NullAlarmHistorianSink.Instance,
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
loggerFactory: NullLoggerFactory.Instance);
result.ShouldBeSameAs(Phase7ComposedSources.Empty);
result.VirtualReadable.ShouldBeNull();
result.ScriptedAlarmReadable.ShouldBeNull();
}
[Fact]
public void Compose_VirtualTag_rows_returns_non_null_VirtualReadable()
{
var scripts = new[] { ScriptRow("scr-1", "return 1;") };
var vtags = new[] { VtRow("vt-1", "scr-1") };
var result = Phase7EngineComposer.Compose(
scripts, vtags, [],
upstream: new CachedTagUpstreamSource(),
alarmStateStore: new InMemoryAlarmStateStore(),
historianSink: NullAlarmHistorianSink.Instance,
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
loggerFactory: NullLoggerFactory.Instance);
result.VirtualReadable.ShouldNotBeNull();
result.Disposables.Count.ShouldBeGreaterThan(0);
}
[Fact]
public void Compose_missing_script_reference_throws_with_actionable_message()
{
var vtags = new[] { VtRow("vt-1", "scr-missing") };
Should.Throw<InvalidOperationException>(() =>
Phase7EngineComposer.Compose(
scripts: [],
vtags, [],
upstream: new CachedTagUpstreamSource(),
alarmStateStore: new InMemoryAlarmStateStore(),
historianSink: NullAlarmHistorianSink.Instance,
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
loggerFactory: NullLoggerFactory.Instance))
.Message.ShouldContain("scr-missing");
}
[Fact]
public void Compose_disabled_VirtualTag_is_skipped()
{
var scripts = new[] { ScriptRow("scr-1", "return 1;") };
var disabled = VtRow("vt-1", "scr-1");
disabled.Enabled = false;
var defs = Phase7EngineComposer.ProjectVirtualTags(
new[] { disabled },
new Dictionary<string, Script> { ["scr-1"] = scripts[0] }).ToList();
defs.ShouldBeEmpty();
}
[Fact]
public void ProjectVirtualTags_maps_timer_interval_milliseconds_to_TimeSpan()
{
var scripts = new[] { ScriptRow("scr-1", "return 1;") };
var vt = VtRow("vt-1", "scr-1");
vt.TimerIntervalMs = 2500;
var def = Phase7EngineComposer.ProjectVirtualTags(
new[] { vt },
new Dictionary<string, Script> { ["scr-1"] = scripts[0] }).Single();
def.TimerInterval.ShouldBe(TimeSpan.FromMilliseconds(2500));
}
[Fact]
public void ProjectScriptedAlarms_maps_Severity_numeric_to_AlarmSeverity_bucket()
{
var scripts = new[] { ScriptRow("scr-1", "return true;") };
var buckets = new[] { (1, AlarmSeverity.Low), (250, AlarmSeverity.Low),
(251, AlarmSeverity.Medium), (500, AlarmSeverity.Medium),
(501, AlarmSeverity.High), (750, AlarmSeverity.High),
(751, AlarmSeverity.Critical), (1000, AlarmSeverity.Critical) };
foreach (var (input, expected) in buckets)
{
var row = AlarmRow("a1", "scr-1");
row.Severity = input;
var def = Phase7EngineComposer.ProjectScriptedAlarms(
new[] { row },
new Dictionary<string, Script> { ["scr-1"] = scripts[0] }).Single();
def.Severity.ShouldBe(expected, $"severity {input} should map to {expected}");
}
}
}