using Opc.Ua; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers; namespace ZB.MOM.WW.OtOpcUa.Client.Shared.Tests.Helpers; /// /// Pins the client-side HistoryRead paging loop shared by the raw and modified reads: continuation pages are read /// (never released) while the caller wants more, and a read stopped by the caller's own cap releases the server cursor. /// public class HistoryReadPagingTests { private static readonly NodeId Node = new("t", 2); /// One recorded call to the page reader. private sealed record Call(byte[]? ContinuationPoint, bool Release); /// A scripted server: each read returns the next page; a release returns an empty Good result. private sealed class ScriptedServer { private readonly Queue<(int Count, byte[]? Cp)> _pages; public List Calls { get; } = []; public ScriptedServer(params (int Count, byte[]? Cp)[] pages) => _pages = new Queue<(int, byte[]?)>(pages); public Task ReadPageAsync(byte[]? cp, bool release, CancellationToken ct) { Calls.Add(new Call(cp, release)); if (release) return Task.FromResult(new HistoryReadResult { StatusCode = StatusCodes.Good }); var (count, next) = _pages.Dequeue(); var data = new HistoryData(); for (var i = 0; i < count; i++) data.DataValues.Add(new DataValue(new Variant(i))); return Task.FromResult(new HistoryReadResult { StatusCode = StatusCodes.Good, ContinuationPoint = next, HistoryData = new ExtensionObject(data) }); } } private static IEnumerable Values(HistoryReadResult r) => r.HistoryData?.Body is HistoryData d ? d.DataValues : []; /// /// A server page limit below the caller's cap hands back a continuation point: the follow-up must READ the next /// page. Releasing it instead (the old raw path passed release = cp != null) silently truncated every raw read /// to its first server page. /// [Fact] public async Task Continuation_pages_are_read_not_released_until_the_server_finishes() { var server = new ScriptedServer((3, [1]), (3, [2]), (2, null)); var values = await HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, maxValues: 100, Node, CancellationToken.None); values.Count.ShouldBe(8); server.Calls.Select(c => c.Release).ShouldAllBe(r => r == false); server.Calls.Select(c => c.ContinuationPoint?.FirstOrDefault()).ShouldBe(new byte?[] { null, 1, 2 }); } /// A read stopped by the caller's cap while the server still holds a cursor releases that cursor. [Fact] public async Task A_read_stopped_by_the_cap_releases_the_outstanding_continuation_point() { var server = new ScriptedServer((5, [7])); var values = await HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, maxValues: 5, Node, CancellationToken.None); values.Count.ShouldBe(5); server.Calls.Count.ShouldBe(2); server.Calls[1].Release.ShouldBeTrue(); server.Calls[1].ContinuationPoint.ShouldBe(new byte[] { 7 }); } /// /// A read the server finished (no continuation point) has nothing to release — and the positive control above /// proves a release would have been recorded had one been owed. /// [Fact] public async Task A_read_the_server_finished_issues_no_release() { var server = new ScriptedServer((5, null)); var values = await HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, maxValues: 5, Node, CancellationToken.None); values.Count.ShouldBe(5); server.Calls.ShouldHaveSingleItem().Release.ShouldBeFalse(); } /// /// Review m3: a non-positive cap is refused. An uncapped client read would materialise a tag's whole window /// (the client side of R08 1.1), and (uint)-1 on the wire asked the server for everything. /// /// The cap. [Theory] [InlineData(0)] [InlineData(-1)] public async Task A_non_positive_cap_is_refused_before_any_read(int max) { var server = new ScriptedServer(); await Should.ThrowAsync(() => HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, max, Node, CancellationToken.None)); server.Calls.ShouldBeEmpty(); } /// Review m2: server pages that overshoot the cap are trimmed to it (60 + 60 with a cap of 100 is 100). [Fact] public async Task The_result_is_trimmed_to_the_cap() { var server = new ScriptedServer((60, [1]), (60, [2])); var values = await HistoryReadPaging.ReadAllAsync(server.ReadPageAsync, Values, maxValues: 100, Node, CancellationToken.None); values.Count.ShouldBe(100); server.Calls.Last().Release.ShouldBeTrue(); // the outstanding cursor is still released } /// /// Review m1: a read cancelled between pages releases the outstanding cursor (best effort, not on the caller's /// cancelled token) and still surfaces the cancellation. Positive control: the release IS recorded. /// [Fact] public async Task A_read_cancelled_mid_paging_releases_the_cursor_and_rethrows() { using var cts = new CancellationTokenSource(); var calls = new List<(byte[]? Cp, bool Release, bool TokenCancelled)>(); Task Read(byte[]? cp, bool release, CancellationToken ct) { calls.Add((cp, release, ct.IsCancellationRequested)); if (release) return Task.FromResult(new HistoryReadResult { StatusCode = StatusCodes.Good }); if (cp != null) { cts.Cancel(); ct.ThrowIfCancellationRequested(); } var data = new HistoryData(); data.DataValues.Add(new DataValue(new Variant(1))); return Task.FromResult(new HistoryReadResult { StatusCode = StatusCodes.Good, ContinuationPoint = [5], HistoryData = new ExtensionObject(data) }); } await Should.ThrowAsync(() => HistoryReadPaging.ReadAllAsync(Read, Values, maxValues: 10, Node, cts.Token)); var release = calls.Where(c => c.Release).ShouldHaveSingleItem(); release.Cp.ShouldBe(new byte[] { 5 }); release.TokenCancelled.ShouldBeFalse(); } /// A cancellation that lands while the cap release is in flight does not lose the values read. [Fact] public async Task The_cap_release_does_not_ride_the_callers_token() { using var cts = new CancellationTokenSource(); var releaseSawCancelled = (bool?)null; Task Read(byte[]? cp, bool release, CancellationToken ct) { if (release) { releaseSawCancelled = ct.IsCancellationRequested; return Task.FromResult(new HistoryReadResult { StatusCode = StatusCodes.Good }); } var data = new HistoryData(); data.DataValues.Add(new DataValue(new Variant(1))); cts.Cancel(); // the caller gives up right as the capped page arrives return Task.FromResult(new HistoryReadResult { StatusCode = StatusCodes.Good, ContinuationPoint = [5], HistoryData = new ExtensionObject(data) }); } var values = await HistoryReadPaging.ReadAllAsync(Read, Values, maxValues: 1, Node, cts.Token); values.Count.ShouldBe(1); releaseSawCancelled.ShouldBe(false); } /// A failed release is best effort: the values already read are still returned. [Fact] public async Task A_failed_release_does_not_lose_the_values_read() { var calls = 0; Task Read(byte[]? cp, bool release, CancellationToken ct) { calls++; if (release) throw new ServiceResultException(StatusCodes.BadSessionClosed); var data = new HistoryData(); data.DataValues.Add(new DataValue(new Variant(1))); return Task.FromResult(new HistoryReadResult { StatusCode = StatusCodes.Good, ContinuationPoint = [9], HistoryData = new ExtensionObject(data) }); } var values = await HistoryReadPaging.ReadAllAsync(Read, Values, maxValues: 1, Node, CancellationToken.None); values.Count.ShouldBe(1); calls.ShouldBe(2); } /// A refused page still surfaces as an exception, not an empty read. [Fact] public async Task A_refused_page_throws() { Task Read(byte[]? cp, bool release, CancellationToken ct) => Task.FromResult(new HistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }); await Should.ThrowAsync(() => HistoryReadPaging.ReadAllAsync(Read, Values, maxValues: 10, Node, CancellationToken.None)); } }