/* ======================================================================== * 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.Concurrent; using System.Collections.Generic; using System.Globalization; using System.Linq; using System.Threading; using System.Threading.Tasks; using NUnit.Framework; using Opc.Ua.Client; using Opc.Ua.Client.TestFramework; using ISession = Opc.Ua.Client.ISession; using Subscription = Opc.Ua.Client.Subscription; #pragma warning disable CS0618 // tests exercise the obsolete classic API namespace Opc.Ua.Sessions.Tests { /// /// Long-running connection stability test. /// [TestFixture] [SetCulture("en-us")] [SetUICulture("en-us")] [Category("Client")] public class ConnectionStabilityTest : ClientTestFramework { /// /// 5 minutes for CI /// private const int kSecurityTokenLifetimeCIMs = 5 * 60 * 1000; /// /// 10 seconds for local testing /// private const int kSecurityTokenLifetimeLocalMs = 10 * 1000; /// /// Report status every 60 seconds /// private const int kStatusReportIntervalSeconds = 60; /// /// Accept 95% of expected notifications (5% tolerance) /// private const double kNotificationToleranceRatio = 0.95; public ConnectionStabilityTest() : base(Utils.UriSchemeOpcTcp) { SupportsExternalServerUrl = true; } [Test] [Explicit] [Category("ConnectionStability")] public async Task ShortHaulStabilityTestAsync() { try { SecurityTokenLifetime = kSecurityTokenLifetimeLocalMs; await OneTimeSetUpAsync().ConfigureAwait(false); // 2 minutes for local testing await RunStabilityTestAsync(2).ConfigureAwait(false); } finally { await OneTimeTearDownAsync().ConfigureAwait(false); } } [Test] [Explicit] [Category("ConnectionStability")] public async Task LongHaulStabilityTestAsync() { try { SecurityTokenLifetime = kSecurityTokenLifetimeCIMs; await OneTimeSetUpAsync().ConfigureAwait(false); // Configurable duration for CI testing string envValue = Environment.GetEnvironmentVariable("TEST_DURATION_MINUTES")!; if (string.IsNullOrEmpty(envValue) || !int.TryParse(envValue, out int minutes) || minutes <= 0) { minutes = 90; // Default to 90 minutes for CI } await RunStabilityTestAsync(minutes).ConfigureAwait(false); } finally { await OneTimeTearDownAsync().ConfigureAwait(false); } } /// /// Long-running test that verifies connection stability over a configurable duration. /// Tests that: /// - Connection remains stable over extended period /// - Subscriptions deliver all expected values (no message loss) /// - Security token renewals happen correctly /// private async Task RunStabilityTestAsync(int testDurationMinutes) { // Get test duration from environment variable or use default int testDurationSeconds = testDurationMinutes * 60; int tokenLifetimeMs = SecurityTokenLifetime; TestContext.Out.WriteLine( $"Starting connection stability test for {testDurationMinutes} minutes ({testDurationSeconds} seconds)"); TestContext.Out.WriteLine( $"Security token lifetime: {tokenLifetimeMs / 1000} seconds ({tokenLifetimeMs / 60000.0:F1} minutes)"); const int publishingInterval = 1000; // 1 second const int writerInterval = 2000; // 2 seconds const int samplingInterval = 500; // 500 ms var valueChanges = new ConcurrentDictionary(); var clientHandles = new ConcurrentDictionary(); var errors = new ConcurrentBag(); ISession? session = null; Subscription? subscription = null; try { // Get nodes for subscription IDictionary nodeIds = GetTestSetStaticMassNumeric(Session.NamespaceUris); if (nodeIds.Count == 0) { Assert.Ignore("No nodes for simulation found, ignoring test."); } TestContext.Out.WriteLine($"Subscribing to {nodeIds.Count} nodes."); // Create session session = await ClientFixture.ConnectAsync( ServerUrl, SecurityPolicies.Basic256Sha256).ConfigureAwait(false); Assert.That(session, Is.Not.Null, "Failed to create session"); // Create subscription subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = publishingInterval, PublishingEnabled = true, KeepAliveCount = 10, LifetimeCount = 100, MaxNotificationsPerPublish = 1000, Priority = 100 }; // Add monitored items foreach (NodeId nodeId in nodeIds.Keys) { var item = new MonitoredItem(subscription.DefaultItem) { StartNodeId = nodeId, AttributeId = Attributes.Value, MonitoringMode = MonitoringMode.Reporting, SamplingInterval = samplingInterval, QueueSize = 10, DiscardOldest = true }; valueChanges.TryAdd(nodeId, 0); clientHandles.TryAdd(item.ClientHandle, nodeId); subscription.AddItem(item); } // Set up notification callback subscription.FastDataChangeCallback = (sub, item, value) => { try { foreach (MonitoredItemNotification notification in item.MonitoredItems) { if (!StatusCode.IsGood(notification.Value.StatusCode)) { string error = $"Bad status code received: {notification.Value.StatusCode} for client handle {notification.ClientHandle}"; errors.Add(error); TestContext.Out.WriteLine($"ERROR: {error}"); } else if (clientHandles.TryGetValue(notification.ClientHandle, out NodeId nodeId)) { valueChanges.AddOrUpdate(nodeId, 1, (key, count) => count + 1); } } } catch (Exception ex) { string error = $"Exception in data change callback: {ex.Message}"; errors.Add(error); TestContext.Out.WriteLine($"ERROR: {error}"); } }; // Create subscription on server session.AddSubscription(subscription); await subscription.CreateAsync().ConfigureAwait(false); TestContext.Out.WriteLine($"Subscription created with {subscription.MonitoredItemCount} monitored items"); // Wait for the subscription to deliver initial values for all nodes before // starting writes. This prevents the first few writes from being missed when // the CI machine is under load and the subscription is still starting up. TestContext.Out.WriteLine("Waiting for subscription to deliver initial values..."); DateTime settleDeadline = DateTime.UtcNow.AddSeconds(15); while (valueChanges.Values.Sum() < nodeIds.Count && DateTime.UtcNow < settleDeadline) { await Task.Delay(200).ConfigureAwait(false); } int settledCount = valueChanges.Values.Sum(); TestContext.Out.WriteLine($"Subscription settled: {settledCount}/{nodeIds.Count} initial notifications received"); // Reset counters so the measurement period starts from when the subscription // is confirmed active (initial-value notifications don't count as write-triggered). foreach (NodeId nodeId in nodeIds.Keys) { valueChanges[nodeId] = 0; } // Create writer session ISession writerSession = await ClientFixture.ConnectAsync( ServerUrl, SecurityPolicies.Basic256Sha256).ConfigureAwait(false); Assert.That(writerSession, Is.Not.Null, "Failed to create writer session"); // Writer task - continuously write values int writeCount = 0; using var writerCts = new CancellationTokenSource(); var writerTask = Task.Run(async () => { while (!writerCts.IsCancellationRequested) { writeCount++; var nodesToWrite = new List(); foreach (KeyValuePair node in nodeIds) { nodesToWrite.Add(new WriteValue { NodeId = node.Key, AttributeId = Attributes.Value, Value = new DataValue( new Variant( Convert.ChangeType(writeCount, node.Value, CultureInfo.InvariantCulture) ) ) }); } try { await writerSession.WriteAsync(null, nodesToWrite, writerCts.Token).ConfigureAwait(false); } catch (ServiceResultException sre) when (sre.StatusCode == StatusCodes.BadRequestInterrupted) { // BadRequestInterrupted is expected when a write is in-flight during a // secure channel renewal. The channel renewal briefly interrupts the old // channel before the new one activates, so this is not a test failure. TestContext.Out.WriteLine( $"INFO: Write interrupted during channel renewal (expected): {sre.Message}"); } catch (Exception ex) { string error = $"Writer session error: {ex.Message}"; errors.Add(error); TestContext.Out.WriteLine($"ERROR: {error}"); } try { await Task.Delay(writerInterval, writerCts.Token).ConfigureAwait(false); } catch (OperationCanceledException) { break; } } TestContext.Out.WriteLine($"Writer task completed. Total writes: {writeCount}"); }, writerCts.Token); // Status reporting task using var statusReportingCts = new CancellationTokenSource(); var statusTask = Task.Run(async () => { int reportCount = 0; while (!statusReportingCts.IsCancellationRequested) { try { await Task.Delay( TimeSpan.FromSeconds(kStatusReportIntervalSeconds), statusReportingCts.Token).ConfigureAwait(false); } catch (OperationCanceledException) { break; } reportCount++; int totalNotifications = valueChanges.Values.Sum(); int elapsedMinutes = reportCount * kStatusReportIntervalSeconds / 60; TestContext.Out.WriteLine( $"[Status Report {reportCount}] Elapsed: {elapsedMinutes} minutes, " + $"Total notifications: {totalNotifications}, Write count: {writeCount}, Errors: {errors.Count}"); #if DEBUG // Report per-node statistics if (reportCount % 5 == 0) // Every 5 minutes { TestContext.Out.WriteLine("Per-node notification counts:"); foreach (KeyValuePair kvp in valueChanges.OrderBy(x => x.Key.ToString())) { TestContext.Out.WriteLine($" {kvp.Key}: {kvp.Value} notifications"); } } #endif } }, statusReportingCts.Token); // Run test for the specified duration TestContext.Out.WriteLine($"Test running... will complete at {DateTime.UtcNow.AddSeconds(testDurationSeconds):yyyy-MM-dd HH:mm:ss} UTC"); await Task.Delay(TimeSpan.FromSeconds(testDurationSeconds)).ConfigureAwait(false); // Stop tasks TestContext.Out.WriteLine("Test duration elapsed. Stopping writer and status tasks..."); await writerCts.CancelAsync().ConfigureAwait(false); await statusReportingCts.CancelAsync().ConfigureAwait(false); try { await writerTask.ConfigureAwait(false); } catch (OperationCanceledException) { /* expected */ } try { await statusTask.ConfigureAwait(false); } catch (OperationCanceledException) { /* expected */ } // Wait for final notifications to be processed TestContext.Out.WriteLine("Waiting for final notifications to be processed..."); await Task.Delay(publishingInterval * 5).ConfigureAwait(false); // Verification TestContext.Out.WriteLine("=== Final Results ==="); TestContext.Out.WriteLine($"Test duration: {testDurationMinutes} minutes"); TestContext.Out.WriteLine($"Security token lifetime: {tokenLifetimeMs / 1000} seconds ({tokenLifetimeMs / 60000.0:F1} minutes)"); TestContext.Out.WriteLine($"Expected token renewals: ~{testDurationMinutes * 60000 / tokenLifetimeMs} times"); TestContext.Out.WriteLine($"Total write operations: {writeCount}"); TestContext.Out.WriteLine($"Total errors: {errors.Count}"); // Calculate expected and received notifications int totalNotifications = valueChanges.Values.Sum(); int expectedMinNotifications = (writeCount - 1) * nodeIds.Count; // Allow for some timing variance TestContext.Out.WriteLine($"Total notifications received: {totalNotifications}"); TestContext.Out.WriteLine($"Expected minimum notifications: {expectedMinNotifications}"); // Per-node verification TestContext.Out.WriteLine("Per-node results:"); bool allNodesReceivedData = true; foreach (NodeId nodeId in nodeIds.Keys) { if (valueChanges.TryGetValue(nodeId, out int changes)) { #if DEBUG TestContext.Out.WriteLine($" {nodeId}: {changes} notifications"); #endif double expected = Math.Max(0, writeCount - 1) * kNotificationToleranceRatio; if (changes < expected) { allNodesReceivedData = false; TestContext.Out.WriteLine( $" WARNING: Expected at least {expected:F0} but got {changes} on {nodeId}"); } } else { allNodesReceivedData = false; TestContext.Out.WriteLine($" {nodeId}: 0 notifications (ERROR)"); } } // List all errors if (!errors.IsEmpty) { TestContext.Out.WriteLine($"Errors encountered ({errors.Count}):"); foreach (string error in errors.Take(20)) // Show first 20 errors { TestContext.Out.WriteLine($" - {error}"); } if (errors.Count > 20) { TestContext.Out.WriteLine($" ... and {errors.Count - 20} more errors"); } } // Cleanup writer session try { await writerSession.CloseAsync().ConfigureAwait(false); writerSession.Dispose(); } catch (Exception ex) { TestContext.Out.WriteLine($"Failed to close writer session: {ex.Message}"); } // Assertions Assert.That(allNodesReceivedData, Is.True, "Not all nodes received expected data"); Assert.That(errors, Is.Empty, $"Test encountered {errors.Count} errors"); Assert.That( totalNotifications, Is.GreaterThanOrEqualTo(expectedMinNotifications), "Total notifications received is less than expected minimum"); TestContext.Out.WriteLine("Connection stability test PASSED"); } finally { // Cleanup if (subscription != null) { try { await subscription.DeleteAsync(true).ConfigureAwait(false); } catch (Exception ex) { TestContext.Out.WriteLine($"Failed to delete subscription: {ex.Message}"); } subscription.Dispose(); } if (session != null) { try { await session.CloseAsync().ConfigureAwait(false); session.Dispose(); } catch (Exception ex) { TestContext.Out.WriteLine($"Failed to close session: {ex.Message}"); } } } } } }