ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/TransportGateTests.cs

using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Transport;

namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests;

/// <summary>
/// Verifies the static <c>TransportGate</c> single-flight guard (arch-review S4):
/// transport bundle commands (export/preview/import) queue rather than run
/// concurrently, because each import holds ~4 concurrent whole-artifact copies of
/// the payload and two concurrent large imports on one central node are a
/// realistic OOM path. The gate is exercised through the real import handler so
/// the test tracks the actual concurrency shape of <see cref="IBundleImporter"/>.
/// </summary>
public class TransportGateTests : TestKit, IDisposable
{
    private readonly IBundleImporter _importer;
    private readonly ServiceCollection _services;

    public TransportGateTests()
    {
        _importer = Substitute.For<IBundleImporter>();
        _services = new ServiceCollection();
        _services.AddScoped(_ => _importer);
    }

    private IActorRef CreateActor()
    {
        var sp = _services.BuildServiceProvider();
        return Sys.ActorOf(Props.Create(() => new ManagementActor(
            sp, NullLogger<ManagementActor>.Instance)));
    }

    private static ManagementEnvelope Envelope(object command) =>
        new(new AuthenticatedUser("admin", "Admin", new[] { "Administrator" }, Array.Empty<string>()),
            command, Guid.NewGuid().ToString("N"));

    void IDisposable.Dispose() => Shutdown();

    [Fact]
    public async Task ImportBundle_ConcurrentCommands_AreSingleFlighted()
    {
        // LoadAsync blocks on this TCS so the first import stays inside the gate.
        var release = new TaskCompletionSource<BundleSession>(
            TaskCreationOptions.RunContinuationsAsynchronously);
        var loadStarted = new TaskCompletionSource(
            TaskCreationOptions.RunContinuationsAsynchronously);
        var loadCalls = 0;

        var session = new BundleSession { SessionId = Guid.NewGuid() };
        _importer.LoadAsync(Arg.Any<Stream>(), Arg.Any<string?>(), Arg.Any<CancellationToken>(), Arg.Any<string?>())
            .Returns(_ =>
            {
                Interlocked.Increment(ref loadCalls);
                loadStarted.TrySetResult();
                return release.Task;
            });
        _importer.PreviewAsync(Arg.Any<Guid>(), Arg.Any<CancellationToken>())
            .Returns(new ImportPreview(session.SessionId, Array.Empty<ImportPreviewItem>()));
        _importer.ApplyAsync(
                Arg.Any<Guid>(), Arg.Any<IReadOnlyList<ImportResolution>>(),
                Arg.Any<string>(), Arg.Any<CancellationToken>(), Arg.Any<BundleNameMap?>())
            .Returns(new ImportResult(
                Guid.NewGuid(), 0, 0, 0, 0, Array.Empty<int>(), "corr"));

        var actor = CreateActor();
        // A trivially-valid base64 payload; the substitute importer never parses it.
        var bundle = Convert.ToBase64String(new byte[] { 1, 2, 3, 4 });

        actor.Tell(Envelope(new ImportBundleCommand(bundle, null, "skip")));
        actor.Tell(Envelope(new ImportBundleCommand(bundle, null, "skip")));

        // The first import must reach LoadAsync...
        await loadStarted.Task.WaitAsync(TimeSpan.FromSeconds(5));
        // ...and the second must be parked at the gate: give it ample time to
        // (wrongly) proceed, then assert it did not.
        await Task.Delay(750);
        Assert.Equal(1, Volatile.Read(ref loadCalls));

        // Release the first; the second should now proceed and both should answer.
        release.SetResult(session);

        ExpectMsg<ManagementSuccess>(TimeSpan.FromSeconds(5));
        ExpectMsg<ManagementSuccess>(TimeSpan.FromSeconds(5));
        AwaitAssert(() => Assert.Equal(2, Volatile.Read(ref loadCalls)),
            TimeSpan.FromSeconds(5));
    }
}