lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttDriverSessionLifecycleTests.cs

using System.Diagnostics;
using System.Text.Json;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

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

/// <summary>
///     The driver's own session lifecycle, against the <see cref="MiniBroker"/> fake: what
///     <see cref="MqttDriver.InitializeAsync"/> does to the session it is <i>replacing</i>, and what a
///     mis-authored duration knob does to the two paths whose whole job is to not fail
///     (<see cref="MqttDriver.ShutdownAsync"/> / <see cref="MqttDriver.DisposeAsync"/>).
/// </summary>
/// <remarks>
///     Plain mode deliberately: it is the only mode whose initialize needs no SUBSCRIBE, and
///     <see cref="MiniBroker"/> answers CONNECT and PINGREQ and nothing else — so a Sparkplug driver
///     would park on its group SUBSCRIBE until the deadline and prove nothing about connections.
/// </remarks>
[Trait("Category", "Unit")]
public sealed class MqttDriverSessionLifecycleTests
{
    /// <summary>
    ///     <b>Driver.Mqtt-002.</b> <c>InitializeCoreAsync</c> publishes <c>_connection</c> — an
    ///     <see cref="IAsyncDisposable"/> — so re-initializing a live driver without tearing down first
    ///     stranded a connected, supervised broker session that no later dispose could reach, and the
    ///     abandoned session's reconnect supervisor kept it alive against the broker indefinitely.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         Reachable with no teardown of the driver's own: <c>DriverInstanceActor</c> re-runs
    ///         <c>InitializeAsync</c> with no <c>ShutdownAsync</c> when an <c>ApplyDelta</c> arrives while
    ///         the driver is Connecting/Reconnecting (<c>AdoptConfigDuringInit</c>). The
    ///         <c>teardownFirst</c> flag on the #527 retry path covers only the retry path.
    ///     </para>
    ///     <para>
    ///         The assertion is the broker's <b>live socket count</b>, not the driver's own view: a leak
    ///         is precisely the case where the driver believes it holds one session and the broker is
    ///         serving two. <see cref="MiniBroker.AcceptedCount"/> is asserted alongside it as the
    ///         control — without it, a second initialize that never dialled at all would satisfy
    ///         "exactly one live connection" just as well.
    ///     </para>
    /// </remarks>
    [Fact]
    public async Task InitializeAsync_OnALiveDriver_TearsDownThePreviousSession()
    {
        using var broker = new MiniBroker();
        var options = BrokerOptions(broker.Port);
        await using var driver = new MqttDriver(options, "d", null);

        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
        await WaitForAsync(() => broker.LiveConnections == 1, "the first session to be established");

        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);

        broker.AcceptedCount.ShouldBe(2, "the second initialize must genuinely open a new session");
        await WaitForAsync(
            () => broker.LiveConnections == 1,
            "the replaced session to be torn down rather than left live and unreachable");
        driver.GetHealth().State.ShouldBe(DriverState.Healthy);
    }

    /// <summary>
    ///     <b>Driver.Mqtt-003.</b> <c>connectTimeoutSeconds</c> carries a <c>[Range]</c> annotation that
    ///     nothing validates on the runtime path — the blob is deserialized straight out of the deploy
    ///     artifact. A negative value is outside the legal range of both
    ///     <see cref="SemaphoreSlim.WaitAsync(TimeSpan)"/> and <see cref="CancellationTokenSource"/>, so
    ///     it threw <see cref="ArgumentOutOfRangeException"/> out of shutdown and dispose — the two calls
    ///     that must always complete, and the ones a host makes while tearing a bad deployment down.
    /// </summary>
    [Fact]
    public async Task ShutdownAndDispose_WithANegativeConnectTimeout_DoNotThrow()
    {
        var options = BrokerOptions(1) with { ConnectTimeoutSeconds = -1, KeepAliveSeconds = -1 };
        var driver = new MqttDriver(options, "d", null);

        await Should.NotThrowAsync(async () =>
            await driver.ShutdownAsync(TestContext.Current.CancellationToken));
        await Should.NotThrowAsync(async () => await driver.DisposeAsync());
    }

    /// <summary>
    ///     The control for the test above: the normalization must not quietly rewrite a <i>legal</i>
    ///     value, or a deployment's authored connect budget would silently become the default.
    /// </summary>
    [Fact]
    public async Task InitializeAsync_WithAnAuthoredConnectTimeout_KeepsIt()
    {
        using var broker = new MiniBroker { Blackhole = true };
        var options = BrokerOptions(broker.Port) with { ConnectTimeoutSeconds = 2 };
        await using var driver = new MqttDriver(options, "d", null);

        var sw = Stopwatch.StartNew();
        await Should.ThrowAsync<TimeoutException>(async () =>
            await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken));
        sw.Stop();

        sw.Elapsed.ShouldBeLessThan(
            TimeSpan.FromSeconds(10),
            "the authored 2s budget must still bound the connect, not the 15s default");
    }

    private static async Task WaitForAsync(Func<bool> condition, string what)
    {
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
        while (DateTime.UtcNow < deadline)
        {
            if (condition())
            {
                return;
            }

            await Task.Delay(25, TestContext.Current.CancellationToken);
        }

        throw new Xunit.Sdk.XunitException($"Timed out waiting for {what}.");
    }

    private static MqttDriverOptions BrokerOptions(int port) => new()
    {
        Mode = MqttMode.Plain,
        Host = "127.0.0.1",
        Port = port,
        UseTls = false,
        ProtocolVersion = MqttProtocolVersion.V311,
        KeepAliveSeconds = 300,
        ConnectTimeoutSeconds = 5,
        ReconnectMinBackoffSeconds = 1,
        ReconnectMaxBackoffSeconds = 2,
    };

    private static string Json(MqttDriverOptions options) => JsonSerializer.Serialize(options, MqttJson.Options);
}