lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReconnectReplayTests.cs

using System.Collections.Concurrent;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.TwinCAT;

namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests;

/// <summary>
///     Unit tests for the STAB-2 fix: native ADS notifications are re-registered onto a fresh
///     client after a reconnect, instead of being silently orphaned. There is <b>no TwinCAT docker
///     fixture</b> (the 13 integration tests need a real TC3 XAR target), so these fake-client tests
///     are the <i>authoritative</i> automated coverage for the reconnect path.
/// </summary>
[Trait("Category", "Unit")]
public sealed class TwinCATReconnectReplayTests
{
    private const string Host = "ads://5.23.91.23.1.1:851";

    private static (TwinCATDriver drv, FakeTwinCATClientFactory factory) NewNativeDriver(
        bool probe = false, TimeSpan? probeInterval = null, params TwinCATTagDefinition[] tags)
    {
        var factory = new FakeTwinCATClientFactory();
        var drv = new TwinCATDriver(new TwinCATDriverOptions
        {
            Devices = [new TwinCATDeviceOptions(Host)],
            Tags = tags,
            Probe = new TwinCATProbeOptions
            {
                Enabled = probe,
                Interval = probeInterval ?? TimeSpan.FromMilliseconds(30),
                Timeout = TimeSpan.FromMilliseconds(100),
            },
            UseNativeNotifications = true,
        }, "drv-1", factory);
        return (drv, factory);
    }

    /// <summary>
    ///     After a client drop, the next connection (driven here through a read) re-registers the
    ///     native notification with the same symbol / type / interval, a push on the fresh client
    ///     reaches OnDataChange, and the old handle is disposed.
    /// </summary>
    [Fact]
    public async Task Reconnect_replays_native_notifications_onto_the_fresh_client()
    {
        var (drv, factory) = NewNativeDriver(
            tags: new TwinCATTagDefinition("Speed", Host, "MAIN.Speed", TwinCATDataType.DInt));
        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);

        var events = new ConcurrentQueue<DataChangeEventArgs>();
        drv.OnDataChange += (_, e) => events.Enqueue(e);

        var interval = TimeSpan.FromMilliseconds(250);
        _ = await drv.SubscribeAsync(["Speed"], interval, TestContext.Current.CancellationToken);

        var client1 = factory.Clients[0];
        client1.Notifications.Count.ShouldBe(1);
        var oldHandle = client1.Notifications[0];

        // Simulate a wire-level drop, then trigger a reconnect via a read.
        client1.SimulateWireDrop();
        _ = await drv.ReadAsync(["Speed"], TestContext.Current.CancellationToken);

        // A brand-new client was built and the notification replayed onto it — same intent.
        factory.Clients.Count.ShouldBe(2);
        var client2 = factory.Clients[1];
        client2.Notifications.Count.ShouldBe(1);
        var replayed = client2.Notifications[0];
        replayed.SymbolPath.ShouldBe("MAIN.Speed");
        replayed.Type.ShouldBe(TwinCATDataType.DInt);

        // The old handle was disposed; a push on the FRESH client reaches OnDataChange.
        oldHandle.Disposed.ShouldBeTrue();
        client2.FireNotification("MAIN.Speed", 7777);
        events.Count.ShouldBe(1);
        events.Last().Snapshot.Value.ShouldBe(7777);
        events.Last().FullReference.ShouldBe("Speed");
    }

    /// <summary>Every subscribed tag on the device is replayed, not just the first.</summary>
    [Fact]
    public async Task Reconnect_replays_all_registered_tags()
    {
        var (drv, factory) = NewNativeDriver(
            tags:
            [
                new TwinCATTagDefinition("A", Host, "MAIN.A", TwinCATDataType.DInt),
                new TwinCATTagDefinition("B", Host, "MAIN.B", TwinCATDataType.Real),
            ]);
        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
        _ = await drv.SubscribeAsync(["A", "B"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);

        factory.Clients[0].SimulateWireDrop();
        _ = await drv.ReadAsync(["A"], TestContext.Current.CancellationToken);

        factory.Clients.Count.ShouldBe(2);
        factory.Clients[1].Notifications.Select(n => n.SymbolPath)
            .ShouldBe(["MAIN.A", "MAIN.B"], ignoreOrder: true);
    }

    /// <summary>
    ///     After a reconnect + replay, unsubscribing disposes the CURRENT (replayed) handle and the
    ///     registration no longer replays on a subsequent reconnect.
    /// </summary>
    [Fact]
    public async Task Unsubscribe_after_reconnect_disposes_the_current_handle_and_stops_replaying()
    {
        var (drv, factory) = NewNativeDriver(
            tags: new TwinCATTagDefinition("X", Host, "MAIN.X", TwinCATDataType.DInt));
        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
        var handle = await drv.SubscribeAsync(["X"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);

        // First reconnect → replayed onto client2.
        factory.Clients[0].SimulateWireDrop();
        _ = await drv.ReadAsync(["X"], TestContext.Current.CancellationToken);
        var client2 = factory.Clients[1];
        client2.Notifications.Count.ShouldBe(1);

        // Unsubscribe disposes the current (client2) handle.
        await drv.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
        client2.Notifications.ShouldBeEmpty();

        // A second reconnect must NOT resurrect the unsubscribed notification.
        client2.SimulateWireDrop();
        _ = await drv.ReadAsync(["X"], TestContext.Current.CancellationToken);
        factory.Clients.Count.ShouldBe(3);
        factory.Clients[2].Notifications.ShouldBeEmpty();
    }

    /// <summary>
    ///     A wire-level probe failure on a locally-"connected" client forces a recycle so the next
    ///     tick rebuilds the client — the compounding half of STAB-2 (the IsConnected fast-path
    ///     keys on port state, not wire liveness).
    /// </summary>
    [Fact]
    public async Task Probe_failure_recycles_the_client_and_rebuilds()
    {
        // First client's wire probe fails; the recycle nulls it and the next tick builds a second
        // client whose probe succeeds.
        var buildCount = 0;
        var factory = new FakeTwinCATClientFactory
        {
            Customise = () =>
            {
                buildCount++;
                // Client #1 reports a failed wire probe; #2+ succeed.
                return new FakeTwinCATClient { ProbeResult = buildCount > 1 };
            },
        };
        var drv = new TwinCATDriver(new TwinCATDriverOptions
        {
            Devices = [new TwinCATDeviceOptions(Host)],
            Tags = [new TwinCATTagDefinition("X", Host, "MAIN.X", TwinCATDataType.DInt)],
            Probe = new TwinCATProbeOptions
            {
                Enabled = true,
                Interval = TimeSpan.FromMilliseconds(30),
                Timeout = TimeSpan.FromMilliseconds(100),
            },
            UseNativeNotifications = true,
        }, "drv-probe", factory);
        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);

        // The probe loop should recycle the wire-dead client and rebuild a healthy one.
        await WaitForAsync(() => factory.Clients.Count >= 2, TimeSpan.FromSeconds(3));
        factory.Clients.Count.ShouldBeGreaterThanOrEqualTo(2);
        factory.Clients[0].DisposeCount.ShouldBeGreaterThanOrEqualTo(1);

        await drv.ShutdownAsync(TestContext.Current.CancellationToken);
    }

    private static async Task WaitForAsync(Func<bool> condition, TimeSpan timeout)
    {
        var deadline = DateTime.UtcNow + timeout;
        while (!condition() && DateTime.UtcNow < deadline)
            await Task.Delay(20);
    }
}