using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Server;
using Shouldly;
using Xunit;
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
using SdkHistoryReadResult = Opc.Ua.HistoryReadResult;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
///
/// Review R08:1.1 (#736): the server's raw cap (ServerHistorian:MaxRawValuesPerRead) bounds the Gateway
/// backend's max_values independently of the node manager. An uncapped raw read (NumValuesPerNode = 0)
/// used to be sent as and materialise the whole window twice (the protobuf list and the
/// snapshot list); a read above the cap now asks for cap + 1 rows and is REFUSED (,
/// BadTooManyOperations) when the window holds more — never answered truncated, because a short answer is what the
/// server's paging reads as "the last page". The data source reports the cap as its
/// (or a strict gateway's lower learned ceiling), so the node
/// manager's existing ceiling paging serves NumValuesPerNode = 0 as pages of at most the cap, each with a
/// continuation point.
///
public sealed class GatewayRawCapTests : IDisposable
{
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private static readonly DateTime T0 = new(2026, 10, 22, 10, 0, 0, DateTimeKind.Utc);
private const int Cap = 1000;
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-gwrawcap-{Guid.NewGuid():N}");
private static DateTime At(int seconds) => T0.AddSeconds(seconds);
private static HistorianSample Row(DateTime t, double v, uint quality = 0) =>
new() { Tag = "T", Timestamp = Timestamp.FromDateTime(t), NumericValue = v, OpcQuality = 192, Quality = quality };
/// AVEVA-shaped: a boundary row at the query start, [start, end] inclusive, capped at max.
private static FakeHistorianGatewayClient Store(int count, int? ceiling = null)
{
var stored = Enumerable.Range(0, count).Select(i => (T: At(i), V: (double)i)).ToList();
return new FakeHistorianGatewayClient
{
MaxValuesCeiling = ceiling,
RawSource = (start, end, max) =>
{
if (max <= 0) throw new ArgumentOutOfRangeException(nameof(max), "max_values must be positive"); // the client package
var rows = new List();
if (stored.All(s => s.T != start))
{
var prior = stored.LastOrDefault(s => s.T < start);
rows.Add(prior.T == default
? new HistorianSample { Tag = "T", Timestamp = Timestamp.FromDateTime(start), NumericValue = 0, OpcQuality = 0, Quality = 1 }
: Row(start, prior.V, 133));
}
rows.AddRange(stored.Where(s => s.T >= start && s.T <= end).Select(s => Row(s.T, s.V)));
return rows.Take(max);
},
};
}
private static GatewayHistorianDataSource Source(FakeHistorianGatewayClient fake, int cap = Cap) =>
new(fake, NullLogger.Instance, maxRawValuesPerRead: cap);
private static int Probe => Cap + 1 + RawAggregateReader.PreWindowAllowance; // cap + 1 rows + the boundary-row allowance
// ---- the data source alone ----------------------------------------------------------------------------------------
[Fact]
public void Without_a_cap_the_source_is_unchanged()
{
new GatewayHistorianDataSource(Store(10), NullLogger.Instance).MaxRawValuesPerRead.ShouldBe(0);
Source(Store(10), cap: 0).MaxRawValuesPerRead.ShouldBe(0);
}
[Fact]
public async Task The_cap_is_the_page_limit_until_a_lower_gateway_ceiling_is_learned()
{
Source(Store(3600)).MaxRawValuesPerRead.ShouldBe(Cap);
// A strict gateway whose ceiling is ABOVE the cap: the cap wins, nothing is refused by the gateway.
var higher = Store(3600, ceiling: 2000);
var ds = Source(higher);
await Should.ThrowAsync(() => ds.ReadRawAsync("T", At(0), At(3600), 0, Ct));
higher.RawCalls.ShouldAllBe(c => c.MaxValues <= 2000);
ds.MaxRawValuesPerRead.ShouldBe(Cap);
// A strict gateway whose ceiling is BELOW the cap: the learned ceiling (less the allowance) wins once known.
var lower = Store(3600, ceiling: 500);
var ds2 = Source(lower);
ds2.MaxRawValuesPerRead.ShouldBe(Cap);
await Should.ThrowAsync(() => ds2.ReadRawAsync("T", At(0), At(3600), 0, Ct));
ds2.MaxRawValuesPerRead.ShouldBe(500 - RawAggregateReader.PreWindowAllowance);
}
[Fact]
public async Task An_uncapped_read_over_the_cap_is_refused_never_truncated_and_never_asks_for_everything()
{
var fake = Store(3600);
var ds = Source(fake);
var ex = await Should.ThrowAsync(() => ds.ReadRawAsync("T", At(0), At(3600), 0, Ct));
ex.Limit.ShouldBe(Cap);
ex.Message.ShouldContain("MaxRawValuesPerRead");
fake.RawCalls.Select(c => c.MaxValues).ShouldBe([Probe]); // never int.MaxValue
fake.RawStreamsCompleted.ShouldBe(1, "the bounded stream is drained, never abandoned");
ds.GetHealthSnapshot().TotalFailures.ShouldBe(0, "a client-request limit is not a backend fault");
}
[Theory]
[InlineData(10)]
[InlineData(Cap - 1)]
[InlineData(Cap)]
public async Task An_uncapped_read_that_fits_the_cap_is_served_in_full(int count)
{
var fake = Store(count);
var r = await Source(fake).ReadRawAsync("T", At(0), At(3600), 0, Ct);
r.Samples.Select(s => s.Value).ShouldBe(Enumerable.Range(0, count).Select(i => (object?)(double)i));
fake.RawCalls.Select(c => c.MaxValues).ShouldBe([Probe]);
}
[Theory]
[InlineData(Cap + 1u)]
[InlineData(5000u)]
[InlineData(uint.MaxValue)]
public async Task A_client_cap_above_the_server_cap_is_bounded_by_it(uint n)
{
await Should.ThrowAsync(() => Source(Store(3600)).ReadRawAsync("T", At(0), At(3600), n, Ct));
var fake = Store(900);
var r = await Source(fake).ReadRawAsync("T", At(0), At(3600), n, Ct);
r.Samples.Count.ShouldBe(900);
fake.RawCalls.ShouldAllBe(c => c.MaxValues == Probe);
}
[Theory]
[InlineData(Cap)]
[InlineData(Cap - 1)]
[InlineData(1)]
public async Task A_client_cap_at_or_under_the_server_cap_reads_as_before(int n)
{
var fake = Store(3600);
var r = await Source(fake).ReadRawAsync("T", At(0), At(3600), (uint)n, Ct);
r.Samples.Count.ShouldBe(n);
fake.RawCalls.Select(c => c.MaxValues).ShouldBe([n + RawAggregateReader.PreWindowAllowance]);
}
[Fact]
public async Task A_point_read_and_an_empty_window_are_unaffected()
{
var fake = Store(3600);
var ds = Source(fake);
(await ds.ReadRawAsync("T", At(5), At(5), 0, Ct)).Samples.Single().Value.ShouldBe(5d);
(await ds.ReadRawAsync("T", At(5000), At(6000), 0, Ct)).Samples.ShouldBeEmpty();
fake.RawCalls.ShouldAllBe(c => c.MaxValues == Probe);
}
// ---- the factory binding --------------------------------------------------------------------------------------------
[Theory]
[InlineData(777, 777)]
[InlineData(0, 0)]
[InlineData(-1, 0)]
[InlineData(1, ServerHistorianOptions.MinMaxRawValuesPerRead)]
public void The_factory_binds_the_effective_option(int configured, int expected)
{
var options = new ServerHistorianOptions
{
Endpoint = "http://localhost:5222", UseTls = false, ApiKey = "histgw_test_secret", MaxRawValuesPerRead = configured,
};
using var ds = GatewayHistorian.CreateDataSource(options, new ServiceCollection().BuildServiceProvider());
ds.ShouldBeOfType().MaxRawValuesPerRead.ShouldBe(expected);
}
[Fact]
public void The_default_option_caps_the_source_at_ten_thousand()
{
var options = new ServerHistorianOptions { Endpoint = "http://localhost:5222", UseTls = false, ApiKey = "histgw_test_secret" };
using var ds = new GatewayHistorianProvider(options).CreateDataSource(new ServiceCollection().BuildServiceProvider());
ds.MaxRawValuesPerRead.ShouldBe(ServerHistorianOptions.DefaultMaxRawValuesPerRead);
}
// ---- through the node manager ---------------------------------------------------------------------------------------
[Theory]
[InlineData(0u, false)]
[InlineData(0u, true)]
[InlineData(5000u, false)]
[InlineData(uint.MaxValue, false)]
public async Task An_uncapped_HistoryRead_pages_at_the_cap_with_continuation_points(uint n, bool reverse)
{
var fake = Store(3600);
var (host, nm, nodeId) = await BootAsync(fake);
try
{
var (values, pages, status) = reverse
? DrainFrom(nm, nodeId, At(3600), At(-1), n, null)
: DrainFrom(nm, nodeId, At(0), At(3600), n, null);
status.Code.ShouldBe(StatusCodes.Good);
var expected = Enumerable.Range(0, 3600).Select(i => (double)i);
values.ShouldBe(reverse ? expected.Reverse() : expected);
pages.Count.ShouldBeGreaterThan(1, "a year-sized window at N = 0 is paged, not one shot");
pages.ShouldAllBe(p => p > 0 && p <= Cap);
pages[0].ShouldBe(Cap - 1); // page + lookahead fills a read
fake.RawCalls.ShouldAllBe(c => c.MaxValues > 0 && c.MaxValues <= Cap + RawAggregateReader.PreWindowAllowance);
}
finally { await host.DisposeAsync(); }
}
[Fact]
public async Task A_bounded_uncapped_HistoryRead_pages_within_the_cap()
{
var fake = Store(3600);
var (host, nm, nodeId) = await BootAsync(fake);
try
{
var (values, pages, status) = DrainFrom(nm, nodeId, At(10).AddMilliseconds(500), At(3000).AddMilliseconds(500), 0, null, returnBounds: true);
status.Code.ShouldBe(StatusCodes.Good);
values.ShouldBe(Enumerable.Range(10, 2992).Select(i => (double)i)); // start bound +10, data +11..+3000, end bound +3001
pages.ShouldAllBe(p => p <= Cap);
fake.RawCalls.ShouldAllBe(c => c.MaxValues > 0 && c.MaxValues <= Cap + RawAggregateReader.PreWindowAllowance);
}
finally { await host.DisposeAsync(); }
}
// ---- helpers ---------------------------------------------------------------------------------------------------------
private async Task<(OpcUaApplicationHost Host, OtOpcUaNodeManager Nm, NodeId NodeId)> BootAsync(FakeHistorianGatewayClient fake)
{
var started = await OpcUaHostStart.StartAsync(
port => new OpcUaApplicationHost(
new OpcUaApplicationHostOptions
{
ApplicationName = "OtOpcUa.GatewayRawCapTest",
ApplicationUri = $"urn:OtOpcUa.GatewayRawCapTest:{Guid.NewGuid():N}",
OpcUaPort = port,
PublicHostname = "localhost",
PkiStoreRoot = _pkiRoot,
},
NullLogger.Instance),
() => new OtOpcUaSdkServer(), Ct);
var nm = started.Server.NodeManager!;
nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
nm.HistorianDataSource = Source(fake);
nm.EnsureVariable("eq-1/pv", parentFolderNodeId: null, displayName: "PV", dataType: "Double",
writable: false, historianTagname: "T", realm: AddressSpaceRealm.Uns);
return (started.Host, nm, nm.TryGetVariable("eq-1/pv")!.NodeId);
}
private static (List Values, List Pages, StatusCode Status) DrainFrom(
OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max, byte[]? cp, int maxPages = 20,
bool returnBounds = false)
{
var values = new List();
var pages = new List();
ServiceResult error;
do
{
(var page, error, cp) = ReadRaw(nm, nodeId, start, end, max, cp, returnBounds);
values.AddRange(page);
pages.Add(page.Count);
pages.Count.ShouldBeLessThanOrEqualTo(maxPages, "paging must terminate");
}
while (cp is not null && StatusCode.IsGood(error.StatusCode));
return (values, pages, error.StatusCode);
}
private static (IReadOnlyList Values, ServiceResult Error, byte[]? Cp) ReadRaw(
OtOpcUaNodeManager nm, NodeId nodeId, DateTime start, DateTime end, uint max, byte[]? inboundCp, bool returnBounds = false)
{
var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
var details = new ReadRawModifiedDetails
{
ReturnBounds = returnBounds, IsReadModified = false, StartTime = start, EndTime = end, NumValuesPerNode = max,
};
var nodesToRead = new List { new() { NodeId = nodeId, ContinuationPoint = inboundCp } };
var results = new List { null! };
var errors = new List { null! };
nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors);
var values = new List();
if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
foreach (var dv in ((HistoryData)ExtensionObject.ToEncodeable(ho)).DataValues)
values.Add(Convert.ToDouble(dv.Value));
return (values, errors[0], results[0]?.ContinuationPoint);
}
public void Dispose()
{
if (Directory.Exists(_pkiRoot))
{
try { Directory.Delete(_pkiRoot, recursive: true); }
catch { /* best-effort */ }
}
}
}