diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/NullHistorianDataSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/NullHistorianDataSource.cs new file mode 100644 index 00000000..582bb53c --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/NullHistorianDataSource.cs @@ -0,0 +1,81 @@ +namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +/// +/// A no-op — the server's default historian backend when no +/// real historian has been registered. Every read returns an EMPTY result (no samples / events, +/// null continuation point) so the node-manager's HistoryRead override surfaces +/// GoodNoData for a historized node rather than faulting, and +/// reports a fully-disabled (zeroed, disconnected) snapshot. +/// +/// A process-wide singleton via (private ctor): it carries no state +/// and is immutable, so one shared instance is safe to assign as the node-manager's +/// HistorianDataSource default until the Host wires a real source post-start (Task 5). +/// +/// +public sealed class NullHistorianDataSource : IHistorianDataSource +{ + /// The shared singleton instance. + public static readonly NullHistorianDataSource Instance = new(); + + private static readonly HistoryReadResult EmptyRead = new(Array.Empty(), null); + private static readonly HistoricalEventsResult EmptyEvents = new(Array.Empty(), null); + + private NullHistorianDataSource() + { + } + + /// + public Task ReadRawAsync( + string fullReference, + DateTime startUtc, + DateTime endUtc, + uint maxValuesPerNode, + CancellationToken cancellationToken) => Task.FromResult(EmptyRead); + + /// + public Task ReadProcessedAsync( + string fullReference, + DateTime startUtc, + DateTime endUtc, + TimeSpan interval, + HistoryAggregateType aggregate, + CancellationToken cancellationToken) => Task.FromResult(EmptyRead); + + /// + public Task ReadAtTimeAsync( + string fullReference, + IReadOnlyList timestampsUtc, + CancellationToken cancellationToken) => Task.FromResult(EmptyRead); + + /// + public Task ReadEventsAsync( + string? sourceName, + DateTime startUtc, + DateTime endUtc, + int maxEvents, + CancellationToken cancellationToken) => Task.FromResult(EmptyEvents); + + /// + /// Returns a fully-disabled snapshot — no connections open, every counter zero, every nullable + /// field null, no cluster nodes. Pure; never blocks. + /// + public HistorianHealthSnapshot GetHealthSnapshot() => new( + TotalQueries: 0, + TotalSuccesses: 0, + TotalFailures: 0, + ConsecutiveFailures: 0, + LastSuccessTime: null, + LastFailureTime: null, + LastError: null, + ProcessConnectionOpen: false, + EventConnectionOpen: false, + ActiveProcessNode: null, + ActiveEventNode: null, + Nodes: Array.Empty()); + + /// No-op — the null source owns no unmanaged resources. + public void Dispose() + { + // Stateless singleton; nothing to release. + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 71a11f0c..294d22ae 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -2,7 +2,13 @@ using System.Collections.Concurrent; using Opc.Ua; using Opc.Ua.Server; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security; +// The SDK's HistoryRead service result (the value the override fills + hands back) and the historian +// data source's read DTO are both named HistoryReadResult. Alias each to keep the two unambiguous: +// the SDK result stays unqualified as the dominant name in the override; the source DTO is HistorianRead. +using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult; +using SdkHistoryReadResult = Opc.Ua.HistoryReadResult; namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; @@ -111,6 +117,32 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 set => _nodeWriteGateway = value ?? NullOpcUaNodeWriteGateway.Instance; } + private volatile IHistorianDataSource _historianDataSource = NullHistorianDataSource.Instance; + + /// + /// Server-side read backend for the OPC UA HistoryRead service over historized variable nodes. + /// When a client issues a HistoryRead (Raw / Processed / AtTime) against a node materialised + /// Historizing (a tag with registered), the + /// HistoryRead override resolves the node's NodeId to its historian tagname and dispatches to + /// this source — so a single registered historian (e.g. Wonderware) serves many drivers' nodes, + /// independent of any driver's lifecycle. + /// + /// Set by the Host at StartAsync (Task 5). The + /// default (assigning null restores it) means "no historian wired" → every read + /// returns empty, so a historized node's HistoryRead surfaces GoodNoData rather than + /// faulting. Backed by a volatile field (auto-properties can't be volatile) to make + /// the startup-write / SDK-read-thread handoff explicit: the Host assigns it once at boot on + /// the start thread and the SDK reads it on HistoryRead request threads. Unlike + /// , the HistoryRead override does NOT run under the + /// node-manager Lock, so the override may block-bridge to this (async) source. + /// + /// + public IHistorianDataSource HistorianDataSource + { + get => _historianDataSource; + set => _historianDataSource = value ?? NullHistorianDataSource.Instance; + } + /// Look up a materialised Part 9 alarm-condition node by its alarm node id (the /// ScriptedAlarmId), or null if not yet materialised. Exposed for tests + diagnostics. /// The alarm node identifier (== ScriptedAlarmId). @@ -975,6 +1007,239 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 return _folders.TryGetValue(parentNodeId, out var existing) ? existing : _root!; } + // --------------------------------------------------------------------------------------------- + // Phase C — OPC UA HistoryRead over historized variable nodes. + // + // The base CustomNodeManager2.HistoryRead (public + protected dispatcher) does the heavy lifting: + // it validates handles under Lock, builds `nodesToProcess` (a NodeHandle list for nodes WE own + // that carry the HistoryRead access bit), validates the timestamp args, handles + // `releaseContinuationPoints`, and dispatches by `details` runtime type to the per-details + // protected virtuals below. We override the three variable-history virtuals; HistoryReadEvents is + // left to the base (Task 4 adds it). Each override receives the pre-filtered handles and fills + // results[handle.Index] / errors[handle.Index] — handle.Index is the original index into the + // service-level results/errors lists, seeded by the base. The base pre-seeds every handle's error + // to BadHistoryOperationUnsupported, so a handle we don't recognise stays "unsupported" by default. + // + // NOTE: unlike OnWriteValue, the SDK does NOT hold the node-manager Lock while invoking these, so + // block-bridging the async data source (GetAwaiter().GetResult()) is safe — it can't freeze the + // address space. Each handle is served in isolation under try/catch so one node's failure (timeout, + // backend throw) never throws out of the batch. + // --------------------------------------------------------------------------------------------- + + /// + /// Serve a HistoryRead-Raw request over the pre-filtered historized variable handles, dispatching + /// each to . Modified-history reads + /// (IsReadModified) are unsupported — we don't serve a modified-value history surface. + /// + protected override void HistoryReadRawModified( + ServerSystemContext context, + ReadRawModifiedDetails details, + TimestampsToReturn timestampsToReturn, + IList nodesToRead, + IList results, + IList errors, + List nodesToProcess, + IDictionary cache) + { + foreach (var handle in nodesToProcess) + { + if (details.IsReadModified) + { + // We never serve modified-value history; mark this node unsupported and move on. + errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; + continue; + } + + ServeNode(handle, results, errors, source => source.ReadRawAsync( + ResolveTagname(handle), + details.StartTime, + details.EndTime, + details.NumValuesPerNode, + CancellationToken.None)); + } + } + + /// + /// Serve a HistoryRead-Processed request, mapping each node's per-node aggregate NodeId (from the + /// parallel AggregateType collection — the base guarantees it is the same length as + /// nodesToRead) to a and dispatching to + /// . An unknown aggregate yields + /// BadAggregateNotSupported for that node. + /// + protected override void HistoryReadProcessed( + ServerSystemContext context, + ReadProcessedDetails details, + TimestampsToReturn timestampsToReturn, + IList nodesToRead, + IList results, + IList errors, + List nodesToProcess, + IDictionary cache) + { + foreach (var handle in nodesToProcess) + { + // AggregateType is a per-node parallel collection (same length as nodesToRead, enforced by + // the base dispatcher). handle.Index is the node's position in that collection. + var aggregateNodeId = details.AggregateType[handle.Index]; + var aggregate = MapAggregate(aggregateNodeId); + if (aggregate is null) + { + errors[handle.Index] = StatusCodes.BadAggregateNotSupported; + continue; + } + + ServeNode(handle, results, errors, source => source.ReadProcessedAsync( + ResolveTagname(handle), + details.StartTime, + details.EndTime, + // OPC UA ProcessingInterval is a Duration in milliseconds. + TimeSpan.FromMilliseconds(details.ProcessingInterval), + aggregate.Value, + CancellationToken.None)); + } + } + + /// + /// Serve a HistoryRead-AtTime request, dispatching the requested timestamps to + /// . + /// + protected override void HistoryReadAtTime( + ServerSystemContext context, + ReadAtTimeDetails details, + TimestampsToReturn timestampsToReturn, + IList nodesToRead, + IList results, + IList errors, + List nodesToProcess, + IDictionary cache) + { + // Snapshot the requested timestamps once — the same list is read for every node. + var timestamps = details.ReqTimes?.ToList() ?? new List(); + foreach (var handle in nodesToProcess) + { + ServeNode(handle, results, errors, source => source.ReadAtTimeAsync( + ResolveTagname(handle), + timestamps, + CancellationToken.None)); + } + } + + /// + /// Block-bridge to the historian source for one node handle and project the result onto the + /// service-level results/errors slots. Resolves the node's registered historian tagname first + /// (a node we don't recognise as historized — which shouldn't reach us, since the base only + /// hands us nodes with the HistoryRead access bit — maps to BadHistoryOperationUnsupported). + /// The callback is invoked only AFTER the tagname is confirmed present; + /// it is wrapped in try/catch so a backend throw / timeout becomes a Bad status for THIS node + /// without throwing out of the batch. + /// + /// The pre-filtered node handle to serve; handle.Index indexes results/errors. + /// The service-level results list to fill at handle.Index. + /// The service-level errors list to fill at handle.Index. + /// Invokes the resolved data-source read; only called once the tagname is known. + private void ServeNode( + NodeHandle handle, + IList results, + IList errors, + Func> read) + { + var idString = handle.NodeId.Identifier?.ToString(); + if (idString is null || !TryGetHistorizedTagname(idString, out _)) + { + // Not a historized node we own a tagname for — unsupported. (The base pre-seeds this same + // status, but set it explicitly so the contract is local + obvious.) + errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; + return; + } + + try + { + // HistoryRead is NOT invoked under the node-manager Lock (unlike OnWriteValue), so blocking + // on the async source here is safe and won't freeze the address space. + var sourceResult = read(HistorianDataSource).GetAwaiter().GetResult(); + var historyData = ToHistoryData(sourceResult); + + results[handle.Index] = new SdkHistoryReadResult + { + // No source samples ⇒ GoodNoData (the node is historized, the window just held no data). + StatusCode = historyData.DataValues.Count == 0 ? StatusCodes.GoodNoData : StatusCodes.Good, + HistoryData = new ExtensionObject(historyData), + // We never issue continuation points — every read returns the full window in one shot. + ContinuationPoint = null, + }; + errors[handle.Index] = ServiceResult.Good; + } + catch (Exception ex) + { + // One node's backend failure (throw / timeout / cancellation) must not throw out of the + // batch — surface a Bad status for THIS node only. This CustomNodeManager2 carries no + // ILogger (see ReportConditionEvent), so log through the SDK's static trace rather than + // swallowing silently. Utils.LogError is [Obsolete] in 1.5.378 (favours an ITelemetryContext + // this manager doesn't wire) — suppress the deprecation, matching the existing pattern. +#pragma warning disable CS0618 // Type or member is obsolete + Utils.LogError(ex, "OtOpcUaNodeManager: HistoryRead failed for node {0}", handle.NodeId); +#pragma warning restore CS0618 + errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; + } + } + + /// Resolve a handle's registered historian tagname. The caller () + /// has already confirmed the node is historized before invoking the read callback, so this is a + /// guaranteed hit; the null-coalesce is a defensive fallback to the bare NodeId string. + private string ResolveTagname(NodeHandle handle) + { + var idString = handle.NodeId.Identifier?.ToString() ?? string.Empty; + return TryGetHistorizedTagname(idString, out var tagname) ? tagname! : idString; + } + + /// + /// Map an OPC UA Part 13 standard-aggregate function NodeId to our + /// . Returns null for any aggregate we don't serve so + /// the caller can surface BadAggregateNotSupported. + /// + /// The per-node aggregate-function NodeId from the request. + /// The mapped aggregate, or null when unsupported. + private static HistoryAggregateType? MapAggregate(NodeId aggregateNodeId) + { + if (aggregateNodeId == ObjectIds.AggregateFunction_Average) return HistoryAggregateType.Average; + if (aggregateNodeId == ObjectIds.AggregateFunction_Minimum) return HistoryAggregateType.Minimum; + if (aggregateNodeId == ObjectIds.AggregateFunction_Maximum) return HistoryAggregateType.Maximum; + if (aggregateNodeId == ObjectIds.AggregateFunction_Total) return HistoryAggregateType.Total; + if (aggregateNodeId == ObjectIds.AggregateFunction_Count) return HistoryAggregateType.Count; + return null; + } + + /// + /// Project the historian source's (Core.Abstractions DTO) into an + /// SDK — one per , + /// carrying value / status / source+server timestamps. A null SourceTimestamp maps to + /// DateTime.MinValue (the SDK's "unset" sentinel for that field). + /// + /// The data source's read result. + /// The populated SDK . + private static HistoryData ToHistoryData(HistorianRead sourceResult) + { + var values = new DataValueCollection(sourceResult.Samples.Count); + foreach (var sample in sourceResult.Samples) + { + values.Add(ToSdkDataValue(sample)); + } + + return new HistoryData { DataValues = values }; + } + + /// Convert one driver-agnostic to an SDK + /// , mirroring value / status code / source + server timestamps. + /// The source sample. + /// The equivalent SDK data value. + private static DataValue ToSdkDataValue(DataValueSnapshot snapshot) => new() + { + WrappedValue = new Variant(snapshot.Value), + StatusCode = new StatusCode(snapshot.StatusCode), + SourceTimestamp = snapshot.SourceTimestampUtc ?? DateTime.MinValue, + ServerTimestamp = snapshot.ServerTimestampUtc, + }; + /// public override void CreateAddressSpace(IDictionary> externalReferences) { diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/NullHistorianDataSourceTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/NullHistorianDataSourceTests.cs new file mode 100644 index 00000000..6ed8afb7 --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/Historian/NullHistorianDataSourceTests.cs @@ -0,0 +1,100 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests.Historian; + +/// +/// The default no-op historian source. Every read returns an empty result with a null +/// continuation point; reports a fully +/// disabled shape; and is the shared singleton. +/// +public sealed class NullHistorianDataSourceTests +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private static readonly NullHistorianDataSource Source = NullHistorianDataSource.Instance; + + /// is a single shared instance. + [Fact] + public void Instance_is_a_singleton() + { + NullHistorianDataSource.Instance.ShouldBeSameAs(NullHistorianDataSource.Instance); + } + + /// ReadRawAsync returns no samples and a null continuation point. + [Fact] + public async Task ReadRawAsync_returns_empty() + { + var result = await Source.ReadRawAsync( + "WW.Tag", DateTime.UtcNow.AddHours(-1), DateTime.UtcNow, maxValuesPerNode: 100, Ct); + + result.Samples.ShouldBeEmpty(); + result.ContinuationPoint.ShouldBeNull(); + } + + /// ReadProcessedAsync returns no samples and a null continuation point. + [Fact] + public async Task ReadProcessedAsync_returns_empty() + { + var result = await Source.ReadProcessedAsync( + "WW.Tag", DateTime.UtcNow.AddHours(-1), DateTime.UtcNow, + TimeSpan.FromSeconds(10), HistoryAggregateType.Average, Ct); + + result.Samples.ShouldBeEmpty(); + result.ContinuationPoint.ShouldBeNull(); + } + + /// ReadAtTimeAsync returns no samples and a null continuation point. + [Fact] + public async Task ReadAtTimeAsync_returns_empty() + { + var result = await Source.ReadAtTimeAsync( + "WW.Tag", new[] { DateTime.UtcNow }, Ct); + + result.Samples.ShouldBeEmpty(); + result.ContinuationPoint.ShouldBeNull(); + } + + /// ReadEventsAsync returns no events and a null continuation point. + [Fact] + public async Task ReadEventsAsync_returns_empty() + { + var result = await Source.ReadEventsAsync( + sourceName: null, DateTime.UtcNow.AddHours(-1), DateTime.UtcNow, maxEvents: 0, Ct); + + result.Events.ShouldBeEmpty(); + result.ContinuationPoint.ShouldBeNull(); + } + + /// GetHealthSnapshot reports a fully-disabled (zeroed, disconnected, no-node) shape. + [Fact] + public void GetHealthSnapshot_is_disabled() + { + var snapshot = Source.GetHealthSnapshot(); + + snapshot.TotalQueries.ShouldBe(0); + snapshot.TotalSuccesses.ShouldBe(0); + snapshot.TotalFailures.ShouldBe(0); + snapshot.ConsecutiveFailures.ShouldBe(0); + snapshot.LastSuccessTime.ShouldBeNull(); + snapshot.LastFailureTime.ShouldBeNull(); + snapshot.LastError.ShouldBeNull(); + snapshot.ProcessConnectionOpen.ShouldBeFalse(); + snapshot.EventConnectionOpen.ShouldBeFalse(); + snapshot.ActiveProcessNode.ShouldBeNull(); + snapshot.ActiveEventNode.ShouldBeNull(); + snapshot.Nodes.ShouldBeEmpty(); + } + + /// Dispose is a safe no-op (idempotent). + [Fact] + public void Dispose_is_a_noop() + { + Should.NotThrow(() => + { + Source.Dispose(); + Source.Dispose(); + }); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs new file mode 100644 index 00000000..3d050b43 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs @@ -0,0 +1,464 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Opc.Ua; +using Opc.Ua.Server; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult; +using SdkHistoryReadResult = Opc.Ua.HistoryReadResult; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests; + +/// +/// Phase C Task 3 — the node-manager's OPC UA HistoryRead override (Raw / Processed / AtTime) over +/// historized variable nodes. Boots a real (the same harness +/// uses), materialises a historized variable via +/// , wires a recording fake +/// , then invokes the node manager's PUBLIC +/// HistoryRead(OperationContext, …) directly. The base CustomNodeManager2 builds the node +/// handles + dispatches to the protected per-details overrides, so this exercises the real dispatch +/// path in-process — fast + deterministic, no client socket. +/// +public sealed class NodeManagerHistoryReadTests : IDisposable +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private readonly string _pkiRoot = Path.Combine( + Path.GetTempPath(), + $"otopcua-historyread-{Guid.NewGuid():N}"); + + /// Raw read: the fake receives the resolved tagname + StartTime/EndTime/NumValuesPerNode, + /// and the returned samples decode to a HistoryData whose DataValues mirror value/status/source+server + /// timestamps. StatusCode is Good when samples are present. + [Fact] + public async Task Raw_dispatches_to_source_and_maps_samples() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource(); + nm.HistorianDataSource = fake; + + nm.EnsureVariable("eq-1/temp", parentFolderNodeId: null, displayName: "Temp", dataType: "Float", + writable: false, historianTagname: "WW.Temp"); + var nodeId = nm.TryGetVariable("eq-1/temp")!.NodeId; + + var src = DateTime.UtcNow.AddSeconds(-5); + var srv = DateTime.UtcNow; + fake.RawResult = new HistorianRead( + new[] { new DataValueSnapshot(42.5f, StatusCodes.Good, src, srv) }, null); + + var start = DateTime.UtcNow.AddHours(-1); + var end = DateTime.UtcNow; + var details = new ReadRawModifiedDetails + { + StartTime = start, + EndTime = end, + NumValuesPerNode = 100, + IsReadModified = false, + }; + + var (results, errors) = InvokeHistoryRead(server, nm, details, nodeId); + + // The source saw the resolved tagname + the request window + cap. + fake.LastCall.ShouldBe("Raw"); + fake.LastTagname.ShouldBe("WW.Temp"); + fake.LastStart.ShouldBe(start); + fake.LastEnd.ShouldBe(end); + fake.LastMaxValues.ShouldBe(100u); + + errors[0].StatusCode.Code.ShouldBe(StatusCodes.Good); + results[0].StatusCode.Code.ShouldBe(StatusCodes.Good); + var data = (HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData); + data.DataValues.Count.ShouldBe(1); + var dv = data.DataValues[0]; + dv.Value.ShouldBe(42.5f); + dv.StatusCode.Code.ShouldBe(StatusCodes.Good); + dv.SourceTimestamp.ShouldBe(src); + dv.ServerTimestamp.ShouldBe(srv); + + await host.DisposeAsync(); + } + + /// The resolved tagname that reaches the source is the OVERRIDE tagname (distinct from the + /// NodeId / FullName) — the override resolves the NodeId→tagname map, not the bare NodeId. + [Fact] + public async Task Raw_resolves_override_tagname_not_node_id() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { RawResult = Empty() }; + nm.HistorianDataSource = fake; + + // Node id "eq-9/flow" but a DISTINCT historian tagname "Plant.Flow.PV". + nm.EnsureVariable("eq-9/flow", parentFolderNodeId: null, displayName: "Flow", dataType: "Double", + writable: false, historianTagname: "Plant.Flow.PV"); + var nodeId = nm.TryGetVariable("eq-9/flow")!.NodeId; + + var details = new ReadRawModifiedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), + EndTime = DateTime.UtcNow, + NumValuesPerNode = 10, + // ReadRawModifiedDetails defaults IsReadModified=true; a raw (non-modified) read clears it. + IsReadModified = false, + }; + + InvokeHistoryRead(server, nm, details, nodeId); + + fake.LastTagname.ShouldBe("Plant.Flow.PV"); + + await host.DisposeAsync(); + } + + /// Empty samples ⇒ the node's StatusCode is GoodNoData (the node is historized, the window + /// just held no data). + [Fact] + public async Task Raw_empty_samples_yields_GoodNoData() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { RawResult = Empty() }; + nm.HistorianDataSource = fake; + + nm.EnsureVariable("eq-1/empty", parentFolderNodeId: null, displayName: "Empty", dataType: "Float", + writable: false, historianTagname: "WW.Empty"); + var nodeId = nm.TryGetVariable("eq-1/empty")!.NodeId; + + var details = new ReadRawModifiedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), + EndTime = DateTime.UtcNow, + NumValuesPerNode = 10, + IsReadModified = false, + }; + + var (results, errors) = InvokeHistoryRead(server, nm, details, nodeId); + + errors[0].StatusCode.Code.ShouldBe(StatusCodes.Good); + results[0].StatusCode.Code.ShouldBe(StatusCodes.GoodNoData); + var data = (HistoryData)ExtensionObject.ToEncodeable(results[0].HistoryData); + data.DataValues.ShouldBeEmpty(); + + await host.DisposeAsync(); + } + + /// A non-historized node (plain variable, no HistoryRead bit) reaching HistoryRead yields + /// BadHistoryOperationUnsupported — the source is never invoked. + [Fact] + public async Task Non_historized_node_yields_BadHistoryOperationUnsupported() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { RawResult = Empty() }; + nm.HistorianDataSource = fake; + + // Plain (non-historized) variable — no HistoryRead access bit. + nm.EnsureVariable("eq-1/plain", parentFolderNodeId: null, displayName: "Plain", dataType: "Int32", + writable: false, historianTagname: null); + var nodeId = nm.TryGetVariable("eq-1/plain")!.NodeId; + + var details = new ReadRawModifiedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), + EndTime = DateTime.UtcNow, + NumValuesPerNode = 10, + }; + + var (_, errors) = InvokeHistoryRead(server, nm, details, nodeId); + + errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported); + fake.LastCall.ShouldBeNull(); // source never reached + + await host.DisposeAsync(); + } + + /// Raw with IsReadModified=true ⇒ BadHistoryOperationUnsupported (we don't serve modified + /// history); the source is never invoked. + [Fact] + public async Task Raw_read_modified_yields_BadHistoryOperationUnsupported() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { RawResult = Empty() }; + nm.HistorianDataSource = fake; + + nm.EnsureVariable("eq-1/mod", parentFolderNodeId: null, displayName: "Mod", dataType: "Float", + writable: false, historianTagname: "WW.Mod"); + var nodeId = nm.TryGetVariable("eq-1/mod")!.NodeId; + + var details = new ReadRawModifiedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), + EndTime = DateTime.UtcNow, + NumValuesPerNode = 10, + IsReadModified = true, + }; + + var (_, errors) = InvokeHistoryRead(server, nm, details, nodeId); + + errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported); + fake.LastCall.ShouldBeNull(); + + await host.DisposeAsync(); + } + + /// Processed read with a known aggregate (Average) reaches the source as + /// HistoryAggregateType.Average + the ProcessingInterval as a TimeSpan. + [Fact] + public async Task Processed_known_aggregate_dispatches_with_interval() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { ProcessedResult = Empty() }; + nm.HistorianDataSource = fake; + + nm.EnsureVariable("eq-1/avg", parentFolderNodeId: null, displayName: "Avg", dataType: "Float", + writable: false, historianTagname: "WW.Avg"); + var nodeId = nm.TryGetVariable("eq-1/avg")!.NodeId; + + var details = new ReadProcessedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), + EndTime = DateTime.UtcNow, + ProcessingInterval = 10_000.0, // ms + AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_Average }, + }; + + InvokeHistoryRead(server, nm, details, nodeId); + + fake.LastCall.ShouldBe("Processed"); + fake.LastTagname.ShouldBe("WW.Avg"); + fake.LastAggregate.ShouldBe(HistoryAggregateType.Average); + fake.LastInterval.ShouldBe(TimeSpan.FromMilliseconds(10_000.0)); + + await host.DisposeAsync(); + } + + /// Processed read with an UNKNOWN aggregate NodeId ⇒ BadAggregateNotSupported; the source + /// is never invoked. + [Fact] + public async Task Processed_unknown_aggregate_yields_BadAggregateNotSupported() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { ProcessedResult = Empty() }; + nm.HistorianDataSource = fake; + + nm.EnsureVariable("eq-1/sd", parentFolderNodeId: null, displayName: "Sd", dataType: "Float", + writable: false, historianTagname: "WW.Sd"); + var nodeId = nm.TryGetVariable("eq-1/sd")!.NodeId; + + var details = new ReadProcessedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), + EndTime = DateTime.UtcNow, + ProcessingInterval = 10_000.0, + // StandardDeviationSample is a real OPC UA aggregate we do not serve. + AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_StandardDeviationSample }, + }; + + var (_, errors) = InvokeHistoryRead(server, nm, details, nodeId); + + errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadAggregateNotSupported); + fake.LastCall.ShouldBeNull(); + + await host.DisposeAsync(); + } + + /// AtTime read: the requested timestamps reach ReadAtTimeAsync in order. + [Fact] + public async Task AtTime_dispatches_requested_timestamps() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { AtTimeResult = Empty() }; + nm.HistorianDataSource = fake; + + nm.EnsureVariable("eq-1/at", parentFolderNodeId: null, displayName: "At", dataType: "Float", + writable: false, historianTagname: "WW.At"); + var nodeId = nm.TryGetVariable("eq-1/at")!.NodeId; + + var t1 = DateTime.UtcNow.AddMinutes(-2); + var t2 = DateTime.UtcNow.AddMinutes(-1); + var details = new ReadAtTimeDetails + { + ReqTimes = new DateTimeCollection { t1, t2 }, + }; + + InvokeHistoryRead(server, nm, details, nodeId); + + fake.LastCall.ShouldBe("AtTime"); + fake.LastTagname.ShouldBe("WW.At"); + fake.LastTimestamps.ShouldBe(new[] { t1, t2 }); + + await host.DisposeAsync(); + } + + /// A backend that throws ⇒ that node's error is Bad (not GoodNoData) and no exception + /// escapes the HistoryRead call. + [Fact] + public async Task Backend_throw_yields_bad_status_and_does_not_escape() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { ThrowOnRead = true }; + nm.HistorianDataSource = fake; + + nm.EnsureVariable("eq-1/boom", parentFolderNodeId: null, displayName: "Boom", dataType: "Float", + writable: false, historianTagname: "WW.Boom"); + var nodeId = nm.TryGetVariable("eq-1/boom")!.NodeId; + + var details = new ReadRawModifiedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), + EndTime = DateTime.UtcNow, + NumValuesPerNode = 10, + IsReadModified = false, + }; + + // The call must not throw even though the backend does. + var (results, errors) = InvokeHistoryRead(server, nm, details, nodeId); + + StatusCode.IsBad(errors[0].StatusCode).ShouldBeTrue(); + errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported); + // The result slot was never filled with a GoodNoData success. + (results[0].StatusCode.Code == StatusCodes.GoodNoData).ShouldBeFalse(); + + await host.DisposeAsync(); + } + + /// Invoke the node manager's public HistoryRead with a single node, returning the filled + /// results + errors. Uses a session-less (the + /// (RequestHeader, SecureChannelContext, RequestType, IUserIdentity) ctor) — HistoryRead's + /// handle-building only needs the NodeId + namespace, not a session. + private static (IList Results, IList Errors) InvokeHistoryRead( + OtOpcUaSdkServer server, OtOpcUaNodeManager nm, HistoryReadDetails details, NodeId nodeId) + { + var context = new OperationContext( + new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null); + + var nodesToRead = new List + { + new() { NodeId = nodeId }, + }; + var results = new List { null! }; + var errors = new List { null! }; + + nm.HistoryRead( + context, + details, + TimestampsToReturn.Both, + releaseContinuationPoints: false, + nodesToRead, + results, + errors); + + return (results, errors); + } + + private static HistorianRead Empty() => new(Array.Empty(), null); + + /// A recording fake historian source — captures the last call's kind + arguments and returns + /// a configured result (or throws when is set). + private sealed class RecordingHistorianDataSource : IHistorianDataSource + { + public bool ThrowOnRead { get; init; } + public HistorianRead RawResult { get; set; } = new(Array.Empty(), null); + public HistorianRead ProcessedResult { get; set; } = new(Array.Empty(), null); + public HistorianRead AtTimeResult { get; set; } = new(Array.Empty(), null); + + public string? LastCall { get; private set; } + public string? LastTagname { get; private set; } + public DateTime LastStart { get; private set; } + public DateTime LastEnd { get; private set; } + public uint LastMaxValues { get; private set; } + public TimeSpan LastInterval { get; private set; } + public HistoryAggregateType LastAggregate { get; private set; } + public IReadOnlyList? LastTimestamps { get; private set; } + + public Task ReadRawAsync( + string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, + CancellationToken cancellationToken) + { + if (ThrowOnRead) throw new InvalidOperationException("backend boom"); + LastCall = "Raw"; + LastTagname = fullReference; + LastStart = startUtc; + LastEnd = endUtc; + LastMaxValues = maxValuesPerNode; + return Task.FromResult(RawResult); + } + + public Task ReadProcessedAsync( + string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, + HistoryAggregateType aggregate, CancellationToken cancellationToken) + { + if (ThrowOnRead) throw new InvalidOperationException("backend boom"); + LastCall = "Processed"; + LastTagname = fullReference; + LastStart = startUtc; + LastEnd = endUtc; + LastInterval = interval; + LastAggregate = aggregate; + return Task.FromResult(ProcessedResult); + } + + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) + { + if (ThrowOnRead) throw new InvalidOperationException("backend boom"); + LastCall = "AtTime"; + LastTagname = fullReference; + LastTimestamps = timestampsUtc; + return Task.FromResult(AtTimeResult); + } + + public Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, + CancellationToken cancellationToken) => + Task.FromResult(new HistoricalEventsResult(Array.Empty(), null)); + + public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); + + public void Dispose() + { + } + } + + private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync() + { + var host = new OpcUaApplicationHost( + new OpcUaApplicationHostOptions + { + ApplicationName = "OtOpcUa.HistoryReadTest", + ApplicationUri = $"urn:OtOpcUa.HistoryReadTest:{Guid.NewGuid():N}", + OpcUaPort = AllocateFreePort(), + PublicHostname = "localhost", + PkiStoreRoot = _pkiRoot, + }, + NullLogger.Instance); + + var server = new OtOpcUaSdkServer(); + await host.StartAsync(server, Ct); + return (host, server); + } + + private static int AllocateFreePort() + { + using var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0); + listener.Start(); + var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + /// Cleans up the PKI root directory. + public void Dispose() + { + if (Directory.Exists(_pkiRoot)) + { + try { Directory.Delete(_pkiRoot, recursive: true); } + catch { /* best-effort cleanup */ } + } + } +}