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

using System.Buffers.Binary;
using Google.Protobuf;
using Org.Eclipse.Tahu.Protobuf;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;

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

/// <summary>
///     lmxopcua#754 (R06:M-5): Sparkplug B <c>*Array</c> metrics are read into 1-D OPC UA array values. Every
///     test builds real Tahu protobuf payloads, encodes them to wire bytes, decodes them through the real
///     <see cref="SparkplugCodec"/> and feeds <see cref="SparkplugIngestor.Dispatch"/> — the method MQTTnet's
///     dispatcher thread reaches — so this is the driver's production ingest path end to end, short of a broker.
///     (The project's <c>MiniBroker</c> never forwards a PUBLISH, so there is no in-process broker to route
///     through; the live Mosquitto + simulator suite is <c>Driver.Mqtt.IntegrationTests</c>.)
/// </summary>
public sealed class SparkplugArrayIngestTests
{
    private const uint Good = 0x00000000u;
    private const uint BadDecodingError = 0x80070000u;
    private const uint BadTypeMismatch = 0x80740000u;

    private const string Group = "Plant1";
    private const string Node = "EdgeA";
    private const string Device = "Filler1";
    private const string ArrPath = "Plant/Mqtt/spb/Profile";
    private const ulong ArrAlias = 11UL;

    // ---- the value: a 1-D array of the tag's element type ----

    /// <summary>An unauthored dataType takes the birth's ELEMENT type, and the value is an int[] — not a byte[].</summary>
    [Fact]
    public async Task Int32Array_OnAnArrayTag_PublishesAnInt32Array_AtGood()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob());

        Birth(ing, TahuDataType.Int32Array, Int32s(-42, 0, 70000));

        var snapshot = seen.ShouldHaveSingleItem();
        snapshot.StatusCode.ShouldBe(Good);
        snapshot.Value.ShouldBeOfType<int[]>().ShouldBe(new[] { -42, 0, 70000 });
    }

    /// <summary>The DATA path (alias only, no datatype on the wire) decodes by the birth's binding, like a scalar.</summary>
    [Fact]
    public async Task Ddata_ByAlias_PublishesTheNewVector()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob());
        Birth(ing, TahuDataType.Int32Array, Int32s(1, 2));

        var data = new Payload { Seq = 2UL, Timestamp = 1721822405000UL };
        data.Metrics.Add(new Payload.Types.Metric { Alias = ArrAlias, BytesValue = ByteString.CopyFrom(Int32s(3, 4, 5)) });
        ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(data));

        seen.Count.ShouldBe(2);
        seen[1].StatusCode.ShouldBe(Good);
        seen[1].Value.ShouldBe(new[] { 3, 4, 5 });
        ing.Values.Read(ArrPath).Value.ShouldBe(new[] { 3, 4, 5 });
    }

    /// <summary>An authored element type wins over the birth's, element by element through the scalar coercion.</summary>
    [Fact]
    public async Task AnAuthoredElementType_Wins_OverTheBirth()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob(dataType: "Float64"));

        Birth(ing, TahuDataType.Int32Array, Int32s(-1, 2));

        seen.ShouldHaveSingleItem().Value.ShouldBeOfType<double[]>().ShouldBe(new[] { -1d, 2d });
    }

    /// <summary>Int8 widens to Int16, exactly as the scalar map does (no OPC UA SByte in DriverDataType).</summary>
    [Fact]
    public async Task Int8Array_WidensToAnInt16Array()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob());

        Birth(ing, TahuDataType.Int8Array, [0xE9, 0x7B]);

        seen.ShouldHaveSingleItem().Value.ShouldBeOfType<short[]>().ShouldBe(new short[] { -23, 123 });
    }

    [Fact]
    public async Task BooleanArray_PublishesABoolArray()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob());

        Birth(ing, TahuDataType.BooleanArray, [0x03, 0x00, 0x00, 0x00, 0xA0]);

        seen.ShouldHaveSingleItem().Value.ShouldBeOfType<bool[]>().ShouldBe(new[] { true, false, true });
    }

    [Fact]
    public async Task StringArray_PublishesAStringArray()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob());

        Birth(ing, TahuDataType.StringArray, [0x41, 0x42, 0x43, 0x00, 0x00, 0x7A, 0x00]);

        seen.ShouldHaveSingleItem().Value.ShouldBeOfType<string[]>().ShouldBe(new[] { "ABC", "", "z" });
    }

    [Fact]
    public async Task DateTimeArray_PublishesUtcDateTimes()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob());
        var bytes = new byte[8];
        BinaryPrimitives.WriteInt64LittleEndian(bytes, 1256102875335L);

        Birth(ing, TahuDataType.DateTimeArray, bytes);

        var value = seen.ShouldHaveSingleItem().Value.ShouldBeOfType<DateTime[]>().ShouldHaveSingleItem();
        value.ShouldBe(DateTimeOffset.FromUnixTimeMilliseconds(1256102875335L).UtcDateTime);
        value.Kind.ShouldBe(DateTimeKind.Utc);
    }

    // ---- a declared arrayLength is the node's ArrayDimensions ----

    [Fact]
    public async Task AMatchingDeclaredLength_IsServed()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob(arrayLength: 3));

        Birth(ing, TahuDataType.Int32Array, Int32s(1, 2, 3));

        seen.ShouldHaveSingleItem().StatusCode.ShouldBe(Good);
    }

    /// <summary>A vector that is not the node's declared length is not a value of that node (MTConnect's #627 verdict).</summary>
    [Fact]
    public async Task AVectorOfAnotherLength_IsBadDecodingError_NotServedUnderTheWrongDimensions()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob(arrayLength: 4));

        Birth(ing, TahuDataType.Int32Array, Int32s(1, 2, 3));

        var snapshot = seen.ShouldHaveSingleItem();
        snapshot.StatusCode.ShouldBe(BadDecodingError);
        snapshot.Value.ShouldBeNull();
    }

    // ---- refusals: all-or-nothing, and never the packed buffer ----

    [Fact]
    public async Task AMalformedBody_IsBadDecodingError()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob());

        Birth(ing, TahuDataType.Int32Array, [1, 2, 3, 4, 5]);

        var snapshot = seen.ShouldHaveSingleItem();
        snapshot.StatusCode.ShouldBe(BadDecodingError);
        snapshot.Value.ShouldBeNull();
    }

    /// <summary>One element that does not fit the effective type refuses the WHOLE vector — never a partial array.</summary>
    [Fact]
    public async Task OneElementThatDoesNotFit_RefusesTheWholeVector()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob(dataType: "Int16"));

        Birth(ing, TahuDataType.Int32Array, Int32s(1, 70000));

        var snapshot = seen.ShouldHaveSingleItem();
        snapshot.StatusCode.ShouldBe(BadTypeMismatch);
        snapshot.Value.ShouldBeNull();
    }

    /// <summary>A SCALAR metric on an array tag: the shapes disagree, refused rather than wrapped in a one-element array.</summary>
    [Fact]
    public async Task AScalarMetric_OnAnArrayTag_IsBadTypeMismatch()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob());
        Nbirth(ing);

        var birth = new Payload { Seq = 1UL, Timestamp = 1UL };
        birth.Metrics.Add(new Payload.Types.Metric
        {
            Name = "Profile", Alias = ArrAlias, Datatype = (uint)TahuDataType.Int32, IntValue = 5,
        });
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Decode(birth));

        var snapshot = seen.ShouldHaveSingleItem();
        snapshot.StatusCode.ShouldBe(BadTypeMismatch);
        snapshot.Value.ShouldBeNull();
    }

    /// <summary>A publisher that puts a scalar oneof on an array-typed metric: there is no packed body to decode.</summary>
    [Fact]
    public async Task ANonBytesValue_OnAnArrayMetric_IsBadTypeMismatch()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob());
        Nbirth(ing);

        var birth = new Payload { Seq = 1UL, Timestamp = 1UL };
        birth.Metrics.Add(new Payload.Types.Metric
        {
            Name = "Profile", Alias = ArrAlias, Datatype = (uint)TahuDataType.Int32Array, IntValue = 5,
        });
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Decode(birth));

        seen.ShouldHaveSingleItem().StatusCode.ShouldBe(BadTypeMismatch);
    }

    [Fact]
    public async Task AnExplicitNull_OnAnArrayMetric_IsBadTypeMismatch()
    {
        var (ing, seen) = await SubscribedAsync(ArrayBlob());
        Nbirth(ing);

        var birth = new Payload { Seq = 1UL, Timestamp = 1UL };
        birth.Metrics.Add(new Payload.Types.Metric
        {
            Name = "Profile", Alias = ArrAlias, Datatype = (uint)TahuDataType.Int32Array, IsNull = true,
        });
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Decode(birth));

        seen.ShouldHaveSingleItem().StatusCode.ShouldBe(BadTypeMismatch);
    }

    /// <summary>
    ///     A template member that is itself an array re-enters the same publish funnel, so it is decoded too —
    ///     the structured path cannot silently lag the top-level one.
    /// </summary>
    [Fact]
    public async Task AnArrayTemplateMember_IsDecodedLikeATopLevelArray()
    {
        const string MemberPath = "Plant/Mqtt/spb/MotorProfile";
        var blob = $$"""{"groupId":"{{Group}}","edgeNodeId":"{{Node}}","deviceId":"{{Device}}","metricName":"Motor/Profile","isArray":true}""";
        var ing = NewIngestor([new RawTagEntry(MemberPath, blob, WriteIdempotent: false)]);
        await ing.SubscribeAsync([MemberPath], TimeSpan.Zero, TestContext.Current.CancellationToken);
        var seen = new List<DataValueSnapshot>();
        ing.OnDataChange += (_, e) => seen.Add(e.Snapshot);
        Nbirth(ing);

        var template = new Payload.Types.Template { IsDefinition = false };
        template.Metrics.Add(new Payload.Types.Metric
        {
            Name = "Profile", Datatype = (uint)TahuDataType.Int16Array, BytesValue = ByteString.CopyFrom(0x01, 0x00, 0xFF, 0xFF),
        });
        var birth = new Payload { Seq = 1UL, Timestamp = 1UL };
        birth.Metrics.Add(new Payload.Types.Metric
        {
            Name = "Motor", Alias = 30UL, Datatype = (uint)TahuDataType.Template, TemplateValue = template,
        });
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Decode(birth));

        var snapshot = seen.ShouldHaveSingleItem();
        snapshot.StatusCode.ShouldBe(Good);
        snapshot.Value.ShouldBeOfType<short[]>().ShouldBe(new short[] { 1, -1 });
    }

    // ---- helpers ----

    private static string ArrayBlob(string? dataType = null, uint? arrayLength = null)
    {
        var typeKey = dataType is null ? "" : $""","dataType":"{dataType}" """;
        var lengthKey = arrayLength is null ? "" : $""","arrayLength":{arrayLength}""";
        return $$"""{"groupId":"{{Group}}","edgeNodeId":"{{Node}}","deviceId":"{{Device}}","metricName":"Profile","isArray":true{{typeKey}}{{lengthKey}}}""";
    }

    private static async Task<(SparkplugIngestor Ingestor, List<DataValueSnapshot> Seen)> SubscribedAsync(string blob)
    {
        var ing = NewIngestor([new RawTagEntry(ArrPath, blob, WriteIdempotent: false)]);
        await ing.SubscribeAsync([ArrPath], TimeSpan.Zero, TestContext.Current.CancellationToken);
        var seen = new List<DataValueSnapshot>();
        ing.OnDataChange += (_, e) => seen.Add(e.Snapshot);
        return (ing, seen);
    }

    private static SparkplugIngestor NewIngestor(RawTagEntry[] tags)
    {
        var ingestor = new SparkplugIngestor(
            new MqttDriverOptions
            {
                Mode = MqttMode.SparkplugB,
                Sparkplug = new MqttSparkplugOptions { GroupId = Group, HostId = "otopcua-host-1" },
            },
            "mqtt-spb-1",
            null,
            null,
            logger: null,
            cache: null,
            maxPayloadBytes: MqttSubscriptionManager.DefaultMaxPayloadBytes,
            rebirthDebounce: TimeSpan.Zero);
        ingestor.Register(tags);
        return ingestor;
    }

    private static void Nbirth(SparkplugIngestor ing)
    {
        var payload = new Payload { Seq = 0UL, Timestamp = 1721822400000UL };
        payload.Metrics.Add(new Payload.Types.Metric { Name = "bdSeq", Datatype = (uint)TahuDataType.Uint64, LongValue = 1UL });
        ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Decode(payload));
    }

    /// <summary>NBIRTH, then a DBIRTH declaring <c>Profile</c> as <paramref name="type"/> carrying <paramref name="packed"/>.</summary>
    private static void Birth(SparkplugIngestor ing, TahuDataType type, byte[] packed)
    {
        Nbirth(ing);
        var birth = new Payload { Seq = 1UL, Timestamp = 1721822400000UL };
        birth.Metrics.Add(new Payload.Types.Metric
        {
            Name = "Profile", Alias = ArrAlias, Datatype = (uint)type, BytesValue = ByteString.CopyFrom(packed),
        });
        ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Decode(birth));
    }

    private static byte[] Int32s(params int[] values)
    {
        var bytes = new byte[values.Length * 4];
        for (var i = 0; i < values.Length; i++)
        {
            BinaryPrimitives.WriteInt32LittleEndian(bytes.AsSpan(i * 4), values[i]);
        }

        return bytes;
    }

    private static SparkplugTopic Topic(SparkplugMessageType type, string? device = null) => new(type, Group, Node, device, null);

    /// <summary>Through the wire: encode the generated message, then decode it with the production codec.</summary>
    private static SparkplugPayload Decode(Payload payload) => SparkplugCodec.Decode(payload.ToByteArray());
}