ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayBatchSeamTests.cs

using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;

namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;

/// <summary>
/// WP2.1b — the MxGateway adapter's batch subscribe seam and the alarm-stream union
/// filter. The plain-vs-supervisory advise choice itself lives inside
/// <c>RealMxGatewayClient</c> (it needs a live MXAccess session); its bounded-pipeline
/// mechanism is pinned separately by <see cref="BatchSeamPrimitiveTests"/>.
/// </summary>
[Collection("DataConnectionManagerActor")]
public class MxGatewayBatchSeamTests
{
    private static MxGatewayDataConnection NewAdapter(FakeMxGatewayClient fake) =>
        new(fake, NullLogger<MxGatewayDataConnection>.Instance);

    private static Dictionary<string, string> Details() => new()
    {
        ["Endpoint"] = "http://gw:5000",
        ["ApiKey"] = "key",
        ["ClientName"] = "client-a",
        ["WriteUserId"] = "0",
        ["ReadTimeoutMs"] = "2000",
    };

    [Fact]
    public async Task SubscribeBatch_IssuesExactlyOneBulkCallPerBatch()
    {
        var fake = new FakeMxGatewayClient();
        var adapter = NewAdapter(fake);
        await adapter.ConnectAsync(Details());

        var results = await adapter.SubscribeBatchAsync(
            ["A.x", "A.y", "A.z"], (_, _) => { });

        // ONE bulk RPC for the whole batch — replacing the historical AddItem + Advise
        // pair PER TAG (6 RPCs for these three tags).
        Assert.Single(fake.BulkSubscribeCalls);
        Assert.Equal(3, fake.BulkSubscribeCalls[0].Count);
        Assert.Equal(3, results.Count);
        Assert.All(results, r => Assert.True(r.Success));
        Assert.All(results, r => Assert.NotNull(r.SubscriptionId));
    }

    [Fact]
    public async Task SubscribeBatch_ReportsPerTagFailuresWithoutFailingTheBatch()
    {
        var fake = new FakeMxGatewayClient();
        fake.BulkSubscribeFailures.Add("A.y");
        var adapter = NewAdapter(fake);
        await adapter.ConnectAsync(Details());

        var results = await adapter.SubscribeBatchAsync(["A.x", "A.y"], (_, _) => { });

        Assert.True(results.Single(r => r.TagPath == "A.x").Success);
        var failed = results.Single(r => r.TagPath == "A.y");
        Assert.False(failed.Success);
        Assert.Null(failed.SubscriptionId);
        Assert.NotNull(failed.ErrorMessage);
    }

    [Fact]
    public async Task UnsubscribeBatch_ReleasesEveryIdInOneBulkCall()
    {
        var fake = new FakeMxGatewayClient();
        var adapter = NewAdapter(fake);
        await adapter.ConnectAsync(Details());

        var results = await adapter.SubscribeBatchAsync(["A.x", "A.y"], (_, _) => { });
        await adapter.UnsubscribeBatchAsync(results.Select(r => r.SubscriptionId!).ToList());

        Assert.Single(fake.BulkUnsubscribeCalls);
        Assert.Equal(2, fake.BulkUnsubscribeCalls[0].Count);
    }

    [Fact]
    public async Task AlarmStream_OpensOnTheLongestCommonPrefix_AndOnlyRestartsForAnEscapingSource()
    {
        var fake = new FakeMxGatewayClient();
        var adapter = NewAdapter(fake);
        await adapter.ConnectAsync(Details());

        void NoTransition(NativeAlarmTransition _) { }

        // First source → the stream opens scoped to it.
        await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, NoTransition);
        await WaitForPrefixCountAsync(fake, 1);
        Assert.Equal("Area1.Tank1", fake.AlarmStreamPrefixes[0]);

        // A sibling under the same prefix widens the union → one restart on "Area1.Tank".
        await adapter.SubscribeAlarmsAsync("Area1.Tank2", null, NoTransition);
        await WaitForPrefixCountAsync(fake, 2);
        Assert.Equal("Area1.Tank", fake.AlarmStreamPrefixes[1]);

        // A source ALREADY covered by the live prefix must NOT restart the stream.
        await adapter.SubscribeAlarmsAsync("Area1.Tank3.Sub", null, NoTransition);
        await Task.Delay(100);
        Assert.Equal(2, fake.AlarmStreamPrefixes.Count);

        // A source outside the prefix widens it again — here down to gateway-wide.
        await adapter.SubscribeAlarmsAsync("Zone9.Pump", null, NoTransition);
        await WaitForPrefixCountAsync(fake, 3);
        Assert.Null(fake.AlarmStreamPrefixes[2]); // "" → gateway-wide
    }

    [Fact]
    public async Task AlarmStream_UnsubscribeNeverRestartsTheStream()
    {
        var fake = new FakeMxGatewayClient();
        var adapter = NewAdapter(fake);
        await adapter.ConnectAsync(Details());

        var first = await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
        // Wait for the first stream to actually open before widening: the open runs on a
        // Task.Run whose token the restart cancels, so a restart racing an unstarted task
        // would leave the first prefix unrecorded.
        await WaitForPrefixCountAsync(fake, 1);
        await adapter.SubscribeAlarmsAsync("Area1.Tank2", null, _ => { });
        await WaitForPrefixCountAsync(fake, 2);

        // Dropping a source leaves the prefix too BROAD at worst — bandwidth, not
        // correctness — so no restart is issued.
        await adapter.UnsubscribeAlarmsAsync(first);
        await Task.Delay(100);
        Assert.Equal(2, fake.AlarmStreamPrefixes.Count);
    }

    private static async Task WaitForPrefixCountAsync(FakeMxGatewayClient fake, int expected)
    {
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
        while (DateTime.UtcNow < deadline)
        {
            lock (fake.AlarmStreamPrefixes)
            {
                if (fake.AlarmStreamPrefixes.Count >= expected)
                    return;
            }
            await Task.Delay(20);
        }

        Assert.Fail($"alarm stream was opened {fake.AlarmStreamPrefixes.Count} time(s), expected {expected}");
    }
}