/* ======================================================================== * 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.IO; using System.Linq; using NUnit.Framework; using Opc.Ua.Tests; using Quickstarts.Servers; namespace Opc.Ua.Server.Tests { /// /// Round-trip tests for the BinaryEncoder/BinaryDecoder based /// serialization in and /// . /// Tests call the internal encode/decode methods directly. /// Persistor cleanup clears the process-wide batch directory. /// [TestFixture] [Category("Server")] [SetCulture("en-us")] [SetUICulture("en-us")] [NonParallelizable] public class DurableSubscriptionSerializationTests { private ServiceMessageContext m_context; /// /// Creates the serialization message context and registers the two fixture namespaces. /// [OneTimeSetUp] public void OneTimeSetUp() { ITelemetryContext telemetry = NUnitTelemetryContext.Create(); m_context = ServiceMessageContext.Create(telemetry); m_context.NamespaceUris.GetIndexOrAppend( "urn:test:namespace1"); m_context.NamespaceUris.GetIndexOrAppend( "urn:test:namespace2"); } /// /// Verifies that an empty durable subscription preserves its settings and empty item and message collections. /// [Test] public void RoundTripEmptySubscription() { StoredSubscription result = RoundTripSubscription(CreateMinimalSubscription(id: 42)); Assert.That(result.Id, Is.EqualTo(42u)); Assert.That(result.IsDurable, Is.True); Assert.That(result.PublishingInterval, Is.EqualTo(1000.0)); Assert.That(result.Priority, Is.EqualTo((byte)5)); Assert.That(result.MonitoredItems.Count(), Is.Zero); Assert.That(result.SentMessages, Has.Count.EqualTo(0)); } /// /// Verifies that the current sample format preserves publishing state and anonymous owner application URI. /// [TestCase(false)] [TestCase(true)] public void RoundTripSubscriptionState(bool publishingEnabled) { StoredSubscription original = CreateMinimalSubscription(id: 43); original.PublishingEnabled = publishingEnabled; original.OwnerClientApplicationUri = "urn:test:client"; StoredSubscription result = RoundTripSubscription(original); Assert.Multiple(() => { Assert.That(result.PublishingEnabled, Is.EqualTo(publishingEnabled)); Assert.That(result.OwnerClientApplicationUri, Is.EqualTo("urn:test:client")); }); } /// /// Verifies that the current sample format preserves the triggering links between /// monitored items (OPC 10000-4 ยง5.13.1.6). /// [Test] public void RoundTripSubscriptionTriggeringLinks() { StoredSubscription original = CreateMinimalSubscription(id: 45); original.TriggeringLinks = new Dictionary> { [1] = [2u, 3u], [4] = [5u] }; StoredSubscription result = RoundTripSubscription(original); Assert.That(result.TriggeringLinks, Is.Not.Null); Assert.Multiple(() => { Assert.That(result.TriggeringLinks.Keys, Is.EquivalentTo(new uint[] { 1, 4 })); Assert.That(result.TriggeringLinks[1], Is.EqualTo(new uint[] { 2, 3 })); Assert.That(result.TriggeringLinks[4], Is.EqualTo(new uint[] { 5 })); }); } /// /// Verifies that version-one sample records migrate to the safe default publishing state. /// [Test] public void DecodeVersionOneSubscriptionDefaultsState() { using var encoder = new BinaryEncoder(m_context); StoredSubscription original = CreateMinimalSubscription(id: 44); WriteLegacySubscription(encoder, original); using var decoder = new BinaryDecoder(encoder.CloseAndReturnBuffer()!, m_context); StoredSubscription result = SubscriptionStore.DecodeSubscription(decoder, version: 1); Assert.Multiple(() => { Assert.That(result.PublishingEnabled, Is.True); Assert.That(result.OwnerClientApplicationUri, Is.Null); }); } /// /// Verifies that durable subscription serialization preserves its monitored items and their identifiers. /// [Test] public void RoundTripSubscriptionWithMonitoredItems() { StoredSubscription original = CreateMinimalSubscription(id: 100); original.MonitoredItems = [ CreateMonitoredItem(id: 1, subscriptionId: 100), CreateMonitoredItem(id: 2, subscriptionId: 100), CreateMonitoredItem(id: 3, subscriptionId: 100) ]; StoredSubscription result = RoundTripSubscription(original); var items = result.MonitoredItems.ToList(); Assert.That(items, Has.Count.EqualTo(3)); Assert.That(items[0].Id, Is.EqualTo(1u)); Assert.That(items[1].Id, Is.EqualTo(2u)); Assert.That(items[2].Id, Is.EqualTo(3u)); Assert.That(items[0].SubscriptionId, Is.EqualTo(100u)); } /// /// Verifies that durable subscription serialization preserves monitored-item properties. /// [Test] public void RoundTripMonitoredItemProperties() { StoredSubscription original = CreateMinimalSubscription(id: 1); StoredMonitoredItem mi = CreateMonitoredItem(id: 7, subscriptionId: 1); mi.NodeId = new NodeId("TestNode", 2); mi.AttributeId = Attributes.Value; mi.QueueSize = 10; mi.SamplingInterval = 500.0; mi.DiscardOldest = true; mi.MonitoringMode = MonitoringMode.Reporting; mi.IsDurable = true; mi.LastValue = new DataValue( new Variant(42), StatusCodes.Good, DateTime.UtcNow); original.MonitoredItems = [mi]; var restored = RoundTripSubscription(original) .MonitoredItems.First() as StoredMonitoredItem; Assert.That(restored, Is.Not.Null); Assert.That(restored.Id, Is.EqualTo(7u)); Assert.That(restored.NodeId, Is.EqualTo(new NodeId("TestNode", 2))); Assert.That(restored.AttributeId, Is.EqualTo(Attributes.Value)); Assert.That(restored.QueueSize, Is.EqualTo(10u)); Assert.That(restored.SamplingInterval, Is.EqualTo(500.0)); Assert.That(restored.DiscardOldest, Is.True); Assert.That(restored.MonitoringMode, Is.EqualTo(MonitoringMode.Reporting)); Assert.That(restored.IsDurable, Is.True); Assert.That(restored.LastValue.IsNull, Is.False); Assert.That((int)restored.LastValue.WrappedValue, Is.EqualTo(42)); Assert.That(restored.LastValue, Is.EqualTo(mi.LastValue)); } /// /// Verifies that the retired interim notification-state format is rejected before decoding a subscription. /// [Test] public void DecodeRejectsInterimVersionTwoSubscriptions() { StoredMonitoredItem first = CreateMonitoredItem(id: 7, subscriptionId: 1); first.LastValue = new DataValue(Variant.From(42), StatusCodes.Good); StoredMonitoredItem second = CreateMonitoredItem(id: 8, subscriptionId: 2); second.LastValue = new DataValue(Variant.From(43), StatusCodes.Good); using var encoder = new BinaryEncoder(m_context); foreach (StoredMonitoredItem item in new[] { first, second }) { StoredSubscription subscription = CreateMinimalSubscription(item.SubscriptionId); subscription.MonitoredItems = [item]; SubscriptionStore.EncodeSubscription(encoder, subscription); encoder.WriteBoolean(null, true); encoder.WriteDataValue(null, new DataValue(Variant.Null, StatusCodes.BadCommunicationError)); encoder.WriteStatusCode(null, StatusCodes.BadCommunicationError); } byte[] bytes = encoder.CloseAndReturnBuffer()!; using var decoder = new BinaryDecoder(bytes!, m_context); Assert.That( () => SubscriptionStore.DecodeSubscription(decoder, version: 2), Throws.TypeOf() .With.Message.Contains("Unsupported durable subscription store version 2")); Assert.That(decoder.Position, Is.Zero); } /// /// Verifies that new subscription records contain raw monitored-item state rather than transient notifications. /// [Test] public void NewSubscriptionRecordContainsOnlyRawMonitoredItemState() { StoredSubscription subscription = CreateMinimalSubscription(id: 1); StoredMonitoredItem item = CreateMonitoredItem(id: 7, subscriptionId: 1); item.LastValue = new DataValue(Variant.From(42), StatusCodes.Good); subscription.MonitoredItems = [item]; using var encoder = new BinaryEncoder(m_context); SubscriptionStore.EncodeSubscription(encoder, subscription); byte[] bytes = encoder.CloseAndReturnBuffer()!; using var decoder = new BinaryDecoder(bytes!, m_context); IStoredMonitoredItem restored = SubscriptionStore.DecodeSubscription(decoder, version: 4) .MonitoredItems.Single(); Assert.That(restored.Id, Is.EqualTo(item.Id)); Assert.That(restored.LastValue, Is.EqualTo(item.LastValue)); Assert.That(decoder.Position, Is.EqualTo(bytes.Length)); } /// /// Verifies that durable subscription serialization preserves the user identity token. /// [Test] public void RoundTripSubscriptionWithUserIdentityToken() { StoredSubscription original = CreateMinimalSubscription(id: 55); original.UserIdentityToken = new UserNameIdentityToken { PolicyId = "username_policy", UserName = "testuser" }; StoredSubscription result = RoundTripSubscription(original); Assert.That(result.UserIdentityToken, Is.Not.Null); Assert.That(result.UserIdentityToken, Is.InstanceOf()); var token = (UserNameIdentityToken)result.UserIdentityToken; Assert.That(token.UserName, Is.EqualTo("testuser")); Assert.That(token.PolicyId, Is.EqualTo("username_policy")); } /// /// Verifies that multiple durable subscriptions round-trip independently. /// [Test] public void RoundTripMultipleSubscriptions() { var subs = new List { CreateMinimalSubscription(id: 1), CreateMinimalSubscription(id: 2), CreateMinimalSubscription(id: 3) }; subs[0].MonitoredItems = [ CreateMonitoredItem(id: 10, subscriptionId: 1), CreateMonitoredItem(id: 11, subscriptionId: 1) ]; subs[1].MonitoredItems = [ CreateMonitoredItem(id: 20, subscriptionId: 2) ]; List results = RoundTripSubscriptions(subs); Assert.That(results, Has.Count.EqualTo(3)); Assert.That(results[0].Id, Is.EqualTo(1u)); Assert.That(results[1].Id, Is.EqualTo(2u)); Assert.That(results[2].Id, Is.EqualTo(3u)); Assert.That(results[0].MonitoredItems.Count(), Is.EqualTo(2)); Assert.That(results[1].MonitoredItems.Count(), Is.EqualTo(1)); Assert.That(results[2].MonitoredItems.Count(), Is.Zero); } /// /// Verifies that a storable data-change queue round-trips through serialization. /// [Test] public void RoundTripStorableDataChangeQueue() { StorableDataChangeQueue result = RoundTripDataChangeQueue( new StorableDataChangeQueue { IsDurable = true, MonitoredItemId = 42, ItemsInQueue = 5, QueueSize = 100 }); Assert.That(result.IsDurable, Is.True); Assert.That(result.MonitoredItemId, Is.EqualTo(42u)); Assert.That(result.ItemsInQueue, Is.EqualTo(5)); Assert.That(result.QueueSize, Is.EqualTo(100u)); Assert.That(result.EnqueueBatch, Is.Null); Assert.That(result.DequeueBatch, Is.Null); } /// /// Verifies that a storable data-change queue preserves its batches during serialization. /// [Test] public void RoundTripStorableDataChangeQueueWithBatches() { var values = new List<(DataValue, ServiceResult?)> { (new DataValue(new Variant(1), StatusCodes.Good), null), (new DataValue(new Variant("hello"), StatusCodes.Good), new ServiceResult(StatusCodes.BadTimeout)) }; var batch = new DataChangeBatch(values, 10, 42); StorableDataChangeQueue result = RoundTripDataChangeQueue( new StorableDataChangeQueue { IsDurable = true, MonitoredItemId = 42, ItemsInQueue = 2, QueueSize = 100, EnqueueBatch = batch, DataChangeBatches = [batch] }); Assert.That(result.EnqueueBatch, Is.Not.Null); Assert.That(result.EnqueueBatch.Values, Has.Count.EqualTo(2)); Assert.That( (int)result.EnqueueBatch.Values[0].Item1.WrappedValue, Is.EqualTo(1)); Assert.That(result.EnqueueBatch.Values[1].Item2, Is.Not.Null); Assert.That(result.DataChangeBatches, Has.Count.EqualTo(1)); } /// /// Verifies that a storable event queue round-trips through serialization. /// [Test] public void RoundTripStorableEventQueue() { StorableEventQueue result = RoundTripEventQueue( new StorableEventQueue { IsDurable = true, MonitoredItemId = 99, QueueSize = 50 }); Assert.That(result.IsDurable, Is.True); Assert.That(result.MonitoredItemId, Is.EqualTo(99u)); Assert.That(result.QueueSize, Is.EqualTo(50u)); Assert.That(result.EnqueueBatch, Is.Null); Assert.That(result.DequeueBatch, Is.Null); } private static StoredSubscription CreateMinimalSubscription( uint id) { return new StoredSubscription { Id = id, IsDurable = true, LifetimeCounter = 10, MaxLifetimeCount = 100, MaxKeepaliveCount = 10, MaxMessageCount = 5, MaxNotificationsPerPublish = 1000, PublishingInterval = 1000.0, Priority = 5, LastSentMessage = 0, SequenceNumber = 1, SentMessages = [], MonitoredItems = [] }; } private static StoredMonitoredItem CreateMonitoredItem( uint id, uint subscriptionId) { return new StoredMonitoredItem { Id = id, SubscriptionId = subscriptionId, NodeId = new NodeId(id, 0), AttributeId = Attributes.Value, QueueSize = 1, SamplingInterval = 1000.0, MonitoringMode = MonitoringMode.Reporting, TimestampsToReturn = TimestampsToReturn.Both, DiagnosticsMasks = DiagnosticsMasks.None }; } private static void WriteLegacySubscription( BinaryEncoder encoder, StoredSubscription subscription) { encoder.WriteUInt32(null, subscription.Id); encoder.WriteBoolean(null, subscription.IsDurable); encoder.WriteUInt32(null, subscription.LifetimeCounter); encoder.WriteUInt32(null, subscription.MaxLifetimeCount); encoder.WriteUInt32(null, subscription.MaxKeepaliveCount); encoder.WriteUInt32(null, subscription.MaxMessageCount); encoder.WriteUInt32(null, subscription.MaxNotificationsPerPublish); encoder.WriteDouble(null, subscription.PublishingInterval); encoder.WriteByte(null, subscription.Priority); encoder.WriteInt32(null, subscription.LastSentMessage); encoder.WriteUInt32(null, subscription.SequenceNumber); encoder.WriteExtensionObject(null, ExtensionObject.Null); encoder.WriteExtensionObjectArray(null, []); encoder.WriteInt32(null, 0); } private StoredSubscription RoundTripSubscription( StoredSubscription original) { return RoundTripSubscriptions([original]).Single(); } /// /// Encodes and decodes a list of subscriptions using /// the internal SubscriptionStore methods. /// private List RoundTripSubscriptions( List originals) { using var stream = new MemoryStream(); using (var encoder = new BinaryEncoder( stream, m_context, true)) { encoder.WriteStringArray( null, m_context.NamespaceUris.ToArrayOf()); encoder.WriteStringArray( null, m_context.ServerUris.ToArrayOf()); encoder.WriteInt32(null, originals.Count); foreach (StoredSubscription sub in originals) { SubscriptionStore.EncodeSubscription( encoder, sub); } } stream.Position = 0; using var decoder = new BinaryDecoder( stream, m_context, true); ArrayOf nsUris = decoder.ReadStringArray(null); ArrayOf srvUris = decoder.ReadStringArray(null); decoder.SetMappingTables( new NamespaceTable(nsUris.Memory.ToArray().Select(uri => uri!)), new StringTable(srvUris.Memory.ToArray().Select(uri => uri!))); int count = decoder.ReadInt32(null); var results = new List(count); for (int i = 0; i < count; i++) { results.Add( SubscriptionStore.DecodeSubscription(decoder)); } return results; } /// /// Encodes and decodes a StorableDataChangeQueue using /// the internal DurableMonitoredItemQueueFactory methods. /// private StorableDataChangeQueue RoundTripDataChangeQueue( StorableDataChangeQueue original) { using var stream = new MemoryStream(); using (var encoder = new BinaryEncoder( stream, m_context, true)) { encoder.WriteStringArray( null, m_context.NamespaceUris.ToArrayOf()); encoder.WriteStringArray( null, m_context.ServerUris.ToArrayOf()); DurableMonitoredItemQueueFactory .EncodeDataChangeQueue(encoder, original); } stream.Position = 0; using var decoder = new BinaryDecoder( stream, m_context, true); ArrayOf nsUris = decoder.ReadStringArray(null); ArrayOf srvUris = decoder.ReadStringArray(null); decoder.SetMappingTables( new NamespaceTable(nsUris.Memory.ToArray().Select(uri => uri!)), new StringTable(srvUris.Memory.ToArray().Select(uri => uri!))); return DurableMonitoredItemQueueFactory .DecodeDataChangeQueue(decoder); } /// /// Encodes and decodes a StorableEventQueue using /// the internal DurableMonitoredItemQueueFactory methods. /// private StorableEventQueue RoundTripEventQueue( StorableEventQueue original) { using var stream = new MemoryStream(); using (var encoder = new BinaryEncoder( stream, m_context, true)) { encoder.WriteStringArray( null, m_context.NamespaceUris.ToArrayOf()); encoder.WriteStringArray( null, m_context.ServerUris.ToArrayOf()); DurableMonitoredItemQueueFactory .EncodeEventQueue(encoder, original); } stream.Position = 0; using var decoder = new BinaryDecoder( stream, m_context, true); ArrayOf nsUris = decoder.ReadStringArray(null); ArrayOf srvUris = decoder.ReadStringArray(null); decoder.SetMappingTables( new NamespaceTable(nsUris.Memory.ToArray().Select(uri => uri!)), new StringTable(srvUris.Memory.ToArray().Select(uri => uri!))); return DurableMonitoredItemQueueFactory .DecodeEventQueue(decoder); } /// /// Verifies that a data-change batch round-trips through the queue persistor. /// [Test] public void RoundTripDataChangeBatchViaPersistor() { using IDisposable scope = AmbientMessageContext.SetScopedContext(m_context); var values = new List<(DataValue, ServiceResult?)> { (new DataValue(new Variant(42), StatusCodes.Good), null), (new DataValue(new Variant("test"), StatusCodes.Good), new ServiceResult(StatusCodes.BadTimeout)) }; var batch = new DataChangeBatch(values, 10, 99); var persistor = new BatchPersistor( NUnitTelemetryContext.Create()); // Persist persistor.PersistSynchronously(batch); Assert.That(batch.IsPersisted, Is.True); Assert.That(batch.Values, Is.Null); // Restore persistor.RestoreSynchronously(batch); Assert.That(batch.IsPersisted, Is.False); Assert.That(batch.Values, Is.Not.Null); Assert.That(batch.Values, Has.Count.EqualTo(2)); Assert.That( (int)batch.Values[0].Item1.WrappedValue, Is.EqualTo(42)); Assert.That(batch.Values[1].Item2, Is.Not.Null); Assert.That( batch.Values[1].Item2.StatusCode, Is.EqualTo(StatusCodes.BadTimeout)); // Clean up persistor.DeleteBatches([]); } /// /// Verifies that an event batch round-trips through the queue persistor. /// [Test] public void RoundTripEventBatchViaPersistor() { using IDisposable scope = AmbientMessageContext.SetScopedContext(m_context); var events = new List { new() { ClientHandle = 1 }, new() { ClientHandle = 2 } }; var batch = new EventBatch(events, 5, 77); var persistor = new BatchPersistor( NUnitTelemetryContext.Create()); // Persist persistor.PersistSynchronously(batch); Assert.That(batch.IsPersisted, Is.True); // Restore persistor.RestoreSynchronously(batch); Assert.That(batch.IsPersisted, Is.False); Assert.That(batch.Events, Is.Not.Null); Assert.That(batch.Events, Has.Count.EqualTo(2)); // Clean up persistor.DeleteBatches([]); } /// /// Verifies that deleting a persisted queue batch removes its backing file. /// [Test] public void PersistorDeleteBatchRemovesFile() { using IDisposable scope = AmbientMessageContext.SetScopedContext(m_context); var values = new List<(DataValue, ServiceResult?)> { (new DataValue(new Variant(1), StatusCodes.Good), null) }; var batch = new DataChangeBatch(values, 5, 50); var persistor = new BatchPersistor( NUnitTelemetryContext.Create()); persistor.PersistSynchronously(batch); Assert.That(batch.IsPersisted, Is.True); // Delete all batches persistor.DeleteBatches([]); // Restore should not crash (file gone) batch.Restore(values); batch.SetPersisted(); persistor.RestoreSynchronously(batch); // RestoreInProgress is reset even when file is missing Assert.That(batch.RestoreInProgress, Is.False); } } }