/* ========================================================================
* Copyright (c) 2005-2026 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 Microsoft.Extensions.Options;
using Moq;
using NUnit.Framework;
using Opc.Ua.PubSub.Adapter.DependencyInjection;
using Opc.Ua.PubSub.Adapter.Diagnostics;
using Opc.Ua.PubSub.Adapter.Publisher;
using Opc.Ua.PubSub.Adapter.Session;
using Opc.Ua.PubSub.Application;
using Opc.Ua.PubSub.Configuration;
using Opc.Ua.PubSub.DataSets;
namespace Opc.Ua.PubSub.Adapter.Tests
{
///
/// Unit tests for the dependency-injection runtime types:
/// ,
/// and
/// .
///
[TestFixture]
public sealed class ServerAdapterRuntimeTests
{
private static SubscriptionCoordinator CreateCoordinator(
List created)
{
PublishedDataSetDataType pds = AdapterTestHelpers.PublishedDataSet(
"PDS", AdapterTestHelpers.Variable.Value(new NodeId(11u)));
PubSubConfigurationDataType config = AdapterTestHelpers.Configuration(
500, [pds]);
Mock session = AdapterTestHelpers.ConnectedSession();
session
.Setup(s => s.CreateDataChangeSubscriptionAsync(
It.IsAny(), It.IsAny()))
.Returns((double interval, CancellationToken ct) =>
{
var sub = new FakeDataChangeSubscription();
created.Add(sub);
return new ValueTask(sub);
});
session
.Setup(s => s.ReadAsync(
It.IsAny>(), It.IsAny()))
.Returns((ArrayOf reads, CancellationToken ct) =>
{
var values = new DataValue[reads.Count];
for (int i = 0; i < reads.Count; i++)
{
values[i] = new DataValue(new Variant(i));
}
return new ValueTask>(values.ToArrayOf());
});
return new SubscriptionCoordinator(
config,
session.Object,
SubscriptionAffinity.WriterGroup,
AdapterTestHelpers.Telemetry());
}
private static ServerAdapterReloadCoordinator CreateReloadCoordinator(
ServerAdapterRuntime runtime)
{
return new ServerAdapterReloadCoordinator(
new FakeConfigurationStore(new PubSubConfigurationDataType()),
new OptionsMonitorStub(new ServerPublisherOptions()),
new OptionsMonitorStub(new ServerSubscriberOptions()),
new OptionsMonitorStub(
new ServerActionResponderOptions()),
new MutableDataSetSourceProvider(),
new MutableDataSetSinkProvider(),
runtime,
AdapterTestHelpers.Telemetry(),
new AdapterMetrics());
}
[Test]
public void FactoryCreateNullOptionsThrows()
{
var factory = new ServerSessionFactory();
Assert.That(
() => factory.Create(null!, AdapterTestHelpers.Telemetry()),
Throws.ArgumentNullException.With.Property("ParamName").EqualTo("options"));
}
[Test]
public void FactoryCreateNullTelemetryThrows()
{
var factory = new ServerSessionFactory();
Assert.That(
() => factory.Create(
new ServerConnectionOptions { EndpointUrl = "opc.tcp://host:4840" },
null!),
Throws.ArgumentNullException.With.Property("ParamName").EqualTo("telemetry"));
}
[Test]
public async Task FactoryCreateReturnsExternalServerSessionAsync()
{
var factory = new ServerSessionFactory();
IServerSession session = factory.Create(
new ServerConnectionOptions { EndpointUrl = "opc.tcp://host:4840" },
AdapterTestHelpers.Telemetry());
Assert.That(session, Is.InstanceOf());
Assert.That(session.IsConnected, Is.False);
await session.DisposeAsync().ConfigureAwait(false);
}
[Test]
public void RuntimeAddSessionNullThrows()
{
var runtime = new ServerAdapterRuntime();
Assert.That(
() => runtime.AddSession(null!),
Throws.ArgumentNullException.With.Property("ParamName").EqualTo("session"));
}
[Test]
public void RuntimeAddCoordinatorNullThrows()
{
var runtime = new ServerAdapterRuntime();
Assert.That(
() => runtime.AddCoordinator(null!),
Throws.ArgumentNullException.With.Property("ParamName").EqualTo("coordinator"));
}
[Test]
public async Task RuntimeStartStartsRegisteredCoordinatorsAsync()
{
var created = new List();
SubscriptionCoordinator coordinator = CreateCoordinator(created);
var runtime = new ServerAdapterRuntime();
runtime.AddCoordinator(coordinator);
await runtime.StartAsync().ConfigureAwait(false);
Assert.That(created, Has.Count.EqualTo(1));
await runtime.DisposeAsync().ConfigureAwait(false);
}
[Test]
public async Task RuntimeStartIsIdempotentAsync()
{
var created = new List();
SubscriptionCoordinator coordinator = CreateCoordinator(created);
var runtime = new ServerAdapterRuntime();
runtime.AddCoordinator(coordinator);
await runtime.StartAsync().ConfigureAwait(false);
await runtime.StartAsync().ConfigureAwait(false);
Assert.That(created, Has.Count.EqualTo(1));
await runtime.DisposeAsync().ConfigureAwait(false);
}
[Test]
public async Task RuntimeDisposeDisposesSessionsAsync()
{
var session = new Mock();
session.Setup(s => s.DisposeAsync()).Returns(default(ValueTask));
var runtime = new ServerAdapterRuntime();
runtime.AddSession(session.Object);
await runtime.DisposeAsync().ConfigureAwait(false);
session.Verify(s => s.DisposeAsync(), Times.Once);
}
[Test]
public async Task RuntimeAcquireSessionReusesEqualConnectionOptionsAsync()
{
var session = new Mock();
session.Setup(s => s.DisposeAsync()).Returns(default(ValueTask));
var factory = new Mock();
factory
.Setup(f => f.Create(
It.IsAny(), It.IsAny()))
.Returns(session.Object);
var runtime = new ServerAdapterRuntime(factory.Object);
await using ServerAdapterRuntime.ServerSessionLease first = runtime.AcquireSession(
new ServerConnectionOptions { EndpointUrl = "opc.tcp://host:4840" },
AdapterTestHelpers.Telemetry());
await using ServerAdapterRuntime.ServerSessionLease second = runtime.AcquireSession(
new ServerConnectionOptions { EndpointUrl = "opc.tcp://host:4840" },
AdapterTestHelpers.Telemetry());
Assert.That(second.Session, Is.SameAs(first.Session));
factory.Verify(f => f.Create(
It.IsAny(), It.IsAny()), Times.Once);
await runtime.DisposeAsync().ConfigureAwait(false);
}
[Test]
public async Task RuntimeAcquireSessionCreatesDistinctSessionsForDifferentEndpointsAsync()
{
var sessions = new Queue(
[new Mock().Object, new Mock().Object]);
var factory = new Mock();
factory
.Setup(f => f.Create(
It.IsAny(), It.IsAny()))
.Returns(() => sessions.Dequeue());
var runtime = new ServerAdapterRuntime(factory.Object);
await using ServerAdapterRuntime.ServerSessionLease first = runtime.AcquireSession(
new ServerConnectionOptions { EndpointUrl = "opc.tcp://one:4840" },
AdapterTestHelpers.Telemetry());
await using ServerAdapterRuntime.ServerSessionLease second = runtime.AcquireSession(
new ServerConnectionOptions { EndpointUrl = "opc.tcp://two:4840" },
AdapterTestHelpers.Telemetry());
Assert.That(second.Session, Is.Not.SameAs(first.Session));
factory.Verify(f => f.Create(
It.IsAny(), It.IsAny()), Times.Exactly(2));
await runtime.DisposeAsync().ConfigureAwait(false);
}
[Test]
public async Task RuntimeReleaseLastSessionLeaseDisposesSessionAsync()
{
var session = new Mock();
session.Setup(s => s.DisposeAsync()).Returns(default(ValueTask));
var factory = new Mock();
factory
.Setup(f => f.Create(
It.IsAny(), It.IsAny()))
.Returns(session.Object);
var runtime = new ServerAdapterRuntime(factory.Object);
ServerAdapterRuntime.ServerSessionLease first = runtime.AcquireSession(
new ServerConnectionOptions { EndpointUrl = "opc.tcp://host:4840" },
AdapterTestHelpers.Telemetry());
ServerAdapterRuntime.ServerSessionLease second = runtime.AcquireSession(
new ServerConnectionOptions { EndpointUrl = "opc.tcp://host:4840" },
AdapterTestHelpers.Telemetry());
await first.DisposeAsync().ConfigureAwait(false);
session.Verify(s => s.DisposeAsync(), Times.Never);
await second.DisposeAsync().ConfigureAwait(false);
session.Verify(s => s.DisposeAsync(), Times.Once);
await runtime.DisposeAsync().ConfigureAwait(false);
}
[Test]
public async Task RuntimeAddSessionAfterDisposeThrowsAsync()
{
var runtime = new ServerAdapterRuntime();
await runtime.DisposeAsync().ConfigureAwait(false);
var session = new Mock();
Assert.That(
() => runtime.AddSession(session.Object),
Throws.TypeOf());
}
[Test]
public async Task RuntimeDisposeIsIdempotentAsync()
{
var session = new Mock();
session.Setup(s => s.DisposeAsync()).Returns(default(ValueTask));
var runtime = new ServerAdapterRuntime();
runtime.AddSession(session.Object);
await runtime.DisposeAsync().ConfigureAwait(false);
await runtime.DisposeAsync().ConfigureAwait(false);
session.Verify(s => s.DisposeAsync(), Times.Once);
}
[Test]
public void HostedServiceNullApplicationThrows()
{
var runtime = new ServerAdapterRuntime();
Assert.That(
() => new ServerAdapterHostedService(
null!, runtime, null!),
Throws.ArgumentNullException.With.Property("ParamName").EqualTo("application"));
}
[Test]
public void HostedServiceNullRuntimeThrows()
{
IPubSubApplication application = new Mock().Object;
Assert.That(
() => new ServerAdapterHostedService(
application, null!, null!),
Throws.ArgumentNullException.With.Property("ParamName").EqualTo("runtime"));
}
[Test]
public void HostedServiceNullReloadCoordinatorThrows()
{
IPubSubApplication application = new Mock().Object;
var runtime = new ServerAdapterRuntime();
Assert.That(
() => new ServerAdapterHostedService(application, runtime, null!),
Throws.ArgumentNullException.With.Property("ParamName").EqualTo("reloadCoordinator"));
}
[Test]
public async Task HostedServiceStartStartsCoordinatorsAndStopDisposesAsync()
{
var created = new List();
SubscriptionCoordinator coordinator = CreateCoordinator(created);
var session = new Mock();
session.Setup(s => s.DisposeAsync()).Returns(default(ValueTask));
var runtime = new ServerAdapterRuntime();
runtime.AddCoordinator(coordinator);
runtime.AddSession(session.Object);
var hosted = new ServerAdapterHostedService(
new Mock().Object,
runtime,
CreateReloadCoordinator(runtime));
await hosted.StartAsync(CancellationToken.None).ConfigureAwait(false);
Assert.That(created, Has.Count.EqualTo(1));
await hosted.StopAsync(CancellationToken.None).ConfigureAwait(false);
session.Verify(s => s.DisposeAsync(), Times.Once);
}
private sealed class OptionsMonitorStub : IOptionsMonitor
{
public OptionsMonitorStub(T value)
{
CurrentValue = value;
}
public T CurrentValue { get; }
public T Get(string? name)
{
return CurrentValue;
}
public IDisposable? OnChange(Action listener)
{
return null;
}
}
private sealed class FakeConfigurationStore : IPubSubConfigurationStore
{
public FakeConfigurationStore(PubSubConfigurationDataType configuration)
{
m_configuration = configuration;
}
public event EventHandler? Changed;
public ValueTask LoadAsync(
CancellationToken cancellationToken = default)
{
return new ValueTask(m_configuration);
}
public ValueTask SaveAsync(
PubSubConfigurationDataType configuration,
CancellationToken cancellationToken = default)
{
PubSubConfigurationDataType previous = m_configuration;
m_configuration = configuration;
Changed?.Invoke(this, new PubSubConfigurationChangedEventArgs(previous, configuration));
return default;
}
public ValueTask GetConfigurationVersionAsync(
CancellationToken cancellationToken = default)
{
return new ValueTask((ConfigurationVersionDataType?)null);
}
public ValueTask SetConfigurationVersionAsync(
ConfigurationVersionDataType configurationVersion,
CancellationToken cancellationToken = default)
{
return default;
}
public ValueTask GetPublishedDataSetConfigurationVersionAsync(
string publishedDataSetName,
CancellationToken cancellationToken = default)
{
return new ValueTask((ConfigurationVersionDataType?)null);
}
public ValueTask SetPublishedDataSetConfigurationVersionAsync(
string publishedDataSetName,
ConfigurationVersionDataType configurationVersion,
CancellationToken cancellationToken = default)
{
return default;
}
private PubSubConfigurationDataType m_configuration;
}
}
}