Merge ph-T13 (pluggable historian)
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
+291
@@ -0,0 +1,291 @@
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
using CoreHistoryReadResult = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
|
||||
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// OPC UA HistoryUpdate over a real SDK client session (pluggable historian, Task 13): the encoding and
|
||||
/// dispatch the node-manager unit tests drive around. An in-memory updater and data source share ONE
|
||||
/// dictionary, so an insert is observable through a wire HistoryRead.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The updater is installed BEFORE the historized node materialises — the HistoryWrite AccessLevel bit
|
||||
/// (which the SDK base requires to admit a node to HistoryUpdate) is fixed at creation. The control test
|
||||
/// pins the reverse order.
|
||||
/// </remarks>
|
||||
public sealed class HistoryUpdateOverTheWireTests
|
||||
{
|
||||
private const string Tagname = "WW.Speed";
|
||||
private const NodePermissions Editor =
|
||||
NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe
|
||||
| NodePermissions.HistoryRead | NodePermissions.HistoryUpdate;
|
||||
private const NodePermissions ReaderOnly =
|
||||
NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe | NodePermissions.HistoryRead;
|
||||
|
||||
private static readonly DateTime T1 = new(2026, 10, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
private static readonly DateTime T2 = T1.AddMinutes(1);
|
||||
|
||||
/// <summary>Insert x2, read back, duplicate insert refused, DeleteAtTime one, read back one.</summary>
|
||||
[Fact]
|
||||
public async Task Insert_read_duplicate_and_delete_round_trip_over_the_wire()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var store = new SharedStore();
|
||||
var rig = await StartAsync("HuWireRoundTrip", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Nm.HistorianDataUpdater = store;
|
||||
rig.Nm.HistorianDataSource = store;
|
||||
rig.Enforce(true, [Key()], ClusterGrant(KeyPrincipal, "acl-editor", Editor));
|
||||
var id = rig.EnsureRawVariable("Speed", "Double", writable: false, Tagname);
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var node = RawNode(session, id);
|
||||
|
||||
var insert = await UpdateAsync(session, node, PerformUpdateType.Insert, ct, (T1, 1.5), (T2, 2.5));
|
||||
insert.StatusCode.Code.ShouldBe(StatusCodes.Good);
|
||||
insert.OperationResults.Select(c => c.Code).ShouldBe([StatusCodes.GoodEntryInserted, StatusCodes.GoodEntryInserted]);
|
||||
|
||||
var read = await ReadRawAsync(session, node, ct);
|
||||
read.Select(v => (v.SourceTimestamp, (double)v.Value)).ShouldBe([(T1, 1.5), (T2, 2.5)]);
|
||||
|
||||
var dup = await UpdateAsync(session, node, PerformUpdateType.Insert, ct, (T1, 9.9));
|
||||
dup.OperationResults.Select(c => c.Code).ShouldBe([StatusCodes.BadEntryExists]);
|
||||
(await ReadRawAsync(session, node, ct)).Count.ShouldBe(2, "the refused insert changed nothing");
|
||||
|
||||
var del = await DeleteAtTimeAsync(session, node, ct, T1);
|
||||
del.OperationResults.Select(c => c.Code).ShouldBe([StatusCodes.Good]);
|
||||
|
||||
var after = await ReadRawAsync(session, node, ct);
|
||||
after.Select(v => (v.SourceTimestamp, (double)v.Value)).ShouldBe([(T2, 2.5)]);
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ALLOW beside DENY: the same server, same node — the editor session is admitted, the session
|
||||
/// without the HistoryUpdate grant gets BadUserAccessDenied and the updater is never called.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_session_without_the_grant_is_denied_and_the_updater_is_not_called()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var store = new SharedStore();
|
||||
var rig = await StartAsync("HuWireDeny", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Nm.HistorianDataUpdater = store;
|
||||
rig.Nm.HistorianDataSource = store;
|
||||
rig.Enforce(
|
||||
true, [Key()],
|
||||
ClusterGrant(KeyPrincipal, "acl-editor", Editor),
|
||||
ClusterGrant(OpcUaPrincipals.Anonymous, "acl-anon", ReaderOnly));
|
||||
var id = rig.EnsureRawVariable("Speed", "Double", writable: false, Tagname);
|
||||
|
||||
using var denied = await rig.ConnectAsync(anonymous: true, ct);
|
||||
var denial = await UpdateAsync(denied, RawNode(denied, id), PerformUpdateType.Insert, ct, (T1, 1.0));
|
||||
denial.StatusCode.Code.ShouldBe(StatusCodes.BadUserAccessDenied);
|
||||
store.UpdateCalls.ShouldBe(0, "a denied session never reaches the updater");
|
||||
store.Count.ShouldBe(0);
|
||||
|
||||
using var allowed = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var ok = await UpdateAsync(allowed, RawNode(allowed, id), PerformUpdateType.Insert, ct, (T1, 1.0));
|
||||
ok.OperationResults.Select(c => c.Code).ShouldBe([StatusCodes.GoodEntryInserted]);
|
||||
store.UpdateCalls.ShouldBe(1);
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>The HistoryWrite bit is visible on both AccessLevel and UserAccessLevel when an updater is wired.</summary>
|
||||
[Fact]
|
||||
public async Task The_history_write_bit_is_visible_over_the_wire()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync("HuWireBit", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Nm.HistorianDataUpdater = new SharedStore();
|
||||
rig.Enforce(true, [Key()], ClusterGrant(KeyPrincipal, "acl-editor", Editor));
|
||||
var id = rig.EnsureRawVariable("Speed", "Double", writable: false, Tagname);
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var node = RawNode(session, id);
|
||||
(await ReadByteAsync(session, node, Attributes.AccessLevel, ct) & AccessLevels.HistoryWrite)
|
||||
.ShouldBe(AccessLevels.HistoryWrite);
|
||||
(await ReadByteAsync(session, node, Attributes.UserAccessLevel, ct) & AccessLevels.HistoryWrite)
|
||||
.ShouldBe(AccessLevels.HistoryWrite);
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Control (no retro-update): materialising the node BEFORE the updater is installed leaves it without
|
||||
/// the HistoryWrite bit, so the SDK base answers BadHistoryOperationUnsupported — even for an editor.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task An_updater_installed_after_the_node_materialised_is_unsupported()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var store = new SharedStore();
|
||||
var rig = await StartAsync("HuWireLate", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(true, [Key()], ClusterGrant(KeyPrincipal, "acl-editor", Editor));
|
||||
var id = rig.EnsureRawVariable("Speed", "Double", writable: false, Tagname);
|
||||
rig.Nm.HistorianDataUpdater = store;
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var node = RawNode(session, id);
|
||||
(await ReadByteAsync(session, node, Attributes.AccessLevel, ct) & AccessLevels.HistoryWrite)
|
||||
.ShouldBe(0, "the bit is fixed at materialisation");
|
||||
|
||||
var result = await UpdateAsync(session, node, PerformUpdateType.Insert, ct, (T1, 1.0));
|
||||
result.StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
|
||||
store.UpdateCalls.ShouldBe(0);
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
// ───────────────────────────── helpers ─────────────────────────────
|
||||
|
||||
private static NodeId RawNode(ISession s, string id) =>
|
||||
new(id, (ushort)s.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri));
|
||||
|
||||
private static async Task<byte> ReadByteAsync(ISession session, NodeId node, uint attribute, CancellationToken ct)
|
||||
{
|
||||
var response = await session.ReadAsync(null, 0, TimestampsToReturn.Neither, new ReadValueIdCollection
|
||||
{
|
||||
new ReadValueId { NodeId = node, AttributeId = attribute },
|
||||
}, ct);
|
||||
return (byte)response.Results.ShouldHaveSingleItem().Value;
|
||||
}
|
||||
|
||||
private static async Task<HistoryUpdateResult> UpdateAsync(
|
||||
ISession session, NodeId node, PerformUpdateType mode, CancellationToken ct,
|
||||
params (DateTime Time, double Value)[] items)
|
||||
{
|
||||
var details = new UpdateDataDetails { NodeId = node, PerformInsertReplace = mode };
|
||||
foreach (var (time, value) in items)
|
||||
details.UpdateValues.Add(new DataValue(new Variant(value), StatusCodes.Good, time, time));
|
||||
return await SendAsync(session, details, ct);
|
||||
}
|
||||
|
||||
private static Task<HistoryUpdateResult> DeleteAtTimeAsync(
|
||||
ISession session, NodeId node, CancellationToken ct, params DateTime[] times)
|
||||
{
|
||||
var details = new DeleteAtTimeDetails { NodeId = node };
|
||||
foreach (var t in times) details.ReqTimes.Add(t);
|
||||
return SendAsync(session, details, ct);
|
||||
}
|
||||
|
||||
private static async Task<HistoryUpdateResult> SendAsync(ISession session, object details, CancellationToken ct)
|
||||
{
|
||||
var response = await session.HistoryUpdateAsync(
|
||||
null, new ExtensionObjectCollection { new ExtensionObject(details) }, ct);
|
||||
return response.Results.ShouldHaveSingleItem();
|
||||
}
|
||||
|
||||
private static async Task<List<DataValue>> ReadRawAsync(ISession session, NodeId node, CancellationToken ct)
|
||||
{
|
||||
var details = new ReadRawModifiedDetails
|
||||
{
|
||||
StartTime = T1.AddHours(-1),
|
||||
EndTime = T1.AddHours(1),
|
||||
NumValuesPerNode = 100,
|
||||
IsReadModified = false,
|
||||
ReturnBounds = false,
|
||||
};
|
||||
var response = await session.HistoryReadAsync(
|
||||
null, new ExtensionObject(details), TimestampsToReturn.Source, false,
|
||||
new HistoryReadValueIdCollection { new HistoryReadValueId { NodeId = node } }, ct);
|
||||
var result = response.Results.ShouldHaveSingleItem();
|
||||
StatusCode.IsBad(result.StatusCode).ShouldBeFalse($"HistoryRead failed: {result.StatusCode}");
|
||||
var data = (HistoryData)ExtensionObject.ToEncodeable(result.HistoryData);
|
||||
return data.DataValues.ToList();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// One dictionary behind both seams, thread-safe and completing on a pool thread (as a real backend
|
||||
/// would), so the node manager's blocking bridge is exercised the way production runs it.
|
||||
/// </summary>
|
||||
private sealed class SharedStore : IHistorianDataUpdater, IHistorianDataSource
|
||||
{
|
||||
private readonly object _gate = new();
|
||||
private readonly SortedDictionary<DateTime, double> _samples = [];
|
||||
private int _updateCalls;
|
||||
|
||||
public int UpdateCalls => Volatile.Read(ref _updateCalls);
|
||||
|
||||
public int Count { get { lock (_gate) return _samples.Count; } }
|
||||
|
||||
public async Task<IReadOnlyList<uint>> UpdateDataAsync(
|
||||
string tag, HistoryUpdateMode mode, IReadOnlyList<HistorizationValue> values, CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref _updateCalls);
|
||||
await Task.Delay(1, ct).ConfigureAwait(false);
|
||||
tag.ShouldBe(Tagname);
|
||||
mode.ShouldBe(HistoryUpdateMode.Insert);
|
||||
var codes = new List<uint>();
|
||||
lock (_gate)
|
||||
{
|
||||
foreach (var v in values)
|
||||
{
|
||||
var t = v.TimestampUtc!.Value;
|
||||
if (_samples.ContainsKey(t)) { codes.Add(StatusCodes.BadEntryExists); continue; }
|
||||
v.Value.TryGetDouble(out var d).ShouldBeTrue();
|
||||
_samples[t] = d;
|
||||
codes.Add(StatusCodes.GoodEntryInserted);
|
||||
}
|
||||
}
|
||||
return codes;
|
||||
}
|
||||
|
||||
public Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, CancellationToken ct) =>
|
||||
throw new NotSupportedException();
|
||||
|
||||
public async Task<IReadOnlyList<uint>> DeleteAtTimeAsync(
|
||||
string tag, IReadOnlyList<DateTime> timesUtc, CancellationToken ct)
|
||||
{
|
||||
await Task.Delay(1, ct).ConfigureAwait(false);
|
||||
lock (_gate)
|
||||
return timesUtc.Select(t => _samples.Remove(t) ? StatusCodes.Good : StatusCodes.BadNoEntryExists).ToList();
|
||||
}
|
||||
|
||||
public async Task<CoreHistoryReadResult> ReadRawAsync(
|
||||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken)
|
||||
{
|
||||
await Task.Delay(1, cancellationToken).ConfigureAwait(false);
|
||||
fullReference.ShouldBe(Tagname);
|
||||
lock (_gate)
|
||||
{
|
||||
var rows = _samples.Where(kv => kv.Key >= startUtc && kv.Key <= endUtc)
|
||||
.Select(kv => new DataValueSnapshot(kv.Value, StatusCodes.Good, kv.Key, kv.Key)).ToList();
|
||||
return new CoreHistoryReadResult(rows, null);
|
||||
}
|
||||
}
|
||||
|
||||
public Task<CoreHistoryReadResult> ReadProcessedAsync(
|
||||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, HistoryAggregateType aggregate,
|
||||
CancellationToken cancellationToken) => throw new NotSupportedException();
|
||||
|
||||
public Task<CoreHistoryReadResult> ReadAtTimeAsync(
|
||||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
|
||||
throw new NotSupportedException();
|
||||
|
||||
public Task<HistoricalEventsResult> ReadEventsAsync(
|
||||
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) =>
|
||||
throw new NotSupportedException();
|
||||
|
||||
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
||||
|
||||
public void Dispose() { }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user