/* ========================================================================
* 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.Tasks;
using Microsoft.Extensions.Options;
using Moq;
using NUnit.Framework;
using Opc.Ua.Client.Subscriptions;
using Opc.Ua.Client.Subscriptions.MonitoredItems;
using Opc.Ua.Client.Subscriptions.Streaming;
// Tests run on the default TaskScheduler so CA2007's sync-context risk does not apply.
#pragma warning disable CA2007
namespace Opc.Ua.Client.Tests.Streaming
{
///
/// Deterministic null-guard tests for .
///
///
/// Full lazy EnsureSubscriptionAsync / monitored-item routing /
/// dispose coverage requires a real ISubscriptionManager +
/// ISubscription pipeline and lives in the integration test
/// suite. Only the cheap argument-validation guards are exercised
/// here.
///
[TestFixture]
[Category("Client")]
[Category("StreamingExtensions")]
[Parallelizable]
public sealed class StreamingSubscriptionTests
{
[Test]
public void ConstructorWithNullSubscriptionManagerThrowsArgumentNullException()
{
Assert.That(
() => new StreamingSubscription(null!),
Throws.ArgumentNullException);
}
[Test]
public async Task SubscribeDataChangesAsyncWithNullNodeIdThrowsArgumentNullException()
{
await using var sub = new StreamingSubscription(NewStubManager());
// NodeId.Null is the canonical "null" value of the INullable
// struct. The production guard checks nodeId.IsNull and throws
// synchronously before any subscription-manager interaction.
Assert.That(
() => sub.SubscribeDataChangesAsync(NodeId.Null),
Throws.ArgumentNullException);
}
[Test]
public async Task SubscribeDataChangesAsyncWithNullNodeIdsListThrowsArgumentNullException()
{
await using var sub = new StreamingSubscription(NewStubManager());
Assert.That(
() => sub.SubscribeDataChangesAsync(null!),
Throws.ArgumentNullException);
}
[Test]
public async Task SubscribeEventsAsyncWithNullNotifierIdThrowsArgumentNullException()
{
await using var sub = new StreamingSubscription(NewStubManager());
Assert.That(
() => sub.SubscribeEventsAsync(NodeId.Null, new EventFilter()),
Throws.ArgumentNullException);
}
[Test]
public async Task SubscribeEventsAsyncWithNullFilterThrowsArgumentNullException()
{
await using var sub = new StreamingSubscription(NewStubManager());
Assert.That(
() => sub.SubscribeEventsAsync(ObjectIds.Server, null!),
Throws.ArgumentNullException);
}
[Test]
public async Task SubscribeEventsAsyncWhenMonitoredItemCannotBeAddedThrowsAsync()
{
IMonitoredItem? monitoredItem = null;
var monitoredItems = new Mock();
monitoredItems
.Setup(items => items.TryAdd(
It.IsAny(),
It.IsAny>(),
out monitoredItem))
.Returns(false);
var subscription = new Mock();
subscription.SetupGet(value => value.MonitoredItems).Returns(monitoredItems.Object);
var manager = new Mock();
IOptionsMonitor? capturedOptions =
null;
manager
.Setup(value => value.Add(
It.IsAny(),
It.IsAny<
IOptionsMonitor>()))
.Callback<
ISubscriptionNotificationHandler,
IOptionsMonitor>(
(_, options) => capturedOptions = options)
.Returns(subscription.Object);
await using var streaming = new StreamingSubscription(
manager.Object,
new Subscriptions.SubscriptionOptions
{
PublishingInterval = TimeSpan.FromSeconds(2)
});
await using IAsyncEnumerator enumerator = streaming
.SubscribeEventsAsync(ObjectIds.Server, new EventFilter())
.GetAsyncEnumerator();
InvalidOperationException? exception = null;
try
{
await enumerator.MoveNextAsync().ConfigureAwait(false);
}
catch (InvalidOperationException caught)
{
exception = caught;
}
Assert.That(exception, Is.Not.Null);
Assert.That(
capturedOptions?.CurrentValue.PublishingEnabled,
Is.True);
}
///
/// DisposeAsync racing the first subscriber (which is still creating
/// the shared subscription) must dispose the subscription that
/// subscriber creates and fail the subscriber with
/// , instead of leaking it.
///
[Test]
public async Task DisposeRacingFirstSubscribeDisposesCreatedSubscriptionAsync()
{
var subscription = new Mock();
var manager = new Mock();
StreamingSubscription? streaming = null;
Task? disposal = null;
manager
.Setup(value => value.Add(
It.IsAny(),
It.IsAny>()))
.Callback(() => disposal = streaming!.DisposeAsync().AsTask())
.Returns(subscription.Object);
streaming = new StreamingSubscription(manager.Object);
await using (streaming)
{
await using IAsyncEnumerator enumerator = streaming
.SubscribeEventsAsync(ObjectIds.Server, new EventFilter())
.GetAsyncEnumerator();
Assert.That(
async () => await enumerator.MoveNextAsync().ConfigureAwait(false),
Throws.InstanceOf());
Assert.That(disposal, Is.Not.Null);
await disposal!.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
subscription.Verify(value => value.DisposeAsync(), Times.Once);
}
}
///
/// Stub subscription manager — its methods are never invoked
/// because the argument guards trip synchronously before
/// EnsureSubscriptionAsync is entered. Using a loose
/// avoids hand-implementing every
/// member of the surface.
///
private static ISubscriptionManager NewStubManager()
{
return new Mock(MockBehavior.Loose).Object;
}
}
}