using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Browsing;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser;

// See SparkplugTopicTests.cs for why this alias is redeclared locally in every consuming project
// rather than inherited: `SparkplugDataType` is a `global using` alias for the generated
// `Org.Eclipse.Tahu.Protobuf.DataType` (declared in Contracts/SparkplugDataType.cs), and `global using`
// scope does not cross a ProjectReference.
using SparkplugDataType = Org.Eclipse.Tahu.Protobuf.DataType;

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

/// <summary>
///     Covers the passive <c>#</c>-observation browse session and the browser that opens it. The
///     session is exercised through its test seam (<c>ObserveTopicForTest</c>) rather than a live
///     broker: the accumulating tree is the whole of the P1 logic, and the MQTTnet plumbing that
///     feeds it is a one-line handler.
/// </summary>
public sealed class MqttBrowseSessionTests
{
    // ---------------------------------------------------------------- tree shape

    [Fact]
    public async Task ExpandAsync_BuildsTopicSegmentTree_FromObservedTopics()
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("factory/line3/oven/temp");

        var root = await s.RootAsync(TestContext.Current.CancellationToken);
        root.ShouldContain(n => n.DisplayName == "factory");

        var lvl2 = await s.ExpandAsync("factory", TestContext.Current.CancellationToken);
        lvl2.ShouldContain(n => n.DisplayName == "line3");

        var lvl3 = await s.ExpandAsync("factory/line3", TestContext.Current.CancellationToken);
        lvl3.ShouldContain(n => n.DisplayName == "oven");

        var lvl4 = await s.ExpandAsync("factory/line3/oven", TestContext.Current.CancellationToken);
        var leaf = lvl4.ShouldHaveSingleItem();
        leaf.DisplayName.ShouldBe("temp");
        leaf.NodeId.ShouldBe("factory/line3/oven/temp");
        leaf.Kind.ShouldBe(BrowseNodeKind.Leaf);
        leaf.HasChildrenHint.ShouldBeFalse();
    }

    [Fact]
    public async Task ObservedTopics_AccumulateAcrossMessages_AndDoNotDuplicate()
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("factory/line3/oven/temp");
        s.ObserveTopicForTest("factory/line3/oven/temp");   // repeat — must not duplicate
        s.ObserveTopicForTest("factory/line3/oven/pressure");
        s.ObserveTopicForTest("factory/line4/mixer/rpm");
        s.ObserveTopicForTest("plant/power");

        var root = await s.RootAsync(TestContext.Current.CancellationToken);
        root.Select(n => n.DisplayName).ShouldBe(["factory", "plant"]);   // sorted, deduped

        var lines = await s.ExpandAsync("factory", TestContext.Current.CancellationToken);
        lines.Select(n => n.DisplayName).ShouldBe(["line3", "line4"]);

        var oven = await s.ExpandAsync("factory/line3/oven", TestContext.Current.CancellationToken);
        oven.Select(n => n.DisplayName).ShouldBe(["pressure", "temp"]);
    }

    [Fact]
    public async Task IntermediateNode_IsFolder_WithChildrenHint()
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("factory/line3/oven/temp");

        var root = await s.RootAsync(TestContext.Current.CancellationToken);
        var factory = root.ShouldHaveSingleItem();
        factory.Kind.ShouldBe(BrowseNodeKind.Folder);
        factory.HasChildrenHint.ShouldBeTrue();
    }

    [Fact]
    public async Task ExpandAsync_UnknownNode_ReturnsEmpty()
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("factory/line3");

        var children = await s.ExpandAsync("nope/not/here", TestContext.Current.CancellationToken);
        children.ShouldBeEmpty();
    }

    [Fact]
    public async Task RootAsync_BeforeAnyTraffic_IsEmpty()
    {
        // MQTT has no discovery protocol: an observation window that has seen nothing shows nothing.
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        (await s.RootAsync(TestContext.Current.CancellationToken)).ShouldBeEmpty();
    }

    // ---------------------------------------------------------------- read-only guarantee

    [Fact]
    public async Task BrowseCalls_PublishNothing()   // browse is read-only
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("factory/line3/oven/temp", "23.5");

        await s.RootAsync(TestContext.Current.CancellationToken);
        await s.ExpandAsync("factory", TestContext.Current.CancellationToken);
        await s.AttributesAsync("factory/line3/oven/temp", TestContext.Current.CancellationToken);

        s.PublishCountForTest.ShouldBe(0);
    }

    // ---------------------------------------------------------------- attributes

    [Theory]
    [InlineData("23.5", nameof(Core.Abstractions.DriverDataType.Float64))]
    [InlineData("42", nameof(Core.Abstractions.DriverDataType.Int64))]
    [InlineData("true", nameof(Core.Abstractions.DriverDataType.Boolean))]
    [InlineData("{\"v\":1}", nameof(Core.Abstractions.DriverDataType.String))]
    public async Task AttributesAsync_InfersTypeFromLastPayload(string payload, string expectedType)
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("factory/oven/temp", payload);

        var attrs = await s.AttributesAsync("factory/oven/temp", TestContext.Current.CancellationToken);
        var a = attrs.ShouldHaveSingleItem();
        a.Name.ShouldBe("temp");
        a.DriverDataType.ShouldBe(expectedType);
        a.IsArray.ShouldBeFalse();
        a.IsAlarm.ShouldBeFalse();
        a.SecurityClass.ShouldContain(payload);   // the last-payload snippet
    }

    [Fact]
    public async Task AttributesAsync_LongPayload_IsTruncatedToASnippet()
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("t/long", new string('x', 500));

        var a = (await s.AttributesAsync("t/long", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
        a.SecurityClass.Length.ShouldBeLessThanOrEqualTo(MqttBrowseSession.PayloadSnippetMaxChars + 1);
    }

    [Fact]
    public async Task AttributesAsync_UnobservedFolder_ReturnsEmpty()
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("factory/oven/temp", "1");

        // "factory" is a synthesised path segment — nothing was ever published to it.
        (await s.AttributesAsync("factory", TestContext.Current.CancellationToken)).ShouldBeEmpty();
        (await s.AttributesAsync("no/such/node", TestContext.Current.CancellationToken)).ShouldBeEmpty();
    }

    // ---------------------------------------------------------------- concurrency

    [Fact]
    public async Task ObserveDuringBrowse_IsThreadSafe()
    {
        // Messages arrive on MQTTnet's dispatcher thread while the picker calls ExpandAsync.
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        using var stop = new CancellationTokenSource();

        var observed = 0;
        var writer = Task.Run(() =>
        {
            var i = 0;
            while (!stop.Token.IsCancellationRequested)
            {
                s.ObserveTopicForTest($"factory/line{i % 8}/dev{i % 32}/m{i}", i.ToString());
                Interlocked.Increment(ref observed);
                i++;
            }
        }, CancellationToken.None);

        // The read loop is pure in-memory work and would otherwise finish before the writer thread
        // is even scheduled, leaving the two never actually overlapping.
        while (Volatile.Read(ref observed) == 0) await Task.Yield();

        for (var i = 0; i < 400; i++)
        {
            await s.RootAsync(TestContext.Current.CancellationToken);
            await s.ExpandAsync("factory", TestContext.Current.CancellationToken);
            await s.ExpandAsync("factory/line3", TestContext.Current.CancellationToken);
        }

        await stop.CancelAsync();
        await writer;   // rethrows anything the observer thread hit

        (await s.RootAsync(TestContext.Current.CancellationToken)).ShouldNotBeEmpty();
    }

    // ---------------------------------------------------------------- lifetime

    [Fact]
    public async Task LastUsedUtc_AdvancesOnEveryCall()
    {
        var s = new MqttBrowseSession(MqttMode.Plain);
        var before = s.LastUsedUtc;
        await Task.Delay(15, TestContext.Current.CancellationToken);

        await s.RootAsync(TestContext.Current.CancellationToken);
        s.LastUsedUtc.ShouldBeGreaterThan(before);

        var afterRoot = s.LastUsedUtc;
        await Task.Delay(15, TestContext.Current.CancellationToken);
        await s.ExpandAsync("x", TestContext.Current.CancellationToken);
        s.LastUsedUtc.ShouldBeGreaterThan(afterRoot);

        await s.DisposeAsync();
    }

    [Fact]
    public async Task DisposeAsync_OnNeverOpenedSession_IsIdempotentAndDoesNotThrow()
    {
        var s = new MqttBrowseSession(MqttMode.Plain);
        await Should.NotThrowAsync(async () => await s.DisposeAsync());
        await Should.NotThrowAsync(async () => await s.DisposeAsync());
    }

    [Fact]
    public async Task AfterDispose_BrowseCallsThrowObjectDisposed()
    {
        var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("a/b");
        await s.DisposeAsync();

        await Should.ThrowAsync<ObjectDisposedException>(
            async () => await s.RootAsync(TestContext.Current.CancellationToken));
        await Should.ThrowAsync<ObjectDisposedException>(
            async () => await s.ExpandAsync("a", TestContext.Current.CancellationToken));
        await Should.ThrowAsync<ObjectDisposedException>(
            async () => await s.AttributesAsync("a/b", TestContext.Current.CancellationToken));
    }

    [Fact]
    public async Task ObserveAfterDispose_IsIgnored()
    {
        // A message already in MQTTnet's dispatcher when the reaper disposes the session.
        var s = new MqttBrowseSession(MqttMode.Plain);
        await s.DisposeAsync();
        Should.NotThrow(() => s.ObserveTopicForTest("late/arrival", "1"));
    }

    // ---------------------------------------------------------------- node cap

    [Fact]
    public async Task NodeCap_StopsRecording_AndSurfacesATruncationMarker()
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain, nodeCap: 4);
        s.ObserveTopicForTest("a/b/c");        // 3 nodes
        s.ObserveTopicForTest("d/e/f");        // would take it past 4

        var root = await s.RootAsync(TestContext.Current.CancellationToken);
        root.ShouldContain(n => n.NodeId == MqttBrowseSession.TruncatedNodeId);
    }

    [Fact]
    public async Task OversizedTopic_IsRejectedWhole_AndSurfacesAMarker()
    {
        // A truncated topic is a DIFFERENT topic — the picker would commit it as a tag address.
        // Every segment here is 4 chars, well under MaxSegmentChars, so ONLY the whole-topic bound
        // can reject it — the segment guard must not be what makes this test pass.
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("good/topic", "1");

        var manyShortSegments = "huge/" + string.Join('/', Enumerable.Repeat("abc", MqttBrowseSession.MaxTopicChars / 4 + 8));
        manyShortSegments.Length.ShouldBeGreaterThan(MqttBrowseSession.MaxTopicChars);
        manyShortSegments.Split('/').ShouldAllBe(seg => seg.Length <= MqttBrowseSession.MaxSegmentChars);
        s.ObserveTopicForTest(manyShortSegments, "1");

        var root = await s.RootAsync(TestContext.Current.CancellationToken);
        root.ShouldContain(n => n.NodeId == MqttBrowseSession.OversizedNodeId);
        root.ShouldContain(n => n.DisplayName == "good");
        root.ShouldNotContain(n => n.DisplayName == "huge");   // no half-built path left behind
    }

    [Fact]
    public async Task OversizedSegment_IsRejectedWhole_EvenWhenTheTopicFits()
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("a/" + new string('y', MqttBrowseSession.MaxSegmentChars + 1) + "/b", "1");

        s.OversizedRejected.ShouldBeTrue();
        (await s.ExpandAsync("a", TestContext.Current.CancellationToken)).ShouldBeEmpty();
        var root = await s.RootAsync(TestContext.Current.CancellationToken);
        root.ShouldNotContain(n => n.DisplayName == "a");
    }

    [Fact]
    public async Task TopicAtExactlyTheCap_IsAccepted()
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest(new string('z', MqttBrowseSession.MaxSegmentChars), "1");

        s.OversizedRejected.ShouldBeFalse();
        (await s.RootAsync(TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
    }

    // ---------------------------------------------------------------- payload bounds

    [Fact]
    public async Task PayloadBeyondTheInspectionWindow_IsNeverExamined()
    {
        // The discriminating test for the byte cap: content past PayloadInspectMaxBytes cannot
        // influence the result at all. With an unbounded decode the trailing scalar would be found
        // and typed Int64, and would show up in the snippet.
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest(
            "t/tail",
            new string(' ', MqttBrowseSession.PayloadInspectMaxBytes + 64) + "42");

        var a = (await s.AttributesAsync("t/tail", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
        a.DriverDataType.ShouldBe(nameof(Core.Abstractions.DriverDataType.String));
        a.SecurityClass.ShouldNotContain("42");
    }

    [Fact]
    public async Task HugePayload_YieldsABoundedSnippetMarkedAsClipped()
    {
        // A '#' subscription receives multi-MB retained blobs; the node cap never engages for a
        // large payload on an already-known topic, so the decode itself must be bounded.
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("t/blob", new string('a', 5_000_000));

        var a = (await s.AttributesAsync("t/blob", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
        a.SecurityClass.Length.ShouldBeLessThanOrEqualTo(MqttBrowseSession.PayloadSnippetMaxChars + 1);
        a.SecurityClass.ShouldEndWith("…");
        a.DriverDataType.ShouldBe(nameof(Core.Abstractions.DriverDataType.String));
    }

    [Fact]
    public async Task ClippedMultiByteUtf8_IsNotMislabelledAsBinary()
    {
        // The inspection window can land mid-sequence; backing off to a whole sequence keeps a
        // perfectly good UTF-8 payload from being reported as binary.
        await using var s = new MqttBrowseSession(MqttMode.Plain);

        // 3 bytes per char, so 512 / 3 leaves a partial sequence at the window edge.
        s.ObserveTopicForTest("t/utf8", new string('☃', 4000));

        var a = (await s.AttributesAsync("t/utf8", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
        a.SecurityClass.ShouldStartWith("☃");
        a.SecurityClass.ShouldNotContain("binary");
        a.SecurityClass.ShouldNotContain("�");   // no replacement char from a split sequence
    }

    [Fact]
    public async Task BinaryPayload_IsReportedAsOpaque_WithItsTrueLength()
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        var binary = new byte[2048];
        Array.Fill(binary, (byte)0xC0);   // never legal UTF-8
        s.Observe("t/bin", binary);

        var a = (await s.AttributesAsync("t/bin", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
        a.SecurityClass.ShouldBe("(2048 bytes, binary)");   // full length, not the clipped window
    }

    [Fact]
    public async Task EmbeddedControlCharacters_AreScrubbedToSpaces()
    {
        // Trim() only strips the ends; an embedded newline would break the picker's single-line row.
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("t/multiline", "line one\nline two\ttabbed");

        var a = (await s.AttributesAsync("t/multiline", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
        a.SecurityClass.ShouldNotContain("\n");
        a.SecurityClass.ShouldNotContain("\t");
        a.SecurityClass.ShouldNotContain("");
        a.SecurityClass.ShouldStartWith("line one line two tabbed");
    }

    // ---------------------------------------------------------------- the browser

    [Fact]
    public void Ctor_IsConnectionFree_AndReportsTheMqttDriverType()
    {
        // The universal-browser CanBrowse pattern constructs a throwaway instance purely to ask
        // what driver type it handles — the ctor must never touch the network.
        var browser = new MqttDriverBrowser();
        browser.DriverType.ShouldBe("Mqtt");
    }

    [Theory]
    [InlineData(null, "#")]
    [InlineData("", "#")]
    [InlineData("   ", "#")]
    [InlineData("factory", "factory/#")]
    [InlineData("factory/", "factory/#")]
    [InlineData("factory/line3", "factory/line3/#")]
    public void BuildRootFilter_ScopesTheWildcardToTheConfiguredPrefix(string? prefix, string expected)
    {
        var opts = new MqttDriverOptions
        {
            Mode = MqttMode.Plain,
            Plain = prefix is null ? null : new MqttPlainOptions { TopicPrefix = prefix },
        };
        MqttDriverBrowser.BuildRootFilter(opts).ShouldBe(expected);
    }

    [Fact]
    public void BrowseClientIdSuffix_IsUnique_AndIsAppliedToTheConnectClientId()
    {
        // A broker disconnects an existing client when a new one CONNECTs with the same id —
        // browsing must never knock the running driver offline.
        var a = MqttDriverBrowser.BuildBrowseClientIdSuffix();
        var b = MqttDriverBrowser.BuildBrowseClientIdSuffix();
        a.ShouldNotBe(b);
        a.ShouldStartWith("-browse-");
        a.Length.ShouldBe("-browse-".Length + 8);

        var opts = new MqttDriverOptions { ClientId = "otopcua-node1" };
        MqttConnection.BuildClientOptions(opts, a).ClientId.ShouldBe("otopcua-node1" + a);
    }

    [Fact]
    public void BrowseConnect_ForcesACleanSession()
    {
        // A transient browse identity must not leave queued-message state behind on the broker.
        var opts = new MqttDriverOptions { CleanSession = false };
        MqttDriverBrowser.ToBrowseOptions(opts).CleanSession.ShouldBeTrue();
    }

    [Theory]
    [InlineData(1, 5)]      // clamped up
    [InlineData(15, 15)]
    [InlineData(600, 30)]   // clamped down
    public void OpenBudget_IsClampedTo5To30Seconds(int configured, int expected)
    {
        var opts = new MqttDriverOptions { ConnectTimeoutSeconds = configured };
        MqttDriverBrowser.OpenBudget(opts).ShouldBe(TimeSpan.FromSeconds(expected));
    }

    // ------------------------------------------------------- Sparkplug: the Last-Will landmine

    [Fact]
    public void MqttDriverOptions_CarryNothingWillShaped_SoABrowseSessionCanNeverFireAnNdeath()
    {
        // THE landmine (see the marked block on MqttDriverBrowser.ToBrowseOptions). A Sparkplug
        // NDEATH is a Last Will: it is published by the BROKER, not by us, so it is invisible to
        // PublishCountForTest — and any ungraceful end to a browse session would fire an NDEATH
        // under the plant's real edge-node identity. ToBrowseOptions is where a will must be
        // stripped; today there is nothing to strip, and this test is what makes that a checked
        // fact rather than a claim: adding a will-shaped property anywhere in the options graph
        // reddens it and forces the author to handle it in ToBrowseOptions.
        static IEnumerable<string> Names(Type t) => t.GetProperties().Select(p => p.Name);

        var suspects = Names(typeof(MqttDriverOptions))
            .Concat(Names(typeof(MqttSparkplugOptions)))
            .Concat(Names(typeof(MqttPlainOptions)))
            .Where(n => n.Contains("will", StringComparison.OrdinalIgnoreCase)
                        || n.Contains("testament", StringComparison.OrdinalIgnoreCase)
                        || n.Contains("death", StringComparison.OrdinalIgnoreCase))
            .ToList();

        suspects.ShouldBeEmpty();
    }

    // ------------------------------------------------------- Sparkplug: birth-driven tree

    [Fact]
    public async Task SparkplugTree_FillsFromObservedBirths_BrowseNeverPublishes()
    {
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "Temperature", SparkplugDataType.Float);

        var groups = await s.RootAsync(TestContext.Current.CancellationToken);
        groups.ShouldContain(n => n.DisplayName == "Plant1");

        var nodes = await s.ExpandAsync("Plant1", TestContext.Current.CancellationToken);
        nodes.ShouldContain(n => n.DisplayName == "EdgeA");

        var devices = await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken);
        var device = devices.ShouldHaveSingleItem();
        device.DisplayName.ShouldBe("Filler1");
        device.Kind.ShouldBe(BrowseNodeKind.Folder);

        var metrics = await s.ExpandAsync("Plant1/EdgeA/Filler1", TestContext.Current.CancellationToken);
        var metric = metrics.ShouldHaveSingleItem();
        metric.DisplayName.ShouldBe("Temperature");
        metric.Kind.ShouldBe(BrowseNodeKind.Leaf);
        metric.NodeId.ShouldBe("Plant1/EdgeA/Filler1::Temperature");

        await s.AttributesAsync(metric.NodeId, TestContext.Current.CancellationToken);

        s.PublishCountForTest.ShouldBe(0);   // passive — browsing a live plant broker publishes nothing
    }

    [Fact]
    public async Task SparkplugTree_NodeLevelMetrics_HangDirectlyUnderTheEdgeNode()
    {
        // An NBIRTH metric has no device. It is NOT parked under a synthesised device folder —
        // a fake segment would be indistinguishable from a real device of the same name, and the
        // authored address tuple carries deviceId = null for exactly these.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("Plant1", "EdgeA", device: null, "Node Control/Rebirth", SparkplugDataType.Boolean);
        s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "Temperature", SparkplugDataType.Float);

        var underEdge = await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken);
        underEdge.Select(n => n.DisplayName).ShouldBe(["Filler1", "Node Control/Rebirth"]);

        var device = underEdge.Single(n => n.DisplayName == "Filler1");
        device.Kind.ShouldBe(BrowseNodeKind.Folder);

        // A metric name may itself contain '/'; it stays ONE leaf (the metric name is the tag's
        // stable binding key, so a split would invite committing a partial name).
        var nodeMetric = underEdge.Single(n => n.DisplayName == "Node Control/Rebirth");
        nodeMetric.Kind.ShouldBe(BrowseNodeKind.Leaf);
        nodeMetric.NodeId.ShouldBe("Plant1/EdgeA::Node Control/Rebirth");
        (await s.ExpandAsync(nodeMetric.NodeId, TestContext.Current.CancellationToken)).ShouldBeEmpty();
    }

    [Fact]
    public async Task SparkplugTree_ADeviceAndANodeMetricSharingALabel_AreBothKept()
    {
        // The two are siblings under the edge node and CAN share a label. They are distinct nodes with
        // distinct ids, so neither may displace the other — a tree keyed by label would silently drop
        // whichever arrived second, and the picker would offer a folder where a metric should be.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("Plant1", "EdgeA", "Temp", "Value", SparkplugDataType.Float);  // device "Temp"
        s.ObserveBirthForTest("Plant1", "EdgeA", null, "Temp", SparkplugDataType.Float);     // metric "Temp"

        var children = await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken);
        children.Count.ShouldBe(2);
        children.ShouldContain(n => n.NodeId == "Plant1/EdgeA/Temp" && n.Kind == BrowseNodeKind.Folder);
        children.ShouldContain(n => n.NodeId == "Plant1/EdgeA::Temp" && n.Kind == BrowseNodeKind.Leaf);

        (await s.AttributesAsync("Plant1/EdgeA::Temp", TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem().Name.ShouldBe("Temp");
    }

    [Fact]
    public async Task SparkplugTree_EmptyBirth_StillRendersItsDeviceAsAFolder()
    {
        // A DBIRTH carrying no metrics leaves a childless device node; it must not masquerade as a
        // committable metric leaf.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveEmptyBirthForTest("Plant1", "EdgeA", "Filler1");

        var device = (await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem();
        device.DisplayName.ShouldBe("Filler1");
        device.Kind.ShouldBe(BrowseNodeKind.Folder);
    }

    [Fact]
    public async Task SparkplugObserve_NonBirthMessages_BuildNoTree()
    {
        // Only NBIRTH/DBIRTH are self-describing. A DDATA metric carries an alias and no name at
        // all, so anything built from one would be a tree of unnamed nodes.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveRawForTest("spBv1.0/Plant1/DDATA/EdgeA/Filler1", [0x08, 0x01]);
        s.ObserveRawForTest("spBv1.0/Plant1/NCMD/EdgeA", [0x08, 0x01]);
        s.ObserveRawForTest("spBv1.0/Plant1/NDEATH/EdgeA", [0x08, 0x01]);
        s.ObserveRawForTest("spBv1.0/STATE/host1", "ONLINE"u8.ToArray());
        s.ObserveRawForTest("some/plain/topic", "23.5"u8.ToArray());          // not Sparkplug at all

        (await s.RootAsync(TestContext.Current.CancellationToken)).ShouldBeEmpty();
        s.PublishCountForTest.ShouldBe(0);
    }

    [Fact]
    public void SparkplugObserve_GarbageBirthPayload_IsIgnoredAndNeverThrows()
    {
        // Runs on MQTTnet's dispatcher thread behind an unauthenticated firehose.
        var s = new MqttBrowseSession(MqttMode.SparkplugB);
        Should.NotThrow(() => s.ObserveRawForTest("spBv1.0/Plant1/NBIRTH/EdgeA", [0xFF, 0xFF, 0xFF, 0xFF]));
        Should.NotThrow(() => s.ObserveRawForTest("spBv1.0/Plant1/NBIRTH/EdgeA", []));
    }

    // ------------------------------------------------------- Sparkplug: AttributesAsync

    [Fact]
    public async Task SparkplugAttributes_AreBirthDerived_NotPayloadSniffed()
    {
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "Temperature", SparkplugDataType.Float);

        var a = (await s.AttributesAsync(
            "Plant1/EdgeA/Filler1::Temperature", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();

        a.Name.ShouldBe("Temperature");
        a.DriverDataType.ShouldBe(nameof(Core.Abstractions.DriverDataType.Float32));   // from the birth
        a.IsArray.ShouldBeFalse();
        a.SecurityClass.ShouldContain("DBIRTH");
        a.SecurityClass.ShouldContain("Float");
        // The plain-mode payload-snippet machinery must not run: a Sparkplug body is protobuf, and
        // sniffing it would report "(N bytes, binary)" for every metric in the plant.
        a.SecurityClass.ShouldNotContain("binary");
        a.SecurityClass.ShouldNotContain("bytes");
    }

    [Fact]
    public async Task SparkplugAttributes_ArrayMetric_ReportsElementTypePlusIsArray()
    {
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("Plant1", "EdgeA", null, "Profile", SparkplugDataType.Int32Array);

        var a = (await s.AttributesAsync(
            "Plant1/EdgeA::Profile", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
        a.DriverDataType.ShouldBe(nameof(Core.Abstractions.DriverDataType.Int32));   // element type
        a.IsArray.ShouldBeTrue();
    }

    [Fact]
    public async Task SparkplugAttributes_UnsupportedDataType_IsReportedNotDefaulted()
    {
        // ToDriverDataType() returns null for DataSet/Template/PropertySet/PropertySetList/Unknown.
        // A null must never be coerced into a guessed DriverDataType — the operator has to see that
        // the metric is not bindable.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("Plant1", "EdgeA", null, "Curve", SparkplugDataType.DataSet);

        var a = (await s.AttributesAsync(
            "Plant1/EdgeA::Curve", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();

        Enum.TryParse<Core.Abstractions.DriverDataType>(a.DriverDataType, ignoreCase: true, out _)
            .ShouldBeFalse();                       // not a guessed type — unmappable, and visibly so
        a.SecurityClass.ShouldContain("unsupported");
    }

    [Fact]
    public async Task SparkplugAttributes_StateTheBindingTuple_SoTheCommitNeverParsesTheNodeId()
    {
        // The address of a Sparkplug tag is (group, edgeNode, device?, metric) — NOT the node id. The
        // session already holds the decomposition (it parsed the birth topic), so it states it; the
        // AdminUI's browse-commit mapper reads these keys and never splits the id.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("OtOpcUaSim", "EdgeA", "Filler1", "Temperature", SparkplugDataType.Float);

        var a = (await s.AttributesAsync(
            "OtOpcUaSim/EdgeA/Filler1::Temperature", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();

        a.AddressFields.ShouldNotBeNull();
        a.AddressFields![MqttTagConfigKeys.GroupId].ShouldBe("OtOpcUaSim");
        a.AddressFields[MqttTagConfigKeys.EdgeNodeId].ShouldBe("EdgeA");
        a.AddressFields[MqttTagConfigKeys.DeviceId].ShouldBe("Filler1");
        a.AddressFields[MqttTagConfigKeys.MetricName].ShouldBe("Temperature");
    }

    [Fact]
    public async Task SparkplugAttributes_NodeLevelMetric_StatesNoDeviceId()
    {
        // An NBIRTH metric belongs to the edge node itself. An invented/blank device id would be a
        // device that does not exist — the same reason the tree hangs it directly under the edge node.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("OtOpcUaSim", "EdgeA", null, "Uptime", SparkplugDataType.Int64);

        var a = (await s.AttributesAsync(
            "OtOpcUaSim/EdgeA::Uptime", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();

        a.AddressFields.ShouldNotBeNull();
        a.AddressFields!.ContainsKey(MqttTagConfigKeys.DeviceId).ShouldBeFalse();
        a.AddressFields[MqttTagConfigKeys.EdgeNodeId].ShouldBe("EdgeA");
    }

    [Fact]
    public async Task SparkplugAttributes_MetricNameContainingASlash_IsStatedWhole()
    {
        // 'Node Control/Rebirth' is a canonical Sparkplug metric name. Its node id is
        // 'OtOpcUaSim/EdgeA::Node Control/Rebirth' — un-splittable back into the tuple, which is why the
        // tuple travels rather than the id.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("OtOpcUaSim", "EdgeA", null, "Node Control/Rebirth", SparkplugDataType.Boolean);

        var a = (await s.AttributesAsync(
            "OtOpcUaSim/EdgeA::Node Control/Rebirth", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();

        a.AddressFields.ShouldNotBeNull();
        a.AddressFields![MqttTagConfigKeys.MetricName].ShouldBe("Node Control/Rebirth");
        a.AddressFields[MqttTagConfigKeys.EdgeNodeId].ShouldBe("EdgeA");
    }

    [Fact]
    public async Task SparkplugAttributes_DeviceAndNodeLevelMetricsOfTheSameName_StateDifferentAddresses()
    {
        // The two are siblings sharing a label ('EdgeA/Temp' the device folder vs 'EdgeA::Temp' the
        // node-level metric). Anything that re-derived the address from the label would collapse them.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("OtOpcUaSim", "EdgeA", null, "Temp", SparkplugDataType.Float);
        s.ObserveBirthForTest("OtOpcUaSim", "EdgeA", "Filler1", "Temp", SparkplugDataType.Float);

        var nodeLevel = (await s.AttributesAsync(
            "OtOpcUaSim/EdgeA::Temp", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
        var deviceLevel = (await s.AttributesAsync(
            "OtOpcUaSim/EdgeA/Filler1::Temp", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();

        nodeLevel.AddressFields!.ContainsKey(MqttTagConfigKeys.DeviceId).ShouldBeFalse();
        deviceLevel.AddressFields![MqttTagConfigKeys.DeviceId].ShouldBe("Filler1");
    }

    [Fact]
    public async Task PlainAttributes_StateTheTopic_AsTheAddress()
    {
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("otopcua/fixture/oven/temp", "21.5");

        var a = (await s.AttributesAsync(
            "otopcua/fixture/oven/temp", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();

        a.AddressFields.ShouldNotBeNull();
        a.AddressFields![MqttTagConfigKeys.Topic].ShouldBe("otopcua/fixture/oven/temp");
        a.AddressFields.ContainsKey(MqttTagConfigKeys.MetricName).ShouldBeFalse();  // Plain has no metric
    }

    [Theory]
    [InlineData(MqttMode.SparkplugB, true)]
    [InlineData(MqttMode.Plain, false)]
    public void RebirthAvailable_IsSparkplugOnly(MqttMode mode, bool expected)
    {
        // The picker draws its Request-rebirth affordance off this. A plain window has no re-announce
        // action at all — RequestRebirthAsync refuses one — so offering the button would be a lie.
        new MqttBrowseSession(mode).RebirthAvailable.ShouldBe(expected);
    }

    [Fact]
    public async Task SparkplugAttributes_ForAFolderNode_AreEmpty()
    {
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "Temperature", SparkplugDataType.Float);

        (await s.AttributesAsync("Plant1", TestContext.Current.CancellationToken)).ShouldBeEmpty();
        (await s.AttributesAsync("Plant1/EdgeA", TestContext.Current.CancellationToken)).ShouldBeEmpty();
        (await s.AttributesAsync("Plant1/EdgeA/Filler1", TestContext.Current.CancellationToken)).ShouldBeEmpty();
        (await s.AttributesAsync("no/such/node", TestContext.Current.CancellationToken)).ShouldBeEmpty();
    }

    // ------------------------------------------------------- Sparkplug: the one sanctioned publish

    [Fact]
    public async Task RequestRebirthAsync_ScopedToNode_PublishesOneNcmd()
    {
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "T", SparkplugDataType.Float);

        var published = await s.RequestRebirthAsync("Plant1/EdgeA", TestContext.Current.CancellationToken);

        published.ShouldBe(1);
        s.PublishCountForTest.ShouldBe(1);
        s.PublishedTopicsForTest.ShouldBe(["spBv1.0/Plant1/NCMD/EdgeA"]);
    }

    [Fact]
    public async Task RequestRebirthAsync_ForANeverSeenEdgeNode_StillPublishes()
    {
        // A node that has never birthed is the PRIME rebirth target — refusing to address one
        // because the observation window never saw it would defeat the whole feature.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);

        (await s.RequestRebirthAsync("Plant1/EdgeGhost", TestContext.Current.CancellationToken)).ShouldBe(1);
        s.PublishedTopicsForTest.ShouldBe(["spBv1.0/Plant1/NCMD/EdgeGhost"]);
    }

    [Fact]
    public async Task RequestRebirthAsync_FromADeviceOrMetricNodeId_ResolvesUpToItsEdgeNode()
    {
        // NCMD is node-addressed; there is no device-scoped rebirth. Selecting a metric and
        // clicking Request-rebirth must address the metric's OWNING edge node — resolved from the
        // tuple stored on the node, never re-parsed out of the id string.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "T", SparkplugDataType.Float);

        (await s.RequestRebirthAsync("Plant1/EdgeA/Filler1", TestContext.Current.CancellationToken)).ShouldBe(1);
        (await s.RequestRebirthAsync("Plant1/EdgeA/Filler1::T", TestContext.Current.CancellationToken)).ShouldBe(1);

        s.PublishedTopicsForTest.ShouldBe(["spBv1.0/Plant1/NCMD/EdgeA", "spBv1.0/Plant1/NCMD/EdgeA"]);
    }

    [Fact]
    public async Task RequestRebirthAsync_GroupScope_PublishesExactlyOneNcmdPerObservedEdgeNode()
    {
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        s.ObserveBirthForTest("Plant1", "EdgeB", null, "T", SparkplugDataType.Float);
        s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "T", SparkplugDataType.Float);
        s.ObserveBirthForTest("Plant1", "EdgeA", "Filler2", "T", SparkplugDataType.Float);   // same node
        s.ObserveBirthForTest("Plant2", "EdgeZ", null, "T", SparkplugDataType.Float);        // other group

        var published = await s.RequestRebirthAsync("Plant1", TestContext.Current.CancellationToken);

        published.ShouldBe(2);
        s.PublishCountForTest.ShouldBe(2);
        s.PublishedTopicsForTest.ShouldBe(["spBv1.0/Plant1/NCMD/EdgeA", "spBv1.0/Plant1/NCMD/EdgeB"]);
    }

    [Fact]
    public async Task RequestRebirthAsync_GroupScope_BeyondTheCap_PublishesNothingAtAll()
    {
        // A 200-node group must not emit 200 NCMDs. The refusal is all-or-nothing: a partial fan-out
        // would leave the operator unable to say which half of the plant was asked to rebirth.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
        for (var i = 0; i <= MqttBrowseSession.MaxGroupRebirthNodes; i++)
            s.ObserveBirthForTest("Plant1", $"Edge{i:D3}", null, "T", SparkplugDataType.Float);

        var ex = await Should.ThrowAsync<InvalidOperationException>(
            async () => await s.RequestRebirthAsync("Plant1", TestContext.Current.CancellationToken));

        ex.Message.ShouldContain(MqttBrowseSession.MaxGroupRebirthNodes.ToString());
        s.PublishCountForTest.ShouldBe(0);
    }

    [Fact]
    public async Task RequestRebirthAsync_GroupScope_TruncatedBeforeAnyEdgeNode_PublishesNothing()
    {
        // The node cap can leave a group node with no edge node recorded beneath it. A group that
        // resolves to zero targets is refused rather than quietly succeeding with a count of 0 — the
        // operator clicked Request-rebirth and is entitled to know nothing was sent.
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB, nodeCap: 1);
        s.ObserveBirthForTest("Plant1", "EdgeA", null, "T", SparkplugDataType.Float);

        s.Truncated.ShouldBeTrue();
        (await s.ExpandAsync("Plant1", TestContext.Current.CancellationToken)).ShouldBeEmpty();

        await Should.ThrowAsync<InvalidOperationException>(
            async () => await s.RequestRebirthAsync("Plant1", TestContext.Current.CancellationToken));
        s.PublishCountForTest.ShouldBe(0);
    }

    [Theory]
    [InlineData("")]
    [InlineData("   ")]
    [InlineData("Plant1")]                       // a group this window has never observed
    [InlineData("Plant1/EdgeA/Filler1/Extra")]   // unknown, and not resolvable to an edge node
    public async Task RequestRebirthAsync_UnusableScope_PublishesNothing(string scope)
    {
        await using var s = new MqttBrowseSession(MqttMode.SparkplugB);

        await Should.ThrowAsync<ArgumentException>(
            async () => await s.RequestRebirthAsync(scope, TestContext.Current.CancellationToken));
        s.PublishCountForTest.ShouldBe(0);
    }

    [Fact]
    public async Task RequestRebirthAsync_InPlainMode_IsRefused()
    {
        // P1's guarantee is unconditional: a plain-topic browse window publishes nothing, ever.
        await using var s = new MqttBrowseSession(MqttMode.Plain);
        s.ObserveTopicForTest("factory/oven/temp", "1");

        await Should.ThrowAsync<NotSupportedException>(
            async () => await s.RequestRebirthAsync("Plant1/EdgeA", TestContext.Current.CancellationToken));
        s.PublishCountForTest.ShouldBe(0);
    }

    [Fact]
    public async Task RequestRebirthAsync_AfterDispose_Throws()
    {
        var s = new MqttBrowseSession(MqttMode.SparkplugB);
        await s.DisposeAsync();

        await Should.ThrowAsync<ObjectDisposedException>(
            async () => await s.RequestRebirthAsync("Plant1/EdgeA", TestContext.Current.CancellationToken));
        s.PublishCountForTest.ShouldBe(0);
    }

    // ------------------------------------------------------- Sparkplug: the browser

    [Fact]
    public async Task OpenAsync_SparkplugMode_NoLongerRefuses_ButStillReachesTheBroker()
    {
        // Task 23 unseals the Sparkplug branch. The open still has to CONNECT, so an unreachable
        // broker fails the way any other open does — what must NOT come back is the Task-10
        // NotSupportedException seal.
        var browser = new MqttDriverBrowser();
        const string cfg = """
            {"host":"broker.invalid","port":8883,"mode":"SparkplugB","connectTimeoutSeconds":5,
             "sparkplug":{"groupId":"Plant1"}}
            """;

        var ex = await Should.ThrowAsync<Exception>(
            async () => await browser.OpenAsync(cfg, TestContext.Current.CancellationToken));
        ex.ShouldNotBeOfType<NotSupportedException>();
    }

    [Theory]
    [InlineData("Plant1", "spBv1.0/Plant1/#")]
    [InlineData("", "spBv1.0/#")]        // group not authored yet — discover every group
    [InlineData("   ", "spBv1.0/#")]
    public void BuildRootFilter_SparkplugMode_ScopesToTheSparkplugNamespace(string groupId, string expected)
    {
        var opts = new MqttDriverOptions
        {
            Mode = MqttMode.SparkplugB,
            Sparkplug = new MqttSparkplugOptions { GroupId = groupId },
            Plain = new MqttPlainOptions { TopicPrefix = "ignored/in/sparkplug/mode" },
        };
        MqttDriverBrowser.BuildRootFilter(opts).ShouldBe(expected);
    }

    [Fact]
    public void BuildRootFilter_SparkplugMode_WithNoSparkplugSection_DiscoversEveryGroup()
    {
        var opts = new MqttDriverOptions { Mode = MqttMode.SparkplugB, Sparkplug = null };
        MqttDriverBrowser.BuildRootFilter(opts).ShouldBe("spBv1.0/#");
    }
}