Run a test as a streaming function, so an event stream can be asserted

docs/aws/testing.md

@@ -213,6 +213,67 @@ One behaviour does differ, because the transport does. `[LambdaWebTesting]` is t
 has nothing behind it to hand an unmatched path to, so a path with no route is a 404 from the host
 rather than a fall-through.
 
+### Testing a streaming function
+
+`ResponseMode` runs the test as a function URL in `RESPONSE_STREAM` invoke mode, which is the mode
+`[ServerSentEvents]` handlers have to be deployed in:
+
+```csharp
+[LambdaWebTesting(ResponseMode = LambdaResponseMode.Stream)]
+public class OrderStreamTests {
+
+    [HardenedTest]
+    public async Task TheEventsArriveAsFrames(ITestWebApp app) {
+        var response = await app.Get("/orders/live");
+
+        response.Assert.Ok();
+        Assert.Equal("text/event-stream", response.Headers[KnownHeaders.ContentType].ToString());
+    }
+}
+```
+
+A streamed invocation answers nothing through its output stream. It opens a Lambda response stream
+at the first byte and writes there, so the mode also registers `StreamedResponseCapture` over the
+runtime's stream factory, and the host builds the response from what that recorded: the prelude's
+status and headers, and the bytes.
+
+Take `IResponseStreamFactory` as a parameter to assert on the stream itself:
+
+```csharp
+[HardenedTest]
+public async Task TheInvocationStreams(ITestWebApp app, IResponseStreamFactory streams) {
+    await app.Get("/orders/live");
+
+    var capture = Assert.IsType<StreamedResponseCapture>(streams);
+
+    Assert.True(capture.Opened);
+    Assert.Equal(HttpStatusCode.OK, capture.Prelude!.StatusCode);
+}
+```
+
+**That assertion is the one worth writing**, because the frames are not a discriminator. A buffered
+invocation writes the same bytes - that is what the buffered-mode warning is about, every event
+delivered at the end rather than as it happens - so a test asserting only on the body passes with
+the mode ignored.
+
+An adapter with no caller holding a connection stays buffered under the same mode rather than
+failing, so an SQS function under `Stream` opens nothing and the host reads its envelope as usual.
+
+Sends are one at a time: the capture is reset before each request and read after it, so two requests
+issued concurrently from one test would interleave into one capture. `ITestWebApp` and a typed
+client both send sequentially.
+
+### The Lambda Test Tool cannot do this
+
+There is no equivalent by hand. The tool runs every function on one port and routes by a
+`/{FunctionName}` prefix, so `AWS_LAMBDA_RUNTIME_API` carries a path - and AWS's streaming client
+reads that variable as `host:port` and parses everything after the colon as a port number, which
+throws. Shortening it does not help either: the same client writes a request line with no function
+name in it, so the tool could not route the request. Both are in
+`Amazon.Lambda.RuntimeSupport`, and neither is reachable from here.
+
+Run locally in buffered mode, and test the streaming mode with `ResponseMode` above.
+
 ## Next
 
 - [Triggers](/guide/triggers): the façades, and what each source delivers

docs/guide/streaming.md

@@ -218,6 +218,11 @@ A buffered deployment accumulates the whole body before returning it, so a strea
 all at once at the end. An application with `[ServerSentEvents]` handlers deployed in buffered mode
 logs a warning at startup naming them. See [Response mode](/aws/lambda-web#response-mode).
 
+Streaming on Lambda is tested with `[LambdaWebTesting(ResponseMode = LambdaResponseMode.Stream)]`,
+which runs the real streamed invocation and hands the test what the response stream was given. It
+cannot be exercised by hand against the Lambda Test Tool, for two reasons that are both AWS's; see
+[Testing a streaming function](/aws/testing#testing-a-streaming-function).
+
 The Azure Functions worker hands the whole body to the host when the invocation returns, so a
 streamed response there arrives at the end whatever the handler does.
 

src/Clouds/Aws/Hardened.Aws.Lambda.Testing/LambdaWebHost.cs

@@ -4,10 +4,12 @@
 using Amazon.Lambda.Core;
 using DependencyModules.Testing.Attributes.Interfaces;
 using Hardened.Aws.Lambda.Runtime.Hosting;
+using Hardened.Aws.Lambda.Runtime.Streaming;
 using Hardened.Requests.Testing;
 using Hardened.Shared.Runtime.Application;
 using Hardened.Web.Testing;
 using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Primitives;
 using Hardened.Web.Runtime.Responses;
 
@@ -31,18 +33,83 @@ namespace Hardened.Aws.Lambda.Testing;
 /// </remarks>
 [AttributeUsage(AttributeTargets.Method | AttributeTargets.Class | AttributeTargets.Assembly, AllowMultiple = false)]
 public sealed class LambdaWebTestingAttribute : TestHostAttribute {
-    public override ITestHost CreateHost(ITestMethodContext testMethod, IServiceCollection services) =>
-        new LambdaWebHost();
+
+    /// <summary>
+    /// The response mode the function is deployed in. Buffered unless the test says otherwise.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <c>Stream</c> is what a function URL in <c>RESPONSE_STREAM</c> invoke mode runs as, and it is
+    /// the mode <c>[ServerSentEvents]</c> handlers have to be deployed in - buffered, every event
+    /// arrives when the invocation ends, or never if it times out first. So it is the mode an event
+    /// stream has to be tested in, and until this existed there was no way to ask for it: the host
+    /// read the invocation's output stream, and a streamed invocation writes nothing there.
+    /// </para>
+    /// <para>
+    /// Setting it registers <see cref="StreamedResponseCapture"/> over the runtime's stream factory
+    /// and amends the mode, so the test drives the same <c>Streamed</c> path a deployed function
+    /// takes and reads what it wrote.
+    /// </para>
+    /// </remarks>
+    /// <example>
+    /// <code>
+    /// [LambdaWebTesting(ResponseMode = LambdaResponseMode.Stream)]
+    /// public class OrderStreamTests {
+    ///     [HardenedTest]
+    ///     public async Task TheEventsArriveAsFrames(ITestWebApp app) {
+    ///         var response = await app.Get("/orders/live");
+    ///
+    ///         Assert.Equal(KnownContentType.EventStream, response.Headers[KnownHeaders.ContentType]);
+    ///     }
+    /// }
+    /// </code>
+    /// </example>
+    public LambdaResponseMode ResponseMode { get; set; } = LambdaResponseMode.Buffered;
+
+    public override ITestHost CreateHost(ITestMethodContext testMethod, IServiceCollection services) {
+        if (ResponseMode != LambdaResponseMode.Stream) {
+            return new LambdaWebHost();
+        }
+
+        var capture = new StreamedResponseCapture();
+
+        // Over the runtime module's registration rather than beside it. Test setup attributes run
+        // after the application's modules, so the last IResponseStreamFactory registered is the one
+        // resolved - removing the other is what makes that a statement rather than an ordering
+        // accident, the way [WebTesting] removes the resource-not-found handler it replaces.
+        services.RemoveAll<IResponseStreamFactory>();
+        services.AddSingleton<IResponseStreamFactory>(capture);
+        services.ConfigureLambdaResponseMode(mode => mode.Mode = LambdaResponseMode.Stream);
+
+        return new LambdaWebHost(capture);
+    }
 }
 
 /// <summary>
 /// API Gateway as a test host: a request in as a proxy event, a proxy response back out.
 /// </summary>
 public sealed class LambdaWebHost : ITestHost {
+    private readonly StreamedResponseCapture? _capture;
     private IServiceProvider? _provider;
     private ITestContainerSource? _source;
     private bool _started;
 
+    public LambdaWebHost() { }
+
+    /// <summary>
+    /// The streaming host: the same invocation, reading what the response stream was given rather
+    /// than the proxy envelope the buffered path writes.
+    /// </summary>
+    /// <remarks>
+    /// Internal because the capture has to be the one the attribute registered, and a host built
+    /// with a capture nothing resolves would read an empty response for every request.
+    /// <c>[LambdaWebTesting(ResponseMode = LambdaResponseMode.Stream)]</c> is how a test asks for
+    /// this.
+    /// </remarks>
+    internal LambdaWebHost(StreamedResponseCapture capture) {
+        _capture = capture;
+    }
+
     /// <summary>
     /// Terminal. API Gateway has nothing behind it to hand an unmatched path to, so a path with no
     /// route is a 404 here exactly as it is in a deployed function.
@@ -119,15 +186,52 @@ public async Task<TestWebResponse> SendAsync(
 
         var handler = provider.GetRequiredService<LambdaInvocationHandler>();
 
+        _capture?.Reset();
+
         using var input = new MemoryStream(Encoding.UTF8.GetBytes(Event(request)));
 
         var output = await handler.Invoke(input, new TestContext());
 
+        // A streamed invocation returns Stream.Null and wrote its answer to the response stream, so
+        // the capture is the response. An adapter that cannot stream stays buffered under the same
+        // mode, and that invocation opened nothing - so this asks what happened rather than assuming
+        // the mode decided it.
+        if (_capture is { Opened: true }) {
+            return new TestWebResponse(Streamed(_capture));
+        }
+
         using var proxy = JsonDocument.Parse(output);
 
         return new TestWebResponse(Response(proxy.RootElement));
     }
 
+    /// <summary>
+    /// The response a streamed invocation wrote: its prelude's status and headers, and the bytes
+    /// that went to the stream.
+    /// </summary>
+    /// <remarks>
+    /// A stream opened with no prelude is a function invoked through the Lambda API rather than a
+    /// function URL. There is no status on the wire in that shape at all, so 200 is what the bytes
+    /// arriving means.
+    /// </remarks>
+    private static TestExecutionResponse Streamed(StreamedResponseCapture capture) {
+        var response = new TestExecutionResponse(new MemoryStream(capture.Body, writable: false)) {
+            Status = (int?)capture.Prelude?.StatusCode ?? 200
+        };
+
+        if (capture.Prelude is { } prelude) {
+            foreach (var header in prelude.Headers) {
+                response.Headers[header.Key] = new StringValues(header.Value);
+            }
+
+            if (prelude.Cookies.Count > 0) {
+                response.Headers["Set-Cookie"] = new StringValues(prelude.Cookies.ToArray());
+            }
+        }
+
+        return response;
+    }
+
     /// <summary>
     /// The request as API Gateway would have delivered it.
     /// </summary>

src/Clouds/Aws/Hardened.Aws.Lambda.Testing/StreamedResponseCapture.cs

@@ -0,0 +1,72 @@
+using Amazon.Lambda.Core.ResponseStreaming;
+using Hardened.Aws.Lambda.Runtime.Streaming;
+
+namespace Hardened.Aws.Lambda.Testing;
+
+/// <summary>
+/// The stream a function in <c>RESPONSE_STREAM</c> mode wrote, kept so a test can read it.
+/// </summary>
+/// <remarks>
+/// <para>
+/// A streamed invocation answers nothing through its output stream - it opens a Lambda response
+/// stream at the first byte and writes there, and the bootstrap ignores what <c>Invoke</c> returns
+/// once one has been created. So a test host that only reads the output stream sees an empty
+/// response for every streamed request, which is why <c>[LambdaWebTesting]</c> could not run in
+/// stream mode at all.
+/// </para>
+/// <para>
+/// <see cref="IResponseStreamFactory"/> is the seam that makes this reachable.
+/// <c>LambdaResponseStreamFactory</c> is static and its setter is internal to the AWS packages, so
+/// nothing built on it can be driven from a test; the runtime opens every stream through the
+/// interface instead, and this is the test's implementation of it.
+/// </para>
+/// <para>
+/// <b>One invocation at a time.</b> <c>LambdaWebHost</c> resets this before each request and reads
+/// it after, so two requests sent concurrently from one test would interleave into one capture.
+/// Sequential sends - which is what <c>ITestWebApp</c> and a typed client both do - are unaffected.
+/// </para>
+/// </remarks>
+public sealed class StreamedResponseCapture : IResponseStreamFactory {
+    private MemoryStream _body = new();
+
+    /// <summary>The prelude the stream opened with, or null where nothing opened one.</summary>
+    public HttpResponseStreamPrelude? Prelude { get; private set; }
+
+    /// <summary>Whether this invocation opened a Lambda response stream at all.</summary>
+    public bool Opened => Prelude != null || PlainStreams > 0;
+
+    /// <summary>
+    /// Streams opened with no prelude, which is what a function invoked through the Lambda API
+    /// rather than a function URL gets.
+    /// </summary>
+    public int PlainStreams { get; private set; }
+
+    /// <summary>Every byte written to the stream this invocation opened.</summary>
+    public byte[] Body => _body.ToArray();
+
+    public Stream CreateStream() {
+        PlainStreams++;
+
+        return _body;
+    }
+
+    public Stream CreateHttpStream(HttpResponseStreamPrelude prelude) {
+        Prelude = prelude;
+
+        return _body;
+    }
+
+    /// <summary>
+    /// Clears what the previous invocation wrote.
+    /// </summary>
+    /// <remarks>
+    /// A new stream rather than <c>SetLength(0)</c>, because the runtime's <c>ResponseStream</c>
+    /// keeps whatever it was handed for the life of the invocation and a test asserting on the
+    /// previous body should not see it grow.
+    /// </remarks>
+    internal void Reset() {
+        _body = new MemoryStream();
+        Prelude = null;
+        PlainStreams = 0;
+    }
+}