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
This commit is contained in:
Joseph Doherty
@@ -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-<tag>-<date>.jsonl)
|
||||
```
|
||||
@@ -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')
|
||||
@@ -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)")
|
||||
@@ -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 <aveva-gate-*.json> <dir with hgcli-*.jsonl>
|
||||
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}")
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
-/// Gitea #801 (SDK 2 live gate, "real AVEVA history"): the booted OtOpcUa server, with the <b>production</b>
|
||||
+/// Gitea #801 BASELINE PORT (SDK 1.5.378.182, master efb9197a) (SDK 2 live gate, "real AVEVA history"): the booted OtOpcUa server, with the <b>production</b>
|
||||
/// Gateway backend (<see cref="HistorianBackendRegistry"/> → <c>GatewayHistorianDataSource</c>) 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 <see cref="IHistorianDataSource"/> 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<Row>();
|
||||
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<string> 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<NodeId> { aggregate },
|
||||
+ AggregateType = new NodeIdCollection { aggregate },
|
||||
AggregateConfiguration = new AggregateConfiguration { UseServerCapabilitiesDefaults = true },
|
||||
};
|
||||
var all = new List<Row>();
|
||||
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<WireRead> ReadAtTimeAsync(ISession session, NodeId node, IReadOnlyList<DateTime> 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<Row> Rows, ByteString Cp)> HistoryReadAsync(
|
||||
- ISession session, IEncodeable details, NodeId node, ByteString cp, CancellationToken ct)
|
||||
+ private static async Task<(string Status, List<Row> 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<Row>();
|
||||
- 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<EventRead> 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<EventRow>();
|
||||
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<List<ReferenceDescription>> 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}";
|
||||
@@ -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")
|
||||
+578
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #801 (SDK 2 live gate, "real AVEVA history"): the booted OtOpcUa server, with the <b>production</b>
|
||||
/// Gateway backend (<see cref="HistorianBackendRegistry"/> → <c>GatewayHistorianDataSource</c>) 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 <see cref="IHistorianDataSource"/> 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 <c>Runtime</c> DB.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para><b>Read-only by construction.</b> The Gateway backend has no <see cref="IHistorianDataUpdater"/> (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 (<c>HISTGW_WRITE_SANDBOX_TAG</c>) and must be
|
||||
/// refused by the server before any backend is reached. Without that variable the HistoryUpdate case is skipped.</para>
|
||||
/// <para><b>Env (skips when the endpoint or key is unset):</b> <c>HISTGW_GATEWAY_ENDPOINT</c> (e.g.
|
||||
/// <c>http://wonder-app-vd03.zmr.zimmer.com:5221</c>), <c>HISTGW_GATEWAY_APIKEY</c> (a key with <c>historian:read</c>;
|
||||
/// never logged), optional <c>HISTGW_GATEWAY_ALLOW_UNTRUSTED</c>, <c>HISTGW_WRITE_SANDBOX_TAG</c>,
|
||||
/// <c>OTOPCUA_AVEVA_GATE_OUT</c> (output directory for <c>aveva-gate-*.json</c>), <c>OTOPCUA_AVEVA_GATE_LABEL</c>
|
||||
/// (a run label recorded in the output, e.g. the SDK version), <c>OTOPCUA_AVEVA_GATE_TAGS</c> (comma list of
|
||||
/// historized analog tags), <c>OTOPCUA_AVEVA_GATE_WINDOWS</c> (<c>start/end;start/end</c>, ISO-8601 UTC),
|
||||
/// <c>OTOPCUA_AVEVA_GATE_EVENT_SOURCES</c> (comma list) and <c>OTOPCUA_AVEVA_GATE_EVENT_WINDOW</c>. The defaults are the DEV
|
||||
/// (wonder-app-vd03 / wonder-sql-vd03) tags and fixed past windows used for the 2026-11-12 run.</para>
|
||||
/// <para>The local deterministic fixtures are NOT this test; only a real gateway in front of a real AVEVA
|
||||
/// Historian makes its output AVEVA evidence.</para>
|
||||
/// </remarks>
|
||||
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<string>();
|
||||
|
||||
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<string, string>(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<DateTime> { 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 ?? "<null>").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<WireRead> 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<WireRead> 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<Row>();
|
||||
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<string> 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<WireRead> 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<NodeId> { aggregate },
|
||||
AggregateConfiguration = new AggregateConfiguration { UseServerCapabilitiesDefaults = true },
|
||||
};
|
||||
var all = new List<Row>();
|
||||
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<WireRead> ReadAtTimeAsync(ISession session, NodeId node, IReadOnlyList<DateTime> 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<Row> 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<Row>();
|
||||
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<EventRead> 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<EventRow>();
|
||||
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<Dictionary<string, string?>> 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<Dictionary<string, string?>> ReadHistoryCapabilitiesAsync(ISession session, CancellationToken ct)
|
||||
{
|
||||
var caps = new Dictionary<string, string?>(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<List<ReferenceDescription>> 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<Row> Rows(IReadOnlyList<DataValueSnapshot> 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<string> failures, string what, IReadOnlyList<Row> wire, IReadOnlyList<Row> 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<string> failures, string what, IReadOnlyList<EventRow> wire, IReadOnlyList<EventRow> 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<T>(IReadOnlyList<T> a, IReadOnlyList<T> b)
|
||||
{
|
||||
for (var i = 0; i < Math.Min(a.Count, b.Count); i++) if (!EqualityComparer<T>.Default.Equals(a[i], b[i])) return i;
|
||||
return Math.Min(a.Count, b.Count);
|
||||
}
|
||||
|
||||
private static void Check(List<string> 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<Row> Rows, int Pages);
|
||||
|
||||
private sealed record EventRead(string Status, List<EventRow> Rows, int Pages);
|
||||
|
||||
private sealed record Case(
|
||||
string Name, string? Tag, string? Window, string Path, string Status, string? Note,
|
||||
Dictionary<string, string?>? Attributes, List<Row>? Rows = null, List<EventRow>? 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<Case> Cases { get; } = [];
|
||||
public List<string> Failures { get; set; } = [];
|
||||
public void Add(Case c) => Cases.Add(c);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user