lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7ProbeContentionAndTimeoutTests.cs

using System.Net;
using System.Net.Sockets;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

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

/// <summary>
///     R07:H7 — (a) the background probe used to report the host <c>Stopped</c> whenever a long read
///     pass held the gate past the probe deadline: contention is the connection being USED, not lost;
///     (b) Test Connect ignored the config's <c>timeoutMs</c>. Waits are condition waits on counters and
///     events; bounds are generous.
/// </summary>
[Trait("Category", "Unit")]
public sealed class S7ProbeContentionAndTimeoutTests
{
    private static readonly TimeSpan Bound = TimeSpan.FromSeconds(15);

    private static async Task WaitUntilAsync(Func<bool> condition, CancellationToken ct)
    {
        using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
        cts.CancelAfter(Bound);
        while (!condition())
            await Task.Delay(20, cts.Token); // polling a condition, not asserting on elapsed time
    }

    /// <summary>A read pass holding the gate past the probe deadline does not make the host Stopped.</summary>
    [Fact]
    public async Task Gate_contention_is_no_verdict_but_a_real_probe_failure_still_is()
    {
        var ct = TestContext.Current.CancellationToken;
        var factory = new ScriptedS7PlcFactory();
        var readEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var releaseRead = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        using var drv = new S7Driver(new S7DriverOptions
        {
            Host = "192.0.2.1",
            Timeout = TimeSpan.FromSeconds(30),
            Probe = new S7ProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(50), Timeout = TimeSpan.FromSeconds(1) },
            RawTags = S7RawTags.Entries(new S7TagDefinition("W0", "DB1.DBW0", S7DataType.UInt16, Writable: false)),
        }, "s7-probe-contention", factory);
        var transitions = new List<HostState>();
        drv.OnHostStatusChanged += (_, e) => { lock (transitions) transitions.Add(e.NewState); };
        await drv.InitializeAsync("{}", ct);

        await WaitUntilAsync(() => drv.GetHostStatuses()[0].State == HostState.Running, ct);

        // A read that holds the gate until released.
        factory.Read = async (_, token) =>
        {
            readEntered.TrySetResult();
            await releaseRead.Task.WaitAsync(token);
            return (ushort)1;
        };
        var read = drv.ReadAsync(["W0"], ct);
        await readEntered.Task.WaitAsync(Bound, ct);

        // At least two probe ticks give up on the gate...
        var contentionsBefore = drv.ProbeGateContentions;
        await WaitUntilAsync(() => drv.ProbeGateContentions >= contentionsBefore + 2, ct);
        // ...and the host is still Running — no Stopped transition was published for them.
        drv.GetHostStatuses()[0].State.ShouldBe(HostState.Running);
        lock (transitions) transitions.ShouldNotContain(HostState.Stopped);

        releaseRead.SetResult();
        (await read.WaitAsync(Bound, ct))[0].StatusCode.ShouldBe(0u);

        // Positive control: a genuine probe failure still reports Stopped.
        factory.ReadStatus = _ => throw new System.IO.IOException("connection reset");
        await WaitUntilAsync(() => drv.GetHostStatuses()[0].State == HostState.Stopped, ct);
    }

    /// <summary>Test Connect's handshake is bounded by the config's timeoutMs, not the caller's whole budget.</summary>
    [Fact]
    public async Task Test_connect_honours_the_config_timeoutMs()
    {
        var ct = TestContext.Current.CancellationToken;
        // A TCP endpoint that accepts and then never answers the ISO-on-TCP handshake.
        var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        var held = new List<TcpClient>();
        _ = Task.Run(async () =>
        {
            try { while (true) { var c = await listener.AcceptTcpClientAsync(ct); lock (held) held.Add(c); } }
            catch { /* stopped */ }
        }, ct);
        try
        {
            var port = ((IPEndPoint)listener.LocalEndpoint).Port;
            var json = $$"""{"host":"127.0.0.1","port":{{port}},"timeoutMs":1000}""";

            // The caller's budget (60 s) is far beyond the safety token (20 s): only a probe that honours
            // timeoutMs returns its own "handshake failed: timed out"; one that waits the caller's budget
            // is cut by the safety token and reports the caller-timeout message instead.
            using var safety = CancellationTokenSource.CreateLinkedTokenSource(ct);
            safety.CancelAfter(TimeSpan.FromSeconds(20));
            var r = await new S7DriverProbe().ProbeAsync(json, TimeSpan.FromSeconds(60), safety.Token);

            r.Ok.ShouldBeFalse();
            r.Message!.ShouldContain("handshake failed: timed out");
            safety.IsCancellationRequested.ShouldBeFalse();
        }
        finally
        {
            listener.Stop();
            lock (held) foreach (var c in held) c.Dispose();
        }
    }

    /// <summary>The config reader: any casing, positive integers only.</summary>
    [Fact]
    public void ReadTimeoutMs_reads_the_driver_key()
    {
        S7DriverProbe.ReadTimeoutMs("""{"TimeoutMs":2500}""").ShouldBe(2500);
        S7DriverProbe.ReadTimeoutMs("""{"timeoutms":10}""").ShouldBe(10);
        S7DriverProbe.ReadTimeoutMs("""{"timeoutMs":0}""").ShouldBeNull();
        S7DriverProbe.ReadTimeoutMs("""{"host":"x"}""").ShouldBeNull();
        S7DriverProbe.ReadTimeoutMs("not json").ShouldBeNull();
    }
}