UA-.NETStandard/tests/Opc.Ua.Server.Tests/DurableSubscriptionSerializationTests.cs

/* ========================================================================
 * 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
{
    /// <summary>
    /// Round-trip tests for the BinaryEncoder/BinaryDecoder based
    /// serialization in <see cref="SubscriptionStore"/> and
    /// <see cref="DurableMonitoredItemQueueFactory"/>.
    /// Tests call the internal encode/decode methods directly.
    /// Persistor cleanup clears the process-wide batch directory.
    /// </summary>
    [TestFixture]
    [Category("Server")]
    [SetCulture("en-us")]
    [SetUICulture("en-us")]
    [NonParallelizable]
    public class DurableSubscriptionSerializationTests
    {
        private ServiceMessageContext m_context;

        /// <summary>
        /// Creates the serialization message context and registers the two fixture namespaces.
        /// </summary>
        [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");
        }

        /// <summary>
        /// Verifies that an empty durable subscription preserves its settings and empty item and message collections.
        /// </summary>
        [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));
        }

        /// <summary>
        /// Verifies that the current sample format preserves publishing state and anonymous owner application URI.
        /// </summary>
        [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"));
            });
        }

        /// <summary>
        /// Verifies that the current sample format preserves the triggering links between
        /// monitored items (OPC 10000-4 §5.13.1.6).
        /// </summary>
        [Test]
        public void RoundTripSubscriptionTriggeringLinks()
        {
            StoredSubscription original = CreateMinimalSubscription(id: 45);
            original.TriggeringLinks = new Dictionary<uint, IReadOnlyList<uint>>
            {
                [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 }));
            });
        }

        /// <summary>
        /// Verifies that version-one sample records migrate to the safe default publishing state.
        /// </summary>
        [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);
            });
        }

        /// <summary>
        /// Verifies that durable subscription serialization preserves its monitored items and their identifiers.
        /// </summary>
        [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));
        }

        /// <summary>
        /// Verifies that durable subscription serialization preserves monitored-item properties.
        /// </summary>
        [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));
        }

        /// <summary>
        /// Verifies that the retired interim notification-state format is rejected before decoding a subscription.
        /// </summary>
        [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<InvalidDataException>()
                    .With.Message.Contains("Unsupported durable subscription store version 2"));
            Assert.That(decoder.Position, Is.Zero);
        }

        /// <summary>
        /// Verifies that new subscription records contain raw monitored-item state rather than transient notifications.
        /// </summary>
        [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));
        }

        /// <summary>
        /// Verifies that durable subscription serialization preserves the user identity token.
        /// </summary>
        [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<UserNameIdentityToken>());
            var token = (UserNameIdentityToken)result.UserIdentityToken;
            Assert.That(token.UserName, Is.EqualTo("testuser"));
            Assert.That(token.PolicyId, Is.EqualTo("username_policy"));
        }

        /// <summary>
        /// Verifies that multiple durable subscriptions round-trip independently.
        /// </summary>
        [Test]
        public void RoundTripMultipleSubscriptions()
        {
            var subs = new List<StoredSubscription>
            {
                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<StoredSubscription> 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);
        }

        /// <summary>
        /// Verifies that a storable data-change queue round-trips through serialization.
        /// </summary>
        [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);
        }

        /// <summary>
        /// Verifies that a storable data-change queue preserves its batches during serialization.
        /// </summary>
        [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));
        }

        /// <summary>
        /// Verifies that a storable event queue round-trips through serialization.
        /// </summary>
        [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();
        }

        /// <summary>
        /// Encodes and decodes a list of subscriptions using
        /// the internal SubscriptionStore methods.
        /// </summary>
        private List<StoredSubscription> RoundTripSubscriptions(
            List<StoredSubscription> 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<string?> nsUris = decoder.ReadStringArray(null);
            ArrayOf<string?> 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<StoredSubscription>(count);
            for (int i = 0; i < count; i++)
            {
                results.Add(
                    SubscriptionStore.DecodeSubscription(decoder));
            }
            return results;
        }

        /// <summary>
        /// Encodes and decodes a StorableDataChangeQueue using
        /// the internal DurableMonitoredItemQueueFactory methods.
        /// </summary>
        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<string?> nsUris = decoder.ReadStringArray(null);
            ArrayOf<string?> 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);
        }

        /// <summary>
        /// Encodes and decodes a StorableEventQueue using
        /// the internal DurableMonitoredItemQueueFactory methods.
        /// </summary>
        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<string?> nsUris = decoder.ReadStringArray(null);
            ArrayOf<string?> 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);
        }

        /// <summary>
        /// Verifies that a data-change batch round-trips through the queue persistor.
        /// </summary>
        [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([]);
        }

        /// <summary>
        /// Verifies that an event batch round-trips through the queue persistor.
        /// </summary>
        [Test]
        public void RoundTripEventBatchViaPersistor()
        {
            using IDisposable scope =
                AmbientMessageContext.SetScopedContext(m_context);

            var events = new List<EventFieldList>
            {
                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([]);
        }

        /// <summary>
        /// Verifies that deleting a persisted queue batch removes its backing file.
        /// </summary>
        [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);
        }
    }
}