Closes Phase 7 Gap 5: VirtualTagEngine called IHistoryWriter.Record per evaluation when Historize=true but Phase7EngineComposer always passed NullHistoryWriter, so virtual-tag history was computed but never persisted. The fix: - New RingBufferHistoryWriter implements both IHistoryWriter (write port for the evaluation pipeline) and IHistorianDataSource (read port for IHistoryRouter so OPC UA HistoryRead on virtual-tag nodes resolves here). Maintains one bounded ring buffer (1000 samples, configurable) per tag path; Record() is O(1) and never blocks evaluation. - Phase7EngineComposer.Compose now accepts IHistoryRouter? and, when any VirtualTagDefinition.Historize=true, creates a RingBufferHistoryWriter, passes it to VirtualTagEngine as historyWriter, adds it to the disposables list, and registers it under the "virtual:" prefix in the router for HistoryRead dispatch. - Phase7Composer accepts IHistoryRouter? from DI (already registered as singleton in Program.cs) and threads it through to Phase7EngineComposer.Compose. - NullHistoryWriter remains as fallback when no tags request historization. - 16 new unit tests in RingBufferHistoryWriterTests.cs cover ring-buffer semantics, eviction, per-tag isolation, ReadRawAsync windowing, IHistorianDataSource stubs, router registration, and the Historize=false / null-router fallback paths. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
309 lines
12 KiB
C#
309 lines
12 KiB
C#
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.History;
|
|
using ZB.MOM.WW.OtOpcUa.Server.Phase7;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Server.Tests.Phase7;
|
|
|
|
/// <summary>
|
|
/// Task #28 — Gap 5 closure: verifies that <see cref="RingBufferHistoryWriter"/>
|
|
/// correctly records virtual-tag evaluation results and returns them via the
|
|
/// <see cref="IHistorianDataSource"/> read interface; and that
|
|
/// <see cref="Phase7EngineComposer.Compose"/> wires the writer and registers it
|
|
/// with an <see cref="IHistoryRouter"/> when <c>Historize=true</c> tags are present.
|
|
/// </summary>
|
|
[Trait("Category", "Unit")]
|
|
public sealed class RingBufferHistoryWriterTests
|
|
{
|
|
private static readonly DateTime T0 = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
|
|
private static readonly DateTime T1 = T0.AddSeconds(1);
|
|
private static readonly DateTime T2 = T0.AddSeconds(2);
|
|
private static readonly DateTime T3 = T0.AddSeconds(3);
|
|
|
|
private static DataValueSnapshot Snap(double value, DateTime ts) =>
|
|
new(value, 0u, ts, ts);
|
|
|
|
// ===== RingBufferHistoryWriter unit tests =====
|
|
|
|
[Fact]
|
|
public void Record_stores_sample_retrievable_via_GetSnapshots()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter();
|
|
writer.Record("/area/line/eq/Tag1", Snap(42.0, T0));
|
|
|
|
var snaps = writer.GetSnapshots("/area/line/eq/Tag1");
|
|
snaps.Length.ShouldBe(1);
|
|
snaps[0].Value.ShouldBe(42.0);
|
|
}
|
|
|
|
[Fact]
|
|
public void Record_multiple_samples_preserves_insertion_order()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter();
|
|
writer.Record("/t", Snap(1.0, T0));
|
|
writer.Record("/t", Snap(2.0, T1));
|
|
writer.Record("/t", Snap(3.0, T2));
|
|
|
|
var snaps = writer.GetSnapshots("/t");
|
|
snaps.Length.ShouldBe(3);
|
|
snaps[0].Value.ShouldBe(1.0);
|
|
snaps[1].Value.ShouldBe(2.0);
|
|
snaps[2].Value.ShouldBe(3.0);
|
|
}
|
|
|
|
[Fact]
|
|
public void Record_evicts_oldest_when_capacity_exceeded()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter(capacity: 3);
|
|
writer.Record("/t", Snap(1.0, T0));
|
|
writer.Record("/t", Snap(2.0, T1));
|
|
writer.Record("/t", Snap(3.0, T2));
|
|
writer.Record("/t", Snap(4.0, T3)); // evicts 1.0
|
|
|
|
var snaps = writer.GetSnapshots("/t");
|
|
snaps.Length.ShouldBe(3);
|
|
snaps[0].Value.ShouldBe(2.0, "oldest evicted");
|
|
snaps[2].Value.ShouldBe(4.0, "newest present");
|
|
}
|
|
|
|
[Fact]
|
|
public void Record_maintains_separate_buffers_per_tag_path()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter();
|
|
writer.Record("/area/eq/TagA", Snap(10.0, T0));
|
|
writer.Record("/area/eq/TagB", Snap(20.0, T0));
|
|
|
|
writer.GetSnapshots("/area/eq/TagA").Single().Value.ShouldBe(10.0);
|
|
writer.GetSnapshots("/area/eq/TagB").Single().Value.ShouldBe(20.0);
|
|
writer.TagCount.ShouldBe(2);
|
|
}
|
|
|
|
[Fact]
|
|
public void GetSnapshots_returns_empty_for_unknown_path()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter();
|
|
writer.GetSnapshots("/not/a/path").ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public void Dispose_clears_buffers_and_subsequent_Record_is_silently_ignored()
|
|
{
|
|
var writer = new RingBufferHistoryWriter();
|
|
writer.Record("/t", Snap(1.0, T0));
|
|
writer.Dispose();
|
|
|
|
// After dispose, Record must silently drop (no exception).
|
|
Should.NotThrow(() => writer.Record("/t", Snap(2.0, T1)));
|
|
// GetSnapshots post-dispose returns empty (buffers cleared).
|
|
writer.GetSnapshots("/t").ShouldBeEmpty();
|
|
}
|
|
|
|
// ===== IHistorianDataSource.ReadRawAsync tests =====
|
|
|
|
[Fact]
|
|
public async Task ReadRawAsync_returns_empty_for_unknown_path()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter();
|
|
var result = await writer.ReadRawAsync("notexists", T0, T3, 100, default);
|
|
result.Samples.ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReadRawAsync_returns_samples_in_time_window()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter();
|
|
writer.Record("/t", Snap(1.0, T0));
|
|
writer.Record("/t", Snap(2.0, T1));
|
|
writer.Record("/t", Snap(3.0, T2));
|
|
|
|
// Window [T0, T2) — T2 excluded (half-open interval).
|
|
var result = await writer.ReadRawAsync("/t", T0, T2, 100, default);
|
|
result.Samples.Count.ShouldBe(2);
|
|
result.Samples[0].Value.ShouldBe(1.0);
|
|
result.Samples[1].Value.ShouldBe(2.0);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReadRawAsync_respects_maxValuesPerNode_cap()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter();
|
|
writer.Record("/t", Snap(1.0, T0));
|
|
writer.Record("/t", Snap(2.0, T1));
|
|
writer.Record("/t", Snap(3.0, T2));
|
|
|
|
var result = await writer.ReadRawAsync("/t", T0, T3, maxValuesPerNode: 2, default);
|
|
result.Samples.Count.ShouldBe(2);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReadProcessedAsync_returns_empty_result()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter();
|
|
writer.Record("/t", Snap(1.0, T0));
|
|
var result = await writer.ReadProcessedAsync("/t", T0, T3, TimeSpan.FromSeconds(1),
|
|
HistoryAggregateType.Average, default);
|
|
result.Samples.ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReadAtTimeAsync_returns_empty_result()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter();
|
|
writer.Record("/t", Snap(1.0, T0));
|
|
var result = await writer.ReadAtTimeAsync("/t", [T0, T1], default);
|
|
result.Samples.ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReadEventsAsync_returns_empty_result()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter();
|
|
var result = await writer.ReadEventsAsync(null, T0, T3, 100, default);
|
|
result.Events.ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public void GetHealthSnapshot_returns_connected_non_null_snapshot()
|
|
{
|
|
using var writer = new RingBufferHistoryWriter();
|
|
var health = writer.GetHealthSnapshot();
|
|
health.ShouldNotBeNull();
|
|
health.ProcessConnectionOpen.ShouldBeTrue("ring buffer is always available in-process");
|
|
}
|
|
|
|
// ===== Phase7EngineComposer wiring tests =====
|
|
|
|
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, bool historize = false) => new()
|
|
{
|
|
VirtualTagRowId = Guid.NewGuid(), GenerationId = 1,
|
|
VirtualTagId = id, EquipmentId = "eq-1", Name = id,
|
|
DataType = "Float32", ScriptId = scriptId,
|
|
Historize = historize,
|
|
};
|
|
|
|
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_without_Historize_uses_NullHistoryWriter_and_skips_router_registration()
|
|
{
|
|
using var router = new HistoryRouter();
|
|
var scripts = new[] { ScriptRow("s1", "return 1;") };
|
|
var vtags = new[] { VtRow("vt-1", "s1", historize: false) };
|
|
|
|
var result = Phase7EngineComposer.Compose(
|
|
scripts, vtags, [],
|
|
upstream: new CachedTagUpstreamSource(),
|
|
alarmStateStore: new InMemoryAlarmStateStore(),
|
|
historianSink: NullAlarmHistorianSink.Instance,
|
|
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
|
|
loggerFactory: NullLoggerFactory.Instance,
|
|
historyRouter: router);
|
|
|
|
// Router should not have a "virtual:" prefix entry when no Historize=true tags.
|
|
router.Resolve(Phase7EngineComposer.VirtualTagHistoryPrefix + "vt-1").ShouldBeNull();
|
|
result.VirtualReadable.ShouldNotBeNull();
|
|
}
|
|
|
|
[Fact]
|
|
public void Compose_with_Historize_true_registers_RingBufferHistoryWriter_in_router()
|
|
{
|
|
using var router = new HistoryRouter();
|
|
var scripts = new[] { ScriptRow("s1", "return 1.0f;") };
|
|
var vtags = new[] { VtRow("vt-hist", "s1", historize: true) };
|
|
|
|
var result = Phase7EngineComposer.Compose(
|
|
scripts, vtags, [],
|
|
upstream: new CachedTagUpstreamSource(),
|
|
alarmStateStore: new InMemoryAlarmStateStore(),
|
|
historianSink: NullAlarmHistorianSink.Instance,
|
|
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
|
|
loggerFactory: NullLoggerFactory.Instance,
|
|
historyRouter: router);
|
|
|
|
// The "virtual:" prefix must resolve to a RingBufferHistoryWriter instance.
|
|
var source = router.Resolve(Phase7EngineComposer.VirtualTagHistoryPrefix + "vt-hist");
|
|
source.ShouldNotBeNull("router should have the ring-buffer source registered under 'virtual:' prefix");
|
|
source.ShouldBeOfType<RingBufferHistoryWriter>();
|
|
|
|
result.VirtualReadable.ShouldNotBeNull();
|
|
}
|
|
|
|
[Fact]
|
|
public void Compose_with_Historize_true_but_no_router_does_not_throw()
|
|
{
|
|
var scripts = new[] { ScriptRow("s1", "return 1.0f;") };
|
|
var vtags = new[] { VtRow("vt-hist", "s1", historize: true) };
|
|
|
|
// historyRouter = null — should still work, just no registration.
|
|
Should.NotThrow(() => Phase7EngineComposer.Compose(
|
|
scripts, vtags, [],
|
|
upstream: new CachedTagUpstreamSource(),
|
|
alarmStateStore: new InMemoryAlarmStateStore(),
|
|
historianSink: NullAlarmHistorianSink.Instance,
|
|
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
|
|
loggerFactory: NullLoggerFactory.Instance,
|
|
historyRouter: null));
|
|
}
|
|
|
|
[Fact]
|
|
public void Compose_with_Historize_true_router_already_registered_does_not_throw()
|
|
{
|
|
// Simulate a reload scenario where the prefix is already registered.
|
|
using var router = new HistoryRouter();
|
|
using var priorWriter = new RingBufferHistoryWriter();
|
|
router.Register(Phase7EngineComposer.VirtualTagHistoryPrefix, priorWriter);
|
|
|
|
var scripts = new[] { ScriptRow("s1", "return 1.0f;") };
|
|
var vtags = new[] { VtRow("vt-hist", "s1", historize: true) };
|
|
|
|
// Second compose call — should tolerate the duplicate without throwing.
|
|
Should.NotThrow(() => Phase7EngineComposer.Compose(
|
|
scripts, vtags, [],
|
|
upstream: new CachedTagUpstreamSource(),
|
|
alarmStateStore: new InMemoryAlarmStateStore(),
|
|
historianSink: NullAlarmHistorianSink.Instance,
|
|
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
|
|
loggerFactory: NullLoggerFactory.Instance,
|
|
historyRouter: router));
|
|
}
|
|
|
|
[Fact]
|
|
public void Compose_RingBufferHistoryWriter_is_in_disposables_list()
|
|
{
|
|
using var router = new HistoryRouter();
|
|
var scripts = new[] { ScriptRow("s1", "return 1.0f;") };
|
|
var vtags = new[] { VtRow("vt-hist", "s1", historize: true) };
|
|
|
|
var result = Phase7EngineComposer.Compose(
|
|
scripts, vtags, [],
|
|
upstream: new CachedTagUpstreamSource(),
|
|
alarmStateStore: new InMemoryAlarmStateStore(),
|
|
historianSink: NullAlarmHistorianSink.Instance,
|
|
rootScriptLogger: new LoggerConfiguration().CreateLogger(),
|
|
loggerFactory: NullLoggerFactory.Instance,
|
|
historyRouter: router);
|
|
|
|
// The RingBufferHistoryWriter must be tracked in Disposables so Phase7Composer.DisposeAsync
|
|
// clears the ring buffer on shutdown.
|
|
result.Disposables.ShouldContain(d => d is RingBufferHistoryWriter);
|
|
}
|
|
}
|