using System.Collections.Concurrent;
using System.Globalization;
using System.IO.Compression;
using System.Net;
using System.Net.Sockets;
using System.Text;
namespace AvParser.Infrastructure.Tests.Collecting;
/// One request as the server saw it.
/// Requested path.
/// Request headers, lower-cased keys.
/// When the request line was read.
/// When the response finished being written.
internal sealed record RecordedRequest(
string Path,
IReadOnlyDictionary Headers,
DateTimeOffset ArrivedUtc,
DateTimeOffset CompletedUtc
);
/// How the server should answer one path.
internal sealed record Reply
{
/// Status code to send.
public int Status { get; init; } = 200;
/// Body bytes. Ignored for statuses that carry none.
public byte[] Body { get; init; } = [];
/// Value for the Content-Type header, if any.
public string? ContentType { get; init; }
/// Extra headers, sent verbatim.
public IReadOnlyDictionary Headers { get; init; } =
new Dictionary(StringComparer.OrdinalIgnoreCase);
/// Send the body with chunked transfer encoding instead of a length.
public bool Chunked { get; init; }
/// Compress the body and declare Content-Encoding: gzip.
public bool Gzip { get; init; }
/// Declare this length instead of the real one — for testing truncation.
public long? DeclaredLength { get; init; }
/// Write only this many body bytes, then close the connection mid-transfer.
public int? TruncateAfter { get; init; }
/// Wait this long before sending the status line.
public TimeSpan HeaderDelay { get; init; }
/// Wait this long between body writes.
public TimeSpan BodyDelay { get; init; }
/// Bytes per body write, when dripping.
public int BodyChunkSize { get; init; } = int.MaxValue;
}
///
/// A deliberately badly behaved HTTP server on loopback.
///
///
///
/// Raw sockets rather than HttpListener or Kestrel, because most of what needs testing is
/// below the level either of those will let you reach: declaring four kilobytes and sending three
/// hundred bytes, dripping one byte at a time, or closing mid-chunk. A well-behaved server cannot
/// produce the failures a collector has to survive.
///
///
/// Port 0 lets the OS assign a free port, so test classes never collide and nothing has to be
/// reserved or cleaned up between runs.
///
///
internal sealed class LoopbackServer : IAsyncDisposable
{
private readonly TcpListener _listener;
private readonly CancellationTokenSource _shutdown = new();
private readonly ConcurrentDictionary> _routes = new(StringComparer.Ordinal);
private readonly ConcurrentBag _requests = [];
private readonly Task _acceptLoop;
public LoopbackServer()
{
_listener = new TcpListener(IPAddress.Loopback, 0);
_listener.Start();
var port = ((IPEndPoint)_listener.LocalEndpoint).Port;
BaseAddress = new Uri($"http://127.0.0.1:{port.ToString(CultureInfo.InvariantCulture)}/");
_acceptLoop = Task.Run(AcceptAsync);
}
/// Root of this server.
public Uri BaseAddress { get; }
/// Every request served, in no particular order.
public IReadOnlyCollection Requests => [.. _requests];
/// How many requests were served.
public int RequestCount => _requests.Count;
/// Registers a fixed reply for a path.
public Uri Map(string path, Reply reply) => Map(path, () => reply);
/// Registers a reply computed per request, for sequences.
public Uri Map(string path, Func reply)
{
_routes[path] = reply;
return new Uri(BaseAddress, path);
}
/// Registers a body served as-is with a 200.
public Uri MapBody(string path, byte[] body, string? contentType = null) =>
Map(path, new Reply { Body = body, ContentType = contentType });
///
public async ValueTask DisposeAsync()
{
await _shutdown.CancelAsync();
_listener.Stop();
try
{
await _acceptLoop;
}
catch (Exception ex) when (ex is OperationCanceledException or SocketException or ObjectDisposedException)
{
// Shutting down a listener out from under its accept loop is the normal way to stop it.
}
_shutdown.Dispose();
}
private async Task AcceptAsync()
{
while (!_shutdown.IsCancellationRequested)
{
TcpClient client;
try
{
client = await _listener.AcceptTcpClientAsync(_shutdown.Token);
}
catch (Exception ex) when (ex is OperationCanceledException or SocketException or ObjectDisposedException)
{
return;
}
_ = Task.Run(() => ServeAsync(client));
}
}
private async Task ServeAsync(TcpClient client)
{
using (client)
{
try
{
await using var stream = client.GetStream();
var (path, headers) = await ReadRequestAsync(stream);
var arrived = DateTimeOffset.UtcNow;
try
{
var reply = _routes.TryGetValue(path, out var factory)
? factory()
: new Reply { Status = 404, Body = "not found"u8.ToArray() };
if (reply.HeaderDelay > TimeSpan.Zero)
{
await Task.Delay(reply.HeaderDelay, _shutdown.Token);
}
await WriteReplyAsync(stream, reply);
}
finally
{
// Recorded even when the client hung up mid-response: several tests arrange
// exactly that, and they still need to know the request arrived.
_requests.Add(new RecordedRequest(path, headers, arrived, DateTimeOffset.UtcNow));
}
}
catch (Exception ex) when (ex is IOException or OperationCanceledException or SocketException)
{
// A client that hangs up mid-response is exactly what several tests arrange.
}
}
}
private static async Task<(string Path, Dictionary Headers)> ReadRequestAsync(NetworkStream stream)
{
var buffer = new List(1024);
var single = new byte[1];
while (buffer.Count < 16 * 1024)
{
var read = await stream.ReadAsync(single);
if (read == 0)
{
break;
}
buffer.Add(single[0]);
if (
buffer.Count >= 4
&& buffer[^4] == (byte)'\r'
&& buffer[^3] == (byte)'\n'
&& buffer[^2] == (byte)'\r'
&& buffer[^1] == (byte)'\n'
)
{
break;
}
}
var text = Encoding.ASCII.GetString([.. buffer]);
var lines = text.Split("\r\n", StringSplitOptions.RemoveEmptyEntries);
var path =
lines.Length > 0
? lines[0].Split(' ') is { Length: >= 2 } parts
? parts[1]
: "/"
: "/";
var headers = new Dictionary(StringComparer.OrdinalIgnoreCase);
foreach (var line in lines.Skip(1))
{
var colon = line.IndexOf(':', StringComparison.Ordinal);
if (colon > 0)
{
headers[line[..colon].Trim()] = line[(colon + 1)..].Trim();
}
}
return (path, headers);
}
private async Task WriteReplyAsync(NetworkStream stream, Reply reply)
{
var body = reply.Gzip ? Compress(reply.Body) : reply.Body;
var head = new StringBuilder();
head.Append(CultureInfo.InvariantCulture, $"HTTP/1.1 {reply.Status} {Describe(reply.Status)}\r\n");
head.Append("Connection: close\r\n");
if (reply.ContentType is not null)
{
head.Append(CultureInfo.InvariantCulture, $"Content-Type: {reply.ContentType}\r\n");
}
if (reply.Gzip)
{
head.Append("Content-Encoding: gzip\r\n");
}
foreach (var (name, value) in reply.Headers)
{
head.Append(CultureInfo.InvariantCulture, $"{name}: {value}\r\n");
}
if (reply.Chunked)
{
head.Append("Transfer-Encoding: chunked\r\n");
}
else
{
var declared = reply.DeclaredLength ?? body.Length;
head.Append(CultureInfo.InvariantCulture, $"Content-Length: {declared}\r\n");
}
head.Append("\r\n");
await stream.WriteAsync(Encoding.ASCII.GetBytes(head.ToString()), _shutdown.Token);
await stream.FlushAsync(_shutdown.Token);
var limit = Math.Min(reply.TruncateAfter ?? body.Length, body.Length);
var chunk = Math.Max(1, Math.Min(reply.BodyChunkSize, limit == 0 ? 1 : limit));
for (var offset = 0; offset < limit; offset += chunk)
{
var count = Math.Min(chunk, limit - offset);
if (reply.Chunked)
{
var header = count.ToString("x", CultureInfo.InvariantCulture) + "\r\n";
await stream.WriteAsync(Encoding.ASCII.GetBytes(header), _shutdown.Token);
}
await stream.WriteAsync(body.AsMemory(offset, count), _shutdown.Token);
if (reply.Chunked)
{
await stream.WriteAsync("\r\n"u8.ToArray(), _shutdown.Token);
}
await stream.FlushAsync(_shutdown.Token);
if (reply.BodyDelay > TimeSpan.Zero)
{
await Task.Delay(reply.BodyDelay, _shutdown.Token);
}
}
// A truncated reply deliberately omits the terminal chunk: the client must notice.
if (reply.Chunked && reply.TruncateAfter is null)
{
await stream.WriteAsync("0\r\n\r\n"u8.ToArray(), _shutdown.Token);
await stream.FlushAsync(_shutdown.Token);
}
}
private static byte[] Compress(byte[] body)
{
using var output = new MemoryStream();
using (var gzip = new GZipStream(output, CompressionLevel.Fastest, leaveOpen: true))
{
gzip.Write(body);
}
return output.ToArray();
}
private static string Describe(int status) =>
status switch
{
200 => "OK",
301 => "Moved Permanently",
302 => "Found",
403 => "Forbidden",
404 => "Not Found",
410 => "Gone",
429 => "Too Many Requests",
500 => "Internal Server Error",
503 => "Service Unavailable",
_ => "Status",
};
}