/* ======================================================================== * 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.Threading; using System.Threading.Tasks; using Moq; using NUnit.Framework; using Opc.Ua.Client.Subscriptions; using Opc.Ua.Client.Subscriptions.Engine; using Opc.Ua.Tests; // CA2000: test code; many disposables are ownership-transferred to test fixtures or short-lived, // making CA2000 noisy without a real leak risk. Disabled file-level for the suite (the existing // narrow-scope pragma below at line ~163 predates the file-level decision). #pragma warning disable CA2000 namespace Opc.Ua.Client.Tests { [TestFixture] [Parallelizable] [Category("Client")] [Category("DefaultSubscriptionEngine")] public class DefaultSubscriptionEngineTests { private ITelemetryContext m_telemetry; private Mock m_mockContext; [SetUp] public void SetUp() { m_telemetry = NUnitTelemetryContext.Create(); m_mockContext = new Mock(); m_mockContext .Setup(c => c.SessionId) .Returns(new NodeId(1)); m_mockContext .Setup(c => c.Disposed) .Returns(false); m_mockContext .Setup(c => c.Reconnecting) .Returns(false); m_mockContext .Setup(c => c.Connected) .Returns(false); m_mockContext .Setup(c => c.Closing) .Returns(false); m_mockContext .Setup(c => c.Telemetry) .Returns(m_telemetry); m_mockContext .Setup(c => c.OperationTimeout) .Returns(60000); m_mockContext .Setup(c => c.ReturnDiagnostics) .Returns(DiagnosticsMasks.None); m_mockContext .Setup(c => c.ReconnectLock) .Returns(new SemaphoreSlim(1, 1)); m_mockContext .Setup(c => c.GoodPublishRequestCount) .Returns(0); #if OPCUA_V1_CLIENT m_mockContext .Setup(c => c.Subscriptions) .Returns(new List()); #endif } [Test] public void FactoryCreateReturnsV2Engine() { ISubscriptionEngine engine = DefaultSubscriptionEngineFactory.Instance.Create( m_mockContext.Object); Assert.That(engine, Is.Not.Null); Assert.That( engine, Is.InstanceOf()); engine.Dispose(); } [Test] public void FactoryInstanceIsSingleton() { DefaultSubscriptionEngineFactory first = DefaultSubscriptionEngineFactory.Instance; DefaultSubscriptionEngineFactory second = DefaultSubscriptionEngineFactory.Instance; Assert.That(first, Is.SameAs(second)); } [Test] public void ConstructorThrowsOnNullContext() { Assert.That( () => new DefaultSubscriptionEngine(null!), Throws.TypeOf()); } [Test] public void SubscriptionContextReportsRevisedSessionTimeout() { m_mockContext .Setup(c => c.OperationTimeout) .Returns(120000); m_mockContext .Setup(c => c.SessionTimeout) .Returns(60000); var context = new DefaultSubscriptionEngine.SubscriptionContextAdapter( m_mockContext.Object); Assert.That( context.SessionTimeout, Is.EqualTo(TimeSpan.FromSeconds(60))); } [TestCase(0.0)] [TestCase(-1.0)] [TestCase(double.NaN)] [TestCase(double.PositiveInfinity)] [TestCase(1e300)] public void SubscriptionContextReportsZeroForUnusableSessionTimeout( double sessionTimeout) { m_mockContext .Setup(c => c.SessionTimeout) .Returns(sessionTimeout); var context = new DefaultSubscriptionEngine.SubscriptionContextAdapter( m_mockContext.Object); Assert.That(context.SessionTimeout, Is.EqualTo(TimeSpan.Zero)); } [Test] public void StartPublishingDoesNotThrow() { using var engine = new DefaultSubscriptionEngine(m_mockContext.Object); Assert.That( () => engine.StartPublishing(timeout: 5000, fullQueue: false), Throws.Nothing); } [Test] public void PauseAndResumeDoNotThrow() { using var engine = new DefaultSubscriptionEngine(m_mockContext.Object); Assert.That(engine.PausePublishing, Throws.Nothing); Assert.That(engine.ResumePublishing, Throws.Nothing); } [Test] public async Task StopPublishingDisposesCleanlyAsync() { using var engine = new DefaultSubscriptionEngine(m_mockContext.Object); await engine.StopPublishingAsync(CancellationToken.None) .ConfigureAwait(false); } [Test] public void DisposeIsIdempotent() { #pragma warning disable CA2000 // Dispose objects before losing scope var engine = new DefaultSubscriptionEngine(m_mockContext.Object); #pragma warning restore CA2000 // Dispose objects before losing scope Assert.That(engine.Dispose, Throws.Nothing); Assert.That(engine.Dispose, Throws.Nothing); } [Test] public void NotifySubscriptionsChangedDoesNotThrow() { using var engine = new DefaultSubscriptionEngine(m_mockContext.Object); Assert.That( engine.NotifySubscriptionsChanged, Throws.Nothing); } [Test] public void GoodPublishRequestCountDefaultsToZero() { using var engine = new DefaultSubscriptionEngine(m_mockContext.Object); Assert.That( engine.GoodPublishRequestCount, Is.Zero); } [Test] public void BadPublishRequestCountDefaultsToZero() { using var engine = new DefaultSubscriptionEngine(m_mockContext.Object); Assert.That( engine.BadPublishRequestCount, Is.Zero); } [Test] public void MinMaxPublishRequestCountCanBeSet() { using var engine = new DefaultSubscriptionEngine(m_mockContext.Object); engine.MinPublishRequestCount = 3; engine.MaxPublishRequestCount = 20; Assert.That( engine.MinPublishRequestCount, Is.EqualTo(3)); Assert.That( engine.MaxPublishRequestCount, Is.EqualTo(20)); } [Test] public void BridgeConstructorThrowsOnNullSink() { Assert.That( () => new SubscriptionBridge(null!), Throws.TypeOf()); } [Test] public void BridgeCanBeInstantiated() { var sink = new RecordingSubscriptionMessageSink(); var bridge = new SubscriptionBridge(sink); Assert.That(bridge, Is.Not.Null); } [Test] public async Task BridgeOnKeepAliveForwardsToCacheSinkAsync() { var sink = new RecordingSubscriptionMessageSink(); var bridge = new SubscriptionBridge(sink); var mockSubscription = new Mock(); await bridge.OnKeepAliveNotificationAsync( mockSubscription.Object, sequenceNumber: 1, publishTime: DateTime.UtcNow, publishStateMask: PublishState.None) .ConfigureAwait(false); Assert.That(sink.Calls, Has.Count.EqualTo(1)); Assert.That(sink.Calls[0].Message.SequenceNumber, Is.EqualTo(1u)); } [Test] public async Task BridgeOnDataChangeForwardsToCacheSinkAsync() { var sink = new RecordingSubscriptionMessageSink(); var bridge = new SubscriptionBridge(sink); var mockSubscription = new Mock(); var changes = new DataValueChange[] { new( MonitoredItem: null, Value: new DataValue(new Variant(42)), DiagnosticInfo: null) }; await bridge.OnDataChangeNotificationAsync( mockSubscription.Object, sequenceNumber: 5, publishTime: DateTime.UtcNow, notification: changes.AsMemory(), publishStateMask: PublishState.None, stringTable: []) .ConfigureAwait(false); Assert.That(sink.Calls, Has.Count.EqualTo(1)); Assert.That(sink.Calls[0].Message.SequenceNumber, Is.EqualTo(5u)); } [Test] public async Task BridgeOnEventNotificationForwardsToCacheSinkAsync() { var sink = new RecordingSubscriptionMessageSink(); var bridge = new SubscriptionBridge(sink); var mockSubscription = new Mock(); var events = new EventNotification[] { new( MonitoredItem: null, Fields: new ArrayOf( new[] { new Variant("event-data") })) }; await bridge.OnEventDataNotificationAsync( mockSubscription.Object, sequenceNumber: 10, publishTime: DateTime.UtcNow, notification: events.AsMemory(), publishStateMask: PublishState.None, stringTable: ["ns-entry"]) .ConfigureAwait(false); Assert.That(sink.Calls, Has.Count.EqualTo(1)); Assert.That(sink.Calls[0].Message.SequenceNumber, Is.EqualTo(10u)); } [Test] public async Task BridgeDataChangeWithDiagnosticsPreservesDiagnosticInfoAsync() { var sink = new RecordingSubscriptionMessageSink(); var bridge = new SubscriptionBridge(sink); var mockSubscription = new Mock(); var diag = new DiagnosticInfo { AdditionalInfo = "test-diag" }; var changes = new DataValueChange[] { new( MonitoredItem: null, Value: new DataValue(new Variant(1)), DiagnosticInfo: diag) }; await bridge.OnDataChangeNotificationAsync( mockSubscription.Object, sequenceNumber: 7, publishTime: DateTime.UtcNow, notification: changes.AsMemory(), publishStateMask: PublishState.None, stringTable: []) .ConfigureAwait(false); Assert.That(sink.Calls, Has.Count.EqualTo(1)); NotificationMessage captured = sink.Calls[0].Message; Assert.That( captured.NotificationData.Count, Is.EqualTo(1)); ExtensionObject ext = captured.NotificationData[0]; Assert.That( ext.TryGetValue( out DataChangeNotification? dcn), Is.True); Assert.That( dcn!.DiagnosticInfos.Count, Is.EqualTo(1)); Assert.That( dcn.DiagnosticInfos[0].AdditionalInfo, Is.EqualTo("test-diag")); } /// /// Hand-rolled fake for /// (internal interface) that records every SaveMessageInCache /// call. Replaces a Moq-driven mock so that the test assembly does /// not need InternalsVisibleTo("DynamicProxyGenAssembly2") /// on the production assembly. /// private sealed class RecordingSubscriptionMessageSink : ISubscriptionMessageSink { public List<(ArrayOf AvailableSequenceNumbers, NotificationMessage Message)> Calls { get; } = []; public void SaveMessageInCache( ArrayOf availableSequenceNumbers, NotificationMessage message) { Calls.Add((availableSequenceNumbers, message)); } } } }