Joseph Doherty
94 lines
4.0 KiB
C#
94 lines
4.0 KiB
C#
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>())
|
|
.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));
|
|
}
|
|
}
|