Joseph Doherty
1e1fad76c4
v2-ci / unit-tests (push) Has been cancelled
Since #808 an UNS folder with no placement in the generation is denied (fail closed), so the matrix's ad hoc "Eq" folder made every UNS case fail with BadUserAccessDenied on the candidate server. The fixture now supplies unsFolderScopes (+ the alarm's tagReferencingEquipment) as a deployed generation does. An unplaced "Unplaced" UNS folder is kept as a negative control: the Shared client leg requires it denied against a #808+ server with the ACL on. No server authorization code changed. Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi
465 lines
36 KiB
C#
465 lines
36 KiB
C#
#pragma warning disable OTOPCUA0001 // This wire harness deliberately exercises public driver capabilities directly.
|
|
using System.Collections.Concurrent;
|
|
using System.Net;
|
|
using System.Net.Sockets;
|
|
using System.Reflection;
|
|
using System.Text.Json;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Opc.Ua;
|
|
using Opc.Ua.Client;
|
|
using ZB.MOM.WW.OtOpcUa.Client.Shared;
|
|
using ZB.MOM.WW.OtOpcUa.Client.Shared.Models;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;
|
|
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
|
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
|
using AlarmEventArgs = ZB.MOM.WW.OtOpcUa.Client.Shared.Models.AlarmEventArgs;
|
|
using WriteRequest = ZB.MOM.WW.OtOpcUa.Core.Abstractions.WriteRequest;
|
|
using SharedSecurityMode = ZB.MOM.WW.OtOpcUa.Client.Shared.Models.SecurityMode;
|
|
using CoreHistory = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
|
|
|
|
internal static class ProcessHost
|
|
{
|
|
static readonly DateTime Epoch = new(2026, 10, 1, 0, 0, 0, DateTimeKind.Utc);
|
|
const string RawUri = "https://zb.com/otopcua/raw", UnsUri = "https://zb.com/otopcua/uns";
|
|
const string Folder = "Matrix/dev", Scalar = Folder + "/Scalar", Vector = Folder + "/Vector", Alarm = Folder + "/Alarm";
|
|
static readonly CancellationToken Ct = new CancellationTokenSource(TimeSpan.FromMinutes(4)).Token;
|
|
static string Password => Environment.GetEnvironmentVariable("OPCUA_MATRIX_PASSWORD") ?? throw new InvalidOperationException("Missing ephemeral credential");
|
|
static readonly List<object> Cases = [];
|
|
static readonly List<object> Observations = [];
|
|
static int Failures;
|
|
static string Root = "";
|
|
static object? Clr(DataValue v)
|
|
{
|
|
#if SDK2
|
|
return ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(v.WrappedValue);
|
|
#else
|
|
return v.Value;
|
|
#endif
|
|
}
|
|
static DateTime SourceTime(DataValue value)
|
|
{
|
|
#if SDK2
|
|
return value.SourceTimestamp.ToDateTime();
|
|
#else
|
|
return value.SourceTimestamp;
|
|
#endif
|
|
}
|
|
static DateTime ServerTime(DataValue value)
|
|
{
|
|
#if SDK2
|
|
return value.ServerTimestamp.ToDateTime();
|
|
#else
|
|
return value.ServerTimestamp;
|
|
#endif
|
|
}
|
|
static void Require(bool yes, string message) { if (!yes) throw new InvalidOperationException(message); }
|
|
static async Task Case(string name, Func<Task> action)
|
|
{
|
|
try { await action(); Cases.Add(new { name, status = "passed" }); }
|
|
catch (Exception e) { Failures++; Cases.Add(new { name, status = "failed", error = e.ToString() }); }
|
|
}
|
|
static async Task Wait(Func<bool> predicate)
|
|
{
|
|
for (int i = 0; i < 200; i++) { if (predicate()) return; await Task.Delay(50, Ct); }
|
|
Require(false, "Notification deadline exceeded");
|
|
}
|
|
static ConnectionSettings Settings(string endpoint) => new()
|
|
{
|
|
EndpointUrl = endpoint,
|
|
Username = "apikey",
|
|
Password = Password,
|
|
SecurityMode = SharedSecurityMode.SignAndEncrypt,
|
|
AutoAcceptCertificates = false,
|
|
CertificateStorePath = Environment.GetEnvironmentVariable("OPCUA_MATRIX_CLIENT_PKI") ?? Path.Combine(Root, "client-pki")
|
|
};
|
|
public static async Task<int> Main(string[] args)
|
|
{
|
|
Root = Path.GetFullPath(args[1]); Directory.CreateDirectory(Root);
|
|
try
|
|
{
|
|
switch (args[0])
|
|
{
|
|
case "local-data-path": Console.WriteLine(Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData)); return 0;
|
|
case "prepare":
|
|
var type = typeof(OpcUaClientService).Assembly.GetType("ZB.MOM.WW.OtOpcUa.Client.Shared.Adapters.DefaultApplicationConfigurationFactory")!;
|
|
var factory = Activator.CreateInstance(type, true)!;
|
|
await (Task<ApplicationConfiguration>)type.GetMethod("CreateAsync")!.Invoke(factory, [Settings("opc.tcp://127.0.0.1:1"), Ct])!;
|
|
Console.WriteLine("PREPARED"); return 0;
|
|
case "prepare-driver": await PrepareDriver(); return 0;
|
|
case "driver": await Driver(args[2]); break;
|
|
case "server": await Serve(); return 0;
|
|
case "client": await Client(args[2], args.Length > 3 ? args[3] : null); break;
|
|
case "performance": await Performance(args[2]); break;
|
|
default: throw new ArgumentException("Expected prepare/server/client");
|
|
}
|
|
}
|
|
catch (Exception e) { Failures++; Cases.Add(new { name = "process", status = "failed", error = e.ToString() }); }
|
|
File.WriteAllText(Path.Combine(Root, "result.json"), JsonSerializer.Serialize(new { sdk = typeof(NodeId).Assembly.GetName().Version?.ToString(), process = Environment.ProcessId, cases = Cases, observations = Observations, failures = Failures, loadedSdkAssemblies = AppDomain.CurrentDomain.GetAssemblies().Where(a => a.GetName().Name!.StartsWith("Opc.Ua", StringComparison.Ordinal)).Select(a => new { name = a.GetName().Name, version = a.GetName().Version?.ToString(), path = a.Location }) }, new JsonSerializerOptions { WriteIndented = true }));
|
|
return Failures == 0 ? 0 : 1;
|
|
}
|
|
static async Task Serve()
|
|
{
|
|
var options = new OpcUaApplicationHostOptions
|
|
{
|
|
ApplicationName = "OtOpcUa.Matrix",
|
|
ApplicationUri = $"urn:OtOpcUa.Matrix.{Guid.NewGuid():N}",
|
|
PublicHostname = "127.0.0.1",
|
|
PkiStoreRoot = Path.Combine(Root, "server-pki"),
|
|
AllowAnonymous = false,
|
|
AutoAcceptUntrustedClientCertificates = false,
|
|
EnabledSecurityProfiles = Environment.GetEnvironmentVariable("OPCUA_MATRIX_SERVER_SECURITY") switch
|
|
{
|
|
"none" => [OpcUaSecurityProfile.None],
|
|
"sign" => [OpcUaSecurityProfile.Basic256Sha256Sign],
|
|
_ => [OpcUaSecurityProfile.Basic256Sha256SignAndEncrypt],
|
|
}
|
|
};
|
|
using (var listener = new TcpListener(IPAddress.Loopback, 0)) { listener.Start(); options.OpcUaPort = ((IPEndPoint)listener.LocalEndpoint).Port; }
|
|
var server = new OtOpcUaSdkServer();
|
|
await using var host = new OpcUaApplicationHost(options, NullLogger<OpcUaApplicationHost>.Instance, new Auth());
|
|
await host.StartAsync(server, Ct);
|
|
var nm = server.NodeManager!; var sink = new SdkAddressSpaceSink(nm);
|
|
var authority = new DataPlaneAuthorityState();
|
|
authority.Apply("matrix", "rev1", [new ApiKey { ApiKeyId = "matrix", Name = "matrix", SecretHash = "fixture", Principal = "matrix", Enabled = true, CreatedUtc = DateTime.UtcNow.AddDays(-1) }],
|
|
[new NodeAcl { NodeAclId = "all", ClusterId = "matrix", Principal = "matrix", ScopeKind = NodeAclScopeKind.Cluster, PermissionFlags = (NodePermissions)int.MaxValue }], new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>()
|
|
#if SDK2
|
|
// lmxopcua#888: since #808 an UNS folder with no placement in the authority snapshot is denied (fail closed, no
|
|
// realm/cluster fallback). A real deployment's generation places every Area/Line/Equipment folder; do the same
|
|
// for the fixture's "Eq" equipment folder (a root folder, so its id chain is just ["Eq"]). The "Unplaced" folder
|
|
// below is deliberately left out of the map: it is the negative control that must stay denied.
|
|
, unsFolderScopes: new Dictionary<string, UnsFolderScope> { ["Eq"] = new UnsFolderScope(new NodeAclPath([NodeAclRealms.Uns, "Eq"]), ["Eq"]) },
|
|
tagReferencingEquipment: new Dictionary<string, IReadOnlySet<string>> { [Alarm] = new HashSet<string>(StringComparer.OrdinalIgnoreCase) { "Eq" } }
|
|
#endif
|
|
);
|
|
nm.DataPlaneAuthority = authority;
|
|
// OPCUA_MATRIX_ACL=off mirrors the shipped default (Security:Acl:Enabled=false): no node authorizer.
|
|
if (Environment.GetEnvironmentVariable("OPCUA_MATRIX_ACL") != "off")
|
|
nm.Authorizer = new DataPlaneAuthorizer(authority, new TriePermissionEvaluator(authority.Tries), true, NullLogger.Instance, TimeProvider.System);
|
|
var history = new Source(Root); nm.HistorianDataSource = history;
|
|
sink.EnsureFolder(Folder, null, "dev", AddressSpaceRealm.Raw);
|
|
sink.EnsureFolder("Eq", null, "Eq", AddressSpaceRealm.Uns);
|
|
// Negative control (#888): an UNS folder + variable the generation does not place must stay denied under the ACL.
|
|
sink.EnsureFolder("Unplaced", null, "Unplaced", AddressSpaceRealm.Uns);
|
|
sink.EnsureVariable("Unplaced/Scalar", "Unplaced", "Scalar", "Int32", true, AddressSpaceRealm.Uns);
|
|
sink.WriteValue("Unplaced/Scalar", 5, 0, Epoch, AddressSpaceRealm.Uns);
|
|
foreach (var realm in new[] { AddressSpaceRealm.Raw, AddressSpaceRealm.Uns })
|
|
{
|
|
var scalar = realm == AddressSpaceRealm.Raw ? Scalar : "Eq/Scalar";
|
|
var vector = realm == AddressSpaceRealm.Raw ? Vector : "Eq/Vector";
|
|
var parent = realm == AddressSpaceRealm.Raw ? Folder : "Eq";
|
|
sink.EnsureVariable(scalar, parent, "Scalar", "Int32", true, realm, "matrix.scalar");
|
|
sink.EnsureVariable(vector, parent, "Vector", "Int32", true, realm, isArray: true, arrayLength: 3);
|
|
sink.WriteValue(scalar, 17, 0, Epoch, realm); sink.WriteValue(vector, new[] { 1, 2, 3 }, 0, Epoch, realm);
|
|
}
|
|
sink.AddReference("Eq/Scalar", AddressSpaceRealm.Uns, Scalar, AddressSpaceRealm.Raw);
|
|
sink.MaterialiseAlarmCondition(Alarm, Folder, "Alarm", "OffNormalAlarm", 800, AddressSpaceRealm.Raw, isNative: true);
|
|
sink.WireAlarmNotifiers(Alarm, AddressSpaceRealm.Raw, ["Eq"], AddressSpaceRealm.Uns);
|
|
sink.EnsureVariable(Folder + "/UnsetTimestamp", Folder, "UnsetTimestamp", "Int32", false, AddressSpaceRealm.Raw);
|
|
sink.WriteValue(Folder + "/UnsetTimestamp", 3, 0, DateTime.MinValue, AddressSpaceRealm.Raw);
|
|
nm.NodeWriteGateway = new Gateway();
|
|
var state = new AlarmConditionSnapshot(true, false, true, true, AlarmShelvingKind.Unshelved, 800, "matrix-alarm");
|
|
nm.NativeAlarmAckRouter = _ => Task.Run(() => { state = state with { Acknowledged = true }; sink.WriteAlarmCondition(Alarm, state, DateTime.UtcNow, AddressSpaceRealm.Raw); });
|
|
File.WriteAllText(Path.Combine(Root, "endpoint.txt"), $"opc.tcp://127.0.0.1:{options.OpcUaPort}/OtOpcUa");
|
|
File.WriteAllText(Path.Combine(Root, "ready.json"), JsonSerializer.Serialize(new { endpoint = $"opc.tcp://127.0.0.1:{options.OpcUaPort}/OtOpcUa", process = Environment.ProcessId, sdk = typeof(NodeId).Assembly.GetName().Version?.ToString(), rawUri = RawUri, unsUri = UnsUri }));
|
|
int tick = 0;
|
|
while (!File.Exists(Path.Combine(Root, "stop")))
|
|
{
|
|
sink.WriteAlarmCondition(Alarm, state with { Message = ++tick % 2 == 0 ? "matrix-alarm" : "matrix-tick" }, DateTime.UtcNow, AddressSpaceRealm.Raw);
|
|
await Task.Delay(1000, Ct);
|
|
}
|
|
using var process = System.Diagnostics.Process.GetCurrentProcess();
|
|
File.WriteAllText(Path.Combine(Root, "server-resources.json"), JsonSerializer.Serialize(new { process.WorkingSet64, process.PrivateMemorySize64, process.HandleCount, threads=process.Threads.Count, cpuMilliseconds=process.TotalProcessorTime.TotalMilliseconds, allocatedBytes=GC.GetTotalAllocatedBytes(), collections=new[]{GC.CollectionCount(0),GC.CollectionCount(1),GC.CollectionCount(2)} }, new JsonSerializerOptions {WriteIndented=true}));
|
|
}
|
|
// #880: the first ServiceResultException status in a failed connect's exception chain (null = no OPC UA status).
|
|
static StatusCode? FailureStatus(Exception e)
|
|
{
|
|
for (Exception? x = e; x is not null; x = x.InnerException)
|
|
{
|
|
if (x is ServiceResultException sre) return sre.StatusCode;
|
|
if (x is AggregateException ae && ae.InnerExceptions.Count > 0) return FailureStatus(ae.InnerExceptions[0]);
|
|
}
|
|
return null;
|
|
}
|
|
// A wrong secret of the production length: same key id, one changed character, never logged.
|
|
static string WrongSecret => Password[..^1] + (Password[^1] == 'A' ? 'B' : 'A');
|
|
static async Task ApiKeyFormatCases(string endpoint, string? serverSdk)
|
|
{
|
|
Observations.Add(new { apiKeyLength = Password.Length, serverSdk, note = "ApiKeySecret.Generate shape; SDK 2.0.1 sends a UserName password over 64 bytes as RsaEncryptedSecret unless the client forces the legacy format (#880)" });
|
|
// The production client (Shared, i.e. the CLI/UI) with a wrong key of the same length must reach the server's
|
|
// real key check: BadIdentityTokenRejected. BadIdentityTokenInvalid means the server could not even decrypt
|
|
// the token: the #880 format failure.
|
|
await Case("production-length-API-key-wrong-secret-reaches-key-check", async () =>
|
|
{
|
|
using var probe = new OpcUaClientService();
|
|
var settings = Settings(endpoint); settings.Password = WrongSecret;
|
|
Exception? failure = null;
|
|
try { await probe.ConnectAsync(settings, Ct); } catch (Exception e) { failure = e; }
|
|
Require(failure is not null, "A wrong key was admitted");
|
|
var status = FailureStatus(failure!);
|
|
Observations.Add(new { wrongSecretStatus = status?.ToString() });
|
|
Require(status == StatusCodes.BadIdentityTokenRejected, "Expected BadIdentityTokenRejected (key check), got " + (status?.ToString() ?? failure!.GetType().Name));
|
|
});
|
|
#if SDK2
|
|
// Negative control: the SDK's own UserName handler, same valid key. Against an SDK 1.x server this must fail on
|
|
// format (BadIdentityTokenInvalid); this is the bug the production clients avoid. Against SDK 2 it is admitted.
|
|
await Case("sdk2-default-RsaEncryptedSecret-control", async () =>
|
|
{
|
|
var serverIsSdk1 = serverSdk?.StartsWith("1.", StringComparison.Ordinal) ?? throw new InvalidOperationException("Server SDK version not supplied");
|
|
var type = typeof(OpcUaClientService).Assembly.GetType("ZB.MOM.WW.OtOpcUa.Client.Shared.Adapters.DefaultApplicationConfigurationFactory")!;
|
|
var config = await (Task<ApplicationConfiguration>)type.GetMethod("CreateAsync")!.Invoke(Activator.CreateInstance(type, true)!, [Settings(endpoint), Ct])!;
|
|
try
|
|
{
|
|
var telemetry = DefaultTelemetry.Create(_ => { });
|
|
var selected = await CoreClientUtils.SelectEndpointAsync(config, endpoint, true, telemetry, Ct);
|
|
Require(selected.SecurityMode == MessageSecurityMode.SignAndEncrypt, "Control did not negotiate SignAndEncrypt");
|
|
StatusCode? status = null;
|
|
try
|
|
{
|
|
using var session = await new DefaultSessionFactory(telemetry).CreateAsync(config, new ConfiguredEndpoint(null, selected, EndpointConfiguration.Create(config)),
|
|
false, false, "Matrix880Control", 60_000, new UserIdentity("apikey", System.Text.Encoding.UTF8.GetBytes(Password)), default, Ct);
|
|
await session.CloseAsync(Ct);
|
|
}
|
|
catch (Exception e) { status = FailureStatus(e) ?? StatusCodes.Bad; }
|
|
Observations.Add(new { sdkDefaultFormatStatus = status?.ToString() ?? "Good", serverSdk });
|
|
if (serverIsSdk1) Require(status == StatusCodes.BadIdentityTokenInvalid, "Expected the SDK 1.x server to reject RsaEncryptedSecret as BadIdentityTokenInvalid, got " + (status?.ToString() ?? "admitted"));
|
|
else Require(status is null, "SDK 2 server refused the SDK-default format: " + status);
|
|
}
|
|
finally { (config.CertificateManager as IDisposable)?.Dispose(); }
|
|
});
|
|
#endif
|
|
}
|
|
static async Task Client(string endpoint, string? serverSdk)
|
|
{
|
|
using var client = new OpcUaClientService();
|
|
await Case("trusted-SignAndEncrypt-API-key", async () => { var info = await client.ConnectAsync(Settings(endpoint), Ct); Require(info.SecurityMode == "SignAndEncrypt" && info.SecurityPolicyUri == SecurityPolicies.Basic256Sha256, "Unexpected negotiated policy"); Observations.Add(new { info.SecurityMode, info.SecurityPolicyUri }); });
|
|
await ApiKeyFormatCases(endpoint, serverSdk);
|
|
if (!client.IsConnected) return;
|
|
var namespaces = (string[])Clr(await client.ReadValueAsync(VariableIds.Server_NamespaceArray, Ct))!;
|
|
var raw = (ushort)Array.IndexOf(namespaces, RawUri); var uns = (ushort)Array.IndexOf(namespaces, UnsUri);
|
|
NodeId ScalarId(ushort ns) => new(ns == raw ? Scalar : "Eq/Scalar", ns);
|
|
NodeId VectorId(ushort ns) => new(ns == raw ? Vector : "Eq/Vector", ns);
|
|
await Case("raw-UNS-URI-browse", async () =>
|
|
{
|
|
Require(raw > 0 && uns > 0, "Missing namespace URI");
|
|
var rawChildren = await client.BrowseAsync(new NodeId(Folder, raw), Ct);
|
|
var unsChildren = await client.BrowseAsync(new NodeId("Eq", uns), Ct);
|
|
Observations.Add(new { rawChildren, unsChildren });
|
|
// Supplement the production Shared results with unfiltered wire-reference metadata.
|
|
var adapter = typeof(OpcUaClientService).GetField("_session", BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(client)!;
|
|
var sdkSession = (Opc.Ua.Client.Session)adapter.GetType().GetField("_session", BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(adapter)!;
|
|
var (_, _, rawReferences) = await sdkSession.BrowseAsync(null, null, new NodeId(Folder, raw), 0u,
|
|
BrowseDirection.Forward, ReferenceTypeIds.References, true, 0u);
|
|
Observations.Add(new { rawReferences = (rawReferences.ToArray() ?? []).Select(reference => new { nodeId = reference.NodeId.ToString(), browseName = reference.BrowseName.ToString(), nodeClass = reference.NodeClass.ToString(), typeDefinition = reference.TypeDefinition.ToString(), referenceType = reference.ReferenceTypeId.ToString(), reference.IsForward }).ToArray() });
|
|
var expectedRaw = new[] { Scalar, Vector, Alarm, Folder + "/UnsetTimestamp" }.Select(id => new NodeId(id, raw).ToString());
|
|
Require(expectedRaw.All(id => rawChildren.Any(child => child.NodeId == id)), "Missing raw children: " + string.Join(",", expectedRaw.Where(id => !rawChildren.Any(child => child.NodeId == id))));
|
|
Require(new[] { "Eq/Scalar", "Eq/Vector" }.Select(id => new NodeId(id, uns).ToString()).All(id => unsChildren.Any(child => child.NodeId == id)), "Missing UNS scalar/array children");
|
|
});
|
|
await Case("initial-quality-explicit-source-and-server-timestamps", async () => { var initial = await client.ReadValueAsync(ScalarId(raw), Ct); Require(StatusCode.IsGood(initial.StatusCode), "Bad initial quality"); Require(SourceTime(initial) == Epoch && ServerTime(initial) > Epoch, "Lost explicit initial timestamps"); Observations.Add(new { sourceTimestamp = SourceTime(initial).ToString("O"), serverTimestamp = ServerTime(initial).ToString("O"), contract = "Source supplied as 2026-10-01T00:00:00Z; server timestamp observed without normalization or historian algorithm changes" }); });
|
|
// #888 negative control: the unplaced UNS folder's variable must be denied against a #808+ server with the ACL on
|
|
// (a 1.x baseline server predates #808; with OPCUA_MATRIX_ACL=off there is no authorizer, so it reads Good).
|
|
await Case("uns-unplaced-folder-denied-or-open-by-acl-mode", async () =>
|
|
{
|
|
var expectDenied = Environment.GetEnvironmentVariable("OPCUA_MATRIX_ACL") != "off" && (serverSdk?.StartsWith("2.", StringComparison.Ordinal) ?? false);
|
|
string outcome;
|
|
try { var v = await client.ReadValueAsync(new NodeId("Unplaced/Scalar", uns), Ct); outcome = v.StatusCode.ToString(); }
|
|
catch (Exception e) { outcome = e.Message; }
|
|
Observations.Add(new { unplacedRead = outcome, expectDenied });
|
|
var denied = outcome.Contains("BadUserAccessDenied", StringComparison.Ordinal);
|
|
Require(expectDenied ? denied : !denied, $"Unplaced UNS folder read: expected {(expectDenied ? "denied" : "not denied")}, got {outcome}");
|
|
});
|
|
foreach (var ns in new[] { raw, uns })
|
|
{
|
|
await Case($"scalar-read-write-ns-{ns}", async () => { Require(Clr(await client.ReadValueAsync(ScalarId(ns), Ct)) is int, "Scalar lost Int32 type"); Require(StatusCode.IsGood(await client.WriteValueAsync(ScalarId(ns), 42, Ct)), "Scalar write failed"); Require((int)Clr(await client.ReadValueAsync(ScalarId(ns), Ct))! == 42, "Scalar round trip failed"); });
|
|
await Case($"typed-array-read-write-ns-{ns}", async () => { Require(Clr(await client.ReadValueAsync(VectorId(ns), Ct)) is int[], "Array lost Int32[] type"); Require(StatusCode.IsGood(await client.WriteValueAsync(VectorId(ns), new[] { 7, 8, 9 }, Ct)), "Array write failed"); Require(((int[])Clr(await client.ReadValueAsync(VectorId(ns), Ct))!).SequenceEqual(new[] { 7, 8, 9 }), "Array round trip failed"); });
|
|
}
|
|
await Case("monitor-quality-timestamps", async () => { var seen = new ConcurrentQueue<DataValue>(); client.DataChanged += (_, e) => seen.Enqueue(e.Value); await client.SubscribeAsync(ScalarId(raw), 100, Ct); await client.WriteValueAsync(ScalarId(raw), 73, Ct); await Wait(() => seen.Any(v => Equals(Clr(v), 73))); var v = seen.Last(v => Equals(Clr(v), 73)); Require(StatusCode.IsGood(v.StatusCode) && SourceTime(v) > Epoch && ServerTime(v) > Epoch, "Bad monitor quality or missing timestamps"); Observations.Add(new { sourceTimestamp = SourceTime(v).ToString("O"), serverTimestamp = ServerTime(v).ToString("O"), status = v.StatusCode.ToString() }); await client.UnsubscribeAsync(ScalarId(raw), Ct); });
|
|
await Case("event-filter-and-native-ack", async () => { var alarms = new ConcurrentQueue<AlarmEventArgs>(); client.AlarmEvent += (_, a) => alarms.Enqueue(a); await client.SubscribeAlarmsAsync(ObjectIds.Server, 100, Ct); await Wait(() => alarms.Any(a => a.Message == "matrix-alarm" && a.ActiveState && a.Severity == 800)); var alarm = alarms.Last(a => a.Message == "matrix-alarm"); Require(alarm.EventId?.Length > 0, "Missing event ID"); Require(StatusCode.IsGood(await client.AcknowledgeAlarmAsync(new NodeId(Alarm, raw).ToString(), alarm.EventId!, "matrix-ack", Ct)), "Ack refused"); await Wait(() => alarms.Any(a => a.Message == "matrix-alarm" && a.AckedState)); await client.UnsubscribeAlarmsAsync(Ct); });
|
|
await Case("raw-history-CP-paging", async () => { var values = await client.HistoryReadRawAsync(ScalarId(raw), Epoch, Epoch.AddMinutes(1), 100, Ct); Require(values.Count == 25, "Expected 25 raw samples, got " + values.Count); Require(values.Select(v => Convert.ToInt32(Clr(v))).SequenceEqual(Enumerable.Range(0, 25)), "Raw page loss/duplicates/order"); });
|
|
await Case("event-history-CP-paging", async () =>
|
|
{
|
|
var filter = new EventFilter();
|
|
#if SDK2
|
|
filter.SelectClauses += new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, new QualifiedName("EventId")) { AttributeId = Attributes.Value };
|
|
#else
|
|
filter.SelectClauses.Add(new SimpleAttributeOperand {TypeDefinitionId=ObjectTypeIds.BaseEventType,BrowsePath=new QualifiedNameCollection {new QualifiedName("EventId")},AttributeId=Attributes.Value});
|
|
#endif
|
|
var events = await client.HistoryReadEventsAsync(new NodeId(Folder, raw), Epoch, Epoch.AddMinutes(1), filter, 100, Ct); Require(events.Count == 25, "Expected 25 event samples, got " + events.Count);
|
|
#if SDK2
|
|
var ids = events.Select(e => System.Text.Encoding.UTF8.GetString((byte[])ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers.ValueConverter.ToClrObject(e.Fields[0])!));
|
|
#else
|
|
var ids=events.Select(e=>System.Text.Encoding.UTF8.GetString((byte[])e.Fields[0].Value));
|
|
#endif
|
|
Require(ids.SequenceEqual(Enumerable.Range(0, 25).Select(i => $"e{i:00}")), "Event history loss/duplicates/order across timestamp ties");
|
|
});
|
|
await Case("event-history-where-filter-positive-and-negative", async () =>
|
|
{
|
|
foreach (var type in new[] { ObjectTypeIds.BaseEventType, ObjectTypeIds.AuditEventType })
|
|
{
|
|
var filter = new EventFilter();
|
|
#if SDK2
|
|
filter.SelectClauses += new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, new QualifiedName("EventId")) { AttributeId = Attributes.Value };
|
|
filter.WhereClause.Push(FilterOperator.OfType, Variant.From(type));
|
|
#else
|
|
filter.SelectClauses.Add(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType,new QualifiedName("EventId")){AttributeId=Attributes.Value});
|
|
filter.WhereClause.Push(FilterOperator.OfType,new LiteralOperand(new Variant(type)));
|
|
#endif
|
|
var events = await client.HistoryReadEventsAsync(new NodeId(Folder, raw), Epoch, Epoch.AddMinutes(1), filter, 100, Ct);
|
|
Require(events.Count == (type == ObjectTypeIds.BaseEventType ? 25 : 0), $"Where filter for {type} expected {(type == ObjectTypeIds.BaseEventType ? 25 : 0)} got {events.Count}");
|
|
}
|
|
});
|
|
var unset = await client.ReadValueAsync(new NodeId(Folder + "/UnsetTimestamp", raw), Ct);
|
|
Observations.Add(new { sourceTimestamp = SourceTime(unset).ToString("O"), serverTimestamp = ServerTime(unset).ToString("O"), timestampObservation = "Fixture source supplied DateTime.MinValue (unset). These are unmodified application/SDK wire observations; no historian algorithm or expected timestamp was changed." });
|
|
await Benchmark(client, ScalarId(raw), VectorId(raw));
|
|
await client.DisconnectAsync(Ct);
|
|
}
|
|
static OpcUaClientDriverOptions DriverOptions(string endpoint) => new()
|
|
{
|
|
EndpointUrl = endpoint,
|
|
SecurityPolicy = OpcUaSecurityPolicy.Basic256Sha256,
|
|
SecurityMode = OpcUaSecurityMode.SignAndEncrypt,
|
|
AuthType = OpcUaAuthType.Username,
|
|
Username = "apikey",
|
|
Password = Password,
|
|
AutoAcceptCertificates = false,
|
|
RawTags = [new("scalar",JsonSerializer.Serialize(new {nodeId=$"nsu={RawUri};s={Scalar}"}),true),
|
|
new("vector",JsonSerializer.Serialize(new {nodeId=$"nsu={RawUri};s={Vector}"}),true),
|
|
new("uns",JsonSerializer.Serialize(new {nodeId=$"nsu={UnsUri};s=Eq/Scalar"}),true)],
|
|
SessionName = "MatrixDriver",
|
|
ApplicationUri = "urn:localhost:MatrixDriver"
|
|
};
|
|
static string IsolatedLocalData()
|
|
{
|
|
var local = Path.GetFullPath(Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData));
|
|
Require(local.StartsWith(Path.GetFullPath(Root) + Path.DirectorySeparatorChar, StringComparison.Ordinal), "Driver/CLI canonical PKI must be in task-owned root: " + local);
|
|
return local;
|
|
}
|
|
static async Task PrepareDriver()
|
|
{
|
|
var local = IsolatedLocalData();
|
|
await using var driver = new OpcUaClientDriver(DriverOptions("opc.tcp://127.0.0.1:1"), "matrix");
|
|
// The production configuration method creates the application's own certificate without
|
|
// attempting a session. Reflection is confined to preparation; all wire operations below
|
|
// call the unchanged public driver API.
|
|
var type = typeof(OpcUaClientDriver);
|
|
var hook = Activator.CreateInstance(type.GetNestedType("CertValidationHook", BindingFlags.NonPublic)!, true)!;
|
|
await (Task<ApplicationConfiguration>)type.GetMethod("BuildApplicationConfigurationAsync", BindingFlags.Instance | BindingFlags.NonPublic)!.Invoke(driver, [hook, Ct])!;
|
|
File.WriteAllText(Path.Combine(Root, "driver-pki.txt"), Path.Combine(local, "OtOpcUa", "pki"));
|
|
}
|
|
static async Task Driver(string endpoint)
|
|
{
|
|
IsolatedLocalData();
|
|
await using var driver = new OpcUaClientDriver(DriverOptions(endpoint), "matrix");
|
|
await Case("driver-trusted-encrypted-API-key-connect", () => driver.InitializeAsync("{}", Ct));
|
|
await Case("driver-raw-UNS-URI-scalar-and-array-read", async () =>
|
|
{
|
|
var values = await driver.ReadAsync(["scalar", "vector", "uns"], Ct);
|
|
Require(values.Count == 3 && values.All(x => x.StatusCode == 0), "Read failure");
|
|
Require(values[0].Value is int && values[1].Value is int[] && values[2].Value is int, "Lost scalar/array type");
|
|
});
|
|
await Case("driver-scalar-array-write-roundtrip", async () =>
|
|
{
|
|
var writes = await driver.WriteAsync([new WriteRequest("scalar", 91), new WriteRequest("vector", new[] { 11, 12, 13 })], Ct);
|
|
Require(writes.All(x => x.StatusCode == 0), "Write failed");
|
|
var values = await driver.ReadAsync(["scalar", "vector"], Ct);
|
|
Require(Equals(values[0].Value, 91) && ((int[])values[1].Value!).SequenceEqual(new[] { 11, 12, 13 }), "Write round trip mismatch");
|
|
});
|
|
await Case("driver-monitor-quality-timestamps", async () =>
|
|
{
|
|
var seen = new ConcurrentQueue<DataValueSnapshot>();
|
|
driver.OnDataChange += (_, e) => seen.Enqueue(e.Snapshot);
|
|
var subscription = await driver.SubscribeAsync(["scalar"], TimeSpan.FromMilliseconds(100), Ct);
|
|
await driver.WriteAsync([new WriteRequest("scalar", 92)], Ct);
|
|
await Wait(() => seen.Any(x => Equals(x.Value, 92)));
|
|
var value = seen.Last(x => Equals(x.Value, 92));
|
|
Require(value.StatusCode == 0 && value.SourceTimestampUtc.HasValue && value.ServerTimestampUtc > Epoch, "Missing monitor quality/timestamps");
|
|
Observations.Add(new { value.SourceTimestampUtc, value.ServerTimestampUtc, value.StatusCode });
|
|
await driver.UnsubscribeAsync(subscription, Ct);
|
|
});
|
|
Observations.Add(new { capability = "driver-history", status = "not-applicable", evidence = "Driver has no IHistoryProvider/HistoryRead capability; .claude/rules/drivers.md #655. Shared/CLI own history." });
|
|
await driver.ShutdownAsync(Ct);
|
|
}
|
|
static async Task Performance(string endpoint)
|
|
{
|
|
using var client = new OpcUaClientService();
|
|
var info = await client.ConnectAsync(Settings(endpoint), Ct);
|
|
Require(info.SecurityMode == "SignAndEncrypt" && info.SecurityPolicyUri == SecurityPolicies.Basic256Sha256, "Unexpected benchmark endpoint policy");
|
|
var namespaces = (string[])Clr(await client.ReadValueAsync(VariableIds.Server_NamespaceArray, Ct))!;
|
|
var raw = (ushort)Array.IndexOf(namespaces, RawUri);
|
|
Require(raw > 0, "Missing benchmark raw namespace");
|
|
Observations.Add(new {info.SecurityMode, info.SecurityPolicyUri, tieredCompilation=Environment.GetEnvironmentVariable("DOTNET_TieredCompilation"), scope="Two scalar/Int32[3] variables; zero data monitors during measurement; five fixture variables and one native condition"});
|
|
await Case("warmed-production-client-read-write-measurement", () => Benchmark(client, new NodeId(Scalar, raw), new NodeId(Vector, raw)));
|
|
await client.DisconnectAsync(Ct);
|
|
}
|
|
static async Task Benchmark(OpcUaClientService client, NodeId scalar, NodeId vector)
|
|
{
|
|
// Local diagnostic subset; excludes reconnect, history and load.
|
|
var rounds = int.Parse(Environment.GetEnvironmentVariable("OPCUA_MATRIX_PERF_ROUNDS") ?? "3");
|
|
var warmups = int.Parse(Environment.GetEnvironmentVariable("OPCUA_MATRIX_PERF_WARMUPS") ?? "10");
|
|
var measured = int.Parse(Environment.GetEnvironmentVariable("OPCUA_MATRIX_PERF_MEASURED") ?? "50");
|
|
Require(rounds > 0 && warmups >= 0 && measured > 0, "Invalid performance sample counts");
|
|
for (int round = 1; round <= rounds; round++)
|
|
{
|
|
for (int warmup = 0; warmup < warmups; warmup++)
|
|
{
|
|
await client.ReadValueAsync(scalar, Ct); await client.WriteValueAsync(scalar, warmup, Ct);
|
|
await client.ReadValueAsync(vector, Ct); await client.WriteValueAsync(vector, new[] { warmup, warmup + 1, warmup + 2 }, Ct);
|
|
}
|
|
var samples = new Dictionary<string, List<double>>();
|
|
var before = System.Diagnostics.Process.GetCurrentProcess(); var cpu = before.TotalProcessorTime;
|
|
var allocated = GC.GetTotalAllocatedBytes(); var gc = new[] { GC.CollectionCount(0), GC.CollectionCount(1), GC.CollectionCount(2) };
|
|
foreach (var operation in new[] { "scalar-read", "scalar-write", "array-read", "array-write" })
|
|
{
|
|
var times = new List<double>();
|
|
for (int i = 0; i < measured; i++)
|
|
{
|
|
var start = System.Diagnostics.Stopwatch.GetTimestamp();
|
|
if (operation == "scalar-read") await client.ReadValueAsync(scalar, Ct);
|
|
else if (operation == "scalar-write") Require(StatusCode.IsGood(await client.WriteValueAsync(scalar, i, Ct)), "Benchmark scalar write");
|
|
else if (operation == "array-read") await client.ReadValueAsync(vector, Ct);
|
|
else Require(StatusCode.IsGood(await client.WriteValueAsync(vector, new[] { i, i + 1, i + 2 }, Ct)), "Benchmark array write");
|
|
times.Add(System.Diagnostics.Stopwatch.GetElapsedTime(start).TotalMilliseconds);
|
|
}
|
|
samples[operation] = times;
|
|
}
|
|
before.Refresh();
|
|
Observations.Add(new
|
|
{
|
|
benchmarkRound = round,
|
|
warmupOperationsPerKind = 10,
|
|
operationsPerKind = 50,
|
|
latencyMilliseconds = samples.ToDictionary(x => x.Key, x => new { p50 = x.Value.Order().ElementAt((int)Math.Ceiling(measured * 0.50) - 1), p95 = x.Value.Order().ElementAt((int)Math.Ceiling(measured * 0.95) - 1), p99 = x.Value.Order().ElementAt((int)Math.Ceiling(measured * 0.99) - 1), throughputOperationsPerSecond = measured * 1000.0 / x.Value.Sum() }),
|
|
clientCpuMilliseconds = (before.TotalProcessorTime - cpu).TotalMilliseconds,
|
|
clientAllocatedBytes = GC.GetTotalAllocatedBytes() - allocated, clientAllocatedBytesPerOperation = (GC.GetTotalAllocatedBytes() - allocated) / (4.0 * measured),
|
|
clientWorkingSetBytes = before.WorkingSet64,
|
|
clientPrivateBytes = before.PrivateMemorySize64,
|
|
clientThreads = before.Threads.Count,
|
|
clientHandles = before.HandleCount,
|
|
collections = new[] { GC.CollectionCount(0) - gc[0], GC.CollectionCount(1) - gc[1], GC.CollectionCount(2) - gc[2] }
|
|
});
|
|
}
|
|
}
|
|
sealed class Auth : IOpcUaUserAuthenticator
|
|
{
|
|
public Task<OpcUaUserAuthResult> AuthenticateUserNameAsync(string username, string password, CancellationToken ct) => Task.FromResult(username == "apikey" && password == Password ? OpcUaUserAuthResult.AllowKey("matrix", [OpcUaDataPlaneRoles.WriteOperate, OpcUaDataPlaneRoles.AlarmAck], "matrix") : OpcUaUserAuthResult.Deny("Invalid ephemeral fixture key"));
|
|
}
|
|
sealed class Gateway : IOpcUaNodeWriteGateway { public Task<NodeWriteOutcome> WriteAsync(string id, object? value, AddressSpaceRealm realm, CancellationToken ct) => Task.FromResult(new NodeWriteOutcome(true, null)); }
|
|
sealed class Source(string root) : IHistorianDataSource
|
|
{
|
|
void Log(string operation, int rows) => File.AppendAllText(Path.Combine(root, "history-requests.jsonl"), JsonSerializer.Serialize(new { operation, rows }) + "\n");
|
|
public int MaxRawValuesPerRead => 7;
|
|
public Task<CoreHistory> ReadRawAsync(string r, DateTime s, DateTime e, uint m, CancellationToken ct)
|
|
{ var all = Enumerable.Range(0, 25).Select(i => new DataValueSnapshot(i, 0, Epoch.AddSeconds(i), Epoch.AddSeconds(i))).Where(v => v.SourceTimestampUtc >= s && v.SourceTimestampUtc < e).ToList(); var rows = all.Take((int)(m == 0 ? 7 : m)).ToArray(); Log("raw", rows.Length); return Task.FromResult(new CoreHistory(rows, all.Count > rows.Length ? [1] : null)); }
|
|
public Task<HistoricalEventsResult> ReadEventsAsync(string? source, DateTime s, DateTime e, int max, CancellationToken ct)
|
|
{ var all = Enumerable.Range(0, 25).Select(i => new HistoricalEvent($"e{i:00}", "matrix", Epoch.AddSeconds(i / 2), Epoch, "matrix-history", 500)).Where(v => v.EventTimeUtc >= s && v.EventTimeUtc < e).ToList(); var rows = all.Take(Math.Min(7, max <= 0 ? 7 : max)).ToArray(); Log("events", rows.Length); return Task.FromResult(new HistoricalEventsResult(rows, all.Count > rows.Length ? [1] : null)); }
|
|
public Task<CoreHistory> ReadProcessedAsync(string r, DateTime s, DateTime e, TimeSpan i, HistoryAggregateType a, CancellationToken ct) => NullHistorianDataSource.Instance.ReadProcessedAsync(r, s, e, i, a, ct);
|
|
public Task<CoreHistory> ReadAtTimeAsync(string r, IReadOnlyList<DateTime> t, CancellationToken ct) => NullHistorianDataSource.Instance.ReadAtTimeAsync(r, t, ct);
|
|
public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
|
|
public void Dispose() { }
|
|
}
|
|
}
|