dohertj2/lmxopcua
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge b640-622: OpcUaClient history + reconnect wire tests (#622)

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 17:02:21 -04:00
parent 73e54656f8 4a93d4cead
commit 555e8d8047
4 changed files with 332 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverHistoryWireTests.cs
+200
View File
@@ -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");
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs
+80
View File
@@ -0,0 +1,80 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// <summary>
/// Reconnect + resume leg of the OpcUaClient driver's wire coverage (Gitea #622): the in-process server is
/// stopped and a fresh one is booted on the SAME port / application URI / PKI store — a process restart as the
/// driver sees it, with an empty address space the test then re-creates. The driver must detect the loss,
/// rebuild a session, and its existing subscription must deliver again with no caller action.
/// </summary>
[Trait("Category", "Integration")]
public sealed class OpcUaClientDriverReconnectWireTests
{
private static async Task WaitAsync(Func<bool> condition, string because, TimeSpan? timeout = null)
{
var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(60));
while (DateTime.UtcNow < deadline)
{
if (condition()) return;
await Task.Delay(50);
}
condition().ShouldBeTrue(because);
}
[Fact]
public async Task After_a_server_restart_the_driver_reconnects_and_the_subscription_resumes_delivering()
{
var ct = TestContext.Current.CancellationToken;
var rig = await OpcUaClientDriverWireTests.StartRigAsync("drv-reconnect", ct);
await using var _rig = rig;
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 10.0f);
await using var drv = OpcUaClientDriverWireTests.NewDriver(
rig, "wire-reconnect", reconnectPeriod: TimeSpan.FromSeconds(1));
await drv.InitializeAsync("{}", ct);
var seen = new List<float>();
drv.OnDataChange += (_, e) =>
{
if (e.FullReference == OpcUaClientDriverWireTests.SpeedRef && e.Snapshot.StatusCode == 0u)
lock (seen) seen.Add(Convert.ToSingle(e.Snapshot.Value));
};
var states = new List<HostState>();
drv.OnHostStatusChanged += (_, e) => { lock (states) states.Add(e.NewState); };
await drv.SubscribeAsync([OpcUaClientDriverWireTests.SpeedRef], TimeSpan.FromMilliseconds(100), ct);
await WaitAsync(() => { lock (seen) return seen.Contains(10.0f); }, "initial value must arrive before the outage");
// Outage: the loss must be DETECTED (host leaves Running) while the server is genuinely down.
await rig.RestartAsync(ct, whileDown: async () =>
{
await WaitAsync(
() => { lock (states) return states.Any(s => s != HostState.Running); },
"the driver must notice the server is gone", TimeSpan.FromSeconds(30));
});
// The new server is empty: redeploy the nodes and publish a value no earlier delivery could have carried.
rig.EnsureRawVariable("Speed", "Float", writable: true);
rig.EnsureRawVariable("Counter", "Int32", writable: false);
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 77.0f);
await WaitAsync(() => drv.GetHostStatuses().Single().State == HostState.Running,
"the driver must reconnect after the server is back");
await WaitAsync(() => { lock (seen) return seen.Contains(77.0f); },
"the pre-outage subscription must resume and deliver the post-restart value");
// Ongoing delivery, not just a one-off initial value.
OpcUaClientDriverWireTests.Publish(rig, OpcUaClientDriverWireTests.SpeedId, 88.0f);
await WaitAsync(() => { lock (seen) return seen.Contains(88.0f); },
"a later change must also reach the resumed subscription");
// And the data plane works for plain reads on the new session.
var back = await drv.ReadAsync([OpcUaClientDriverWireTests.SpeedRef], ct);
back[0].StatusCode.ShouldBe(0u);
Convert.ToSingle(back[0].Value).ShouldBe(88.0f);
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverWireTests.cs
+14 -12
View File
@@ -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>
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ServerEventRig.cs
+38 -21
View File
@@ -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;
}