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>
    [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 and PINGREQ and nothing else: 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);
    }

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