/* Copyright (c) 2026 The OPC Foundation, Inc. All rights reserved. * Licensed under the MIT License. See LICENSE.txt in the project root. */ using System; using System.Collections.Generic; using System.IO; using System.Linq; using System.Threading; using System.Threading.Tasks; using Moq; using NUnit.Framework; using Opc.Ua.Tests; namespace Opc.Ua.Client.Tests { /// /// Bounded ranged reads are asserted at the Read service and destination boundaries. /// [TestFixture] [Category("Client")] [Parallelizable] public sealed class SessionByteStreamExtensionsTests { [SetUp] public void SetUp() { m_requests.Clear(); m_values.Clear(); m_context = ServiceMessageContext.Create(NUnitTelemetryContext.Create()); m_session = new Mock(MockBehavior.Strict); m_session.SetupGet(s => s.MessageContext).Returns(m_context); m_session.Setup(s => s.ReadAsync( It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>(), It.IsAny())) .Returns, CancellationToken>( (_, age, timestamps, requests, token) => { Assert.That(requests.Count, Is.EqualTo(1)); Assert.That(age, Is.Zero); Assert.That(timestamps, Is.EqualTo(TimestampsToReturn.Neither)); m_requests.Add((requests[0], token)); return new ValueTask(new ReadResponse { ResponseHeader = new ResponseHeader(), Results = [m_values.Dequeue()] }); }); } [TestCase(false)] [TestCase(true)] public async Task ReadBytesToAsyncAdvancesRangesAfterShortReadsUntilExplicitEndAsync(bool noData) { m_values.Enqueue(Value(10, 11)); m_values.Enqueue(Value(12)); m_values.Enqueue(Value(13, 14)); m_values.Enqueue(noData ? DataValue.FromStatusCode(StatusCodes.BadIndexRangeNoData) : Value()); using var destination = new MemoryStream(); using var cancellation = new CancellationTokenSource(); long count = await m_session.Object.ReadBytesToAsync( new NodeId("content", 3), destination, maxBytes: 5, chunkSize: 4, cancellation.Token).ConfigureAwait(false); Assert.Multiple(() => { Assert.That(count, Is.EqualTo(5)); Assert.That(destination.ToArray(), Is.EqualTo(new byte[] { 10, 11, 12, 13, 14 })); Assert.That(destination.CanWrite, Is.True); Assert.That(m_requests.Select(r => r.Request.IndexRange), Is.EqualTo(s_shortReadRanges)); Assert.That(m_requests.All(r => r.Request.NodeId == new NodeId("content", 3)), Is.True); Assert.That(m_requests.All(r => r.Request.AttributeId == Attributes.Value), Is.True); Assert.That(m_requests.All(r => r.Token == cancellation.Token), Is.True); }); } [TestCase(false)] [TestCase(true)] public async Task ReadBytesToAsyncEmptyInitialValueReturnsZeroWithoutAnotherReadAsync(bool noData) { m_values.Enqueue(noData ? DataValue.FromStatusCode(StatusCodes.BadIndexRangeNoData) : Value()); using var destination = new MemoryStream(); long count = await ReadAsync(destination, 1, 1).ConfigureAwait(false); Assert.That(count, Is.Zero); Assert.That(destination.Length, Is.Zero); Assert.That(destination.CanWrite, Is.True); Assert.That(m_requests.Select(r => r.Request.IndexRange), Is.EqualTo(s_singleByteRange)); } [Test] public async Task ReadBytesToAsyncRejectsCumulativeLimitBeforeWritingOffendingChunk() { m_values.Enqueue(Value(1, 2, 3)); m_values.Enqueue(Value(4, 5, 6)); using var destination = new MemoryStream(); await Assert.ThrowsAsync(async () => await ReadAsync(destination, maxBytes: 5, chunkSize: 4).ConfigureAwait(false)); Assert.That(destination.ToArray(), Is.EqualTo(new byte[] { 1, 2, 3 })); Assert.That(m_requests.Select(r => r.Request.IndexRange), Is.EqualTo(s_cumulativeRanges)); Assert.That(destination.CanWrite, Is.True); } [Test] public async Task ReadBytesToAsyncRejectsServerIgnoringRequestedRangeBeforeWriting() { m_values.Enqueue(Value(1, 2, 3, 4, 5)); using var destination = new MemoryStream(); await Assert.ThrowsAsync(async () => await ReadAsync(destination, maxBytes: 20, chunkSize: 4).ConfigureAwait(false)); Assert.That(destination.Length, Is.Zero); Assert.That(destination.CanWrite, Is.True); Assert.That(m_requests.Select(r => r.Request.IndexRange), Is.EqualTo(s_initialRange)); } [Test] public async Task ReadBytesToAsyncHonorsSessionByteStringCapAsync() { m_context.MaxByteStringLength = 2; m_values.Enqueue(Value(21, 22)); m_values.Enqueue(Value(23)); m_values.Enqueue(Value()); using var destination = new MemoryStream(); long count = await ReadAsync(destination, maxBytes: 3, chunkSize: 9).ConfigureAwait(false); Assert.That(count, Is.EqualTo(3)); Assert.That(destination.ToArray(), Is.EqualTo(new byte[] { 21, 22, 23 })); Assert.That(m_requests.Select(r => r.Request.IndexRange), Is.EqualTo(s_cappedRanges)); } [TestCase(0)] [TestCase(-1)] public async Task ReadBytesToAsyncNonPositiveSessionCapDoesNotShrinkRequestAsync(int cap) { m_context.MaxByteStringLength = cap; m_values.Enqueue(Value(31)); m_values.Enqueue(Value()); using var destination = new MemoryStream(); long count = await ReadAsync(destination, 1, 4).ConfigureAwait(false); Assert.That(count, Is.EqualTo(1)); Assert.That(destination.ToArray(), Is.EqualTo(new byte[] { 31 })); Assert.That(m_requests.Select(r => r.Request.IndexRange), Is.EqualTo(s_uncappedRanges)); } [TestCase("string")] [TestCase("integer")] public async Task ReadBytesToAsyncRejectsWrongValueType(string kind) { m_values.Enqueue(new DataValue(kind == "string" ? Variant.From("not bytes") : Variant.From(17))); using var destination = new MemoryStream(); ServiceResultException exception = (await Assert.ThrowsAsync( async () => await ReadAsync(destination, 8, 4).ConfigureAwait(false)))!; Assert.That(exception.StatusCode, Is.EqualTo(StatusCodes.BadTypeMismatch)); Assert.That(destination.Length, Is.Zero); Assert.That(destination.CanWrite, Is.True); Assert.That(m_requests, Has.Count.EqualTo(1)); } [TestCaseSource(nameof(s_badValueStatuses))] public async Task ReadBytesToAsyncPreservesBadValueStatusAndPreviouslyWrittenBytes(StatusCode status) { m_values.Enqueue(Value(41, 42)); m_values.Enqueue(DataValue.FromStatusCode(status)); using var destination = new MemoryStream(); ServiceResultException exception = (await Assert.ThrowsAsync( async () => await ReadAsync(destination, 8, 4).ConfigureAwait(false)))!; Assert.That(exception.StatusCode, Is.EqualTo(status)); Assert.That(destination.ToArray(), Is.EqualTo(new byte[] { 41, 42 })); Assert.That(destination.CanWrite, Is.True); Assert.That(m_requests, Has.Count.EqualTo(2)); } [TestCase(false)] [TestCase(true)] public async Task ReadBytesToAsyncRejectsMalformedResponse(bool diagnostics) { m_session.Setup(s => s.ReadAsync( It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>(), It.IsAny())) .Returns(new ValueTask(new ReadResponse { ResponseHeader = new ResponseHeader(), Results = diagnostics ? new[] { Value(1) }.ToArrayOf() : [], DiagnosticInfos = diagnostics ? new[] { new DiagnosticInfo(), new DiagnosticInfo() }.ToArrayOf() : [] })); using var destination = new MemoryStream(); ServiceResultException exception = (await Assert.ThrowsAsync( async () => await ReadAsync(destination, 8, 4).ConfigureAwait(false)))!; Assert.That(exception.StatusCode, Is.EqualTo(StatusCodes.BadUnexpectedError)); Assert.That(destination.Length, Is.Zero); Assert.That(destination.CanWrite, Is.True); VerifyReads(Times.Once()); } [Test] public async Task ReadBytesToAsyncPropagatesCancellationAndRetainsDestination() { using var destination = new MemoryStream(); using var cancellation = new CancellationTokenSource(); cancellation.Cancel(); var primary = new OperationCanceledException(cancellation.Token); m_session.Setup(s => s.ReadAsync( It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>(), cancellation.Token)) .Throws(primary); OperationCanceledException exception = (await Assert.ThrowsAsync( async () => await m_session.Object.ReadBytesToAsync( new NodeId("content", 3), destination, 8, 4, cancellation.Token).ConfigureAwait(false)))!; Assert.That(exception, Is.SameAs(primary)); Assert.That(destination.Length, Is.Zero); Assert.That(destination.CanWrite, Is.True); VerifyReads(Times.Once()); } [Test] public async Task ReadBytesToAsyncPreservesServiceExceptionAndRetainsDestination() { var primary = new ServiceResultException(StatusCodes.BadSessionClosed); using var destination = new MemoryStream(); m_session.Setup(s => s.ReadAsync( It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>(), It.IsAny())) .Throws(primary); ServiceResultException exception = (await Assert.ThrowsAsync( async () => await ReadAsync(destination, 8, 4).ConfigureAwait(false)))!; Assert.That(exception, Is.SameAs(primary)); Assert.That(destination.Length, Is.Zero); Assert.That(destination.CanWrite, Is.True); VerifyReads(Times.Once()); } [TestCase(0L, 4, "maxBytes")] [TestCase(-1L, 4, "maxBytes")] [TestCase(2147483648L, 4, "maxBytes")] [TestCase(8L, 0, "chunkSize")] [TestCase(8L, -1, "chunkSize")] [TestCase(8L, 1048577, "chunkSize")] public async Task ReadBytesToAsyncValidatesBoundsBeforeRead(long maxBytes, int chunkSize, string parameter) { using var destination = new MemoryStream(); ArgumentOutOfRangeException exception = (await Assert.ThrowsAsync( async () => await ReadAsync(destination, maxBytes, chunkSize).ConfigureAwait(false)))!; Assert.That(exception.ParamName, Is.EqualTo(parameter)); Assert.That(destination.Length, Is.Zero); Assert.That(destination.CanWrite, Is.True); VerifyReads(Times.Never()); } [TestCase(1L, 1)] [TestCase(2147483647L, 1048576)] public async Task ReadBytesToAsyncAcceptsLimitBoundariesAsync(long maxBytes, int chunkSize) { m_context.MaxByteStringLength = 0; m_values.Enqueue(Value(51)); m_values.Enqueue(Value()); using var destination = new MemoryStream(); long count = await ReadAsync(destination, maxBytes, chunkSize).ConfigureAwait(false); Assert.That(count, Is.EqualTo(1)); Assert.That(destination.ToArray(), Is.EqualTo(new byte[] { 51 })); Assert.That(m_requests[0].Request.IndexRange, Is.EqualTo(chunkSize == 1 ? "0:0" : "0:1048575")); } [Test] public async Task ReadBytesToAsyncRejectsReadOnlyDestinationWithoutRead() { using var destination = new MemoryStream([1, 2], writable: false); ArgumentException exception = (await Assert.ThrowsAsync( async () => await ReadAsync(destination, 8, 4).ConfigureAwait(false)))!; Assert.That(exception.ParamName, Is.EqualTo("destination")); Assert.That(destination.ToArray(), Is.EqualTo(new byte[] { 1, 2 })); Assert.That(destination.CanRead, Is.True); VerifyReads(Times.Never()); } [Test] public async Task ReadBytesToAsyncRejectsNullDestinationWithoutRead() { ArgumentNullException exception = (await Assert.ThrowsAsync( async () => await ReadAsync(null!, 8, 4).ConfigureAwait(false)))!; Assert.That(exception.ParamName, Is.EqualTo("destination")); VerifyReads(Times.Never()); } [Test] public async Task ReadBytesToAsyncRejectsNullSession() { using var destination = new MemoryStream(); ArgumentNullException exception = (await Assert.ThrowsAsync( async () => await SessionByteStreamExtensions.ReadBytesToAsync( null!, NodeId.Null, destination, 8).ConfigureAwait(false)))!; Assert.That(exception.ParamName, Is.EqualTo("session")); Assert.That(destination.Length, Is.Zero); } private Task ReadAsync(Stream destination, long maxBytes, int chunkSize) { return m_session.Object.ReadBytesToAsync( new NodeId("content", 3), destination, maxBytes, chunkSize).AsTask(); } private void VerifyReads(Times times) { m_session.Verify(s => s.ReadAsync( It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny>(), It.IsAny()), times); } private static DataValue Value(params byte[] bytes) { return new DataValue(Variant.From(ByteString.From(bytes))); } private static readonly StatusCode[] s_badValueStatuses = [ StatusCodes.BadUserAccessDenied, StatusCodes.BadNodeIdUnknown ]; private static readonly string[] s_shortReadRanges = ["0:3", "2:5", "3:6", "5:8"]; private static readonly string[] s_singleByteRange = ["0:0"]; private static readonly string[] s_cumulativeRanges = ["0:3", "3:6"]; private static readonly string[] s_initialRange = ["0:3"]; private static readonly string[] s_cappedRanges = ["0:1", "2:3", "3:4"]; private static readonly string[] s_uncappedRanges = ["0:3", "1:4"]; private ServiceMessageContext m_context = null!; private Mock m_session = null!; private readonly Queue m_values = new(); private readonly List<(ReadValueId Request, CancellationToken Token)> m_requests = []; } }