/* ======================================================================== * 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.Threading; using System.Threading.Tasks; using Moq; using NUnit.Framework; using Opc.Ua.Tests; namespace Opc.Ua.Client.Tests { [TestFixture] [Parallelizable] [Category("Client")] [Category("ClassicSubscriptionEngine")] public class ClassicSubscriptionEngineTests { private ITelemetryContext m_telemetry; private Mock m_mockContext; private SemaphoreSlim m_lock; [SetUp] public void SetUp() { m_telemetry = NUnitTelemetryContext.Create(); m_lock = new SemaphoreSlim(1, 1); 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.Subscriptions) .Returns([]); m_mockContext.Setup(c => c.OperationTimeout) .Returns(60000); m_mockContext.Setup(c => c.ReturnDiagnostics) .Returns(DiagnosticsMasks.None); m_mockContext.Setup(c => c.ReconnectLock) .Returns(m_lock); m_mockContext.Setup(c => c.GoodPublishRequestCount) .Returns(0); } [TearDown] public void TearDown() { m_lock.Dispose(); } [Test] public void FactoryCreateReturnsClassicEngine() { ISubscriptionEngine engine = ClassicSubscriptionEngineFactory.Instance.Create( m_mockContext.Object); Assert.That(engine, Is.Not.Null); Assert.That( engine, Is.InstanceOf()); engine.Dispose(); } [Test] public void StartPublishingWithNoSubscriptionsDoesNothing() { m_mockContext.Setup(c => c.Subscriptions) .Returns([]); using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); engine.StartPublishing(timeout: 5000, fullQueue: false); m_mockContext.Verify( c => c.PublishAsync( It.IsAny(), It.IsAny>(), It.IsAny()), Times.Never); } [Test] public void NotifySubscriptionsChangedTriggersPublishReEvaluation() { m_mockContext.Setup(c => c.Subscriptions) .Returns([]); m_mockContext.Setup(c => c.Disposed) .Returns(false); using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); // With no subscriptions, NotifySubscriptionsChanged // should not trigger a publish request. engine.NotifySubscriptionsChanged(); m_mockContext.Verify( c => c.PublishAsync( It.IsAny(), It.IsAny>(), It.IsAny()), Times.Never); } [Test] public void MinPublishRequestCountDefaultsCorrectly() { using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); Assert.That( engine.MinPublishRequestCount, Is.EqualTo(1)); Assert.That( engine.MaxPublishRequestCount, Is.EqualTo(ushort.MaxValue)); } [Test] public void MinMaxPublishRequestCountCanBeSet() { using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); engine.MinPublishRequestCount = 5; engine.MaxPublishRequestCount = 50; Assert.That( engine.MinPublishRequestCount, Is.EqualTo(5)); Assert.That( engine.MaxPublishRequestCount, Is.EqualTo(50)); } [Test] public void MinPublishRequestCountRejectsInvalidValues() { using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); Assert.That( () => engine.MinPublishRequestCount = 0, Throws.TypeOf()); Assert.That( () => engine.MinPublishRequestCount = 101, Throws.TypeOf()); } [Test] public void MaxPublishRequestCountRejectsInvalidValues() { using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); Assert.That( () => engine.MaxPublishRequestCount = 0, Throws.TypeOf()); } [Test] public void PauseAndResumePublishing() { using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); Assert.That(engine.PausePublishing, Throws.Nothing); Assert.That(engine.ResumePublishing, Throws.Nothing); } [Test] public async Task StopPublishingAsyncDisposesCleanlyAsync() { using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); await engine.StopPublishingAsync().ConfigureAwait(false); } [Test] public void DisposeIsIdempotent() { #pragma warning disable CA2000 // Dispose objects before losing scope var engine = new ClassicSubscriptionEngine(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 ConstructorThrowsOnNullContext() { Assert.That( () => new ClassicSubscriptionEngine(null), Throws.TypeOf()); } [Test] public void GoodPublishRequestCountDelegatesToContext() { m_mockContext.Setup(c => c.GoodPublishRequestCount) .Returns(42); using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); Assert.That( engine.GoodPublishRequestCount, Is.EqualTo(42)); } [Test] public void BadPublishRequestCountIsZero() { using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); Assert.That( engine.BadPublishRequestCount, Is.Zero); } [Test] public void BeginPublishSkipsWhenSessionKeepAliveStopped() { // Arrange — session looks healthy except KeepAliveStopped is true. m_mockContext.Setup(c => c.Connected).Returns(true); m_mockContext.Setup(c => c.Reconnecting).Returns(false); m_mockContext.Setup(c => c.Closing).Returns(false); m_mockContext.Setup(c => c.KeepAliveStopped).Returns(true); m_mockContext.Setup(c => c.LastKeepAliveTime) .Returns(DateTime.UtcNow.AddSeconds(-30)); using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); // Act bool result = engine.BeginPublish(timeout: 5000); // Assert — guard short-circuited; no PublishAsync was invoked. Assert.That(result, Is.False); m_mockContext.Verify( c => c.PublishAsync( It.IsAny(), It.IsAny>(), It.IsAny()), Times.Never); } [Test] public void MaxPublishRequestCountRespectsMinimum() { using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); engine.MinPublishRequestCount = 10; engine.MaxPublishRequestCount = 5; // MaxPublishRequestCount getter returns the larger of // min and max to ensure min is always respected. Assert.That( engine.MaxPublishRequestCount, Is.GreaterThanOrEqualTo( engine.MinPublishRequestCount)); } /// /// Regression coverage for the V1 ack-pruning helper added by /// #3540. /// drops every queued acknowledgement targeting the given /// subscription id while leaving acks for other subscriptions /// untouched. The recovery path relies on this before recreating /// a subscription so stale acks for the dead id cannot produce /// BadSubscriptionIdInvalid responses on servers that /// re-use subscription identifiers (e.g. Kepware always /// starting at 1). /// [Test] public void RemoveAcknowledgementsForSubscriptionDropsOnlyMatchingAcks() { using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); engine.AddPendingAcknowledgement(1u, 10u); engine.AddPendingAcknowledgement(2u, 11u); engine.AddPendingAcknowledgement(1u, 12u); engine.AddPendingAcknowledgement(1u, 13u); engine.AddPendingAcknowledgement(3u, 14u); int dropped = engine.RemoveAcknowledgementsForSubscription(1u); Assert.That(dropped, Is.EqualTo(3), "All queued acks for the dead subscription id must be dropped."); int droppedAgain = engine.RemoveAcknowledgementsForSubscription(1u); Assert.That(droppedAgain, Is.Zero, "Second call must drop nothing (idempotent)."); int droppedSecond = engine.RemoveAcknowledgementsForSubscription(2u); Assert.That(droppedSecond, Is.EqualTo(1), "Survivors must still be in the queue and drainable per-id."); int droppedThird = engine.RemoveAcknowledgementsForSubscription(3u); Assert.That(droppedThird, Is.EqualTo(1)); } /// /// Calling /// /// on an empty pending-ack list is a no-op and must not throw. /// [Test] public void RemoveAcknowledgementsForSubscriptionOnEmptyQueueReturnsZero() { using var engine = new ClassicSubscriptionEngine(m_mockContext.Object); int dropped = engine.RemoveAcknowledgementsForSubscription(42u); Assert.That(dropped, Is.Zero); } } }