using Opc.Ua; using Opc.Ua.Client; using Microsoft.Extensions.Logging.Abstractions; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Configuration.Entities; using ZB.MOM.WW.OtOpcUa.Configuration.Enums; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.Authorization; using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security; using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig; using CoreReadResult = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult; namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// Uses activated sessions and the production continuation store; denied resumes never reach a backend. public sealed class Sdk2HistoryAuthorityWireTests { private const string ValuePath = RawDeviceFolder + "/History"; private static readonly DateTime Epoch = new(2026, 10, 1, 0, 0, 0, DateTimeKind.Utc); public static TheoryData Cases => new( from kind in new[] { "raw", "modified", "events" } from change in new[] { "alias", "replace", "rebuild", "grant", "key", "expiry" } select (kind, change)); [Theory] [MemberData(nameof(Cases))] public async Task A_saved_cursor_requires_its_live_node_and_current_authority(string kind, string change) { var ct = TestContext.Current.CancellationToken; await using var rig = await StartAsync("HistoryAuthority", encrypted: true, ct); var source = new CountingSource(); var clock = new Clock(); var key = Key(); key.ExpiresUtc = clock.GetUtcNow().UtcDateTime.AddMinutes(10); var grant = ClusterGrant(KeyPrincipal, "history-allow", NodePermissions.ReadOnly); var state = rig.Enforce(true, [key], grant); rig.Nm.Authorizer = new DataPlaneAuthorizer(state, new TriePermissionEvaluator(state.Tries), true, NullLogger.Instance, clock); rig.Nm.HistorianDataSource = source; Materialise(rig); using var session = await rig.ConnectAsync(anonymous: false, ct, encrypted: true); var first = await PageAsync(session, kind, alias: false, default, ct); first.StatusCode.ShouldBe(StatusCodes.Good); first.ContinuationPoint.IsEmpty.ShouldBeFalse("the production session store must own a real next page"); source.Calls.ShouldBeGreaterThan(0); switch (change) { case "replace": if (kind == "events") rig.Nm.RemoveEquipmentSubtree(RawDeviceFolder, AddressSpaceRealm.Raw).ShouldBeTrue(); else rig.Nm.RemoveVariableNode(ValuePath, AddressSpaceRealm.Raw).ShouldBeTrue(); Materialise(rig); break; case "rebuild": rig.Nm.RebuildAddressSpace(); Materialise(rig); rig.Nm.CompleteAddressSpaceApply(); break; case "grant": Apply(state, key, []); break; case "key": var disabled = Key(); disabled.Enabled = false; Apply(state, disabled, [grant]); break; case "expiry": clock.Advance(TimeSpan.FromMinutes(20)); break; } var before = source.Calls; var denied = await PageAsync(session, kind, change == "alias", first.ContinuationPoint, ct); denied.StatusCode.ShouldBe(change is "grant" or "key" or "expiry" ? StatusCodes.BadUserAccessDenied : StatusCodes.BadContinuationPointInvalid); denied.HistoryData.IsNull.ShouldBeTrue("a denied resume must not disclose saved data"); denied.ContinuationPoint.IsEmpty.ShouldBeTrue(); source.Calls.ShouldBe(before, "permission and exact-node binding must be checked before backend dispatch"); // A fresh authorized request remains usable: this is not an always-deny fixture. clock.Reset(); Apply(state, key, [grant]); (await PageAsync(session, kind, alias: false, default, ct)).StatusCode.ShouldBe(StatusCodes.Good); source.Calls.ShouldBeGreaterThan(before); await session.CloseAsync(ct); } private static void Apply(DataPlaneAuthorityState state, ApiKey key, NodeAcl[] grants) => state.Apply(ServerEventRig.Cluster, Guid.NewGuid().ToString("N"), [key], grants, new Dictionary>()); private static void Materialise(ServerEventRig rig) { rig.MaterialiseNative(); rig.Sink.EnsureVariable(ValuePath, RawDeviceFolder, "History", "Double", false, AddressSpaceRealm.Raw, "same-backend-tag"); rig.Sink.EnsureVariable(ValuePath, EqA, "HistoryAlias", "Double", false, AddressSpaceRealm.Uns, "same-backend-tag"); } private static async Task PageAsync(ISession session, string kind, bool alias, ByteString point, CancellationToken ct) { var realm = alias ? V3NodeIds.UnsNamespaceUri : V3NodeIds.RawNamespaceUri; var path = kind == "events" ? alias ? EqA : RawDeviceFolder : ValuePath; var node = new NodeId(path, (ushort)session.NamespaceUris.GetIndex(realm)); HistoryReadDetails details = kind == "events" ? new ReadEventDetails { StartTime = Epoch, EndTime = Epoch.AddMinutes(1), NumValuesPerNode = 2, Filter = new EventFilter { SelectClauses = [new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, QualifiedName.From(BrowseNames.EventId))] }, } : new ReadRawModifiedDetails { StartTime = Epoch, EndTime = Epoch.AddMinutes(1), NumValuesPerNode = 2, IsReadModified = kind == "modified" }; var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, false, [new HistoryReadValueId { NodeId = node, ContinuationPoint = point }], ct); return response.Results.ToList().ShouldHaveSingleItem(); } private sealed class Clock : TimeProvider { private readonly DateTimeOffset _initial = DateTimeOffset.UtcNow; private TimeSpan _offset; public override DateTimeOffset GetUtcNow() => _initial + _offset; public void Advance(TimeSpan amount) => _offset += amount; public void Reset() => _offset = TimeSpan.Zero; } private sealed class CountingSource : IHistorianDataSource { private int _calls; public int Calls => Volatile.Read(ref _calls); private static DataValueSnapshot Value(int i) => new((double)i, OpcUaStatus.Good, Epoch.AddSeconds(i), Epoch.AddSeconds(i)); public Task ReadRawAsync(string reference, DateTime start, DateTime end, uint max, CancellationToken ct) { Interlocked.Increment(ref _calls); var rows = Enumerable.Range(0, 25).Select(Value).Where(x => x.SourceTimestampUtc >= start && x.SourceTimestampUtc < end).ToList(); var count = max == 0 ? rows.Count : checked((int)max); return Task.FromResult(new CoreReadResult(rows.Take(count).ToList(), rows.Count > count ? [1] : null)); } public Task ReadModifiedAsync(string reference, DateTime start, DateTime end, uint max, HistoryModifiedCursor? resume, CancellationToken ct) { Interlocked.Increment(ref _calls); var rows = Enumerable.Range(0, 25).Where(i => Epoch.AddSeconds(i) >= start && Epoch.AddSeconds(i) < end && (resume is null || i > resume.ModifiedId)) .Select(i => new HistoryModifiedValue(Value(i), Epoch.AddMinutes(2), HistoryModificationType.Replace, "fixture", i, Epoch.AddSeconds(i))).ToList(); var count = max == 0 ? rows.Count : checked((int)max); return Task.FromResult(new HistoryModifiedReadResult(rows.Take(count).ToList(), rows.Count > count)); } public Task ReadEventsAsync(string? source, DateTime start, DateTime end, int max, CancellationToken ct) { Interlocked.Increment(ref _calls); var rows = Enumerable.Range(0, 25).Select(i => new HistoricalEvent($"e{i:00}", "src", Epoch.AddSeconds(i), Epoch, "m", 500)) .Where(x => x.EventTimeUtc >= start && x.EventTimeUtc < end).ToList(); var count = max <= 0 ? rows.Count : max; return Task.FromResult(new HistoricalEventsResult(rows.Take(count).ToList(), rows.Count > count ? [1] : null)); } public Task ReadProcessedAsync(string r, DateTime s, DateTime e, TimeSpan interval, HistoryAggregateType a, CancellationToken ct) => NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, interval, a, ct); public Task ReadAtTimeAsync(string r, IReadOnlyList times, CancellationToken ct) => NullHistorianDataSource.Instance.ReadAtTimeAsync(r, times, ct); public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); public void Dispose() { } } }