From f607ee75022e5080b5228f305c52dde23b6104ed Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 12 Nov 2026 09:59:35 -0500 Subject: [PATCH] test(historian): #801 real-AVEVA history gate over the SDK wire (env-gated) AvevaGatewayHistoryLiveGateTests boots the server in-process, wires the production Gateway backend (HistorianBackendRegistry -> GatewayHistorianDataSource) to a real HistorianGateway and drives HistoryRead over a real SDK session: raw (all, paged, reverse, bounds), modified, the ten served aggregates and seven refused ones, AtTime, events (all, 1-event pages, reverse), a missing tag, HistoryServerCapabilities and node metadata, and a HistoryUpdate that must be refused (sandbox-bound node only). Every wire answer is compared with the backend's direct answer; the run is dumped as JSON for cross-SDK and SQL comparison. Skips unless HISTGW_GATEWAY_ENDPOINT and HISTGW_GATEWAY_APIKEY are set; read-only by construction. scripts/validation/aveva-history-801: run/compare instructions, the SDK 1.5.378 baseline port patch (API spelling only), and the diff / SQL / aggregate / histgw-cli comparison scripts. Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi --- .../validation/aveva-history-801/README.md | 44 ++ .../aveva-history-801/aggregate_recompute.py | 25 + .../validation/aveva-history-801/diff_runs.py | 28 + .../aveva-history-801/hgcli_raw_check.py | 19 + .../sdk1-baseline-port.patch | 211 +++++++ .../aveva-history-801/sql_raw_check.py | 41 ++ .../AvevaGatewayHistoryLiveGateTests.cs | 578 ++++++++++++++++++ 7 files changed, 946 insertions(+) create mode 100644 scripts/validation/aveva-history-801/README.md create mode 100644 scripts/validation/aveva-history-801/aggregate_recompute.py create mode 100644 scripts/validation/aveva-history-801/diff_runs.py create mode 100644 scripts/validation/aveva-history-801/hgcli_raw_check.py create mode 100644 scripts/validation/aveva-history-801/sdk1-baseline-port.patch create mode 100644 scripts/validation/aveva-history-801/sql_raw_check.py create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs diff --git a/scripts/validation/aveva-history-801/README.md b/scripts/validation/aveva-history-801/README.md new file mode 100644 index 000000000..83ac2286a --- /dev/null +++ b/scripts/validation/aveva-history-801/README.md @@ -0,0 +1,44 @@ +# #801 real-AVEVA history gate (SDK 2.0.1 vs 1.5.378.182) + +Harness: `tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs`. +It boots the OtOpcUa server in-process (`ServerEventRig`), wires the **production** Gateway backend +(`HistorianBackendRegistry` → `GatewayHistorianDataSource`) to a real HistorianGateway, and drives HistoryRead / +HistoryUpdate over a real SDK client session. It skips unless `HISTGW_GATEWAY_ENDPOINT` and `HISTGW_GATEWAY_APIKEY` +are set. It is read-only: the Gateway backend has no updater, and the only HistoryUpdate goes to a node bound to +`HISTGW_WRITE_SANDBOX_TAG` and is refused by the server before any backend call. + +Record of the 2026-11-12 run: [`docs/plans/evidence/2026-11-10-opcua2/aveva-history-801.md`](../../../docs/plans/evidence/2026-11-10-opcua2/aveva-history-801.md). + +## Run (SDK 2, this tree) + +```bash +export HISTGW_GATEWAY_ENDPOINT=http://wonder-app-vd03.zmr.zimmer.com:5221 +export HISTGW_GATEWAY_APIKEY="$(cat /path/to/key-file)" # never on the command line +export HISTGW_WRITE_SANDBOX_TAG=HistGW.LiveTest.Sandbox +export OTOPCUA_AVEVA_GATE_OUT=/tmp/aveva-gate OTOPCUA_AVEVA_GATE_LABEL=sdk2-$(git rev-parse --short HEAD) +dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests -c Release \ + --filter "FullyQualifiedName~AvevaGatewayHistoryLiveGate" +``` + +## Baseline (SDK 1.5.378.182) + +```bash +git worktree add --detach ../otopcua-sdk1 efb9197a # master's first parent before the SDK 2 merge b36e688f +cp tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs \ + ../otopcua-sdk1/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ +(cd ../otopcua-sdk1 && patch -p1 < "$OLDPWD/scripts/validation/aveva-history-801/sdk1-baseline-port.patch") +# then the same dotnet test command in ../otopcua-sdk1 with OTOPCUA_AVEVA_GATE_LABEL=sdk1-efb9197a +``` + +`sdk1-baseline-port.patch` changes only SDK API spelling: collections in place of `ArrayOf`, `byte[]` in place of `ByteString`, +`DataValue.Value` in place of `ClrValue()`. The queries, windows and comparisons are unchanged. + +## Compare + +```bash +python3 -I diff_runs.py $OUT/aveva-gate-sdk2-*.json $OUT/aveva-gate-sdk1-efb9197a.json # every case, row and event +python3 -I sql_raw_check.py $OUT/aveva-gate-sdk2-*.json # raw rows vs Runtime.dbo.History (Full, UTC); needs + # HISTORIAN_SQL_USER / HISTORIAN_SQL_PASSWORD, read-only +python3 -I aggregate_recompute.py $OUT/aveva-gate-sdk2-*.json # simple aggregates recomputed from the raw rows +python3 -I hgcli_raw_check.py $OUT/aveva-gate-sdk2-*.json $OUT # vs histgw-cli read-raw output (hgcli--.jsonl) +``` diff --git a/scripts/validation/aveva-history-801/aggregate_recompute.py b/scripts/validation/aveva-history-801/aggregate_recompute.py new file mode 100644 index 000000000..9b971947b --- /dev/null +++ b/scripts/validation/aveva-history-801/aggregate_recompute.py @@ -0,0 +1,25 @@ +# Independent recomputation of the simple per-interval aggregates from the wire raw rows (Good samples only, [t, t+60s)). +import json, sys, math +from datetime import datetime, timedelta +rep = json.load(open(sys.argv[1])); C = rep['Cases'] +def ts(s): return datetime.fromisoformat(s.replace('Z', '+00:00')) +def get(name, tag, win): return next(c for c in C if c['Name'] == name and c['Tag'] == tag and c['Window'] == win and c['Path'] == 'wire') +checked = bad = 0 +for raw in [c for c in C if c['Name'] == 'raw.all' and c['Path'] == 'wire' and not c['Tag'].startswith('OtOpcUa.')]: + tag, win = raw['Tag'], raw['Window'] + samples = [(ts(r['Src']), float(r['V'])) for r in raw['Rows'] if r['V'] is not None and r['Status'] & 0xC0000000 == 0] + nongood = [ts(r['Src']) for r in raw['Rows'] if r['V'] is None or r['Status'] & 0xC0000000 != 0] + for agg in ['Count', 'Minimum', 'Maximum', 'Average', 'Range', 'Start', 'End', 'Delta']: + p = get(f'processed.{agg}', tag, win) + for row in p['Rows']: + w0 = ts(win.split('/')[0]); k = int((ts(row['Src']) - w0).total_seconds() // 60) + t0 = w0 + timedelta(seconds=60 * k); t1 = t0 + timedelta(seconds=60) # Start/End carry the sample's own time + xs = [v for (t, v) in samples if t0 <= t < t1] + if not xs or any(t0 <= t < t1 for t in nongood): continue # empty / outage intervals: status semantics are compared against SDK1, not recomputed here + exp = {'Count': len(xs), 'Minimum': min(xs), 'Maximum': max(xs), 'Average': sum(xs) / len(xs), + 'Range': max(xs) - min(xs), 'Start': xs[0], 'End': xs[-1], 'Delta': xs[-1] - xs[0]}[agg] + got = float(str(row['V']).split(':')[-1]) + checked += 1 + if not math.isclose(got, exp, rel_tol=1e-12, abs_tol=1e-9): + bad += 1; print('MISMATCH', tag, win[:19], agg, row, 'expected', exp) +print(f'{checked} non-empty interval values recomputed from raw, {bad} mismatching') diff --git a/scripts/validation/aveva-history-801/diff_runs.py b/scripts/validation/aveva-history-801/diff_runs.py new file mode 100644 index 000000000..15c75c769 --- /dev/null +++ b/scripts/validation/aveva-history-801/diff_runs.py @@ -0,0 +1,28 @@ +import json, sys, re +a = json.load(open(sys.argv[1])); b = json.load(open(sys.argv[2])) +def key(c): return (c['Name'], c['Tag'], c['Window'], c['Path']) +A = {key(c): c for c in a['Cases']}; B = {key(c): c for c in b['Cases']} +print(f"A={a['Label']} sdk={a['SdkServerAssembly']} cases={len(A)} failures={len(a['Failures'])}") +print(f"B={b['Label']} sdk={b['SdkServerAssembly']} cases={len(B)} failures={len(b['Failures'])}") +only = set(A) ^ set(B) +if only: print("cases only in one run:", sorted(only, key=str)) +diffs = 0; rows = 0; events = 0; same = 0 +def norm_note(n): return re.sub(r'elapsedMs=\d+', 'elapsedMs=*', n or '') +for k in sorted(set(A) & set(B), key=str): + x, y = A[k], B[k]; d = [] + if x['Status'] != y['Status']: d.append(f"status {x['Status']} vs {y['Status']}") + if norm_note(x['Note']) != norm_note(y['Note']): d.append(f"note {x['Note']} vs {y['Note']}") + if (x['Attributes'] or {}) != (y['Attributes'] or {}): d.append(f"attrs {x['Attributes']} vs {y['Attributes']}") + xr, yr = x['Rows'] or [], y['Rows'] or [] + rows += len(xr) + if len(xr) != len(yr): d.append(f"rows {len(xr)} vs {len(yr)}") + else: + for i, (p, q) in enumerate(zip(xr, yr)): + if (p['V'], p['Status'], p['Src'], p['Srv'] is None) != (q['V'], q['Status'], q['Src'], q['Srv'] is None): + d.append(f"row[{i}] {p} vs {q}"); break + xe, ye = x['Events'] or [], y['Events'] or [] + events += len(xe) + if xe != ye: d.append(f"events {len(xe)} vs {len(ye)}") + if d: diffs += 1; print("DIFF", k, "; ".join(d)[:500]) + else: same += 1 +print(f"compared {same+diffs} cases: identical={same} differing={diffs}; rows compared={rows}, events compared={events} (ServerTimestamp compared for presence only: it is the read time)") diff --git a/scripts/validation/aveva-history-801/hgcli_raw_check.py b/scripts/validation/aveva-history-801/hgcli_raw_check.py new file mode 100644 index 000000000..0952e0289 --- /dev/null +++ b/scripts/validation/aveva-history-801/hgcli_raw_check.py @@ -0,0 +1,19 @@ +# Compares the gate's raw.all wire rows with HistorianGateway's own smoke CLI (`histgw-cli read-raw`, JSON lines), +# an independent gRPC client. Usage: hgcli_raw_check.py +import json, sys, glob, os +from datetime import datetime +rep = json.load(open(sys.argv[1])) +def ts(s): return datetime.fromisoformat(s.replace('Z', '+00:00').replace('0000000+', '+')) +for f in sorted(glob.glob(os.path.join(sys.argv[2], 'hgcli-*.jsonl'))): + rows = [json.loads(l) for l in open(f)] + tag, start = rows[0]['tag'], rows[0]['timestamp'][:19] + c = next(c for c in rep['Cases'] if c['Name'] == 'raw.all' and c['Path'] == 'wire' and c['Tag'] == tag and c['Window'].startswith(start)) + w0, w1 = (ts(x) for x in c['Window'].split('/')) + # AVEVA's synthetic Quality-133 row at the query start is not a stored sample (#687); the gateway's end is inclusive, + # OtOpcUa's exclusive. + stored = [r for r in rows if not (r['quality'] == 133 and ts(r['timestamp']) == w0) and ts(r['timestamp']) < w1] + n = min(len(stored), len(c['Rows'])) + ok = all(abs((ts(a['timestamp']) - ts(b['Src'])).total_seconds()) < 5e-4 and float(a['value']) == float(b['V']) + for a, b in zip(stored[:n], c['Rows'][:n])) + short = ' (histgw-cli stopped at the stored NULL: its JSON writer cannot emit NaN)' if len(stored) < len(c['Rows']) else '' + print(f"{tag:<16} {start} histgw-cli={len(stored)} wire={len(c['Rows'])} compared={n} {'MATCH' if ok else 'MISMATCH'}{short}") diff --git a/scripts/validation/aveva-history-801/sdk1-baseline-port.patch b/scripts/validation/aveva-history-801/sdk1-baseline-port.patch new file mode 100644 index 000000000..edce1aa1a --- /dev/null +++ b/scripts/validation/aveva-history-801/sdk1-baseline-port.patch @@ -0,0 +1,211 @@ +--- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs ++++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs +@@ -15,7 +15,7 @@ + namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + + /// +-/// Gitea #801 (SDK 2 live gate, "real AVEVA history"): the booted OtOpcUa server, with the production ++/// Gitea #801 BASELINE PORT (SDK 1.5.378.182, master efb9197a) (SDK 2 live gate, "real AVEVA history"): the booted OtOpcUa server, with the production + /// Gateway backend ( → GatewayHistorianDataSource) pointed at a REAL + /// HistorianGateway in front of the real AVEVA Historian, serves HistoryRead over a real SDK client session. + /// Every wire answer is compared with the same backend's direct answer +@@ -283,11 +283,11 @@ + { + var t = DateTime.UtcNow.AddHours(-1); + var insert = new UpdateDataDetails { NodeId = Node(sandboxNode), PerformInsertReplace = PerformUpdateType.Insert }; +- insert.UpdateValues += new DataValue(Variant.From(801.0), StatusCodes.Good, t, t); ++ insert.UpdateValues.Add(new DataValue(new Variant(801.0), StatusCodes.Good, t, t)); + var del = new DeleteRawModifiedDetails { NodeId = Node(sandboxNode), IsDeleteModified = false, StartTime = t, EndTime = t.AddSeconds(1) }; + foreach (var (name, details) in new (string, IEncodeable)[] { ("insert", insert), ("deleteRaw", del) }) + { +- var response = await session.HistoryUpdateAsync(null, new ExtensionObject[] { new(details) }, ct); ++ var response = await session.HistoryUpdateAsync(null, new ExtensionObjectCollection { new ExtensionObject(details) }, ct); + var result = response.Results.ToList().ShouldHaveSingleItem(); + var status = Name(result.StatusCode); + report.Add(new Case($"historyUpdate.{name}", sandboxTag, null, "wire", status, +@@ -316,7 +316,7 @@ + { + StartTime = start, EndTime = end, NumValuesPerNode = max, IsReadModified = false, ReturnBounds = returnBounds, + }; +- var (status, rows, cp) = await HistoryReadAsync(session, details, node, default, ct); ++ var (status, rows, cp) = await HistoryReadAsync(session, details, node, null, ct); + await ReleaseAsync(session, details, node, cp, ct); + return new WireRead(status, rows, 1); + } +@@ -330,7 +330,7 @@ + }; + var all = new List(); + var pages = 0; +- ByteString cp = default; ++ byte[]? cp = null; + string status; + do + { +@@ -339,14 +339,14 @@ + pages++; + if (status.StartsWith("Bad", StringComparison.Ordinal)) break; + } +- while (!cp.IsEmpty && pages < 10_000); ++ while (cp is { Length: > 0 } && pages < 10_000); + return new WireRead(status, all, pages); + } + + private static async Task ReadModifiedAsync(ISession session, NodeId node, DateTime start, DateTime end, CancellationToken ct) + { + var details = new ReadRawModifiedDetails { StartTime = start, EndTime = end, NumValuesPerNode = 0, IsReadModified = true, ReturnBounds = false }; +- var (status, _, cp) = await HistoryReadAsync(session, details, node, default, ct); ++ var (status, _, cp) = await HistoryReadAsync(session, details, node, null, ct); + await ReleaseAsync(session, details, node, cp, ct); + return status; + } +@@ -357,12 +357,12 @@ + var details = new ReadProcessedDetails + { + StartTime = start, EndTime = end, ProcessingInterval = IntervalMs, +- AggregateType = new List { aggregate }, ++ AggregateType = new NodeIdCollection { aggregate }, + AggregateConfiguration = new AggregateConfiguration { UseServerCapabilitiesDefaults = true }, + }; + var all = new List(); + var pages = 0; +- ByteString cp = default; ++ byte[]? cp = null; + string status; + do + { +@@ -371,36 +371,36 @@ + pages++; + if (status.StartsWith("Bad", StringComparison.Ordinal)) break; + } +- while (!cp.IsEmpty && pages < 1000); ++ while (cp is { Length: > 0 } && pages < 1000); + return new WireRead(status, all, pages); + } + + private static async Task ReadAtTimeAsync(ISession session, NodeId node, IReadOnlyList times, CancellationToken ct) + { +- var details = new ReadAtTimeDetails { ReqTimes = times.Select(t => (DateTimeUtc)t).ToList(), UseSimpleBounds = true }; +- var (status, rows, cp) = await HistoryReadAsync(session, details, node, default, ct); ++ var details = new ReadAtTimeDetails { ReqTimes = new DateTimeCollection(times), UseSimpleBounds = true }; ++ var (status, rows, cp) = await HistoryReadAsync(session, details, node, null, ct); + await ReleaseAsync(session, details, node, cp, ct); + return new WireRead(status, rows, 1); + } + +- private static async Task<(string Status, List Rows, ByteString Cp)> HistoryReadAsync( +- ISession session, IEncodeable details, NodeId node, ByteString cp, CancellationToken ct) ++ private static async Task<(string Status, List Rows, byte[]? Cp)> HistoryReadAsync( ++ ISession session, IEncodeable details, NodeId node, byte[]? cp, CancellationToken ct) + { + var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, false, +- new HistoryReadValueId[] { new() { NodeId = node, ContinuationPoint = cp } }, ct); ++ new HistoryReadValueIdCollection { new HistoryReadValueId { NodeId = node, ContinuationPoint = cp } }, ct); + var result = response.Results.ToList().ShouldHaveSingleItem(); + var rows = new List(); +- if (!result.HistoryData.IsNull && ExtensionObject.ToEncodeable(result.HistoryData) is HistoryData data) ++ if (result.HistoryData?.Body is HistoryData data) + rows.AddRange(data.DataValues.ToList().Select(dv => new Row( +- Fmt(dv.ClrValue()), dv.StatusCode.Code, Ts(dv.SourceTimestamp), Ts(dv.ServerTimestamp)))); ++ Fmt(dv.Value), dv.StatusCode.Code, Ts(dv.SourceTimestamp), Ts(dv.ServerTimestamp)))); + return (Name(result.StatusCode), rows, result.ContinuationPoint); + } + +- private static async Task ReleaseAsync(ISession session, IEncodeable details, NodeId node, ByteString cp, CancellationToken ct) ++ private static async Task ReleaseAsync(ISession session, IEncodeable details, NodeId node, byte[]? cp, CancellationToken ct) + { +- if (cp.IsEmpty) return; ++ if (cp is not { Length: > 0 }) return; + await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, true, +- new HistoryReadValueId[] { new() { NodeId = node, ContinuationPoint = cp } }, ct); ++ new HistoryReadValueIdCollection { new HistoryReadValueId { NodeId = node, ContinuationPoint = cp } }, ct); + } + + private static async Task ReadEventsAsync(ISession session, NodeId node, DateTime start, DateTime end, uint page, +@@ -408,33 +408,33 @@ + { + var filter = new EventFilter(); + foreach (var field in new[] { "EventId", "Time", "ReceiveTime", "SourceName", "Message", "Severity" }) +- filter.SelectClauses += new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, new QualifiedName(field)) { AttributeId = Attributes.Value }; ++ filter.SelectClauses.Add(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, new QualifiedName(field)) { AttributeId = Attributes.Value }); + var details = new ReadEventDetails { StartTime = start, EndTime = end, NumValuesPerNode = page, Filter = filter }; + var rows = new List(); + var pages = 0; +- ByteString cp = default; ++ byte[]? cp = null; + string status; + do + { + var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, false, +- new HistoryReadValueId[] { new() { NodeId = node, ContinuationPoint = cp } }, ct); ++ new HistoryReadValueIdCollection { new HistoryReadValueId { NodeId = node, ContinuationPoint = cp } }, ct); + var result = response.Results.ToList().ShouldHaveSingleItem(); + status = Name(result.StatusCode); + cp = result.ContinuationPoint; + pages++; +- if (!result.HistoryData.IsNull && ExtensionObject.ToEncodeable(result.HistoryData) is HistoryEvent he) ++ if (result.HistoryData?.Body is HistoryEvent he) + foreach (var e in he.Events.ToList()) + { + var f = e.EventFields.ToList(); + rows.Add(new EventRow( +- f[0].ClrValue() is byte[] id ? System.Text.Encoding.UTF8.GetString(id) : Fmt(f[0].ClrValue()), +- f[1].ClrValue() as DateTime? ?? default, f[2].ClrValue() as DateTime? ?? default, +- f[3].ClrValue() as string, f[4].ClrValue() is LocalizedText lt ? lt.Text : Fmt(f[4].ClrValue()), +- Convert.ToUInt16(f[5].ClrValue() ?? 0, CultureInfo.InvariantCulture))); ++ f[0].Value is byte[] id ? System.Text.Encoding.UTF8.GetString(id) : Fmt(f[0].Value), ++ f[1].Value as DateTime? ?? default, f[2].Value as DateTime? ?? default, ++ f[3].Value as string, f[4].Value is LocalizedText lt ? lt.Text : Fmt(f[4].Value), ++ Convert.ToUInt16(f[5].Value ?? 0, CultureInfo.InvariantCulture))); + } + if (status.StartsWith("Bad", StringComparison.Ordinal)) break; + } +- while (!cp.IsEmpty && pages < 10_000); ++ while (cp is { Length: > 0 } && pages < 10_000); + return new EventRead(status, rows, pages); + } + +@@ -446,9 +446,9 @@ + ("UserAccessLevel", Attributes.UserAccessLevel), ("DataType", Attributes.DataType), + }; + var response = await session.ReadAsync(null, 0, TimestampsToReturn.Neither, +- ids.Select(a => new ReadValueId { NodeId = node, AttributeId = a.Id }).ToArray(), ct); ++ new ReadValueIdCollection(ids.Select(a => new ReadValueId { NodeId = node, AttributeId = a.Id })), ct); + var results = response.Results.ToList(); +- return ids.Select((a, i) => (a.Name, Value: Fmt(results[i].ClrValue()))) ++ return ids.Select((a, i) => (a.Name, Value: Fmt(results[i].Value))) + .ToDictionary(x => x.Name, x => x.Value); + } + +@@ -460,9 +460,9 @@ + if (vars.Count > 0) + { + var response = await session.ReadAsync(null, 0, TimestampsToReturn.Neither, +- vars.Select(r => new ReadValueId { NodeId = ExpandedNodeId.ToNodeId(r.NodeId, session.NamespaceUris), AttributeId = Attributes.Value }).ToArray(), ct); ++ new ReadValueIdCollection(vars.Select(r => new ReadValueId { NodeId = ExpandedNodeId.ToNodeId(r.NodeId, session.NamespaceUris), AttributeId = Attributes.Value })), ct); + var results = response.Results.ToList(); +- for (var i = 0; i < vars.Count; i++) caps[vars[i].BrowseName.Name!] = Fmt(results[i].ClrValue()); ++ for (var i = 0; i < vars.Count; i++) caps[vars[i].BrowseName.Name!] = Fmt(results[i].Value); + } + var aggs = await BrowseAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ReferenceTypeIds.Organizes, ct); + caps["AggregateFunctions"] = string.Join(",", aggs.Select(r => r.BrowseName.Name).OrderBy(n => n, StringComparer.Ordinal)); +@@ -472,9 +472,9 @@ + private static async Task> BrowseAsync(ISession session, NodeId node, NodeId referenceType, CancellationToken ct) + { + var response = await session.BrowseAsync(null, null, 0, +- new BrowseDescription[] ++ new BrowseDescriptionCollection + { +- new() ++ new BrowseDescription + { + NodeId = node, BrowseDirection = BrowseDirection.Forward, ReferenceTypeId = referenceType, + IncludeSubtypes = true, NodeClassMask = 0, ResultMask = (uint)BrowseResultMask.All, +@@ -483,7 +483,7 @@ + return response.Results[0].References.ToList(); + } + +- private static DateTime? Ts(DateTimeUtc t) => (DateTime)t == DateTime.MinValue ? null : DateTime.SpecifyKind((DateTime)t, DateTimeKind.Utc); ++ private static DateTime? Ts(DateTime t) => t == DateTime.MinValue ? null : DateTime.SpecifyKind(t, DateTimeKind.Utc); + + private static string Name(StatusCode code) => + code.SymbolicId is { Length: > 0 } s ? s : $"0x{code.Code:X8}"; diff --git a/scripts/validation/aveva-history-801/sql_raw_check.py b/scripts/validation/aveva-history-801/sql_raw_check.py new file mode 100644 index 000000000..9843fdeee --- /dev/null +++ b/scripts/validation/aveva-history-801/sql_raw_check.py @@ -0,0 +1,41 @@ +# Compares the gate's raw.all wire rows with direct read-only SQL on the AVEVA Runtime DB (History, Full, UTC). +import json, sys, os, subprocess, math +from datetime import datetime, timezone +rep = json.load(open(sys.argv[1])) +def sql(q): + env = dict(os.environ, SQLCMDPASSWORD=os.environ['HISTORIAN_SQL_PASSWORD']) + out = subprocess.run(['sqlcmd', '-S', 'wonder-sql-vd03.zmr.zimmer.com,1433', '-U', os.environ['HISTORIAN_SQL_USER'], '-C', + '-d', 'Runtime', '-h', '-1', '-W', '-s', '|', '-Q', 'SET NOCOUNT ON; ' + q], + env=env, capture_output=True, text=True, timeout=120) + if out.returncode: raise SystemExit(out.stderr + out.stdout) + return [l.split('|') for l in out.stdout.splitlines() if l.strip()] +def iso(s): # '2026-11-10 11:30:01.6200000' -> datetime + return datetime.strptime(s[:23], '%Y-%m-%d %H:%M:%S.%f').replace(tzinfo=timezone.utc) +def wts(s): return datetime.fromisoformat(s.replace('Z', '+00:00')) +total = bad = 0 +for c in rep['Cases']: + if c['Name'] != 'raw.all' or c['Path'] != 'wire' or c['Tag'].startswith('OtOpcUa.'): continue + start, end = c['Window'].split('/') + s0, e0 = wts(start), wts(end) + rows = sql(f"SELECT CONVERT(varchar(27), DateTime, 121), CONVERT(varchar(40), Value, 3), Quality, QualityDetail, OPCQuality " + f"FROM dbo.History WHERE TagName = '{c['Tag']}' AND wwRetrievalMode = 'Full' AND wwTimeZone = 'UTC' " + f"AND DateTime >= '{s0:%Y-%m-%d %H:%M:%S}' AND DateTime < '{e0:%Y-%m-%d %H:%M:%S}' ORDER BY DateTime") + # AVEVA's synthetic boundary row (Quality 133) at the query start is not a stored sample (#687): OtOpcUa drops it. + stored = [r for r in rows if not (r[2] == '133' and iso(r[0]) == s0)] + wire = c['Rows'] + mism = [] + if len(stored) != len(wire): mism.append(f"count sql={len(stored)} wire={len(wire)}") + for i, (r, w) in enumerate(zip(stored, wire)): + t_ok = abs((iso(r[0]) - wts(w['Src'])).total_seconds()) < 0.0005 + sv = None if r[1] == 'NULL' else float(r[1]) + wv = None if w['V'] is None else float(w['V']) + v_ok = (sv is None and wv is None) or (sv is not None and wv is not None and sv == wv) + good = r[4] == '192' + q_ok = (w['Status'] & 0xC0000000) == (0 if good else 0x80000000) + if not (t_ok and v_ok and q_ok): mism.append(f"[{i}] sql={r} wire={w}"); break + total += 1; bad += bool(mism) + q = {} + for r in stored: q[(r[2], r[3], r[4])] = q.get((r[2], r[3], r[4]), 0) + 1 + print(f"{c['Tag']:<30} {start[:19]} sql_stored={len(stored)} (boundary rows dropped={len(rows)-len(stored)}) wire={len(wire)} " + f"quality(Q,QD,OPC)={q} -> {'MATCH' if not mism else 'MISMATCH ' + '; '.join(mism)}") +print(f"{total} tag-windows compared, {bad} mismatching") diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs new file mode 100644 index 000000000..2af2f5689 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AvevaGatewayHistoryLiveGateTests.cs @@ -0,0 +1,578 @@ +using System.Globalization; +using System.Text.Json; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Opc.Ua; +using Opc.Ua.Client; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Host.Configuration; +using ZB.MOM.WW.OtOpcUa.Runtime.Historian; +using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Gitea #801 (SDK 2 live gate, "real AVEVA history"): the booted OtOpcUa server, with the production +/// Gateway backend ( → GatewayHistorianDataSource) pointed at a REAL +/// HistorianGateway in front of the real AVEVA Historian, serves HistoryRead over a real SDK client session. +/// Every wire answer is compared with the same backend's direct answer +/// (value, status, source timestamp), and the whole run is dumped as JSON so the same queries can be diffed +/// against another SDK build and against direct SQL on the historian's Runtime DB. +/// +/// +/// Read-only by construction. The Gateway backend has no (its +/// provider returns none), no provisioner, recorder or alarm writer is built here, and the one HistoryUpdate this +/// test sends targets a node bound to the designated write-sandbox tag (HISTGW_WRITE_SANDBOX_TAG) and must be +/// refused by the server before any backend is reached. Without that variable the HistoryUpdate case is skipped. +/// Env (skips when the endpoint or key is unset): HISTGW_GATEWAY_ENDPOINT (e.g. +/// http://wonder-app-vd03.zmr.zimmer.com:5221), HISTGW_GATEWAY_APIKEY (a key with historian:read; +/// never logged), optional HISTGW_GATEWAY_ALLOW_UNTRUSTED, HISTGW_WRITE_SANDBOX_TAG, +/// OTOPCUA_AVEVA_GATE_OUT (output directory for aveva-gate-*.json), OTOPCUA_AVEVA_GATE_LABEL +/// (a run label recorded in the output, e.g. the SDK version), OTOPCUA_AVEVA_GATE_TAGS (comma list of +/// historized analog tags), OTOPCUA_AVEVA_GATE_WINDOWS (start/end;start/end, ISO-8601 UTC), +/// OTOPCUA_AVEVA_GATE_EVENT_SOURCES (comma list) and OTOPCUA_AVEVA_GATE_EVENT_WINDOW. The defaults are the DEV +/// (wonder-app-vd03 / wonder-sql-vd03) tags and fixed past windows used for the 2026-11-12 run. +/// The local deterministic fixtures are NOT this test; only a real gateway in front of a real AVEVA +/// Historian makes its output AVEVA evidence. +/// +public sealed class AvevaGatewayHistoryLiveGateTests +{ + private static readonly string[] DefaultTags = ["ZIPSEN.WTC1", "ZIPSEN.OvenPressureTransducer", "SysPerfCPUTotal"]; + + // W1 spans the 2026-11-10 11:32:36Z platform stop on DEV (a stored NULL / Bad row, then a gap and a + // recovery row); W2 is a quiet stretch. Both are in the past, so every run (and every SDK build) sees the same data. + private const string DefaultWindows = "2026-11-10T11:30:00Z/2026-11-10T11:41:00Z;2026-11-11T14:00:00Z/2026-11-11T14:10:00Z"; + // AVEVA A&E source names as the gateway reports them. The first must have events in the window (asserted): the + // platform object, whose System.Stop / System.Start bracket the W1 outage. The second is an attribute alarm the + // gateway reports with SourceName "DevWinPlatform.CPULoadAvg" (its Source_ProcessVariable; Source_Object is + // DevWinPlatform), but a per-source ReadEvents for that name returns nothing — on SDK 1.5.378 and 2.0.1 alike. Recorded + // (wire must equal the backend), not required to be non-empty. + private const string DefaultEventSources = "DevWinPlatform,DevWinPlatform.CPULoadAvg"; + private const string DefaultEventWindow = "2026-11-05T15:00:00Z/2026-11-12T00:00:00Z"; + private const string MissingTag = "OtOpcUa.Gate801.DoesNotExist"; + private const double IntervalMs = 60_000; + + // The ten per-sample aggregates the Gateway backend serves (#681) and a set it must refuse (time-weighted / never). + private static readonly (string Name, NodeId Id, HistoryAggregateType? Domain)[] Served = + [ + ("Average", ObjectIds.AggregateFunction_Average, HistoryAggregateType.Average), + ("Count", ObjectIds.AggregateFunction_Count, HistoryAggregateType.Count), + ("Minimum", ObjectIds.AggregateFunction_Minimum, HistoryAggregateType.Minimum), + ("Maximum", ObjectIds.AggregateFunction_Maximum, HistoryAggregateType.Maximum), + ("Range", ObjectIds.AggregateFunction_Range, HistoryAggregateType.Range), + ("Start", ObjectIds.AggregateFunction_Start, HistoryAggregateType.Start), + ("End", ObjectIds.AggregateFunction_End, HistoryAggregateType.End), + ("Delta", ObjectIds.AggregateFunction_Delta, HistoryAggregateType.Delta), + ("MinimumActualTime", ObjectIds.AggregateFunction_MinimumActualTime, HistoryAggregateType.MinimumActualTime), + ("MaximumActualTime", ObjectIds.AggregateFunction_MaximumActualTime, HistoryAggregateType.MaximumActualTime), + ]; + + private static readonly (string Name, NodeId Id)[] Refused = + [ + ("TimeAverage", ObjectIds.AggregateFunction_TimeAverage), + ("Total", ObjectIds.AggregateFunction_Total), + ("TimeAverage2", ObjectIds.AggregateFunction_TimeAverage2), + ("Interpolative", ObjectIds.AggregateFunction_Interpolative), + ("PercentGood", ObjectIds.AggregateFunction_PercentGood), + ("DurationGood", ObjectIds.AggregateFunction_DurationGood), + ("StandardDeviationSample", ObjectIds.AggregateFunction_StandardDeviationSample), + ]; + + [Fact] + public async Task Real_AVEVA_history_served_over_the_wire_matches_the_gateway_backend() + { + var endpoint = Env("HISTGW_GATEWAY_ENDPOINT"); + var apiKey = Env("HISTGW_GATEWAY_APIKEY"); + if (endpoint is null || apiKey is null) + Assert.Skip("Set HISTGW_GATEWAY_ENDPOINT and HISTGW_GATEWAY_APIKEY to run the #801 real-AVEVA history gate."); + + var ct = TestContext.Current.CancellationToken; + var label = Env("OTOPCUA_AVEVA_GATE_LABEL") ?? "unlabelled"; + var tags = (Env("OTOPCUA_AVEVA_GATE_TAGS") ?? string.Join(',', DefaultTags)) + .Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); + var windows = ParseWindows(Env("OTOPCUA_AVEVA_GATE_WINDOWS") ?? DefaultWindows); + var eventSources = (Env("OTOPCUA_AVEVA_GATE_EVENT_SOURCES") ?? DefaultEventSources) + .Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); + var eventWindow = ParseWindows(Env("OTOPCUA_AVEVA_GATE_EVENT_WINDOW") ?? DefaultEventWindow)[0]; + var sandboxTag = Env("HISTGW_WRITE_SANDBOX_TAG"); + + // The production backend path: the same registry + provider + factory the Host uses for Backend=Gateway. + var options = new ServerHistorianOptions + { + Backend = "Gateway", + Enabled = true, + Endpoint = endpoint!, + ApiKey = apiKey!, + UseTls = endpoint!.StartsWith("https://", StringComparison.OrdinalIgnoreCase), + AllowUntrustedServerCertificate = string.Equals(Env("HISTGW_GATEWAY_ALLOW_UNTRUSTED"), "true", StringComparison.OrdinalIgnoreCase), + }; + var provider = HistorianBackendRegistry.Resolve(options, new ConfigurationBuilder().Build()); + provider.CreateUpdater(new ServiceCollection().BuildServiceProvider()) + .ShouldBeNull("the Gateway backend must not offer HistoryUpdate — this gate never writes through it"); + using var source = provider.CreateDataSource(new ServiceCollection().BuildServiceProvider()); + + var report = new Report(label, endpoint!); + var failures = new List(); + + await using var rig = await StartAsync("AvevaGate801", encrypted: false, ct); + rig.Nm.HistorianDataSource = source; + // The Host copies these from ServerHistorianOptions (OtOpcUaServerHostedService); defaults here. + rig.Nm.MaxTieClusterOverfetch = options.MaxTieClusterOverfetch; + rig.Nm.HonorReturnBounds = options.HonorReturnBounds; + rig.Nm.EventReadHorizon = options.EventReadHorizon; + rig.Nm.HistoryReadBatchParallelism = options.HistoryReadBatchParallelism; + rig.Nm.MaxConcurrentHistoryReadBatches = options.MaxConcurrentHistoryReadBatches; + rig.Nm.HistoryReadDeadline = options.HistoryReadDeadline; + rig.Enforce(false, [Key()]); + + var nodes = new Dictionary(StringComparer.Ordinal); + for (var i = 0; i < tags.Length; i++) nodes[tags[i]] = rig.EnsureRawVariable($"Hist{i}", "Double", writable: false, tags[i]); + var missingNode = rig.EnsureRawVariable("HistMissing", "Double", writable: false, MissingTag); + var sysTimeNode = rig.EnsureRawVariable("HistSysTimeSec", "Double", writable: false, "SysTimeSec"); + var sandboxNode = sandboxTag is null ? null : rig.EnsureRawVariable("HistSandbox", "Double", writable: false, sandboxTag); + // An event-history source: a folder whose id is the historian source name, promoted by a native condition. + foreach (var eventSource in eventSources) + { + rig.Sink.EnsureFolder(eventSource, parentNodeId: null, displayName: eventSource, realm: AddressSpaceRealm.Raw); + rig.Sink.MaterialiseAlarmCondition($"{eventSource}/gate801", eventSource, "gate801", "OffNormalAlarm", 500, + AddressSpaceRealm.Raw, isNative: true); + } + + using var session = await rig.ConnectAsync(anonymous: true, ct); + var ns = (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri); + NodeId Node(string id) => new(id, ns); + + // ---- Metadata: node attributes and the HistoryServerCapabilities the server advertises. + foreach (var (tag, id) in nodes.Take(1)) + { + var attrs = await ReadAttributesAsync(session, Node(id), ct); + report.Add(new Case("metadata.node", tag, null, "wire", "Good", null, attrs)); + Check(failures, attrs["Historizing"] == "Boolean:True", "Historizing must be true on a historized node"); + var access = byte.Parse(attrs["AccessLevel"]!.Replace("Byte:", "", StringComparison.Ordinal), CultureInfo.InvariantCulture); + Check(failures, (access & AccessLevels.HistoryRead) != 0, "AccessLevel must carry HistoryRead"); + Check(failures, (access & AccessLevels.HistoryWrite) == 0, "AccessLevel must not carry HistoryWrite on the Gateway backend"); + } + var caps = await ReadHistoryCapabilitiesAsync(session, ct); + report.Add(new Case("metadata.historyServerCapabilities", null, null, "wire", "Good", null, caps)); + Check(failures, caps["AggregateFunctions"] == string.Join(",", Served.Select(a => a.Name).OrderBy(n => n, StringComparer.Ordinal)), + $"advertised history aggregates must be exactly the Gateway's ten, got {caps["AggregateFunctions"]}"); + + foreach (var (tag, id) in nodes) + { + foreach (var (start, end) in windows) + { + var w = $"{start:o}/{end:o}"; + + // ---- Raw, all values, no bounds. + var wire = await ReadRawAsync(session, Node(id), start, end, 0, returnBounds: false, ct); + var direct = await source.ReadRawAsync(tag, start, end, 0, ct); + report.Add(new Case("raw.all", tag, w, "wire", wire.Status, null, null, wire.Rows)); + report.Add(new Case("raw.all", tag, w, "direct", "Good", null, null, Rows(direct.Samples))); + CheckSame(failures, $"raw.all {tag} {w}", wire.Rows, Rows(direct.Samples)); + Check(failures, wire.Rows.Count > 0, $"raw.all {tag} {w}: expected stored samples"); + + // ---- Raw, paged through continuation points (page 37). + var paged = await ReadRawPagedAsync(session, Node(id), start, end, 37, ct); + report.Add(new Case("raw.paged37", tag, w, "wire", paged.Status, $"pages={paged.Pages}", null, paged.Rows)); + CheckSame(failures, $"raw.paged37 {tag} {w}", paged.Rows, wire.Rows); + + // ---- Raw, reverse (StartTime = end > EndTime = start), first page of 25, newest first. Part 11: the + // StartTime is included and the EndTime excluded, so the reverse window is (start, end] — a sample stored + // exactly at `end` is in it (and not in the forward read). Reference: the backend's (start, end] forward. + var reverse = await ReadRawAsync(session, Node(id), end, start, 25, returnBounds: false, ct); + var reverseRef = await source.ReadRawAsync(tag, start.AddTicks(1), end.AddTicks(1), 0, ct); + report.Add(new Case("raw.reverse25", tag, w, "wire", reverse.Status, null, null, reverse.Rows)); + CheckSame(failures, $"raw.reverse25 {tag} {w}", reverse.Rows, + Rows(reverseRef.Samples).AsEnumerable().Reverse().Take(25).ToList()); + + // ---- Raw with ReturnBounds (Part 11 bounding values; served by the server, #691). + var bounds = await ReadRawAsync(session, Node(id), start, end, 0, returnBounds: true, ct); + report.Add(new Case("raw.bounds", tag, w, "wire", bounds.Status, null, null, bounds.Rows)); + Check(failures, bounds.Rows.Count >= wire.Rows.Count, $"raw.bounds {tag} {w}: fewer rows than raw"); + + // ---- Modified (unsupported on Gateway). + var modified = await ReadModifiedAsync(session, Node(id), start, end, ct); + report.Add(new Case("modified", tag, w, "wire", modified, null, null)); + Check(failures, modified == "BadHistoryOperationUnsupported", $"modified {tag} {w}: got {modified}"); + + // ---- Processed: the ten served aggregates, wire vs the backend's own answer. + foreach (var (name, aggId, domain) in Served) + { + var p = await ReadProcessedAsync(session, Node(id), start, end, aggId, ct); + var d = await source.ReadProcessedAsync(tag, start, end, TimeSpan.FromMilliseconds(IntervalMs), domain!.Value, ct); + report.Add(new Case($"processed.{name}", tag, w, "wire", p.Status, null, null, p.Rows)); + report.Add(new Case($"processed.{name}", tag, w, "direct", "Good", null, null, Rows(d.Samples))); + CheckSame(failures, $"processed.{name} {tag} {w}", p.Rows, Rows(d.Samples)); + } + foreach (var (name, aggId) in Refused) + { + var p = await ReadProcessedAsync(session, Node(id), start, end, aggId, ct); + report.Add(new Case($"processed.refused.{name}", tag, w, "wire", p.Status, null, null, p.Rows)); + Check(failures, p.Status == "BadAggregateNotSupported", $"processed.{name} {tag} {w}: got {p.Status}"); + } + + // ---- AtTime: one stored timestamp, a midpoint between two stored samples, and the window start. + var times = new List { start }; + if (wire.Rows.Count >= 2) + { + times.Add(wire.Rows[1].Src!.Value); + times.Add(wire.Rows[0].Src!.Value + (wire.Rows[1].Src!.Value - wire.Rows[0].Src!.Value) / 2); + } + var at = await ReadAtTimeAsync(session, Node(id), times, ct); + var atDirect = await source.ReadAtTimeAsync(tag, times, TypeCode.Double, ct); + report.Add(new Case("attime", tag, w, "wire", at.Status, string.Join(",", times.Select(t => t.ToString("o"))), null, at.Rows)); + report.Add(new Case("attime", tag, w, "direct", "Good", null, null, Rows(atDirect.Samples))); + CheckSame(failures, $"attime {tag} {w}", at.Rows, Rows(atDirect.Samples)); + } + } + + // ---- A tag the historian does not have (#653): no data, not an error. + { + var (start, end) = windows[0]; + var r = await ReadRawAsync(session, Node(missingNode), start, end, 0, returnBounds: false, ct); + report.Add(new Case("raw.missingTag", MissingTag, $"{start:o}/{end:o}", "wire", r.Status, null, null, r.Rows)); + Check(failures, r.Rows.Count == 0 && !r.Status.StartsWith("Bad", StringComparison.Ordinal), $"missing tag: {r.Status} rows={r.Rows.Count}"); + } + + // ---- SysTimeSec raw (historiangw#29 reported server-side timeouts on 2023 R2) — recorded, not asserted. + { + var (start, _) = windows[^1]; + var sw = System.Diagnostics.Stopwatch.StartNew(); + var r = await ReadRawAsync(session, Node(sysTimeNode), start, start.AddSeconds(10), 0, returnBounds: false, ct); + report.Add(new Case("raw.SysTimeSec.known29", "SysTimeSec", $"{start:o}/{start.AddSeconds(10):o}", "wire", r.Status, + $"elapsedMs={sw.ElapsedMilliseconds}", null, r.Rows)); + } + + // ---- Events: the historian's own A&E rows for one source, wire (whole + paged) vs the backend. + { + var (start, end) = eventWindow; + var w = $"{start:o}/{end:o}"; + var discovered = await source.ReadEventsAsync(null, start, end, 0, ct); + report.Add(new Case("events.unfilteredBySource", null, w, "direct", "Good", + string.Join(",", discovered.Events.GroupBy(e => e.SourceName ?? "").OrderBy(g => g.Key, StringComparer.Ordinal) + .Select(g => $"{g.Key}={g.Count()}")), null)); + foreach (var eventSource in eventSources) + { + var directEvents = await source.ReadEventsAsync(eventSource, start, end, 0, ct); + var directRows = directEvents.Events.Select(e => new EventRow(e.EventId, e.EventTimeUtc, e.ReceivedTimeUtc, e.SourceName, + e.Message, e.Severity)).ToList(); + var all = await ReadEventsAsync(session, Node(eventSource), start, end, 0, ct); + var paged = await ReadEventsAsync(session, Node(eventSource), start, end, 1, ct); + var reverse = await ReadEventsAsync(session, Node(eventSource), end, start, 0, ct); + report.Add(new Case("events.all", eventSource, w, "wire", all.Status, $"pages={all.Pages}", null, null, all.Rows)); + report.Add(new Case("events.paged1", eventSource, w, "wire", paged.Status, $"pages={paged.Pages}", null, null, paged.Rows)); + report.Add(new Case("events.reverse", eventSource, w, "wire", reverse.Status, $"pages={reverse.Pages}", null, null, reverse.Rows)); + report.Add(new Case("events.all", eventSource, w, "direct", "Good", null, null, null, directRows)); + if (eventSource == eventSources[0]) + Check(failures, directRows.Count > 0, $"events {eventSource} {w}: expected stored events"); + CheckSameEvents(failures, $"events.all {eventSource}", all.Rows, directRows); + CheckSameEvents(failures, $"events.paged1 {eventSource}", paged.Rows, directRows); + CheckSameEvents(failures, $"events.reverse {eventSource}", reverse.Rows, directRows.AsEnumerable().Reverse().ToList()); + } + } + + // ---- HistoryUpdate: refused by the server (no updater, no HistoryWrite bit). Sandbox-bound node only. + if (sandboxNode is null) + { + report.Add(new Case("historyUpdate", null, null, "wire", "SKIPPED", "HISTGW_WRITE_SANDBOX_TAG not set", null)); + } + else + { + var t = DateTime.UtcNow.AddHours(-1); + var insert = new UpdateDataDetails { NodeId = Node(sandboxNode), PerformInsertReplace = PerformUpdateType.Insert }; + insert.UpdateValues += new DataValue(Variant.From(801.0), StatusCodes.Good, t, t); + var del = new DeleteRawModifiedDetails { NodeId = Node(sandboxNode), IsDeleteModified = false, StartTime = t, EndTime = t.AddSeconds(1) }; + foreach (var (name, details) in new (string, IEncodeable)[] { ("insert", insert), ("deleteRaw", del) }) + { + var response = await session.HistoryUpdateAsync(null, new ExtensionObject[] { new(details) }, ct); + var result = response.Results.ToList().ShouldHaveSingleItem(); + var status = Name(result.StatusCode); + report.Add(new Case($"historyUpdate.{name}", sandboxTag, null, "wire", status, + string.Join(",", result.OperationResults.ToList().Select(Name)), null)); + Check(failures, status == "BadHistoryOperationUnsupported", $"historyUpdate.{name}: got {status}"); + } + } + + var outDir = Env("OTOPCUA_AVEVA_GATE_OUT") ?? Path.GetTempPath(); + Directory.CreateDirectory(outDir); + var file = Path.Combine(outDir, $"aveva-gate-{Sanitize(label)}.json"); + report.Failures = failures; + await File.WriteAllTextAsync(file, JsonSerializer.Serialize(report, new JsonSerializerOptions { WriteIndented = true }), ct); + TestContext.Current.TestOutputHelper?.WriteLine($"#801 AVEVA gate report: {file} ({report.Cases.Count} cases, {failures.Count} failures)"); + + failures.ShouldBeEmpty(); + await session.CloseAsync(ct); + } + + // ------------------------------------------------------------------ wire helpers + + private static async Task ReadRawAsync(ISession session, NodeId node, DateTime start, DateTime end, uint max, + bool returnBounds, CancellationToken ct) + { + var details = new ReadRawModifiedDetails + { + StartTime = start, EndTime = end, NumValuesPerNode = max, IsReadModified = false, ReturnBounds = returnBounds, + }; + var (status, rows, cp) = await HistoryReadAsync(session, details, node, default, ct); + await ReleaseAsync(session, details, node, cp, ct); + return new WireRead(status, rows, 1); + } + + private static async Task ReadRawPagedAsync(ISession session, NodeId node, DateTime start, DateTime end, uint page, + CancellationToken ct) + { + var details = new ReadRawModifiedDetails + { + StartTime = start, EndTime = end, NumValuesPerNode = page, IsReadModified = false, ReturnBounds = false, + }; + var all = new List(); + var pages = 0; + ByteString cp = default; + string status; + do + { + (status, var rows, cp) = await HistoryReadAsync(session, details, node, cp, ct); + all.AddRange(rows); + pages++; + if (status.StartsWith("Bad", StringComparison.Ordinal)) break; + } + while (!cp.IsEmpty && pages < 10_000); + return new WireRead(status, all, pages); + } + + private static async Task ReadModifiedAsync(ISession session, NodeId node, DateTime start, DateTime end, CancellationToken ct) + { + var details = new ReadRawModifiedDetails { StartTime = start, EndTime = end, NumValuesPerNode = 0, IsReadModified = true, ReturnBounds = false }; + var (status, _, cp) = await HistoryReadAsync(session, details, node, default, ct); + await ReleaseAsync(session, details, node, cp, ct); + return status; + } + + private static async Task ReadProcessedAsync(ISession session, NodeId node, DateTime start, DateTime end, NodeId aggregate, + CancellationToken ct) + { + var details = new ReadProcessedDetails + { + StartTime = start, EndTime = end, ProcessingInterval = IntervalMs, + AggregateType = new List { aggregate }, + AggregateConfiguration = new AggregateConfiguration { UseServerCapabilitiesDefaults = true }, + }; + var all = new List(); + var pages = 0; + ByteString cp = default; + string status; + do + { + (status, var rows, cp) = await HistoryReadAsync(session, details, node, cp, ct); + all.AddRange(rows); + pages++; + if (status.StartsWith("Bad", StringComparison.Ordinal)) break; + } + while (!cp.IsEmpty && pages < 1000); + return new WireRead(status, all, pages); + } + + private static async Task ReadAtTimeAsync(ISession session, NodeId node, IReadOnlyList times, CancellationToken ct) + { + var details = new ReadAtTimeDetails { ReqTimes = times.Select(t => (DateTimeUtc)t).ToList(), UseSimpleBounds = true }; + var (status, rows, cp) = await HistoryReadAsync(session, details, node, default, ct); + await ReleaseAsync(session, details, node, cp, ct); + return new WireRead(status, rows, 1); + } + + private static async Task<(string Status, List Rows, ByteString Cp)> HistoryReadAsync( + ISession session, IEncodeable details, NodeId node, ByteString cp, CancellationToken ct) + { + var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, false, + new HistoryReadValueId[] { new() { NodeId = node, ContinuationPoint = cp } }, ct); + var result = response.Results.ToList().ShouldHaveSingleItem(); + var rows = new List(); + if (!result.HistoryData.IsNull && ExtensionObject.ToEncodeable(result.HistoryData) is HistoryData data) + rows.AddRange(data.DataValues.ToList().Select(dv => new Row( + Fmt(dv.ClrValue()), dv.StatusCode.Code, Ts(dv.SourceTimestamp), Ts(dv.ServerTimestamp)))); + return (Name(result.StatusCode), rows, result.ContinuationPoint); + } + + private static async Task ReleaseAsync(ISession session, IEncodeable details, NodeId node, ByteString cp, CancellationToken ct) + { + if (cp.IsEmpty) return; + await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, true, + new HistoryReadValueId[] { new() { NodeId = node, ContinuationPoint = cp } }, ct); + } + + private static async Task ReadEventsAsync(ISession session, NodeId node, DateTime start, DateTime end, uint page, + CancellationToken ct) + { + var filter = new EventFilter(); + foreach (var field in new[] { "EventId", "Time", "ReceiveTime", "SourceName", "Message", "Severity" }) + filter.SelectClauses += new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, new QualifiedName(field)) { AttributeId = Attributes.Value }; + var details = new ReadEventDetails { StartTime = start, EndTime = end, NumValuesPerNode = page, Filter = filter }; + var rows = new List(); + var pages = 0; + ByteString cp = default; + string status; + do + { + var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, false, + new HistoryReadValueId[] { new() { NodeId = node, ContinuationPoint = cp } }, ct); + var result = response.Results.ToList().ShouldHaveSingleItem(); + status = Name(result.StatusCode); + cp = result.ContinuationPoint; + pages++; + if (!result.HistoryData.IsNull && ExtensionObject.ToEncodeable(result.HistoryData) is HistoryEvent he) + foreach (var e in he.Events.ToList()) + { + var f = e.EventFields.ToList(); + rows.Add(new EventRow( + f[0].ClrValue() is byte[] id ? System.Text.Encoding.UTF8.GetString(id) : Fmt(f[0].ClrValue()), + f[1].ClrValue() as DateTime? ?? default, f[2].ClrValue() as DateTime? ?? default, + f[3].ClrValue() as string, f[4].ClrValue() is LocalizedText lt ? lt.Text : Fmt(f[4].ClrValue()), + Convert.ToUInt16(f[5].ClrValue() ?? 0, CultureInfo.InvariantCulture))); + } + if (status.StartsWith("Bad", StringComparison.Ordinal)) break; + } + while (!cp.IsEmpty && pages < 10_000); + return new EventRead(status, rows, pages); + } + + private static async Task> ReadAttributesAsync(ISession session, NodeId node, CancellationToken ct) + { + var ids = new (string Name, uint Id)[] + { + ("Historizing", Attributes.Historizing), ("AccessLevel", Attributes.AccessLevel), + ("UserAccessLevel", Attributes.UserAccessLevel), ("DataType", Attributes.DataType), + }; + var response = await session.ReadAsync(null, 0, TimestampsToReturn.Neither, + ids.Select(a => new ReadValueId { NodeId = node, AttributeId = a.Id }).ToArray(), ct); + var results = response.Results.ToList(); + return ids.Select((a, i) => (a.Name, Value: Fmt(results[i].ClrValue()))) + .ToDictionary(x => x.Name, x => x.Value); + } + + private static async Task> ReadHistoryCapabilitiesAsync(ISession session, CancellationToken ct) + { + var caps = new Dictionary(StringComparer.Ordinal); + var children = await BrowseAsync(session, ObjectIds.HistoryServerCapabilities, ReferenceTypeIds.HierarchicalReferences, ct); + var vars = children.Where(r => r.NodeClass == NodeClass.Variable).ToList(); + if (vars.Count > 0) + { + var response = await session.ReadAsync(null, 0, TimestampsToReturn.Neither, + vars.Select(r => new ReadValueId { NodeId = ExpandedNodeId.ToNodeId(r.NodeId, session.NamespaceUris), AttributeId = Attributes.Value }).ToArray(), ct); + var results = response.Results.ToList(); + for (var i = 0; i < vars.Count; i++) caps[vars[i].BrowseName.Name!] = Fmt(results[i].ClrValue()); + } + var aggs = await BrowseAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ReferenceTypeIds.Organizes, ct); + caps["AggregateFunctions"] = string.Join(",", aggs.Select(r => r.BrowseName.Name).OrderBy(n => n, StringComparer.Ordinal)); + return caps; + } + + private static async Task> BrowseAsync(ISession session, NodeId node, NodeId referenceType, CancellationToken ct) + { + var response = await session.BrowseAsync(null, null, 0, + new BrowseDescription[] + { + new() + { + NodeId = node, BrowseDirection = BrowseDirection.Forward, ReferenceTypeId = referenceType, + IncludeSubtypes = true, NodeClassMask = 0, ResultMask = (uint)BrowseResultMask.All, + }, + }, ct); + return response.Results[0].References.ToList(); + } + + private static DateTime? Ts(DateTimeUtc t) => (DateTime)t == DateTime.MinValue ? null : DateTime.SpecifyKind((DateTime)t, DateTimeKind.Utc); + + private static string Name(StatusCode code) => + code.SymbolicId is { Length: > 0 } s ? s : $"0x{code.Code:X8}"; + + // ------------------------------------------------------------------ SDK-neutral comparison + report + + private static List Rows(IReadOnlyList samples) => + samples.Select(s => new Row(Fmt(s.Value), s.StatusCode, s.SourceTimestampUtc, s.ServerTimestampUtc)).ToList(); + + private static string? Fmt(object? value) => value switch + { + null => null, + double d => d.ToString("R", CultureInfo.InvariantCulture), + float f => ((double)f).ToString("R", CultureInfo.InvariantCulture), + DateTime t => t.ToUniversalTime().ToString("o", CultureInfo.InvariantCulture), + IFormattable f => $"{value.GetType().Name}:{f.ToString(null, CultureInfo.InvariantCulture)}", + _ => $"{value.GetType().Name}:{value}", + }; + + private static void CheckSame(List failures, string what, IReadOnlyList wire, IReadOnlyList reference) + { + if (wire.Count != reference.Count) + { + failures.Add($"{what}: wire {wire.Count} rows vs reference {reference.Count}"); + return; + } + for (var i = 0; i < wire.Count; i++) + { + var (a, b) = (wire[i], reference[i]); + if (a.V != b.V || a.Status != b.Status || a.Src != b.Src) + { + failures.Add($"{what}[{i}]: wire ({a.V}, 0x{a.Status:X8}, {a.Src:o}) vs reference ({b.V}, 0x{b.Status:X8}, {b.Src:o})"); + return; + } + } + } + + private static void CheckSameEvents(List failures, string what, IReadOnlyList wire, IReadOnlyList reference) + { + var a = wire.Select(e => (e.Id, e.Time, e.Severity)).ToList(); + var b = reference.Select(e => (e.Id, e.Time, e.Severity)).ToList(); + if (!a.SequenceEqual(b)) failures.Add($"{what}: wire {a.Count} events vs reference {b.Count} (first difference at {FirstDiff(a, b)})"); + } + + private static int FirstDiff(IReadOnlyList a, IReadOnlyList b) + { + for (var i = 0; i < Math.Min(a.Count, b.Count); i++) if (!EqualityComparer.Default.Equals(a[i], b[i])) return i; + return Math.Min(a.Count, b.Count); + } + + private static void Check(List failures, bool ok, string message) + { + if (!ok) failures.Add(message); + } + + private static List<(DateTime Start, DateTime End)> ParseWindows(string spec) => + spec.Split(';', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries).Select(w => + { + var parts = w.Split('/'); + return (Utc(parts[0]), Utc(parts[1])); + }).ToList(); + + private static DateTime Utc(string s) => + DateTime.Parse(s, CultureInfo.InvariantCulture, DateTimeStyles.AdjustToUniversal | DateTimeStyles.AssumeUniversal); + + private static string? Env(string name) => Environment.GetEnvironmentVariable(name) is { Length: > 0 } v ? v.Trim() : null; + + private static string Sanitize(string s) => string.Concat(s.Select(c => char.IsLetterOrDigit(c) || c is '.' or '-' ? c : '_')); + + private sealed record Row(string? V, uint Status, DateTime? Src, DateTime? Srv); + + private sealed record EventRow(string? Id, DateTime Time, DateTime Received, string? Source, string? Message, ushort Severity); + + private sealed record WireRead(string Status, List Rows, int Pages); + + private sealed record EventRead(string Status, List Rows, int Pages); + + private sealed record Case( + string Name, string? Tag, string? Window, string Path, string Status, string? Note, + Dictionary? Attributes, List? Rows = null, List? Events = null); + + private sealed class Report(string label, string endpoint) + { + public string Label { get; } = label; + public string Endpoint { get; } = endpoint; + public string SdkServerAssembly { get; } = typeof(Opc.Ua.Server.StandardServer).Assembly.GetName().Version?.ToString() ?? "?"; + public DateTime RunUtc { get; } = DateTime.UtcNow; + public List Cases { get; } = []; + public List Failures { get; set; } = []; + public void Add(Case c) => Cases.Add(c); + } +}