/* ======================================================================== * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved. * * OPC Foundation MIT License 1.00 * * Permission is hereby granted, free of charge, to any person * obtaining a copy of this software and associated documentation * files (the "Software"), to deal in the Software without * restriction, including without limitation the rights to use, * copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the * Software is furnished to do so, subject to the following * conditions: * * The above copyright notice and this permission notice shall be * included in all copies or substantial portions of the Software. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR * OTHER DEALINGS IN THE SOFTWARE. * * The complete license agreement can be found here: * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ using System; using System.Collections.Generic; using System.Globalization; using System.IO; using System.Linq; using System.Threading; using System.Threading.Tasks; using BenchmarkDotNet.Attributes; using Microsoft.Extensions.Logging; using NUnit.Framework; using Opc.Ua.Server.TestFramework; using Opc.Ua.Tests; using Quickstarts.ReferenceServer; namespace Opc.Ua.Client.TestFramework { /// /// Client test framework, common code to setup connections. /// public class ClientTestFramework { public static readonly object[] FixtureArgs = [ new object[] { Utils.UriSchemeOpcTcp }, new object[] { Utils.UriSchemeHttps }, new object[] { Utils.UriSchemeOpcHttps } ]; public const int MaxReferences = 100; public const int MaxTimeout = 10000; public const int TransportQuotaMaxMessageSize = 4 * 1024 * 1024; public const int TransportQuotaMaxStringLength = 1 * 1024 * 1024; public TokenValidatorMock TokenValidator { get; set; } = new TokenValidatorMock(); public bool SingleSession { get; set; } = true; public bool AllNodeManagers { get; set; } public int MaxChannelCount { get; set; } = 103; public int MaxSessionCount { get; set; } = 100; /// /// The server's maximum failed-authentication attempts before a client is /// locked out. Defaults to 0 (disabled) so that transient connection /// failures in load and scale fixtures never lock out the shared client /// certificate. Tests that explicitly validate lockout behaviour must set /// this to a positive value (e.g. 5) before calling /// . /// public int MaxFailedAuthenticationAttempts { get; set; } /// /// The server's maximum number of concurrent request-processing worker /// slots. Each outstanding (long-polled) Publish request occupies one /// slot for the duration it is held waiting for notifications, so a /// server serving N sessions with active subscriptions needs at least /// N slots plus head-room; otherwise held Publish requests exhaust the /// pool and starve all other requests (CreateSession/ActivateSession/ /// Read), surfacing as BadRequestTimeout. Defaults to the production /// default (100); load/scale fixtures that open many concurrent /// sessions must raise it above the session count. /// public int MaxRequestThreadCount { get; set; } = 100; /// /// The server's maximum number of concurrent subscriptions. Each session /// in the many-sessions scale test creates one subscription, so this must /// be raised above the session count for high-session runs. Defaults to /// 1000 (the fixture's prior fixed value). /// public int MaxSubscriptionCount { get; set; } = 1000; /// /// The server's minimum number of pre-spawned request-processing worker /// slots (and thread-pool minimum). Raising this warms the request /// pipeline so a burst of concurrent connects is not throttled by the /// thread pool's slow cold-start ramp, which otherwise surfaces as /// transient BadRequestTimeout on cheap operations while the pool grows. /// Defaults to the production default (10); load/scale fixtures that open /// many concurrent sessions raise it to smooth the connect burst. /// public int MinRequestThreadCount { get; set; } = 10; /// /// The requested session timeout, in milliseconds, of the fixture-shared /// created when is set. /// The shared session lives for the whole fixture and is often idle for /// minutes while earlier tests use their own sessions, so only its keep /// alive reads keep it alive on the server. With the /// default of 10 s, a single stall of the /// test host longer than that (seen on loaded net48 CI runners while /// other fixtures start and stop their servers) lets the server expire /// the session, and every later test in the fixture fails with /// BadSessionIdInvalid. /// Sessions created per test keep the default. /// public uint SharedSessionTimeout { get; set; } = 120_000; public bool SupportsExternalServerUrl { get; set; } /// /// When set before OneTimeSetUpCoreAsync, the tests run /// against the server at this URL instead of an in-process reference /// server, without reading the test run parameters. /// and stay /// null, so only tests that do not use them can run this way. /// public string ExternalServerUrl { get; set; } = null!; public bool UseSamplingGroupsInReferenceNodeManager { get; set; } /// /// When non-null, the fixture's is /// configured to use this engine factory for every session it /// creates. Defaults to /// so /// existing classic-engine tests (which use /// TestableSubscription / Session.AddSubscription) /// continue to work. V2-only fixtures should set this to /// during /// OneTimeSetUpAsync. /// public ISubscriptionEngineFactory ClientFixtureSubscriptionEngineFactory { get; set; } = ClassicSubscriptionEngineFactory.Instance; public ServerFixture ServerFixture { get; set; } = null!; public ClientFixture ClientFixture { get; set; } = null!; public ReferenceServer ReferenceServer { get; set; } = null!; public ArrayOf Endpoints { get; set; } public ArrayOf ReferenceDescriptions { get; set; } public ISession Session { get; protected set; } = null!; public OperationLimits OperationLimits { get; private set; } = null!; public int SecurityTokenLifetime { get; set; } = 3_600_000; public string UriScheme { get; } public string PkiRoot { get; set; } = null!; public Uri ServerUrl { get; private set; } = null!; public int ServerFixturePort { get; set; } public ExpandedNodeId[] TestSetStatic { get; private set; } public (Type Type, ExpandedNodeId[] NodeIds)[] TestSetStaticMassNumeric { get; } public ExpandedNodeId[] TestSetSimulation { get; private set; } public ExpandedNodeId[] TestSetDataSimulation { get; } public ExpandedNodeId[] TestSetHistory { get; } public ITelemetryContext Telemetry { get; } /// /// When set before OneTimeSetUpCoreAsync, the underlying /// hands this registry to the /// via /// ServerBase.TransportBindings immediately after /// construction. Integration fixtures that test a specific binding /// (Kestrel-TCP listener, custom transport, ...) use this hook to /// override the default opc.tcp factories without touching /// any process-wide state. /// public Bindings.ITransportBindingRegistry TransportBindingRegistry { get; set; } = null!; private readonly ILogger m_logger; public ClientTestFramework(string uriScheme = Utils.UriSchemeOpcTcp, ITelemetryContext? telemetry = null) { Telemetry = telemetry ?? NUnitTelemetryContext.Create(); m_logger = Telemetry.CreateLogger(); UriScheme = uriScheme; MaxFailedAuthenticationAttempts = 0; TestSetStatic = CommonTestWorkers.NodeIdTestSetStatic; TestSetStaticMassNumeric = CommonTestWorkers.NodeIdTestSetStaticMassNumeric; TestSetSimulation = CommonTestWorkers.NodeIdTestSetSimulation; TestSetDataSimulation = CommonTestWorkers.NodeIdTestSetDataSimulation; TestSetHistory = CommonTestWorkers.NodeIdTestDataHistory; } public void InitializeSession(ISession session) { Session = session; } [DatapointSource] public static readonly string[] Policies = [ .. GetPolicyUrisForTests() ]; protected static IEnumerable GetSupportedEccPolicyUris( bool includeCurvePolicies = true) { IEnumerable displayNames = SecurityPolicies.Default.GetDisplayNames() .Where(name => name.StartsWith("ECC_", StringComparison.Ordinal)); if (!includeCurvePolicies) { displayNames = displayNames .Where(name => !name.StartsWith("ECC_curve", StringComparison.Ordinal)); } return displayNames.Select(name => SecurityPolicies.Default.GetUri(name)!); } private static IEnumerable GetPolicyUrisForTests() { IEnumerable displayNames = SecurityPolicies.Default.GetDisplayNames(); #if NETFRAMEWORK displayNames = displayNames.Where(name => !name.EndsWith("_AesGcm", StringComparison.Ordinal) && !name.EndsWith("_ChaChaPoly", StringComparison.Ordinal)); #endif return displayNames.Select(name => SecurityPolicies.Default.GetUri(name)!); } protected async Task IgnoreIfPolicyNotAdvertisedAsync(string securityPolicyUri) { if (Endpoints.IsNull) { Endpoints = await ClientFixture.GetEndpointsAsync(ServerUrl).ConfigureAwait(false); } bool isAdvertised = false; foreach (EndpointDescription endpoint in Endpoints) { if (string.Equals( endpoint.SecurityPolicyUri, securityPolicyUri, StringComparison.Ordinal)) { isAdvertised = true; break; } } if (!isAdvertised) { string advertisedPolicies = Endpoints.IsNull ? "" : string.Join( ", ", (Endpoints.ToArray() ?? []) .Select(endpoint => endpoint.SecurityPolicyUri) .Where(policy => !string.IsNullOrEmpty(policy)) .Distinct() .OrderBy(policy => policy, StringComparer.Ordinal)); Assert.Ignore( $"SecurityPolicy '{securityPolicyUri}' is not advertised by the server. " + $"Advertised: {advertisedPolicies}"); } } /// /// Set up a Server and a Client instance. /// public virtual Task OneTimeSetUpAsync() { return OneTimeSetUpCoreAsync(); } /// /// Setup a server and client fixture. /// public virtual async Task OneTimeSetUpCoreAsync( bool securityNone = false, bool enableClientSideTracing = false, bool enableServerSideTracing = false, bool disableActivityLogging = false) { // pki directory root for test runs. PkiRoot = Path.GetTempPath() + Path.GetRandomFileName(); if (m_logger.IsEnabled(LogLevel.Information)) { m_logger.LogInformation("Using the Pki Root {FilePath}", PkiRoot); } // An external server set in code (the interop tests run the // client tests against a 1.5 server process) keeps the default // test sets, which address the same reference server nodes. string customUrl = ExternalServerUrl; if (customUrl != null) { m_logger.UsingExternalServerUrl(customUrl); } // The parameters are read from the .runsettings file else if (SupportsExternalServerUrl) { customUrl = TestContext.Parameters["ServerUrl"]!; if (customUrl?.StartsWith(UriScheme, StringComparison.Ordinal) == true) { m_logger.UsingExternalServerUrl(customUrl); // load custom test sets TestSetStatic = ReadCustomTestSet("TestSetStatic"); TestSetSimulation = ReadCustomTestSet("TestSetSimulation"); } else { customUrl = null!; } } if (customUrl == null) { await CreateReferenceServerFixtureAsync( enableServerSideTracing, disableActivityLogging, securityNone) .ConfigureAwait(false); } ClientFixture = new ClientFixture(enableClientSideTracing, disableActivityLogging, Telemetry); // If a derived fixture pinned the engine factory, propagate // it. Default = null = use Session default (currently the // classic engine; see Session.cs). if (ClientFixtureSubscriptionEngineFactory != null) { ClientFixture.UseSubscriptionEngineFactory( ClientFixtureSubscriptionEngineFactory); } await ClientFixture.LoadClientConfigurationAsync(PkiRoot).ConfigureAwait(false); ClientFixture.Config.TransportQuotas!.MaxMessageSize = TransportQuotaMaxMessageSize; ClientFixture.Config.TransportQuotas.MaxByteStringLength = ClientFixture .Config .TransportQuotas .MaxStringLength = TransportQuotaMaxStringLength; ClientFixture.Config.TransportQuotas.SecurityTokenLifetime = SecurityTokenLifetime; if (!string.IsNullOrEmpty(customUrl)) { ServerUrl = new Uri(customUrl); } else { string url = UriScheme + "://localhost:" + ServerFixturePort.ToString(CultureInfo.InvariantCulture); if (UriScheme.StartsWith(Utils.UriSchemeHttp, StringComparison.Ordinal) || Utils.IsUriHttpsScheme(UriScheme)) { url += ConfiguredEndpoint.DiscoverySuffix; } ServerUrl = new Uri(url); } if (SingleSession) { uint sessionTimeout = ClientFixture.SessionTimeout; ClientFixture.SessionTimeout = SharedSessionTimeout; try { Session = await ClientFixture .ConnectAsync(ServerUrl, SecurityPolicies.Basic256Sha256) .ConfigureAwait(false); Assert.That(Session, Is.Not.Null); } catch (Exception e) { TestContext.Progress.WriteLine( $"OneTimeSetUp failed to create session with {ServerUrl}. Error: {e.Message}"); throw; } finally { ClientFixture.SessionTimeout = sessionTimeout; } } } public virtual async Task CreateReferenceServerFixtureAsync( bool enableTracing, bool disableActivityLogging, bool securityNone) { // start Ref server ServerFixture = new ServerFixture( t => new ReferenceServer(t), enableTracing, disableActivityLogging) { UriScheme = UriScheme, SecurityNone = securityNone, AutoAccept = true, AllNodeManagers = AllNodeManagers, OperationLimits = true, UseSamplingGroupsInReferenceNodeManager = UseSamplingGroupsInReferenceNodeManager, TransportBindingRegistry = TransportBindingRegistry }; await ServerFixture.LoadConfigurationAsync(PkiRoot).ConfigureAwait(false); ServerFixture.Config.TransportQuotas!.MaxMessageSize = TransportQuotaMaxMessageSize; ServerFixture.Config.TransportQuotas.MaxByteStringLength = ServerFixture .Config .TransportQuotas .MaxStringLength = TransportQuotaMaxStringLength; ServerFixture.Config.TransportQuotas.SecurityTokenLifetime = SecurityTokenLifetime; void AddExplicitUserTokenPolicies(string securityPolicyUri) { if (SecurityPolicies.Default.GetInfo(securityPolicyUri) == null) { return; } ServerFixture.Config.ServerConfiguration.UserTokenPolicies += new UserTokenPolicy(UserTokenType.UserName) { SecurityPolicyUri = securityPolicyUri }; ServerFixture.Config.ServerConfiguration.UserTokenPolicies += new UserTokenPolicy(UserTokenType.Certificate) { SecurityPolicyUri = securityPolicyUri }; ServerFixture.Config.ServerConfiguration.UserTokenPolicies += new UserTokenPolicy(UserTokenType.IssuedToken) { IssuedTokenType = Profiles.JwtUserToken, PolicyId = Profiles.JwtUserToken, SecurityPolicyUri = securityPolicyUri }; } ServerFixture.Config.ServerConfiguration!.UserTokenPolicies += new UserTokenPolicy(UserTokenType.UserName); ServerFixture.Config.ServerConfiguration.UserTokenPolicies += new UserTokenPolicy(UserTokenType.Certificate); ServerFixture.Config.ServerConfiguration.UserTokenPolicies += new UserTokenPolicy(UserTokenType.IssuedToken) { IssuedTokenType = Profiles.JwtUserToken }; // Reconnect tests need an explicit legacy RSA token policy in addition to the // generic channel-bound policies to cover alternate-token activation. AddExplicitUserTokenPolicies(SecurityPolicies.Basic128Rsa15); foreach (string securityPolicyUri in GetSupportedEccPolicyUris()) { AddExplicitUserTokenPolicies(securityPolicyUri); } ServerFixture.Config.ServerConfiguration.MaxChannelCount = MaxChannelCount; ServerFixture.Config.ServerConfiguration.MaxSessionCount = MaxSessionCount; ServerFixture.Config.ServerConfiguration.MaxFailedAuthenticationAttempts = MaxFailedAuthenticationAttempts; ServerFixture.Config.ServerConfiguration.MaxRequestThreadCount = MaxRequestThreadCount; ServerFixture.Config.ServerConfiguration.MinRequestThreadCount = MinRequestThreadCount; ServerFixture.Config.ServerConfiguration.MaxSubscriptionCount = MaxSubscriptionCount; ServerFixture.Config.ServerConfiguration.MaxQueuedRequestCount = 100000; ReferenceServer = await ServerFixture.StartAsync() .ConfigureAwait(false); ReferenceServer.TokenValidator = TokenValidator; ServerFixturePort = ServerFixture.Port; } /// /// Tear down the Server and the Client. /// public virtual async Task OneTimeTearDownAsync() { if (Session != null) { try { await Session.CloseAsync(5000, true).ConfigureAwait(false); } catch (Exception e) { m_logger.LogError(e, "Error closing session during teardown."); } Session.Dispose(); Session = null!; } if (ServerFixture != null) { try { await ServerFixture.StopAsync().ConfigureAwait(false); } catch (Exception e) { m_logger.LogError(e, "Error stopping server during teardown."); } await Task.Delay(100).ConfigureAwait(false); } if (ClientFixture != null) { await ClientFixture.DisposeAsync().ConfigureAwait(false); } // Clean up pki try { if (!string.IsNullOrEmpty(PkiRoot) && Directory.Exists(PkiRoot)) { Directory.Delete(PkiRoot, true); } } catch { } } /// /// Test setup. /// public virtual async Task SetUpAsync() { if (!SingleSession) { try { Session = await ClientFixture .ConnectAsync(ServerUrl, SecurityPolicies.Basic256Sha256) .ConfigureAwait(false); } catch (Exception e) { Assert.Ignore( $"OneTimeSetup failed to create session, tests skipped. Error: {e.Message}"); } } } /// /// Test Teardown. /// public virtual async Task TearDownAsync() { if (!SingleSession && Session != null) { await Session.CloseAsync().ConfigureAwait(false); Session.Dispose(); Session = null!; } } /// /// Return a test set of nodes with static character. /// /// The namesapce table used in the session. /// The list of static test nodes. public IList GetTestSetStatic(NamespaceTable namespaceUris) { return [.. TestSetStatic.Select(n => ExpandedNodeId.ToNodeId(n, namespaceUris)) .Where(n => !n.IsNull)]; } /// /// Return a test set of nodes with static character. /// 100 of each numeric data type. /// /// The namespace table used in the session. /// The list of static test nodes. public IDictionary GetTestSetStaticMassNumeric(NamespaceTable namespaceUris) { var result = new Dictionary(); foreach ((Type Type, ExpandedNodeId[] NodeIds) entry in TestSetStaticMassNumeric) { Type type = entry.Type; foreach (ExpandedNodeId expandedNodeId in entry.NodeIds) { var nodeId = ExpandedNodeId.ToNodeId(expandedNodeId, namespaceUris); if (!nodeId.IsNull) { result[nodeId] = type; } } } return result; } /// /// Return a test set of nodes with simulated values. /// /// The namesapce table used in the session. /// The list of simulated test nodes. public IList GetTestSetSimulation(NamespaceTable namespaceUris) { return [.. TestSetSimulation.Select(n => ExpandedNodeId.ToNodeId(n, namespaceUris)) .Where(n => !n.IsNull)]; } /// /// Return a test set of nodes all nodes with simulated values. /// /// The namesapce table used in the session. /// The list of simulated test nodes. public IList GetTestSetFullSimulation(NamespaceTable namespaceUris) { var simulation = TestSetSimulation .Select(n => ExpandedNodeId.ToNodeId(n, namespaceUris)) .Where(n => !n.IsNull) .ToList(); simulation.AddRange( TestSetDataSimulation.Select(n => ExpandedNodeId.ToNodeId(n, namespaceUris)) .Where(n => !n.IsNull)); return simulation; } /// /// Return a test data set of nodes with simulated values. /// /// The namesapce table used in the session. /// The list of simulated test nodes. public IList GetTestSetDataSimulation(NamespaceTable namespaceUris) { return [ .. TestSetDataSimulation.Select(n => ExpandedNodeId.ToNodeId(n, namespaceUris)) .Where(n => !n.IsNull) ]; } /// /// Return a test set of nodes with history values. /// /// The namesapce table used in the session. /// The list of test nodes. public IList GetTestSetHistory(NamespaceTable namespaceUris) { return [.. TestSetHistory.Select(n => ExpandedNodeId.ToNodeId(n, namespaceUris)) .Where(n => !n.IsNull)]; } /// /// Enumerator for security policies. /// public IEnumerable BenchPolicies() { return Policies; } /// /// Helper variable for benchmark. /// [ParamsSource(nameof(BenchPolicies))] public string SecurityPolicy = SecurityPolicies.None; /// /// Global Setup for benchmarks. /// public virtual void GlobalSetup() { m_logger.LogInformation("GlobalSetup: Start Server"); OneTimeSetUpCoreAsync().GetAwaiter().GetResult(); m_logger.LogInformation("GlobalSetup: Connecting"); Session = ClientFixture.ConnectAsync(ServerUrl, SecurityPolicy).GetAwaiter() .GetResult(); m_logger.LogInformation("GlobalSetup: Ready"); } /// /// Global cleanup for benchmarks. /// public virtual void GlobalCleanup() { m_logger.LogInformation("GlobalCleanup: Disconnect and Stop Server"); OneTimeTearDownAsync().GetAwaiter().GetResult(); m_logger.LogInformation("GlobalCleanup: Done"); } public async Task GetOperationLimitsAsync() { OperationLimits = new OperationLimits { MaxNodesPerRead = await GetOperationLimitValueAsync( VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerRead ).ConfigureAwait(false), MaxNodesPerHistoryReadData = await GetOperationLimitValueAsync( VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerHistoryReadData ).ConfigureAwait(false), MaxNodesPerHistoryReadEvents = await GetOperationLimitValueAsync( VariableIds .Server_ServerCapabilities_OperationLimits_MaxNodesPerHistoryReadEvents ).ConfigureAwait(false), MaxNodesPerWrite = await GetOperationLimitValueAsync( VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerWrite ).ConfigureAwait(false), MaxNodesPerHistoryUpdateData = await GetOperationLimitValueAsync( VariableIds .Server_ServerCapabilities_OperationLimits_MaxNodesPerHistoryUpdateData ).ConfigureAwait(false), MaxNodesPerHistoryUpdateEvents = await GetOperationLimitValueAsync( VariableIds .Server_ServerCapabilities_OperationLimits_MaxNodesPerHistoryUpdateEvents ).ConfigureAwait(false), MaxNodesPerBrowse = await GetOperationLimitValueAsync( VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerBrowse ).ConfigureAwait(false), MaxMonitoredItemsPerCall = await GetOperationLimitValueAsync( VariableIds.Server_ServerCapabilities_OperationLimits_MaxMonitoredItemsPerCall ).ConfigureAwait(false), MaxNodesPerNodeManagement = await GetOperationLimitValueAsync( VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerNodeManagement ).ConfigureAwait(false), MaxNodesPerRegisterNodes = await GetOperationLimitValueAsync( VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerRegisterNodes ).ConfigureAwait(false), MaxNodesPerTranslateBrowsePathsToNodeIds = await GetOperationLimitValueAsync( VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerTranslateBrowsePathsToNodeIds ).ConfigureAwait(false), MaxNodesPerMethodCall = await GetOperationLimitValueAsync( VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerMethodCall ).ConfigureAwait(false) }; } public async Task GetOperationLimitValueAsync(NodeId nodeId) { try { return await Session.ReadValueAsync(nodeId).ConfigureAwait(false); } catch (ServiceResultException sre) { if (sre.StatusCode == StatusCodes.BadNodeIdUnknown) { return 0; } throw; } } public void OutputSubscriptionInfo(TextWriter writer, Subscription subscription) { writer.WriteLine("Subscription : {0}", subscription.DisplayName); writer.WriteLine("CurrentKeepAliveCount : {0}", subscription.CurrentKeepAliveCount); writer.WriteLine( "CurrentPublishingEnabled: {0}", subscription.CurrentPublishingEnabled); writer.WriteLine("CurrentPriority : {0}", subscription.CurrentPriority); writer.WriteLine("PublishTime : {0}", subscription.PublishTime); writer.WriteLine("LastNotificationTime : {0}", subscription.LastNotificationTime); writer.WriteLine("SequenceNumber : {0}", subscription.SequenceNumber); writer.WriteLine("NotificationCount : {0}", subscription.NotificationCount); writer.WriteLine("LastNotification : {0}", subscription.LastNotification); writer.WriteLine("Notifications : {0}", subscription.Notifications.Count()); writer.WriteLine( "OutstandingMessageWorker: {0}", subscription.OutstandingMessageWorkers); } private static ExpandedNodeId[] ReadCustomTestSet(string param) { // load custom test sets string testSetParameter = TestContext.Parameters[param]!; string[] testSetParameters = testSetParameter!.Split('#'); if (testSetParameters != null) { // parse the custom content var testSet = new List(); foreach (string parameter in testSetParameters) { testSet.Add(ExpandedNodeId.Parse(parameter)); } return [.. testSet]; } return []; } protected void SessionClosing(object? sender, EventArgs e) { if (sender is ISession session) { TestContext.Out.WriteLine("Session_Closing: {0}", session.SessionId); } } /// /// Create a set of subscriptions with monitored items on the given session. /// protected async Task CreateSubscriptionsAsync( ISession session, Subscription template, SubscriptionCollection originSubscriptions, int[] notificationCounters, int[] fastDataCounters, int subscriptionCount, uint queueSize) { for (int ii = 0; ii < subscriptionCount; ii++) { // create subscription with static monitored items var subscription = new TestableSubscription(template) { Handle = ii, FastDataChangeCallback = (s, n, _) => { TestContext.Out.WriteLine( $"FastDataChangeHandlerOrigin: {s.Id}-{n.SequenceNumber}-{n.MonitoredItems.Count}"); Interlocked.Increment(ref fastDataCounters[(int)s.Handle!]); } }; subscription.StateChanged += (s, e) => TestContext.Out .WriteLine($"StateChanged: {s.Session!.SessionId}-{s.Id}-{e.Status}"); subscription.PublishStatusChanged += (s, e) => TestContext.Out.WriteLine( $"PublishStatusChanged: {s.Session!.SessionId}-{s.Id}-{e.Status}"); originSubscriptions.Add(subscription); session.AddSubscription(subscription); await subscription.CreateAsync().ConfigureAwait(false); // set defaults subscription.DefaultItem.DiscardOldest = true; subscription.DefaultItem.QueueSize = ii == 0 ? 0U : queueSize; subscription.DefaultItem.MonitoringMode = MonitoringMode.Reporting; // create test set using the passed session's namespace URIs NamespaceTable namespaceUris = session.NamespaceUris; var testSet = new List(); if (ii == 0) { testSet.AddRange(GetTestSetStatic(namespaceUris)); } else { testSet.AddRange(GetTestSetSimulation(namespaceUris)); } var list = CreateMonitoredItemTestSet(subscription, testSet).ToList(); list.ForEach(i => i.Notification += (item, _) => { Interlocked.Increment(ref notificationCounters[(int)subscription.Handle]); foreach (DataValue value in item.DequeueValues()) { TestContext.Out.WriteLine( "Org:{0}: {1:20}, {2}, {3}, {4}", subscription.Id, item.DisplayName, value.WrappedValue, value.SourceTimestamp, value.StatusCode); } }); subscription.AddItems(list); await subscription.ApplyChangesAsync().ConfigureAwait(false); } } /// /// Create monitored items from a list of node ids on a subscription. /// protected static List CreateMonitoredItemTestSet( Subscription subscription, IList nodeIds) { var list = new List(); foreach (NodeId nodeId in nodeIds) { var item = new TestableMonitoredItem(subscription.DefaultItem) { StartNodeId = nodeId }; list.Add(item); } return list; } } /// /// Source-generated log messages of . /// internal static partial class ClientTestFrameworkLog { [LoggerMessage(Level = LogLevel.Information, Message = "Using the external Server Url {Url}")] public static partial void UsingExternalServerUrl(this ILogger logger, string url); } }