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.
///     <see cref="MiniBroker"/> answers CONNECT, SUBSCRIBE (unless told not to) and PINGREQ, and never
///     PUBLISHes, so ingest is driven through <c>Subscriptions.HandleMessage</c>.
/// </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>
    ///         The <c>IDriver</c> contract allows re-initialising a live driver with no
    ///         <c>ShutdownAsync</c>, and the #527 retry path's <c>teardownFirst</c> flag does not stop a
    ///         retry overlapping a still-running attempt, so the driver must not rely on a teardown.
    ///     </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>
    [Trait("Category", "SlowUnit")]
    [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");
    }

    /// <summary>
    ///     <b>End-to-end through <see cref="MqttDriver"/> (archreview 05/STAB-18).</b> Deliver a value,
    ///     let the real broker drop the real session, and the driver publishes
    ///     <c>BadNoCommunication</c> with the value retained; feed the <b>identical</b> payload back —
    ///     the retained replay a broker sends on reconnect — and it publishes Good again.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>What this adds over <c>MqttSubscriptionManagerRecoveryTests</c>:</b> the wiring. Those
    ///         drive <c>OnConnectionLost()</c> directly, so they stay green even if
    ///         <c>MqttSubscriptionManager.AttachTo</c> never subscribes to
    ///         <see cref="MqttConnection.ConnectionLost"/> — an unforwarded seam is this repo's
    ///         best-documented way to ship an inert fix. Here the edge comes from a genuine socket drop
    ///         through MQTTnet's own <c>DisconnectedAsync</c>.
    ///     </para>
    ///     <para>
    ///         <b>Vacuity rule:</b> the recovery payload is byte-identical to the pre-outage one. A test
    ///         that replays a <i>different</i> value passes against a build with no stale at all — the
    ///         value change alone would defeat the suppression — and is not the guard.
    ///     </para>
    ///     <para>
    ///         Ingest is fed through <c>Subscriptions.HandleMessage</c>, the method MQTTnet's dispatcher
    ///         calls, because <see cref="MiniBroker"/> answers CONNECT, SUBSCRIBE and PINGREQ only: it
    ///         can drop a session (which is what this test needs from it) but cannot PUBLISH.
    ///     </para>
    /// </remarks>
    [Trait("Category", "SlowUnit")]
    [Fact]
    public async Task ABrokerDrop_MarksCachedRefsBadNoCommunication_AndTheIdenticalReplayRecovers()
    {
        const string rawPath = "Plant/Mqtt/dev1/Temp";
        const string topic = "factory/oven/temp";
        const uint good = 0x00000000u;
        const uint badNoCommunication = 0x80310000u;

        using var broker = new MiniBroker();
        var options = BrokerOptions(broker.Port) with
        {
            RawTags =
            [
                new RawTagEntry(
                    rawPath,
                    $$"""{"topic":"{{topic}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""",
                    WriteIdempotent: false),
            ],
        };
        await using var driver = new MqttDriver(options, "d", null);
        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
        await driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken);

        var raised = new List<DataValueSnapshot>();
        driver.OnDataChange += (_, e) => { lock (raised) raised.Add(e.Snapshot); };

        driver.Subscriptions.HandleMessage(topic, """{"value":21.5}"""u8.ToArray(), retained: false);
        lock (raised) raised.Count.ShouldBe(1, "the first value must publish");

        broker.DropAll();
        await WaitForAsync(
            () => { lock (raised) return raised.Count == 2; },
            "the broker drop to publish a BadNoCommunication snapshot");
        DataValueSnapshot staled;
        lock (raised) staled = raised[1];
        staled.StatusCode.ShouldBe(badNoCommunication);
        staled.Value.ShouldBe(21.5, "the last known value is retained through the outage");

        // The reconnect's retained replay — the IDENTICAL payload.
        driver.Subscriptions.HandleMessage(topic, """{"value":21.5}"""u8.ToArray(), retained: true);

        DataValueSnapshot recovered;
        lock (raised) { raised.Count.ShouldBe(3); recovered = raised[2]; }
        recovered.StatusCode.ShouldBe(good);
        recovered.Value.ShouldBe(21.5);
    }

    /// <summary>
    ///     <b>R06:M-1, end to end.</b> A SUBSCRIBE the broker never answers used to come back from
    ///     <see cref="MqttDriver.SubscribeAsync"/> as a live handle while the topic left every retry set, so
    ///     <c>DriverInstanceActor</c> never re-subscribed and the tag stayed dark behind a Healthy connection.
    ///     It must throw (the actor's reconcile then retries), and the retry against a broker that answers
    ///     must establish and deliver — the positive control.
    /// </summary>
    [Trait("Category", "SlowUnit")]
    [Fact]
    public async Task ASubscribeTheBrokerNeverAnswers_Throws_AndTheRetryEstablishes()
    {
        const string rawPath = "Plant/Mqtt/dev1/Temp";
        const string topic = "factory/oven/temp";

        using var broker = new MiniBroker { AnswerSubscribe = false };
        var options = BrokerOptions(broker.Port) with
        {
            RawTags =
            [
                new RawTagEntry(
                    rawPath,
                    $$"""{"topic":"{{topic}}","payloadFormat":"Scalar","dataType":"Float64"}""",
                    WriteIdempotent: false),
            ],
        };
        await using var driver = new MqttDriver(options, "d", null);
        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);

        await Should.ThrowAsync<InvalidOperationException>(() =>
            driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken));
        driver.GetHealth().State.ShouldBe(DriverState.Healthy, "the connection itself is fine — that was the trap");

        broker.AnswerSubscribe = true;
        var handle = await driver.SubscribeAsync([rawPath], TimeSpan.Zero, TestContext.Current.CancellationToken);

        var raised = new List<DataChangeEventArgs>();
        driver.OnDataChange += (_, e) => { lock (raised) raised.Add(e); };
        driver.Subscriptions.HandleMessage(topic, "7"u8.ToArray(), retained: false);
        lock (raised)
        {
            raised.ShouldHaveSingleItem().SubscriptionHandle.ShouldBe(handle);
            raised[0].Snapshot.StatusCode.ShouldBe(0u);
        }
    }

    /// <summary>
    ///     <b>R06:M-3.</b> A refused initial connect used to rethrow without disposing the
    ///     <see cref="MqttConnection"/> (and its <c>IMqttClient</c>) it had just built. Control: a
    ///     successful initialize leaves its connection live, so the seam is not reporting "disposed" for
    ///     everything.
    /// </summary>
    [Fact]
    public async Task InitializeAsync_RefusedConnect_DisposesTheConnectionItBuilt()
    {
        using var broker = new MiniBroker { ConnAckReturnCode = MiniBroker.ReturnCodeNotAuthorized };
        var options = BrokerOptions(broker.Port);
        await using var driver = new MqttDriver(options, "d", null);
        var built = new List<MqttConnection>();
        driver.ConnectionCreatedForTests = c => built.Add(c);

        await Should.ThrowAsync<Exception>(() => driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken));

        built.ShouldHaveSingleItem().State.ShouldBe(MqttConnectionState.Disposed);

        broker.ConnAckReturnCode = MiniBroker.ReturnCodeAccepted;
        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
        built.Count.ShouldBe(2);
        built[1].State.ShouldBe(MqttConnectionState.Connected);
    }

    /// <summary>
    ///     <b>R06:M-10 + the lifecycle-generation rule.</b> A shutdown that times out on the lifecycle gate
    ///     tears down ungated; the wedged initialize that held the gate must then NOT publish its freshly
    ///     connected session behind that teardown, and its late health write must not overwrite the
    ///     shutdown's. The positive controls: the wedged init really did reach the broker (AcceptedCount),
    ///     and a later initialize connects and reports Healthy.
    /// </summary>
    [Trait("Category", "SlowUnit")]
    [Fact]
    public async Task AWedgedInit_CompletingAfterATimedOutShutdown_PublishesNothing()
    {
        using var broker = new MiniBroker();
        var options = BrokerOptions(broker.Port) with { ConnectTimeoutSeconds = 1 };
        await using var driver = new MqttDriver(options, "d", null);

        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        driver.BeforeConnectHookForTests = async _ =>
        {
            entered.TrySetResult();
            await release.Task;
        };

        var initialize = Task.Run(() => driver.InitializeAsync(Json(options), CancellationToken.None));
        await entered.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);

        // Times out on the gate (1 s) and tears down ungated.
        await driver.ShutdownAsync(TestContext.Current.CancellationToken);
        driver.GetHealth().State.ShouldBe(DriverState.Unknown);

        driver.BeforeConnectHookForTests = null;
        release.SetResult();
        await Should.ThrowAsync<OperationCanceledException>(() => initialize);

        broker.AcceptedCount.ShouldBe(1, "the wedged initialize did connect — the control for the absence below");
        await WaitForAsync(() => broker.LiveConnections == 0, "the superseded session to be torn down");
        driver.GetHealth().State.ShouldBe(DriverState.Unknown, "a superseded init must not overwrite newer health");
        driver.GetHostStatuses().ShouldHaveSingleItem().State.ShouldBe(HostState.Unknown);

        await driver.InitializeAsync(Json(options), TestContext.Current.CancellationToken);
        await WaitForAsync(() => broker.LiveConnections == 1, "the fresh session");
        driver.GetHealth().State.ShouldBe(DriverState.Healthy);
    }

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