lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipReadWindowTests.cs

using Shouldly;
using Xunit;

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

/// <summary>
///     Review 07 P8 / P7: the bounded-parallel read window. On the native transport every concurrent tag read of a
///     device rides ONE shared request scheduler that packs whatever is queued into a Multiple Service Packet, so a
///     window of 8 capped MSP coalescing at 8 services per round trip; it is widened to at least 64 there (the
///     scheduler's reply budget already prevents over-packing). And a cold create waiting on a device's connect gate
///     must not occupy a window slot, or one dead device's backoff-expiry attempt stalls every other device's reads.
/// </summary>
[Trait("Category", "Unit")]
public sealed class AbCipReadWindowTests
{
    private static AbCipDriverOptions Options(int maxConcurrentReads, AbCipTransport? transport = null) => new()
    {
        Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")],
        Probe = new AbCipProbeOptions { Enabled = false },
        MaxConcurrentReads = maxConcurrentReads,
        Transport = transport,
    };

    /// <summary>The window per transport: native ≥ 64; libplctag and injected factories honour the option exactly.</summary>
    [Fact]
    public void The_native_window_is_at_least_64_and_other_transports_keep_the_option()
    {
        new AbCipDriver(Options(8, AbCipTransport.Native), "a").ReadWindow.ShouldBe(AbCipDriver.NativeMinReadWindow);
        AbCipDriver.NativeMinReadWindow.ShouldBeGreaterThanOrEqualTo(64);
        new AbCipDriver(Options(1, AbCipTransport.Native), "a").ReadWindow.ShouldBe(AbCipDriver.NativeMinReadWindow);
        new AbCipDriver(Options(200, AbCipTransport.Native), "a").ReadWindow.ShouldBe(200);
        new AbCipDriver(Options(8), "a", defaultTransport: AbCipTransport.Native).ReadWindow.ShouldBe(AbCipDriver.NativeMinReadWindow);

        // Positive controls: libplctag (per-handle concurrency) and an injected test factory keep the option.
        new AbCipDriver(Options(8, AbCipTransport.Libplctag), "a").ReadWindow.ShouldBe(8);
        new AbCipDriver(Options(0, AbCipTransport.Libplctag), "a").ReadWindow.ShouldBe(1);
        new AbCipDriver(Options(3, AbCipTransport.Native), "a", new FakeAbCipTagFactory()).ReadWindow.ShouldBe(3);
    }

    /// <summary>A tag whose initialisation blocks until released (a cold connect to a slow / dead device).</summary>
    private sealed class BlockingInitTag(AbCipTagCreateParams p, Task release) : FakeAbCipTag(p)
    {
        public override async Task InitializeAsync(CancellationToken cancellationToken)
        {
            await base.InitializeAsync(cancellationToken);
            await release.WaitAsync(cancellationToken);
        }
    }

    /// <summary>
    ///     With a window of ONE, device B's blocked cold create (and its queued sibling) must not stop device A's tag
    ///     from being read: cold waiters sit on B's connect gate, not in the window.
    /// </summary>
    [Fact]
    public async Task A_blocked_cold_create_on_one_device_does_not_hold_the_window_against_another()
    {
        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var factory = new FakeAbCipTagFactory
        {
            Customise = p => p.Gateway == "10.0.0.6"
                ? new BlockingInitTag(p, release.Task) { Value = 2 }
                : new FakeAbCipTag(p) { Value = 1 },
        };
        const string a = "ab://10.0.0.5/1,0", b = "ab://10.0.0.6/1,0";
        var drv = new AbCipDriver(new AbCipDriverOptions
        {
            Devices = [new AbCipDeviceOptions(a), new AbCipDeviceOptions(b)],
            RawTags = AbCipRawTags.From(
                new AbCipTagDefinition("B1", b, "B1", AbCipDataType.DInt),
                new AbCipTagDefinition("B2", b, "B2", AbCipDataType.DInt),
                new AbCipTagDefinition("A1", a, "A1", AbCipDataType.DInt)),
            Probe = new AbCipProbeOptions { Enabled = false },
            MaxConcurrentReads = 1,
        }, "abcip-window", factory);
        await drv.InitializeAsync("{}", CancellationToken.None);

        var read = drv.ReadAsync(["B1", "B2", "A1"], CancellationToken.None);
        var deadline = DateTime.UtcNow.AddSeconds(5);
        while (!(factory.Tags.TryGetValue("A1", out var tag) && tag.ReadCount > 0))
        {
            if (DateTime.UtcNow > deadline)
            {
                release.TrySetResult();
                await read;
                throw new ShouldAssertException("device A's read waited behind device B's blocked cold create");
            }
            await Task.Delay(5);
        }
        read.IsCompleted.ShouldBeFalse(); // B is still blocked — A was read regardless

        release.SetResult();
        var r = await read;
        r.Select(x => x.Value).ShouldBe([2, 2, 1]); // positive control: B's reads complete once it connects
        await drv.DisposeAsync();
    }
}