From 32852382ccfe1d301f73ecfb945420653ad1f507 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 9 Oct 2026 13:07:19 -0400 Subject: [PATCH] test(historian): over-the-wire HistoryUpdate integration tests Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe --- .../HistoryUpdateOverTheWireTests.cs | 291 ++++++++++++++++++ 1 file changed, 291 insertions(+) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/HistoryUpdateOverTheWireTests.cs diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/HistoryUpdateOverTheWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/HistoryUpdateOverTheWireTests.cs new file mode 100644 index 00000000..d7075936 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/HistoryUpdateOverTheWireTests.cs @@ -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; + +/// +/// 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. +/// +/// +/// 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. +/// +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); + + /// Insert x2, read back, duplicate insert refused, DeleteAtTime one, read back one. + [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(); } + } + + /// + /// 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. + /// + [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(); } + } + + /// The HistoryWrite bit is visible on both AccessLevel and UserAccessLevel when an updater is wired. + [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(); } + } + + /// + /// 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. + /// + [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 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 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 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 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> 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(); + } + + /// + /// 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. + /// + private sealed class SharedStore : IHistorianDataUpdater, IHistorianDataSource + { + private readonly object _gate = new(); + private readonly SortedDictionary _samples = []; + private int _updateCalls; + + public int UpdateCalls => Volatile.Read(ref _updateCalls); + + public int Count { get { lock (_gate) return _samples.Count; } } + + public async Task> UpdateDataAsync( + string tag, HistoryUpdateMode mode, IReadOnlyList 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(); + 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 DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, CancellationToken ct) => + throw new NotSupportedException(); + + public async Task> DeleteAtTimeAsync( + string tag, IReadOnlyList 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 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 ReadProcessedAsync( + string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, HistoryAggregateType aggregate, + CancellationToken cancellationToken) => throw new NotSupportedException(); + + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) => + throw new NotSupportedException(); + + public Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) => + throw new NotSupportedException(); + + public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); + + public void Dispose() { } + } +}