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() { } + } }