test(opcuaclient): HistoryRead Raw/AtTime/Processed wire tests against in-process server (#622)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-08 16:52:59 -04:00
parent a8b07ae12d
commit dcc8adcf48
3 changed files with 252 additions and 33 deletions
@@ -0,0 +1,200 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;
+
+namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
+
+/// <summary>
+///     HistoryRead leg of the OpcUaClient driver's wire coverage (Gitea #622): the real
+///     <see cref="OpcUaClientDriver"/> issues Raw / Processed / AtTime reads over loopback to the in-process
+///     OtOpcUa server, whose node manager dispatches to an injected deterministic
+///     <see cref="IHistorianDataSource"/> (<see cref="FakeHistorian"/>) behind a historized raw variable.
+/// </summary>
+[Trait("Category", "Integration")]
+public sealed class OpcUaClientDriverHistoryWireTests
+{
+    private const string HistRef = "Sim/dev1/Hist";
+    private const string HistTag = "Plant.Hist.Series";
+    private const uint Uncertain = 0x40000000u;
+
+    private static readonly DateTime T0 = new(2026, 1, 1, 12, 0, 0, DateTimeKind.Utc);
+
+    private static DateTime At(int seconds) => T0.AddSeconds(seconds);
+
+    /// <summary>Five samples, one per second, value = 1..5; the third is Uncertain.</summary>
+    private static readonly DataValueSnapshot[] Series =
+    [
+        new(1.0, 0u, At(0), At(0)),
+        new(2.0, 0u, At(1), At(1)),
+        new(3.0, Uncertain, At(2), At(2)),
+        new(4.0, 0u, At(3), At(3)),
+        new(5.0, 0u, At(4), At(4)),
+    ];
+
+    private sealed class FakeHistorian : IHistorianDataSource
+    {
+        public List<string> Calls { get; } = [];
+
+        public Task<HistoryReadResult> ReadRawAsync(
+            string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken ct)
+        {
+            lock (Calls) Calls.Add($"raw:{fullReference}");
+            var hits = Series.Where(s => s.SourceTimestampUtc >= startUtc && s.SourceTimestampUtc < endUtc);
+            if (maxValuesPerNode > 0) hits = hits.Take((int)maxValuesPerNode);
+            return Task.FromResult(new HistoryReadResult([.. hits], null));
+        }
+
+        public Task<HistoryReadResult> ReadProcessedAsync(
+            string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
+            HistoryAggregateType aggregate, CancellationToken ct)
+        {
+            lock (Calls) Calls.Add($"processed:{fullReference}:{aggregate}");
+            if (aggregate != HistoryAggregateType.Average)
+                throw new HistoryAggregateNotSupportedException(aggregate);
+
+            var buckets = new List<DataValueSnapshot>();
+            for (var b = startUtc; b < endUtc; b += interval)
+            {
+                var inBucket = Series.Where(s => s.SourceTimestampUtc >= b && s.SourceTimestampUtc < b + interval).ToList();
+                if (inBucket.Count == 0) continue;
+                buckets.Add(new DataValueSnapshot(inBucket.Average(s => (double)s.Value!), 0u, b, b));
+            }
+            return Task.FromResult(new HistoryReadResult(buckets, null));
+        }
+
+        public Task<HistoryReadResult> ReadAtTimeAsync(
+            string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken ct)
+        {
+            lock (Calls) Calls.Add($"attime:{fullReference}");
+            // Step semantics: the last sample at or before the requested instant; BadNoData before the first.
+            var result = timestampsUtc.Select(t =>
+            {
+                var prior = Series.LastOrDefault(s => s.SourceTimestampUtc <= t);
+                return prior is null
+                    ? new DataValueSnapshot(null, 0x809B0000u /* BadNoData */, t, t)
+                    : new DataValueSnapshot(prior.Value, prior.StatusCode, t, t);
+            }).ToList();
+            return Task.FromResult(new HistoryReadResult(result, null));
+        }
+
+        public Task<HistoricalEventsResult> ReadEventsAsync(
+            string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken ct) =>
+            Task.FromResult(new HistoricalEventsResult([], null));
+
+        public HistorianHealthSnapshot GetHealthSnapshot() =>
+            new(0, 0, 0, 0, null, null, null, false, false, null, null, []);
+
+        public void Dispose() { }
+    }
+
+    private static async Task<(ServerEventRig Rig, FakeHistorian Fake, OpcUaClientDriver Drv)> StartAsync(
+        string name, CancellationToken ct)
+    {
+        var rig = await OpcUaClientDriverWireTests.StartRigAsync(name, ct);
+        var fake = new FakeHistorian();
+        rig.Nm.HistorianDataSource = fake;
+        var id = rig.EnsureRawVariable("Hist", "Double", writable: false, historianTagname: HistTag);
+        OpcUaClientDriverWireTests.Publish(rig, id, 5.0);
+
+        var drv = OpcUaClientDriverWireTests.NewDriver(
+            rig, name,
+            extraTags: [OpcUaClientDriverWireTests.Tag(HistRef, ServerEventRig.RawNsu(id))]);
+        await drv.InitializeAsync("{}", ct);
+        return (rig, fake, drv);
+    }
+
+    [Fact]
+    public async Task ReadRaw_returns_the_historians_samples_with_values_timestamps_and_statuses()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var (rig, fake, drv) = await StartAsync("hist-raw", ct);
+        await using var _r = rig;
+        await using var _d = drv;
+
+        var result = await drv.ReadRawAsync(HistRef, At(0), At(10), 0, ct);
+
+        result.Samples.Select(s => Convert.ToDouble(s.Value)).ShouldBe([1.0, 2.0, 3.0, 4.0, 5.0]);
+        result.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(Series.Select(s => s.SourceTimestampUtc));
+        result.Samples.Select(s => s.StatusCode).ShouldBe([0u, 0u, Uncertain, 0u, 0u]);
+        result.ContinuationPoint.ShouldBeNull("the whole window fit in one page");
+        lock (fake.Calls) fake.Calls.ShouldContain($"raw:{HistTag}", "the read must reach the historian under the node's tagname");
+    }
+
+    [Fact]
+    public async Task ReadRaw_honours_the_window_and_pages_with_a_continuation_point_when_the_cap_is_hit()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var (rig, _, drv) = await StartAsync("hist-raw-page", ct);
+        await using var _r = rig;
+        await using var _d = drv;
+
+        // Window [1s, 4s) holds samples 2,3,4 (end exclusive).
+        var window = await drv.ReadRawAsync(HistRef, At(1), At(4), 0, ct);
+        window.Samples.Select(s => Convert.ToDouble(s.Value)).ShouldBe([2.0, 3.0, 4.0]);
+
+        // A cap of 3 against 5 available: first three samples and a continuation point.
+        var paged = await drv.ReadRawAsync(HistRef, At(0), At(10), 3, ct);
+        paged.Samples.Select(s => Convert.ToDouble(s.Value)).ShouldBe([1.0, 2.0, 3.0]);
+        paged.ContinuationPoint.ShouldNotBeNull("a full page must carry a continuation point");
+    }
+
+    [Fact]
+    public async Task ReadAtTime_returns_one_sample_per_requested_instant_with_the_prior_value()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var (rig, fake, drv) = await StartAsync("hist-attime", ct);
+        await using var _r = rig;
+        await using var _d = drv;
+
+        var result = await drv.ReadAtTimeAsync(HistRef, [At(-1), At(1), At(2), At(3).AddMilliseconds(500)], ct);
+
+        result.Samples.Count.ShouldBe(4);
+        result.Samples[0].StatusCode.ShouldNotBe(0u, "before the first sample there is no value");
+        Convert.ToDouble(result.Samples[1].Value).ShouldBe(2.0);
+        result.Samples[1].StatusCode.ShouldBe(0u);
+        Convert.ToDouble(result.Samples[2].Value).ShouldBe(3.0);
+        result.Samples[2].StatusCode.ShouldBe(Uncertain);
+        Convert.ToDouble(result.Samples[3].Value).ShouldBe(4.0);
+        result.Samples.Skip(1).Select(s => s.SourceTimestampUtc)
+            .ShouldBe([At(1), At(2), At(3).AddMilliseconds(500)]);
+        lock (fake.Calls) fake.Calls.ShouldContain($"attime:{HistTag}");
+    }
+
+    [Fact]
+    public async Task ReadProcessed_serves_a_supported_aggregate_and_returns_nothing_for_an_unsupported_one()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var (rig, fake, drv) = await StartAsync("hist-processed", ct);
+        await using var _r = rig;
+        await using var _d = drv;
+
+        // Average over 2s buckets across [0s,6s): {1,2}=1.5, {3,4}=3.5, {5}=5.
+        var avg = await drv.ReadProcessedAsync(HistRef, At(0), At(6), TimeSpan.FromSeconds(2), HistoryAggregateType.Average, ct);
+        avg.Samples.Select(s => Convert.ToDouble(s.Value)).ShouldBe([1.5, 3.5, 5.0]);
+        avg.Samples.Select(s => s.SourceTimestampUtc).ShouldBe([At(0), At(2), At(4)]);
+
+        // The backend refuses Count: the server maps that to BadAggregateNotSupported for the node, which the
+        // driver surfaces as no samples (it does not expose the per-node status). The Average above is the
+        // positive control that the same path serves data, and the call must have REACHED the backend.
+        var count = await drv.ReadProcessedAsync(HistRef, At(0), At(6), TimeSpan.FromSeconds(2), HistoryAggregateType.Count, ct);
+        count.Samples.ShouldBeEmpty();
+        lock (fake.Calls) fake.Calls.ShouldContain($"processed:{HistTag}:Count");
+    }
+
+    [Fact]
+    public async Task A_read_of_a_non_historized_node_returns_nothing_and_never_reaches_the_historian()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var (rig, fake, drv) = await StartAsync("hist-none", ct);
+        await using var _r = rig;
+        await using var _d = drv;
+
+        var result = await drv.ReadRawAsync(OpcUaClientDriverWireTests.SpeedRef, At(0), At(10), 0, ct);
+
+        result.Samples.ShouldBeEmpty();
+        // Positive control: the historized sibling does reach the historian on the same session.
+        (await drv.ReadRawAsync(HistRef, At(0), At(10), 0, ct)).Samples.Count.ShouldBe(5);
+        lock (fake.Calls) fake.Calls.Count.ShouldBe(1, "only the historized read dispatches");
+    }
+}
@@ -13,19 +13,17 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
 ///     fixture" scaffolds in <c>Driver.OpcUaClient.Tests</c> (Gitea #622). No opc-plc and no Docker: every test
 ///     here runs a real secure-channel + session + subscription exchange over loopback. The driver authenticates
 ///     with the rig's API-key credential so writes carry the <c>WriteOperate</c> role.
-///     <para>
-///         HistoryRead is NOT covered here (deferred): the server-side leg needs an <c>IHistorianDataSource</c>
-///         stub plus a historized-variable rig seam, and the driver's passthrough shapes are unit-tested already.
-///     </para>
+///     HistoryRead lives in <see cref="OpcUaClientDriverHistoryWireTests"/>; reconnect/resume in
+///     <see cref="OpcUaClientDriverReconnectWireTests"/>.
 /// </summary>
 [Trait("Category", "Integration")]
 public sealed class OpcUaClientDriverWireTests
 {
-    private const string SpeedRef = "Sim/dev1/Speed";
-    private const string RoRef = "Sim/dev1/Counter";
+    internal const string SpeedRef = "Sim/dev1/Speed";
+    internal const string RoRef = "Sim/dev1/Counter";
    private const string MissingRef = "Sim/dev1/Missing";
-    private const string SpeedId = ServerEventRig.RawDeviceFolder + "/Speed";
-    private const string CounterId = ServerEventRig.RawDeviceFolder + "/Counter";
+    internal const string SpeedId = ServerEventRig.RawDeviceFolder + "/Speed";
+    internal const string CounterId = ServerEventRig.RawDeviceFolder + "/Counter";

    private sealed class RecordingWriteGateway(bool succeed) : IOpcUaNodeWriteGateway
    {
@@ -38,10 +36,12 @@ public sealed class OpcUaClientDriverWireTests
        }
    }

-    private static RawTagEntry Tag(string rawPath, string nodeId) =>
+    internal static RawTagEntry Tag(string rawPath, string nodeId) =>
        new(rawPath, JsonSerializer.Serialize(new { nodeId }), WriteIdempotent: true);

-    private static OpcUaClientDriver NewDriver(ServerEventRig rig, string id, string? browseRoot = null) =>
+    internal static OpcUaClientDriver NewDriver(
+        ServerEventRig rig, string id, string? browseRoot = null, IReadOnlyList<RawTagEntry>? extraTags = null,
+        TimeSpan? reconnectPeriod = null) =>
        new(new OpcUaClientDriverOptions
        {
            EndpointUrl = rig.EndpointUrl,
@@ -50,6 +50,7 @@ public sealed class OpcUaClientDriverWireTests
                Tag(SpeedRef, ServerEventRig.RawNsu(SpeedId)),
                Tag(RoRef, ServerEventRig.RawNsu(CounterId)),
                Tag(MissingRef, ServerEventRig.RawNsu($"{ServerEventRig.RawDeviceFolder}/NoSuchNode")),
+                .. extraTags ?? [],
            ],
            SecurityPolicy = OpcUaSecurityPolicy.None,
            SecurityMode = OpcUaSecurityMode.None,
@@ -62,9 +63,10 @@ public sealed class OpcUaClientDriverWireTests
            SessionTimeout = TimeSpan.FromSeconds(30),
            KeepAliveInterval = TimeSpan.FromSeconds(1),
            BrowseRoot = browseRoot,
+            ReconnectPeriod = reconnectPeriod ?? TimeSpan.FromSeconds(5),
        }, id);

-    private static async Task<ServerEventRig> StartRigAsync(string name, CancellationToken ct)
+    internal static async Task<ServerEventRig> StartRigAsync(string name, CancellationToken ct)
    {
        var rig = await ServerEventRig.StartAsync(name, encrypted: false, ct);
        rig.EnsureRawVariable("Speed", "Float", writable: true);
@@ -72,7 +74,7 @@ public sealed class OpcUaClientDriverWireTests
        return rig;
    }

-    private static void Publish(ServerEventRig rig, string id, object value) =>
+    internal static void Publish(ServerEventRig rig, string id, object value) =>
        rig.Sink.WriteValue(id, value, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);

    /// <summary>Reads round-trip through the real wire; a node the server lacks does not read Good.</summary>
@@ -38,6 +38,7 @@ internal sealed class ServerEventRig : IAsyncDisposable
    private OpcUaApplicationHost _host = null!;
    private OtOpcUaSdkServer _server = null!;
    private int _port;
+    private OpcUaApplicationHostOptions _hostOptions = null!;

    private ServerEventRig(string pkiRoot) => _pkiRoot = pkiRoot;

@@ -51,28 +52,44 @@ internal sealed class ServerEventRig : IAsyncDisposable
    {
        var rig = new ServerEventRig(Path.Combine(Path.GetTempPath(), $"otopcua-{name}-{Guid.NewGuid():N}"));
        rig._port = AllocateFreePort();
-        rig._server = new OtOpcUaSdkServer();
-        rig._host = new OpcUaApplicationHost(
-            new OpcUaApplicationHostOptions
-            {
-                ApplicationName = $"OtOpcUa.{name}",
-                ApplicationUri = $"urn:OtOpcUa.{name}.{Guid.NewGuid():N}",
-                OpcUaPort = rig._port,
-                PublicHostname = "127.0.0.1",
-                PkiStoreRoot = rig._pkiRoot,
-                EnabledSecurityProfiles = encrypted
-                    ? [OpcUaSecurityProfile.None, OpcUaSecurityProfile.Basic256Sha256SignAndEncrypt]
-                    : [OpcUaSecurityProfile.None],
-                AutoAcceptUntrustedClientCertificates = true,
-            },
-            NullLogger<OpcUaApplicationHost>.Instance,
-            new KeyPresentingAuthenticator());
-
-        await rig._host.StartAsync(rig._server, ct);
-        rig.Sink = new SdkAddressSpaceSink(rig.Nm);
+        rig._hostOptions = new OpcUaApplicationHostOptions
+        {
+            ApplicationName = $"OtOpcUa.{name}",
+            ApplicationUri = $"urn:OtOpcUa.{name}.{Guid.NewGuid():N}",
+            OpcUaPort = rig._port,
+            PublicHostname = "127.0.0.1",
+            PkiStoreRoot = rig._pkiRoot,
+            EnabledSecurityProfiles = encrypted
+                ? [OpcUaSecurityProfile.None, OpcUaSecurityProfile.Basic256Sha256SignAndEncrypt]
+                : [OpcUaSecurityProfile.None],
+            AutoAcceptUntrustedClientCertificates = true,
+        };
+        await rig.BootAsync(ct);
        return rig;
    }

+    private async Task BootAsync(CancellationToken ct)
+    {
+        _server = new OtOpcUaSdkServer();
+        _host = new OpcUaApplicationHost(
+            _hostOptions, NullLogger<OpcUaApplicationHost>.Instance, new KeyPresentingAuthenticator());
+        await _host.StartAsync(_server, ct);
+        Sink = new SdkAddressSpaceSink(Nm);
+    }
+
+    /// <summary>
+    ///     Stops the server and boots a fresh one on the SAME port, application URI and PKI store (a process
+    ///     restart as a client sees it). The address space is EMPTY afterwards: the caller re-creates its nodes and
+    ///     re-publishes values, as a redeploy would. <paramref name="whileDown"/> runs once the old server is gone
+    ///     and before the new one binds, so a test can observe the outage.
+    /// </summary>
+    public async Task RestartAsync(CancellationToken ct, Func<Task>? whileDown = null)
+    {
+        await _host.DisposeAsync();
+        if (whileDown is not null) await whileDown();
+        await BootAsync(ct);
+    }
+
    /// <summary>Wires the authorizer over the given keys and grants, enforcing or not.</summary>
    public DataPlaneAuthorityState Enforce(bool enabled, ApiKey[] keys, params NodeAcl[] grants) =>
        Enforce(enabled, keys, ScopePaths(), grants);
@@ -148,11 +165,11 @@ internal sealed class ServerEventRig : IAsyncDisposable
    ///     Adds a raw value variable under <see cref="RawDeviceFolder"/> (the folder is ensured), for tests that
    ///     drive a real client against plain values rather than alarm conditions. Additive to the alarm helpers.
    /// </summary>
-    public string EnsureRawVariable(string name, string dataType, bool writable)
+    public string EnsureRawVariable(string name, string dataType, bool writable, string? historianTagname = null)
    {
        Sink.EnsureFolder(RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
        var id = $"{RawDeviceFolder}/{name}";
-        Sink.EnsureVariable(id, RawDeviceFolder, name, dataType, writable, AddressSpaceRealm.Raw);
+        Sink.EnsureVariable(id, RawDeviceFolder, name, dataType, writable, AddressSpaceRealm.Raw, historianTagname);
        return id;
    }