/* ======================================================================== * 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.Diagnostics; using System.IO; using System.Threading; using System.Threading.Tasks; using NUnit.Framework; using Opc.Ua.PubSub.Adapter.Actions; using Opc.Ua.PubSub.Adapter.Publisher; using Opc.Ua.PubSub.Adapter.Session; using Opc.Ua.PubSub.Adapter.Subscriber; using Opc.Ua.PubSub.Application; using Opc.Ua.PubSub.DataSets; using Opc.Ua.PubSub.Encoding; using Opc.Ua.Server.TestFramework; using Opc.Ua.Tests; using Quickstarts.ReferenceServer; namespace Opc.Ua.PubSub.Adapter.Tests { /// /// End-to-end integration tests that stand up a real in-process OPC UA /// reference server and exercise the external-server PubSub adapter /// components (, /// , /// and /// ) against it through a live /// . /// /// /// A single unsecured (SecurityMode.None) server + session is shared by all /// tests via [OneTimeSetUp]. If the loopback session cannot be /// established in the current environment the setup records the failure and /// every test calls so the gate is not /// broken. /// [TestFixture] [Category("Integration")] [Category("PubSub")] [SetCulture("en-us")] [SetUICulture("en-us")] public sealed class ExternalServerAdapterIntegrationTests { private ServerFixture m_serverFixture = null!; private ServerSession m_session = null!; private ITelemetryContext m_telemetry = null!; private string m_pkiRoot = null!; private string? m_setupError; private ushort m_namespaceIndex; private static readonly TimeSpan s_timeout = TimeSpan.FromSeconds(10); [OneTimeSetUp] public async Task OneTimeSetUpAsync() { m_telemetry = NUnitTelemetryContext.Create(); m_pkiRoot = Path.Combine(Path.GetTempPath(), Path.GetRandomFileName()); try { m_serverFixture = new ServerFixture( t => new ReferenceServer(t)) { UriScheme = Utils.UriSchemeOpcTcp, SecurityNone = true, AutoAccept = true, AllNodeManagers = false, OperationLimits = true }; await m_serverFixture.LoadConfigurationAsync(m_pkiRoot).ConfigureAwait(false); _ = await m_serverFixture.StartAsync().ConfigureAwait(false); string url = $"{Utils.UriSchemeOpcTcp}://localhost:{m_serverFixture.Port}"; m_session = new ServerSession( new ServerConnectionOptions { EndpointUrl = url, SecurityMode = MessageSecurityMode.None, SessionName = "Adapter.IntegrationTests" }, m_telemetry); using var cts = new CancellationTokenSource(s_timeout); await m_session.ConnectAsync(cts.Token).ConfigureAwait(false); m_namespaceIndex = await ResolveReferenceNamespaceIndexAsync(cts.Token) .ConfigureAwait(false); } catch (Exception ex) { m_setupError = ex.Message; } } [OneTimeTearDown] public async Task OneTimeTearDownAsync() { if (m_session != null) { await m_session.DisposeAsync().ConfigureAwait(false); } if (m_serverFixture != null) { await m_serverFixture.StopAsync().ConfigureAwait(false); } TryDeleteDirectory(m_pkiRoot); } [SetUp] public void SkipWhenServerUnavailable() { if (m_setupError != null) { Assert.Ignore( "External server session could not be established in this " + $"environment: {m_setupError}"); } } [Test] public async Task CyclicReadSourceReturnsLiveServerValueAsync() { NodeId nodeId = ScalarNode("Scalar_Static_Int32"); await WriteInt32Async(nodeId, 4242).ConfigureAwait(false); PublishedDataSetDataType pds = AdapterTestHelpers.PublishedDataSet( "IntegrationPDS", AdapterTestHelpers.Variable.Value(nodeId)); var strategy = new CyclicReadStrategy(m_session, m_telemetry); using var metaDataBuilder = new DataSetMetaDataBuilder( pds, m_session, m_telemetry); var source = new ServerPublishedDataSetSource( pds, strategy, metaDataBuilder, m_telemetry); PublishedDataSetSnapshot snapshot = await source .SampleAsync(metaDataBuilder.BuildMetaData()) .ConfigureAwait(false); DataSetField[] fields = (DataSetField[]?)snapshot.Fields ?? []; Assert.That(fields, Has.Length.EqualTo(1)); Assert.That(StatusCode.IsGood(fields[0].StatusCode), Is.True); Assert.That(fields[0].Value.TryGetValue(out int value), Is.True); Assert.That(value, Is.EqualTo(4242)); } [Test] public async Task TargetVariableWriterRoundTripsThroughServerAsync() { NodeId nodeId = ScalarNode("Scalar_Static_Int32"); var writer = new ServerTargetVariableWriter(m_session, m_telemetry); StatusCode status = await writer .WriteAsync(nodeId, Attributes.Value, null, new DataValue(new Variant(13579))) .ConfigureAwait(false); Assert.That(StatusCode.IsGood(status), Is.True); int readBack = await ReadInt32Async(nodeId).ConfigureAwait(false); Assert.That(readBack, Is.EqualTo(13579)); } [Test] public async Task SubscriptionCoordinatorPrimesAndReflectsChangeAsync() { NodeId nodeId = ScalarNode("Scalar_Static_Int32"); await WriteInt32Async(nodeId, 1000).ConfigureAwait(false); PublishedDataSetDataType pds = AdapterTestHelpers.PublishedDataSet( "SubPDS", AdapterTestHelpers.Variable.Value(nodeId)); PubSubConfigurationDataType config = AdapterTestHelpers.Configuration( 200, [pds]); await using var coordinator = new SubscriptionCoordinator( config, m_session, SubscriptionAffinity.WriterGroup, m_telemetry); await coordinator.StartAsync().ConfigureAwait(false); IReadStrategy strategy = coordinator.GetReadStrategy("SubPDS"); ReadValueId[] reads = [ new ReadValueId { NodeId = nodeId, AttributeId = Attributes.Value } ]; ArrayOf primed = await strategy .ReadAsync(reads.ToArrayOf()) .ConfigureAwait(false); Assert.That(primed[0].WrappedValue.TryGetValue(out int primedValue), Is.True); Assert.That(primedValue, Is.EqualTo(1000)); await WriteInt32Async(nodeId, 2000).ConfigureAwait(false); int observed = await WaitForCachedValueAsync(strategy, reads.ToArrayOf(), 2000) .ConfigureAwait(false); Assert.That(observed, Is.EqualTo(2000)); } [Test] public async Task ActionHandlerCallsServerMethodAndReturnsOutputAsync() { const ushort writerId = 7; const ushort targetId = 3; NodeId objectId = ScalarNode("Methods"); NodeId methodId = ScalarNode("Methods_Add"); string[] outputNames = ["Sum"]; ActionMethodMap map = new ActionMethodMap() .Add(writerId, targetId, objectId, methodId, outputNames.ToArrayOf()); var handler = new ServerActionHandler(m_session, map, m_telemetry); PubSubActionHandlerResult result = await handler .HandleAsync(new PubSubActionInvocation { Target = new PubSubActionTarget { DataSetWriterId = writerId, ActionTargetId = targetId }, InputFields = new[] { new DataSetField { Name = "a", Value = new Variant(1.5f) }, new DataSetField { Name = "b", Value = new Variant((uint)2) } }.ToArrayOf() }) .ConfigureAwait(false); Assert.That(StatusCode.IsGood(result.StatusCode), Is.True); Assert.That(result.OutputFields.Count, Is.EqualTo(1)); Assert.That(result.OutputFields[0].Name, Is.EqualTo("Sum")); Assert.That(result.OutputFields[0].Value.TryGetValue(out float sum), Is.True); Assert.That(sum, Is.EqualTo(3.5f).Within(0.001f)); } private async Task WaitForCachedValueAsync( IReadStrategy strategy, ArrayOf reads, int expected) { var stopwatch = Stopwatch.StartNew(); int last = int.MinValue; while (stopwatch.Elapsed < s_timeout) { ArrayOf values = await strategy.ReadAsync(reads).ConfigureAwait(false); if (values[0].WrappedValue.TryGetValue(out int value)) { last = value; if (value == expected) { return value; } } await Task.Delay(100).ConfigureAwait(false); } return last; } private NodeId ScalarNode(string identifier) { return new NodeId(identifier, m_namespaceIndex); } private async Task ResolveReferenceNamespaceIndexAsync(CancellationToken ct) { ReadValueId[] reads = [ new ReadValueId { NodeId = VariableIds.Server_NamespaceArray, AttributeId = Attributes.Value } ]; ArrayOf results = await m_session .ReadAsync(reads.ToArrayOf(), ct) .ConfigureAwait(false); if (!results[0].WrappedValue.TryGetValue(out ArrayOf namespaces) || namespaces.IsNull || namespaces.Count == 0) { throw new InvalidOperationException("Server namespace array is empty."); } string[] namespaceUris = namespaces.ToArray()!; var table = new NamespaceTable(namespaceUris); int index = table.GetIndex(Quickstarts.ReferenceServer.Namespaces.ReferenceServer); if (index < 0) { throw new InvalidOperationException( "Reference server namespace not advertised by the server."); } return (ushort)index; } private async Task WriteInt32Async(NodeId nodeId, int value) { WriteValue[] writes = [ new WriteValue { NodeId = nodeId, AttributeId = Attributes.Value, Value = new DataValue(new Variant(value)) } ]; ArrayOf results = await m_session .WriteAsync(writes.ToArrayOf()) .ConfigureAwait(false); Assert.That(StatusCode.IsGood(results[0]), Is.True, "Writing the test value to the server should succeed."); } private async Task ReadInt32Async(NodeId nodeId) { ReadValueId[] reads = [ new ReadValueId { NodeId = nodeId, AttributeId = Attributes.Value } ]; ArrayOf results = await m_session .ReadAsync(reads.ToArrayOf()) .ConfigureAwait(false); Assert.That(results[0].WrappedValue.TryGetValue(out int value), Is.True); return value; } private static void TryDeleteDirectory(string path) { try { if (!string.IsNullOrEmpty(path) && Directory.Exists(path)) { Directory.Delete(path, recursive: true); } } catch { } } } }