From d178b94d9cc481f83a6541254fc2bf30c8491eac Mon Sep 17 00:00:00 2001 From: Ian Johnson Date: Thu, 10 Sep 2026 13:05:31 -0400 Subject: [PATCH] Run a test as a streaming function, so an event stream can be asserted From the 0.32 trial: C-02, reframed. The reported symptom is that Lambda streaming mode dies on the first request against the AWS Lambda Test Tool, and both causes are AWS's own, confirmed by decompiling RuntimeSupport 2.2.0. RawStreamingHttpClient reads AWS_LAMBDA_RUNTIME_API as host:port and parses the rest as a port number, while the tool needs a function name in the path; and the same client writes a request line with no function name in it, so shortening the endpoint would leave the tool unable to route. Neither is reachable from here. What the finding was really about is that an event stream could not be checked in the mode it deploys in. That did not need those two to talk. The seam already existed - IResponseStreamFactory, which exists because AWS's own factory is static with an internal setter - and the testing library simply had no wiring for it or for the response mode, so an application could not run its own Lambda as a streaming function at all. [LambdaWebTesting(ResponseMode = LambdaResponseMode.Stream)] registers StreamedResponseCapture over the runtime's factory and amends the mode. A streamed invocation returns Stream.Null and writes to the response stream, so the host builds its answer from the prelude's status and headers and the bytes. That closes a hole the repository already named: ServerSentEventManifestTests says the manifest and the buffered-mode warning are each covered and "neither says the two meet". Now something does. The frames are not a discriminator. A buffered invocation writes identical bytes, which is what the buffered-mode warning is about, so three of the four new tests pass with the mode switched off - verified by switching it off. The test that takes IResponseStreamFactory and asserts a stream was opened is the one that fails, and the guide says so. Co-Authored-By: Claude Opus 5 (1M context) --- docs/aws/testing.md | 61 ++++++++++ docs/guide/streaming.md | 5 + .../LambdaWebHost.cs | 108 +++++++++++++++++- .../StreamedResponseCapture.cs | 72 ++++++++++++ .../StreamedEventStreamTests.cs | 108 ++++++++++++++++++ .../Hardened.Aws.Lambda.Testing.approved.txt | 11 ++ 6 files changed, 363 insertions(+), 2 deletions(-) create mode 100644 src/Clouds/Aws/Hardened.Aws.Lambda.Testing/StreamedResponseCapture.cs create mode 100644 src/Clouds/Aws/IntegrationTests/ApiGateway/Hardened.IntegrationTests.ApiGateway.SUT.Tests/StreamedEventStreamTests.cs diff --git a/docs/aws/testing.md b/docs/aws/testing.md index 69b5c662d..ef8e47b44 100644 --- a/docs/aws/testing.md +++ b/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(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 diff --git a/docs/guide/streaming.md b/docs/guide/streaming.md index 867d1d717..a4cb1c665 100644 --- a/docs/guide/streaming.md +++ b/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. diff --git a/src/Clouds/Aws/Hardened.Aws.Lambda.Testing/LambdaWebHost.cs b/src/Clouds/Aws/Hardened.Aws.Lambda.Testing/LambdaWebHost.cs index 4a52013e6..0d0880801 100644 --- a/src/Clouds/Aws/Hardened.Aws.Lambda.Testing/LambdaWebHost.cs +++ b/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; /// [AttributeUsage(AttributeTargets.Method | AttributeTargets.Class | AttributeTargets.Assembly, AllowMultiple = false)] public sealed class LambdaWebTestingAttribute : TestHostAttribute { - public override ITestHost CreateHost(ITestMethodContext testMethod, IServiceCollection services) => - new LambdaWebHost(); + + /// + /// The response mode the function is deployed in. Buffered unless the test says otherwise. + /// + /// + /// + /// Stream is what a function URL in RESPONSE_STREAM invoke mode runs as, and it is + /// the mode [ServerSentEvents] 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. + /// + /// + /// Setting it registers over the runtime's stream factory + /// and amends the mode, so the test drives the same Streamed path a deployed function + /// takes and reads what it wrote. + /// + /// + /// + /// + /// [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]); + /// } + /// } + /// + /// + 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(); + services.AddSingleton(capture); + services.ConfigureLambdaResponseMode(mode => mode.Mode = LambdaResponseMode.Stream); + + return new LambdaWebHost(capture); + } } /// /// API Gateway as a test host: a request in as a proxy event, a proxy response back out. /// public sealed class LambdaWebHost : ITestHost { + private readonly StreamedResponseCapture? _capture; private IServiceProvider? _provider; private ITestContainerSource? _source; private bool _started; + public LambdaWebHost() { } + + /// + /// The streaming host: the same invocation, reading what the response stream was given rather + /// than the proxy envelope the buffered path writes. + /// + /// + /// 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. + /// [LambdaWebTesting(ResponseMode = LambdaResponseMode.Stream)] is how a test asks for + /// this. + /// + internal LambdaWebHost(StreamedResponseCapture capture) { + _capture = capture; + } + /// /// 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 SendAsync( var handler = provider.GetRequiredService(); + _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)); } + /// + /// The response a streamed invocation wrote: its prelude's status and headers, and the bytes + /// that went to the stream. + /// + /// + /// 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. + /// + 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; + } + /// /// The request as API Gateway would have delivered it. /// diff --git a/src/Clouds/Aws/Hardened.Aws.Lambda.Testing/StreamedResponseCapture.cs b/src/Clouds/Aws/Hardened.Aws.Lambda.Testing/StreamedResponseCapture.cs new file mode 100644 index 000000000..7083fc3dc --- /dev/null +++ b/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; + +/// +/// The stream a function in RESPONSE_STREAM mode wrote, kept so a test can read it. +/// +/// +/// +/// 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 Invoke 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 [LambdaWebTesting] could not run in +/// stream mode at all. +/// +/// +/// is the seam that makes this reachable. +/// LambdaResponseStreamFactory 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. +/// +/// +/// One invocation at a time. LambdaWebHost 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 ITestWebApp and a typed client both do - are unaffected. +/// +/// +public sealed class StreamedResponseCapture : IResponseStreamFactory { + private MemoryStream _body = new(); + + /// The prelude the stream opened with, or null where nothing opened one. + public HttpResponseStreamPrelude? Prelude { get; private set; } + + /// Whether this invocation opened a Lambda response stream at all. + public bool Opened => Prelude != null || PlainStreams > 0; + + /// + /// Streams opened with no prelude, which is what a function invoked through the Lambda API + /// rather than a function URL gets. + /// + public int PlainStreams { get; private set; } + + /// Every byte written to the stream this invocation opened. + public byte[] Body => _body.ToArray(); + + public Stream CreateStream() { + PlainStreams++; + + return _body; + } + + public Stream CreateHttpStream(HttpResponseStreamPrelude prelude) { + Prelude = prelude; + + return _body; + } + + /// + /// Clears what the previous invocation wrote. + /// + /// + /// A new stream rather than SetLength(0), because the runtime's ResponseStream + /// keeps whatever it was handed for the life of the invocation and a test asserting on the + /// previous body should not see it grow. + /// + internal void Reset() { + _body = new MemoryStream(); + Prelude = null; + PlainStreams = 0; + } +} diff --git a/src/Clouds/Aws/IntegrationTests/ApiGateway/Hardened.IntegrationTests.ApiGateway.SUT.Tests/StreamedEventStreamTests.cs b/src/Clouds/Aws/IntegrationTests/ApiGateway/Hardened.IntegrationTests.ApiGateway.SUT.Tests/StreamedEventStreamTests.cs new file mode 100644 index 000000000..462d40a67 --- /dev/null +++ b/src/Clouds/Aws/IntegrationTests/ApiGateway/Hardened.IntegrationTests.ApiGateway.SUT.Tests/StreamedEventStreamTests.cs @@ -0,0 +1,108 @@ +using System.Net; +using System.Text; +using Hardened.Aws.Lambda.Runtime.Streaming; +using Hardened.Aws.Lambda.Testing; +using Hardened.Requests.Abstract.Headers; +using Hardened.Shared.Testing.Attributes; +using Hardened.Web.Testing; +using Xunit; + +namespace Hardened.IntegrationTests.ApiGateway.SUT.Tests; + +/// +/// An event stream handler through the mode it has to be deployed in. +/// +/// +/// +/// The half neither existing suite covered. The emitted manifest is checked by the web pipeline +/// fixtures and registered by ServerSentEventManifestTests; the warning for a stream handler +/// deployed buffered is checked by ServerSentEventsResponseModeStartupServiceTests. Neither +/// says the two meet - that an application carrying [ServerSentEvents], invoked as a +/// streaming function, actually writes event frames to the Lambda response stream. +/// +/// +/// It could not be written before: a streamed invocation answers nothing through its output stream, +/// and the host read only that. Nor can it be checked by hand - the Lambda Test Tool needs a +/// function name in AWS_LAMBDA_RUNTIME_API and AWS's streaming client reads that variable as +/// host:port and parses the rest as a port number, so the two cannot be used together at +/// all. This is the check that does not need them. +/// +/// +[LambdaWebTesting(ResponseMode = LambdaResponseMode.Stream)] +public class StreamedEventStreamTests { + + /// + /// That the invocation streamed at all, which nothing else here can tell you. + /// + /// + /// The frames are not the discriminator. A buffered invocation writes the same bytes - that is + /// exactly what the buffered-mode warning is about, every event delivered at the end instead of + /// as it happens - so a test asserting only on the body would pass with the mode ignored. The + /// stream having been opened, with a prelude, is what says which path ran. + /// + [HardenedTest] + public async Task TheInvocationOpensALambdaResponseStream( + ITestWebApp app, IResponseStreamFactory streams) { + await app.Get("/orders/live"); + + var capture = Assert.IsType(streams); + + Assert.True(capture.Opened, "the invocation wrote a proxy envelope rather than streaming"); + Assert.NotNull(capture.Prelude); + Assert.Equal(HttpStatusCode.OK, capture.Prelude!.StatusCode); + } + + [HardenedTest] + public async Task TheEventsArriveAsFramesOnTheResponseStream(ITestWebApp app) { + var response = await app.Get("/orders/live"); + + response.Assert.Ok(); + + Assert.Equal( + KnownContentType.EventStream, response.Headers[KnownHeaders.ContentType].ToString()); + + Assert.Equal( + "data: {\"id\":\"live-1\",\"quantity\":1}\n\n", + Body(response).ReplaceLineEndings("\n")); + } + + /// + /// The prelude carries the status, which is the streamed shape's only way to say one. + /// + /// + /// A buffered invocation puts the status in the proxy envelope it writes at the end. A streamed + /// one has committed to the wire by its first byte, so the status is decided when the stream + /// opens and nothing after can change it. + /// + [HardenedTest] + public async Task AConstrainedTokenStreamsTheSameWay(ITestWebApp app) { + var response = await app.Get("/orders/7/live"); + + response.Assert.Ok(); + + Assert.Contains("\"id\":\"7\"", Body(response)); + } + + /// + /// An ordinary handler under the same mode still answers normally. + /// + /// + /// The control. Stream mode is the function's, not the route's, so every handler in the + /// application runs under it - and one that returns a value rather than a sequence has to keep + /// working, or the mode would be unusable for any application that has both. + /// + [HardenedTest] + public async Task ANonStreamingHandlerStillAnswersUnderStreamMode(ITestWebApp app) { + var response = await app.Get("/orders/o-1"); + + response.Assert.Ok(); + + Assert.Contains("o-1", Body(response)); + } + + private static string Body(TestWebResponse response) { + response.Body.Position = 0; + + return new StreamReader(response.Body, Encoding.UTF8).ReadToEnd(); + } +} diff --git a/src/PublicApi/Hardened.PublicApi.Tests/Approved/Hardened.Aws.Lambda.Testing.approved.txt b/src/PublicApi/Hardened.PublicApi.Tests/Approved/Hardened.Aws.Lambda.Testing.approved.txt index b242aa7bb..c79526d41 100644 --- a/src/PublicApi/Hardened.PublicApi.Tests/Approved/Hardened.Aws.Lambda.Testing.approved.txt +++ b/src/PublicApi/Hardened.PublicApi.Tests/Approved/Hardened.Aws.Lambda.Testing.approved.txt @@ -30,6 +30,17 @@ namespace Hardened.Aws.Lambda.Testing public sealed class LambdaWebTestingAttribute : Hardened.Web.Testing.TestHostAttribute { public LambdaWebTestingAttribute() { } + public Hardened.Aws.Lambda.Runtime.Streaming.LambdaResponseMode ResponseMode { get; set; } public override Hardened.Web.Testing.ITestHost CreateHost(DependencyModules.Testing.Attributes.Interfaces.ITestMethodContext testMethod, Microsoft.Extensions.DependencyInjection.IServiceCollection services) { } } + public sealed class StreamedResponseCapture : Hardened.Aws.Lambda.Runtime.Streaming.IResponseStreamFactory + { + public StreamedResponseCapture() { } + public byte[] Body { get; } + public bool Opened { get; } + public int PlainStreams { get; } + public Amazon.Lambda.Core.ResponseStreaming.HttpResponseStreamPrelude? Prelude { get; } + public System.IO.Stream CreateHttpStream(Amazon.Lambda.Core.ResponseStreaming.HttpResponseStreamPrelude prelude) { } + public System.IO.Stream CreateStream() { } + } }