using System.Text; using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Http.HttpResults; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; using ZB.MOM.WW.ScadaBridge.Security.Auth; namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests; /// /// Review 2026-10-21 05/M2: bundle import/preview bodies used to be materialised as ~4 /// copies at the HTTP edge before the actor's TransportGate was reached. Large /// requests are now single-flight at the edge (decided before the body is read), and the /// command is parsed straight off the request stream. /// public class ManagementEndpointsLargeRequestTests { private static DefaultHttpContext Context(Stream body, long? contentLength) { var services = new ServiceCollection(); services.AddLogging(); services.AddSingleton>( Options.Create(new AuthDisableLoginOptions { DisableLogin = true })); services.AddSingleton(new ManagementActorHolder()); // no actor → 503 after a successful parse var context = new DefaultHttpContext { RequestServices = services.BuildServiceProvider() }; context.Request.Method = "POST"; context.Request.Body = body; context.Request.ContentLength = contentLength; return context; } private static MemoryStream Json(string json) => new(Encoding.UTF8.GetBytes(json)); private static int? StatusOf(IResult result) => (result as IStatusCodeHttpResult)?.StatusCode; /// A second large request is refused with 409 without reading its body. /// A task that represents the asynchronous operation. [Fact] public async Task LargeRequest_WhileAnotherIsInFlight_Returns409_WithoutReadingTheBody() { Assert.True(ManagementEndpoints.LargeRequestGate.Wait(0)); try { var body = new UnreadableStream(); var result = await ManagementEndpoints.HandleRequest( Context(body, ManagementEndpoints.LargeRequestThresholdBytes + 1)); Assert.Equal(409, StatusOf(result)); var json = Assert.IsAssignableFrom(result).Value!.ToString()!; Assert.Contains("IMPORT_IN_PROGRESS", json); Assert.False(body.WasRead); } finally { ManagementEndpoints.LargeRequestGate.Release(); } } /// A request of unknown length (chunked) counts as large. /// A task that represents the asynchronous operation. [Fact] public async Task UnknownLengthRequest_WhileAnotherIsInFlight_Returns409() { Assert.True(ManagementEndpoints.LargeRequestGate.Wait(0)); try { var result = await ManagementEndpoints.HandleRequest(Context(new UnreadableStream(), null)); Assert.Equal(409, StatusOf(result)); } finally { ManagementEndpoints.LargeRequestGate.Release(); } } /// Ordinary small commands are never blocked by an in-flight import. /// A task that represents the asynchronous operation. [Fact] public async Task SmallRequest_WhileLargeIsInFlight_IsNotGated() { Assert.True(ManagementEndpoints.LargeRequestGate.Wait(0)); try { var body = Json("""{"command":"ListSites","payload":{}}"""); var result = await ManagementEndpoints.HandleRequest(Context(body, body.Length)); Assert.Equal(503, StatusOf(result)); // parsed and reached dispatch } finally { ManagementEndpoints.LargeRequestGate.Release(); } } /// The gate is released once a large request completes. /// A task that represents the asynchronous operation. [Fact] public async Task LargeRequest_ReleasesTheGate_WhenDone() { var body = Json("""{"command":"ListSites"}"""); var result = await ManagementEndpoints.HandleRequest( Context(body, ManagementEndpoints.LargeRequestThresholdBytes + 1)); Assert.Equal(503, StatusOf(result)); Assert.True(ManagementEndpoints.LargeRequestGate.Wait(0)); ManagementEndpoints.LargeRequestGate.Release(); } /// The stream parse handles a payload that precedes the command name. /// A task that represents the asynchronous operation. [Fact] public async Task ParseCommandAsync_PayloadBeforeCommand_StillDeserializes() { var result = await ManagementEndpoints.ParseCommandAsync( Json("""{"payload":{"siteId":7},"command":"GetSite"}""")); Assert.True(result.Success, result.ErrorMessage); Assert.Equal(7, Assert.IsType(result.Command).SiteId); } /// The stream parse reports invalid JSON and bad payloads with the existing messages. /// A task that represents the asynchronous operation. [Fact] public async Task ParseCommandAsync_Errors_MatchTheStringParse() { var invalid = await ManagementEndpoints.ParseCommandAsync(Json("{ not json")); Assert.Equal("Invalid JSON body.", invalid.ErrorMessage); var badPayload = await ManagementEndpoints.ParseCommandAsync( Json("""{"command":"GetSite","payload":{"siteId":"not-a-number"}}""")); Assert.False(badPayload.Success); Assert.StartsWith("Failed to deserialize payload:", badPayload.ErrorMessage); var empty = await ManagementEndpoints.ParseCommandAsync(Json("""{"command":" "}""")); Assert.Equal("Empty 'command' field.", empty.ErrorMessage); } private sealed class UnreadableStream : Stream { public bool WasRead { get; private set; } public override bool CanRead => true; public override bool CanSeek => false; public override bool CanWrite => false; public override long Length => throw new NotSupportedException(); public override long Position { get => 0; set => throw new NotSupportedException(); } public override void Flush() { } public override int Read(byte[] buffer, int offset, int count) { WasRead = true; throw new InvalidOperationException("body read"); } public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) { WasRead = true; throw new InvalidOperationException("body read"); } public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); public override void SetLength(long value) => throw new NotSupportedException(); public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); } }