Test streams, uploads and time¶

/// <summary>Streaming calls exercised by the controllable test responses.</summary>
internal interface ITestingStreamingApi
{
    /// <summary>Reads people one at a time as the response body arrives.</summary>
    /// <param name="cancellationToken">The token that cancels the request and the body read.</param>
    /// <returns>The people parsed from the streamed body.</returns>
    [Get("/people/live")]
    IAsyncEnumerable<TestingPerson> WatchAsync(CancellationToken cancellationToken);

    /// <summary>Uploads people as JSON Lines, one serialized person per line.</summary>
    /// <param name="people">The people to upload; enumerated while the body is written.</param>
    /// <returns>A task that completes when the upload has been answered.</returns>
    [Post("/people/import")]
    Task ImportAsync([Body(BodySerializationMethod.JsonLines)] IEnumerable<TestingPerson> people);

    /// <summary>Uploads people as JSON Lines from an asynchronous producer, one serialized person per line as it arrives.</summary>
    /// <param name="people">The people to upload; enumerated once while the body is written.</param>
    /// <param name="cancellationToken">The token that flows to the producer and to every write.</param>
    /// <returns>A task that completes when the upload has been answered.</returns>
    [Post("/people/import-live")]
    Task ImportLiveAsync([Body(BodySerializationMethod.JsonLines)] IAsyncEnumerable<TestingPerson> people, CancellationToken cancellationToken);
}

StreamSource source = new StreamSource(StreamingContentFormat.JsonLines);
using StubHttp http = new StubHttp { { Route.Get("/people/live"), Reply.Stream(source) } };
ITestingStreamingApi api = http.CreateGeneratedClient<ITestingStreamingApi>("https://api.example.com", CreateSettings());
await using IAsyncEnumerator<TestingPerson> people = api.WatchAsync(CancellationToken.None).GetAsyncEnumerator();

source.Release(new TestingPerson(1, "Ada"));
bool gotFirst = await people.MoveNextAsync();
string firstName = people.Current.Name; // "Ada"

ValueTask<bool> second = people.MoveNextAsync();
bool secondReadyEarly = second.IsCompleted; // false: Grace has not been released yet

source.Release(new TestingPerson(2, "Grace"));
bool gotSecond = await second;
string secondName = people.Current.Name; // "Grace"

source.Complete();
bool gotThird = await people.MoveNextAsync(); // false: the source is complete
await source.Closed; // reaching the end disposed the response

StreamSource source = new StreamSource();
using StubHttp http = new StubHttp { { Route.Get("/people/live"), Reply.Stream(source) } };
ITestingStreamingApi api = http.CreateGeneratedClient<ITestingStreamingApi>("https://api.example.com", CreateSettings());
using CancellationTokenSource cancellation = new CancellationTokenSource();
await using IAsyncEnumerator<TestingPerson> people = api.WatchAsync(cancellation.Token).GetAsyncEnumerator();

source.Release(new TestingPerson(1, "Ada"));
bool gotFirst = await people.MoveNextAsync();
Task<bool> stalled = people.MoveNextAsync().AsTask();
bool stalledEarly = stalled.IsCompleted; // false: no second person has arrived

await cancellation.CancelAsync();
try
{
    await stalled; // throws OperationCanceledException
}
catch (OperationCanceledException error)
{
    Console.WriteLine(error.Message); // "A task was canceled."
}

await source.Closed; // the client disposed the response when the read was cancelled

StreamSource source = new StreamSource();
source.Release(new TestingPerson(1, "Ada"));
source.Disconnect();
using StubHttp http = new StubHttp { { Route.Get("/people/live"), Reply.Stream(source) } };
ITestingStreamingApi api = http.CreateGeneratedClient<ITestingStreamingApi>("https://api.example.com", CreateSettings());

List<string> names = [];
try
{
    await foreach (TestingPerson person in api.WatchAsync(CancellationToken.None))
    {
        names.Add(person.Name); // "Ada" arrives before the drop
    }
}
catch (HttpIOException error)
{
    Console.WriteLine(error.Message); // "Refit.Testing simulated disconnect. (ResponseEnded)"
}

StreamSource source = new StreamSource(StreamingContentFormat.ServerSentEvents);
using StubHttp http = new StubHttp { { Route.Get("/events"), Reply.Stream(source) } };
using HttpClient httpClient = new HttpClient(http, disposeHandler: false);
using CancellationTokenSource cancellation = new CancellationTokenSource();

HttpResponseMessage response = await httpClient.GetAsync(new Uri("https://api.example.com/events"), HttpCompletionOption.ResponseHeadersRead);
string? contentType = response.Content.Headers.ContentType?.MediaType; // "text/event-stream": headers arrived
Stream body = await response.Content.ReadAsStreamAsync();
Task<int> read = body.ReadAsync(new byte[64], cancellation.Token).AsTask();
bool readCompletedEarly = read.IsCompleted; // false: the body itself has not arrived

await cancellation.CancelAsync();
try
{
    await read; // never reached: the read is cancelled first
}
catch (OperationCanceledException error)
{
    Console.WriteLine(error.Message); // "A task was canceled."
}

bool closedBeforeDispose = source.IsClosed; // false: cancelling the read did not close the body
response.Dispose();
bool closedAfterDispose = source.IsClosed; // true: disposing the response closed the body

TestingPerson[] people = [new TestingPerson(1, "Ada"), new TestingPerson(2, "Grace")];
using StubHttp http = new StubHttp
{
    {
        Route.Get("/people/live"),
        Reply.JsonLines(people)
    },
    {
        Route.Get("/people/live"),
        Reply.ServerSentEvents(people)
    },
};
ITestingStreamingApi api = http.CreateGeneratedClient<ITestingStreamingApi>("https://api.example.com", CreateSettings());

List<string> jsonLinesNames = [];
await foreach (TestingPerson person in api.WatchAsync(CancellationToken.None))
{
    jsonLinesNames.Add(person.Name);
}

List<string> serverSentEventNames = [];
await foreach (TestingPerson person in api.WatchAsync(CancellationToken.None))
{
    serverSentEventNames.Add(person.Name); // Ada, Grace: same people, delivered as server-sent events this time
}

int pulled = 0;
IEnumerable<TestingPerson> UploadPeople()
{
    pulled++;
    yield return new TestingPerson(1, "Ada");
    pulled++;
    yield return new TestingPerson(2, "Grace");
}

int pulledBeforeReply = -1;
using StubHttp http = new StubHttp
{
    {
        Route.Post("/people/import"),
        Reply.From(async request =>
        {
            pulledBeforeReply = pulled;
            await request.Content!.ReadAsStringAsync(); // read the whole upload
            return new HttpResponseMessage(HttpStatusCode.Accepted);
        })
    },
};
http.RequestCapture = RequestCapture.None; // stops StubHttp from reading the body before the reply above does
ITestingStreamingApi api = http.CreateGeneratedClient<ITestingStreamingApi>("https://api.example.com", CreateSettings());

await api.ImportAsync(UploadPeople());
int pulledWithNoCapture = pulledBeforeReply; // 0: the handler left the whole upload for the reply code
int totalPulled = pulled; // 2
TestingPerson? recorded = await http.LastRequestBodyAsync<TestingPerson>(); // null: RequestCapture.None recorded nothing

pulled = 0; // the second half repeats the upload with RequestCapture.Full
http.RequestCapture = RequestCapture.Full; // buffers the whole upload before any reply code runs
http.Add(Route.Post("/people/import"), Reply.From(request =>
{
    pulledBeforeReply = pulled;
    return new HttpResponseMessage(HttpStatusCode.Accepted);
}));
await api.ImportAsync(UploadPeople());
int pulledWithFullCapture = pulledBeforeReply; // 2: everything was pulled before the reply ran

static async IAsyncEnumerable<TestingPerson> UploadPeopleAsync()
{
    yield return new TestingPerson(1, "Ada");
    yield return new TestingPerson(2, "Grace");
}

List<string> lines = [];
using StubHttp http = new StubHttp
{
    {
        Route.Post("/people/import-live"),
        Reply.From(async (request, cancellationToken) =>
        {
            await using Stream body = await request.Content!.ReadAsStreamAsync(cancellationToken);
            using StreamReader reader = new StreamReader(body);
            string? line;
            while ((line = await reader.ReadLineAsync(cancellationToken)) is not null)
            {
                lines.Add(line);
            }

            return new HttpResponseMessage(HttpStatusCode.Accepted);
        })
    },
};

// RequestCapture.None stops StubHttp from reading the body itself, so the responder above is the
// first (and only) code to read the uploaded stream.
http.RequestCapture = RequestCapture.None;
ITestingStreamingApi api = http.CreateGeneratedClient<ITestingStreamingApi>("https://api.example.com", CreateSettings());

await api.ImportLiveAsync(UploadPeopleAsync(), CancellationToken.None); // lines now has one JSON object per uploaded person

// Reading the body is what fills a bounded capture, so the reply echoes the uploaded person back.
StubResponse echo = Reply.From(static async request =>
{
    string json = await request.Content!.ReadAsStringAsync();
    return new HttpResponseMessage(HttpStatusCode.Created) { Content = new StringContent(json, Encoding.UTF8, "application/json") };
});

// Each route answers once, and this sample sends two requests.
using StubHttp http = new StubHttp
{
    { Route.Post("/people"), echo },
    { Route.Post("/people"), echo },
};
ITestingApi api = http.CreateGeneratedClient<ITestingApi>("https://api.example.com", CreateSettings());

http.RequestCapture = RequestCapture.Bounded(1024); // roomy enough for one serialized person
await api.CreateAsync(new TestingPerson(2, "Grace"));
TestingPerson? recorded = await http.LastRequestBodyAsync<TestingPerson>(); // recorded?.Name == "Grace"

http.RequestCapture = RequestCapture.Bounded(4); // far too small for one serialized person
await api.CreateAsync(new TestingPerson(2, "Grace"));
try
{
    await http.LastRequestBodyAsync<TestingPerson>(); // throws: the capture was truncated
}
catch (InvalidOperationException error)
{
    Console.WriteLine(error.Message); // "The request body exceeded the capture limit of 4 bytes; raise RequestCapture.Bounded or use RequestCapture.Full."
}

FakeTimeProvider clock = new FakeTimeProvider();
NetworkBehavior behavior = new NetworkBehavior(7) // 7 is the random seed, so the random choices repeat on every run
{
    Delay = TimeSpan.FromSeconds(2),
    Variance = 0,
    FailurePercent = 0,
};
using StubHttp http = new StubHttp(behavior) { { Route.Get("/slow"), Reply.Text("done") } };
http.TimeProvider = clock;
using HttpClient httpClient = new HttpClient(http, disposeHandler: false);

Task<HttpResponseMessage> pending = httpClient.GetAsync(new Uri("https://api.example.com/slow"));
clock.Advance(TimeSpan.FromSeconds(1));
bool stillWaiting = pending.IsCompleted; // false: one simulated second is still outstanding

clock.Advance(TimeSpan.FromSeconds(1));
using HttpResponseMessage response = await pending; // arrives once the fake clock has advanced the full 2-second delay
string body = await response.Content.ReadAsStringAsync(); // "done"

FakeTimeProvider clock = new FakeTimeProvider();
using StubHttp http = new StubHttp { { Route.Get("/expected"), Reply.Status(HttpStatusCode.OK) } };
http.TimeProvider = clock;

Task verification = http.VerifyAllCalledAsync(TimeSpan.FromSeconds(5));
bool doneEarly = verification.IsCompleted; // false: nothing has called /expected yet

clock.Advance(TimeSpan.FromSeconds(5));
try
{
    await verification; // throws once the simulated clock reaches the 5-second timeout
}
catch (InvalidOperationException error)
{
    Console.WriteLine(error.Message); // "1 expected request(s) were not made:\n  - GET /expected"
}