From dcc8adcf488d22dbff3b1202096024c25cb1aa87 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 16:52:59 -0400 Subject: [PATCH] test(opcuaclient): HistoryRead Raw/AtTime/Processed wire tests against in-process server (#622) Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe --- .../OpcUaClientDriverHistoryWireTests.cs | 200 ++++++++++++++++++ .../OpcUaClientDriverWireTests.cs | 26 +-- .../ServerEventRig.cs | 59 ++++-- 3 files changed, 252 insertions(+), 33 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverHistoryWireTests.cs diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverHistoryWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverHistoryWireTests.cs new file mode 100644 index 00000000..8be1a1d0 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverHistoryWireTests.cs @@ -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; + +/// +/// HistoryRead leg of the OpcUaClient driver's wire coverage (Gitea #622): the real +/// issues Raw / Processed / AtTime reads over loopback to the in-process +/// OtOpcUa server, whose node manager dispatches to an injected deterministic +/// () behind a historized raw variable. +/// +[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); + + /// Five samples, one per second, value = 1..5; the third is Uncertain. + 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 Calls { get; } = []; + + public Task 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 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(); + 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 ReadAtTimeAsync( + string fullReference, IReadOnlyList 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 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"); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverWireTests.cs index 63cc8b0b..7bfb6768 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverWireTests.cs @@ -13,19 +13,17 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// fixture" scaffolds in Driver.OpcUaClient.Tests (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 WriteOperate role. -/// -/// HistoryRead is NOT covered here (deferred): the server-side leg needs an IHistorianDataSource -/// stub plus a historized-variable rig seam, and the driver's passthrough shapes are unit-tested already. -/// +/// HistoryRead lives in ; reconnect/resume in +/// . /// [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? 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 StartRigAsync(string name, CancellationToken ct) + internal static async Task 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); /// Reads round-trip through the real wire; a node the server lacks does not read Good. diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ServerEventRig.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ServerEventRig.cs index c6e4da42..d5d657be 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ServerEventRig.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ServerEventRig.cs @@ -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.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.Instance, new KeyPresentingAuthenticator()); + await _host.StartAsync(_server, ct); + Sink = new SdkAddressSpaceSink(Nm); + } + + /// + /// 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. runs once the old server is gone + /// and before the new one binds, so a test can observe the outage. + /// + public async Task RestartAsync(CancellationToken ct, Func? whileDown = null) + { + await _host.DisposeAsync(); + if (whileDown is not null) await whileDown(); + await BootAsync(ct); + } + /// Wires the authorizer over the given keys and grants, enforcing or not. 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 (the folder is ensured), for tests that /// drive a real client against plain values rather than alarm conditions. Additive to the alarm helpers. /// - 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; }