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 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 = Commons.OpcUa.EquipmentNodeIds.Variable("eq-1", "", "A");
            sink.WriteValue(nodeIdAString, 41, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow);

            // --- 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);

            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(Commons.OpcUa.EquipmentNodeIds.Variable("eq-1", "", "B"), ns);
            sink.WriteValue(nodeIdB.Identifier.ToString()!, 7, Commons.OpcUa.OpcUaQuality.Good, DateTime.UtcNow);
            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);
        }
    }

    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,
            SecurityConfiguration = new SecurityConfiguration
            {
                ApplicationCertificate = new CertificateIdentifier(),
                AutoAcceptUntrustedCertificates = true,
            },
            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;
    }
}