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;
}