lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/SubscriptionSurvivalTests.cs

using System.Net;
using System.Net.Sockets;
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using Xunit;

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;

/// <summary>
///     R2-07 03/P1 — the HEADLINE proof: an over-the-wire subscription on an existing tag SURVIVES a
///     pure-add deploy (one new tag), instead of being killed by the full-rebuild teardown that pre-R2-07
///     every topology change forced. Boots the real <see cref="OtOpcUaSdkServer"/>, materialises tag A,
///     opens a real client subscription + monitored item on A, then drives a PURE-ADD apply adding tag B
///     (classifier ⇒ PureAdd ⇒ no rebuild ⇒ the idempotent Materialise passes create B and leave A's
///     NodeState untouched), and asserts:
///     <list type="bullet">
///         <item>A's monitored item still delivers a subsequent value write (the item is alive — no
///             <c>BadNodeIdUnknown</c>, no dead subscription);</item>
///         <item>B is browsable + readable (the add landed).</item>
///     </list>
///     Written against the master (pre-T2) behaviour this FAILS (the rebuild recreates A as a new NodeState
///     and the monitored item goes dead); it passes once the classifier routing (T2) + the publish-actor
///     announce (T4b) land.
///     <para><b>Run note:</b> this is a heavy in-process server+client integration test — run it in the
///     serial integration pass (it is NOT part of the lightweight unit sweep). The <see cref="OtOpcUaNodeManager"/>
///     namespace URI is <see cref="OtOpcUaNodeManager.DefaultNamespaceUri"/>; the client resolves the
///     folder-scoped variable NodeIds against it.</para>
/// </summary>
public sealed class SubscriptionSurvivalTests
{
    private const string ServerUri = "urn:OtOpcUa.SubscriptionSurvival";

    /// <summary>A subscription on tag A survives a pure-add deploy adding tag B; B becomes readable.</summary>
    [Fact]
    public async Task Subscription_on_existing_tag_survives_pure_add_of_another_tag()
    {
        var pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-subsurvive-{Guid.NewGuid():N}");
        var port = AllocateFreePort();
        var ct = TestContext.Current.CancellationToken;

        try
        {
            var options = new OpcUaApplicationHostOptions
            {
                ApplicationName = ServerUri,
                ApplicationUri = ServerUri,
                OpcUaPort = port,
                PublicHostname = "127.0.0.1",
                PkiStoreRoot = pkiRoot,
                EnabledSecurityProfiles = new List<OpcUaSecurityProfile> { OpcUaSecurityProfile.None },
                AutoAcceptUntrustedClientCertificates = true,
            };
            var server = new OtOpcUaSdkServer();
            await using var host = new OpcUaApplicationHost(options, NullLogger<OpcUaApplicationHost>.Instance);
            await host.StartAsync(server, ct);

            var nm = server.NodeManager!;
            var sink = new SdkAddressSpaceSink(nm);
            var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);

            // --- Materialise tag A (equipment eq-1, tag "A") and seed it a Good value. ---
            var withA = CompositionWith(
                new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null));
            applier.MaterialiseHierarchy(withA);
            applier.MaterialiseEquipmentTags(withA);
            var nodeIdAString = V3NodeIds.Uns("eq-1", "A");
            sink.WriteValue(nodeIdAString, 41, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);

            // --- Open a client session + subscription + monitored item on A. ---
            using var session = await OpenSessionAsync($"opc.tcp://127.0.0.1:{port}/OtOpcUa", ct);
            var ns = (ushort)session.NamespaceUris.GetIndex(OtOpcUaNodeManager.DefaultNamespaceUri);
            ns.ShouldBeGreaterThan((ushort)0);
            var nodeIdA = new NodeId(nodeIdAString, ns);

            var received = new List<DataValue>();
            var gate = new object();
            var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
            session.AddSubscription(subscription);
            await subscription.CreateAsync(ct);

            var monitored = new MonitoredItem(subscription.DefaultItem)
            {
                StartNodeId = nodeIdA,
                AttributeId = Attributes.Value,
                SamplingInterval = 50,
                QueueSize = 10,
            };
            monitored.Notification += (item, _) =>
            {
                foreach (var v in item.DequeueValues())
                    lock (gate) received.Add(v);
            };
            subscription.AddItem(monitored);
            await subscription.ApplyChangesAsync(ct);

            // Baseline: the initial value (41) is delivered.
            await WaitUntilAsync(() => { lock (gate) return received.Count >= 1; }, TimeSpan.FromSeconds(5));

            // --- Drive a PURE-ADD apply adding tag B (previous={A}, next={A,B}). ---
            var withAB = CompositionWith(
                new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null),
                new EquipmentTagPlan("tag-b", "eq-1", "drv", FolderPath: "", Name: "B", DataType: "Int32", FullName: "2", Writable: false, Alarm: null));
            var plan = AddressSpacePlanner.Compute(withA, withAB);
            plan.AddedEquipmentTags.Count.ShouldBe(1);

            var outcome = applier.Apply(plan);
            outcome.RebuildCalled.ShouldBeFalse(); // PureAdd — no teardown

            // Materialise passes (as the publish actor runs them) then announce — over the FULL composition.
            applier.MaterialiseHierarchy(withAB);
            applier.MaterialiseEquipmentTags(withAB);
            applier.AnnounceAddedNodes(plan);

            // --- A's monitored item must still be alive: a new write to A is delivered. ---
            lock (gate) received.Clear();
            sink.WriteValue(nodeIdAString, 42, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);

            await WaitUntilAsync(() => { lock (gate) return received.Any(v => Equals(v.Value, 42)); }, TimeSpan.FromSeconds(5));
            lock (gate)
            {
                received.ShouldContain(v => Equals(v.Value, 42));
                received.ShouldNotContain(v => v.StatusCode == StatusCodes.BadNodeIdUnknown);
            }

            // --- B is browsable + readable (the add landed). ---
            var nodeIdB = new NodeId(V3NodeIds.Uns("eq-1", "B"), ns);
            sink.WriteValue(nodeIdB.Identifier.ToString()!, 7, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
            var bValue = await session.ReadValueAsync(nodeIdB, ct);
            bValue.StatusCode.ShouldNotBe(StatusCodes.BadNodeIdUnknown);
            bValue.Value.ShouldBe(7);
        }
        finally
        {
            if (Directory.Exists(pkiRoot)) Directory.Delete(pkiRoot, recursive: true);
        }
    }

    /// <summary>R2-07 Phase 2 — a subscription on tag A survives a PURE-REMOVE deploy that removes tag B; B's
    /// monitored item observes a final Bad and the node then disappears (a re-read returns BadNodeIdUnknown),
    /// while A stays alive.</summary>
    [Fact]
    public async Task Subscription_on_surviving_tag_survives_pure_remove_of_another_tag()
    {
        var pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-subsurvive-rm-{Guid.NewGuid():N}");
        var port = AllocateFreePort();
        var ct = TestContext.Current.CancellationToken;

        try
        {
            var options = new OpcUaApplicationHostOptions
            {
                ApplicationName = ServerUri,
                ApplicationUri = ServerUri + ".Remove",
                OpcUaPort = port,
                PublicHostname = "127.0.0.1",
                PkiStoreRoot = pkiRoot,
                EnabledSecurityProfiles = new List<OpcUaSecurityProfile> { OpcUaSecurityProfile.None },
                AutoAcceptUntrustedClientCertificates = true,
            };
            var server = new OtOpcUaSdkServer();
            await using var host = new OpcUaApplicationHost(options, NullLogger<OpcUaApplicationHost>.Instance);
            await host.StartAsync(server, ct);

            var nm = server.NodeManager!;
            var sink = new SdkAddressSpaceSink(nm);
            var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);

            // Materialise A + B and seed both Good.
            var withAB = CompositionWith(
                new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null),
                new EquipmentTagPlan("tag-b", "eq-1", "drv", FolderPath: "", Name: "B", DataType: "Int32", FullName: "2", Writable: false, Alarm: null));
            applier.MaterialiseHierarchy(withAB);
            applier.MaterialiseEquipmentTags(withAB);
            var nodeIdAString = V3NodeIds.Uns("eq-1", "A");
            var nodeIdBString = V3NodeIds.Uns("eq-1", "B");
            sink.WriteValue(nodeIdAString, 10, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
            sink.WriteValue(nodeIdBString, 20, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);

            using var session = await OpenSessionAsync($"opc.tcp://127.0.0.1:{port}/OtOpcUa", ct);
            var ns = (ushort)session.NamespaceUris.GetIndex(OtOpcUaNodeManager.DefaultNamespaceUri);
            var nodeIdA = new NodeId(nodeIdAString, ns);
            var nodeIdB = new NodeId(nodeIdBString, ns);

            var receivedA = new List<DataValue>();
            var receivedB = new List<DataValue>();
            var gate = new object();
            var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
            session.AddSubscription(subscription);
            await subscription.CreateAsync(ct);
            var miA = new MonitoredItem(subscription.DefaultItem) { StartNodeId = nodeIdA, AttributeId = Attributes.Value, SamplingInterval = 50, QueueSize = 10 };
            var miB = new MonitoredItem(subscription.DefaultItem) { StartNodeId = nodeIdB, AttributeId = Attributes.Value, SamplingInterval = 50, QueueSize = 10 };
            miA.Notification += (item, _) => { foreach (var v in item.DequeueValues()) lock (gate) receivedA.Add(v); };
            miB.Notification += (item, _) => { foreach (var v in item.DequeueValues()) lock (gate) receivedB.Add(v); };
            subscription.AddItem(miA);
            subscription.AddItem(miB);
            await subscription.ApplyChangesAsync(ct);
            await WaitUntilAsync(() => { lock (gate) return receivedA.Count >= 1 && receivedB.Count >= 1; }, TimeSpan.FromSeconds(5));

            // Drive a PURE-REMOVE apply removing B (previous={A,B}, next={A}).
            var withA = CompositionWith(
                new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null));
            var plan = AddressSpacePlanner.Compute(withAB, withA);
            plan.RemovedEquipmentTags.Count.ShouldBe(1);

            lock (gate) { receivedA.Clear(); receivedB.Clear(); }
            var outcome = applier.Apply(plan);
            outcome.RebuildCalled.ShouldBeFalse(); // PureRemove — no full rebuild

            // B's monitored item observes a final Bad; the node is then gone from the address space.
            await WaitUntilAsync(() => { lock (gate) return receivedB.Any(v => StatusCode.IsBad(v.StatusCode)); }, TimeSpan.FromSeconds(5));

            // A stays alive: a fresh write to A is delivered.
            sink.WriteValue(nodeIdAString, 11, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
            await WaitUntilAsync(() => { lock (gate) return receivedA.Any(v => Equals(v.Value, 11)); }, TimeSpan.FromSeconds(5));

            // B is gone: a re-read surfaces BadNodeIdUnknown. The 1.5.378 node-cache read path
            // throws ServiceResultException for an unknown node rather than returning a bad DataValue.
            var bReadEx = await Should.ThrowAsync<ServiceResultException>(
                async () => await session.ReadValueAsync(nodeIdB, ct));
            bReadEx.StatusCode.ShouldBe(StatusCodes.BadNodeIdUnknown);
        }
        finally
        {
            if (Directory.Exists(pkiRoot)) Directory.Delete(pkiRoot, recursive: true);
        }
    }

    /// <summary>R2-07 Phase 3 — a subscription on tag A survives a MIXED deploy (−B, +C in one apply): A
    /// stays alive, B is gone (BadNodeIdUnknown), and C is browsable + readable.</summary>
    [Fact]
    public async Task Subscription_survives_mixed_add_remove_deploy()
    {
        var pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-subsurvive-mix-{Guid.NewGuid():N}");
        var port = AllocateFreePort();
        var ct = TestContext.Current.CancellationToken;

        try
        {
            var options = new OpcUaApplicationHostOptions
            {
                ApplicationName = ServerUri,
                ApplicationUri = ServerUri + ".Mixed",
                OpcUaPort = port,
                PublicHostname = "127.0.0.1",
                PkiStoreRoot = pkiRoot,
                EnabledSecurityProfiles = new List<OpcUaSecurityProfile> { OpcUaSecurityProfile.None },
                AutoAcceptUntrustedClientCertificates = true,
            };
            var server = new OtOpcUaSdkServer();
            await using var host = new OpcUaApplicationHost(options, NullLogger<OpcUaApplicationHost>.Instance);
            await host.StartAsync(server, ct);

            var nm = server.NodeManager!;
            var sink = new SdkAddressSpaceSink(nm);
            var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);

            var withAB = CompositionWith(
                new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null),
                new EquipmentTagPlan("tag-b", "eq-1", "drv", FolderPath: "", Name: "B", DataType: "Int32", FullName: "2", Writable: false, Alarm: null));
            applier.MaterialiseHierarchy(withAB);
            applier.MaterialiseEquipmentTags(withAB);
            var nodeIdAString = V3NodeIds.Uns("eq-1", "A");
            sink.WriteValue(nodeIdAString, 10, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);

            using var session = await OpenSessionAsync($"opc.tcp://127.0.0.1:{port}/OtOpcUa", ct);
            var ns = (ushort)session.NamespaceUris.GetIndex(OtOpcUaNodeManager.DefaultNamespaceUri);
            var nodeIdA = new NodeId(nodeIdAString, ns);

            var receivedA = new List<DataValue>();
            var gate = new object();
            var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
            session.AddSubscription(subscription);
            await subscription.CreateAsync(ct);
            var miA = new MonitoredItem(subscription.DefaultItem) { StartNodeId = nodeIdA, AttributeId = Attributes.Value, SamplingInterval = 50, QueueSize = 10 };
            miA.Notification += (item, _) => { foreach (var v in item.DequeueValues()) lock (gate) receivedA.Add(v); };
            subscription.AddItem(miA);
            await subscription.ApplyChangesAsync(ct);
            await WaitUntilAsync(() => { lock (gate) return receivedA.Count >= 1; }, TimeSpan.FromSeconds(5));

            // MIXED apply: remove B, add C (previous={A,B}, next={A,C}).
            var withAC = CompositionWith(
                new EquipmentTagPlan("tag-a", "eq-1", "drv", FolderPath: "", Name: "A", DataType: "Int32", FullName: "1", Writable: false, Alarm: null),
                new EquipmentTagPlan("tag-c", "eq-1", "drv", FolderPath: "", Name: "C", DataType: "Int32", FullName: "3", Writable: false, Alarm: null));
            var plan = AddressSpacePlanner.Compute(withAB, withAC);
            plan.RemovedEquipmentTags.Count.ShouldBe(1);
            plan.AddedEquipmentTags.Count.ShouldBe(1);

            lock (gate) receivedA.Clear();
            var outcome = applier.Apply(plan);
            outcome.RebuildCalled.ShouldBeFalse(); // AddRemoveMix — removes in place, no rebuild
            // Publish-actor sequence: removes ran inside Apply; now materialise the adds + announce.
            applier.MaterialiseEquipmentTags(withAC);
            applier.AnnounceAddedNodes(plan);

            // A stays alive.
            sink.WriteValue(nodeIdAString, 11, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
            await WaitUntilAsync(() => { lock (gate) return receivedA.Any(v => Equals(v.Value, 11)); }, TimeSpan.FromSeconds(5));

            // B is gone. (1.5.378 node-cache read throws for an unknown node — see above.)
            var nodeIdB = new NodeId(V3NodeIds.Uns("eq-1", "B"), ns);
            var bReadEx = await Should.ThrowAsync<ServiceResultException>(
                async () => await session.ReadValueAsync(nodeIdB, ct));
            bReadEx.StatusCode.ShouldBe(StatusCodes.BadNodeIdUnknown);

            // C is browsable + readable.
            var nodeIdC = new NodeId(V3NodeIds.Uns("eq-1", "C"), ns);
            sink.WriteValue(nodeIdC.Identifier.ToString()!, 9, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
            var cRead = await session.ReadValueAsync(nodeIdC, ct);
            cRead.StatusCode.ShouldNotBe((StatusCode)StatusCodes.BadNodeIdUnknown);
            cRead.Value.ShouldBe(9);
        }
        finally
        {
            if (Directory.Exists(pkiRoot)) Directory.Delete(pkiRoot, recursive: true);
        }
    }

    private static AddressSpaceComposition CompositionWith(params EquipmentTagPlan[] tags) =>
        new(
            UnsAreas: Array.Empty<UnsAreaProjection>(),
            UnsLines: Array.Empty<UnsLineProjection>(),
            EquipmentNodes: new[] { new EquipmentNode("eq-1", "Equipment 1", "") },
            DriverInstancePlans: Array.Empty<DriverInstancePlan>(),
            ScriptedAlarmPlans: Array.Empty<ScriptedAlarmPlan>())
        {
            EquipmentTags = tags,
        };

    private static async Task<ISession> OpenSessionAsync(string endpointUrl, CancellationToken ct)
    {
        var appConfig = new ApplicationConfiguration
        {
            ApplicationName = "OtOpcUa.SubscriptionSurvivalClient",
            ApplicationUri = $"urn:OtOpcUa.SubscriptionSurvivalClient.{Guid.NewGuid():N}",
            ApplicationType = ApplicationType.Client,
            // Directory cert stores rooted at a throwaway temp PKI dir — ValidateAsync rejects a
            // bare SecurityConfiguration ("TrustedIssuerCertificates StorePath must be specified").
            // The returned session outlives this method, so the dir is left for the OS temp sweep.
            SecurityConfiguration = TestClientSecurity.Build(TestClientSecurity.AllocatePkiRoot()),
            ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 60_000 },
        };
        await appConfig.ValidateAsync(ApplicationType.Client, ct);
        appConfig.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true;

        var endpoint = await CoreClientUtils.SelectEndpointAsync(
            appConfig, endpointUrl, false, DefaultTelemetry.Create(_ => { }), ct);
        var endpointConfiguration = EndpointConfiguration.Create(appConfig);
        var configuredEndpoint = new ConfiguredEndpoint(null, endpoint, endpointConfiguration);

        return await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(
            appConfig,
            configuredEndpoint,
            updateBeforeConnect: false,
            checkDomain: false,
            sessionName: "SubscriptionSurvivalTests",
            sessionTimeout: 60_000,
            identity: new UserIdentity(new AnonymousIdentityToken()),
            preferredLocales: null,
            ct: ct);
    }

    private static async Task WaitUntilAsync(Func<bool> condition, TimeSpan timeout)
    {
        var deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            if (condition()) return;
            await Task.Delay(50);
        }
        condition().ShouldBeTrue("condition not met within timeout");
    }

    private static int AllocateFreePort()
    {
        var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
        listener.Stop();
        return port;
    }
}