natsdotnet/tests/NATS.Server.Tests/SubszTests.cs

using System.Net;
using System.Net.Http.Json;
using System.Net.Sockets;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Monitoring;

namespace NATS.Server.Tests;

public class SubszTests : IAsyncLifetime
{
    private readonly NatsServer _server;
    private readonly int _natsPort;
    private readonly int _monitorPort;
    private readonly CancellationTokenSource _cts = new();
    private readonly HttpClient _http = new();

    public SubszTests()
    {
        _natsPort = GetFreePort();
        _monitorPort = GetFreePort();
        _server = new NatsServer(
            new NatsOptions { Port = _natsPort, MonitorPort = _monitorPort },
            NullLoggerFactory.Instance);
    }

    public async Task InitializeAsync()
    {
        _ = _server.StartAsync(_cts.Token);
        await _server.WaitForReadyAsync();
        for (int i = 0; i < 50; i++)
        {
            try
            {
                var resp = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
                if (resp.IsSuccessStatusCode) break;
            }
            catch (HttpRequestException) { }
            await Task.Delay(50);
        }
    }

    public async Task DisposeAsync()
    {
        _http.Dispose();
        await _cts.CancelAsync();
        _server.Dispose();
    }

    [Fact]
    public async Task Subz_returns_empty_when_no_subscriptions()
    {
        var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/subz");
        response.StatusCode.ShouldBe(HttpStatusCode.OK);

        var subz = await response.Content.ReadFromJsonAsync<Subsz>();
        subz.ShouldNotBeNull();
        subz.NumSubs.ShouldBe(0u);
    }

    [Fact]
    public async Task Subz_returns_count_with_subscriptions()
    {
        using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
        using var stream = new NetworkStream(sock);
        var buf = new byte[4096];
        _ = await stream.ReadAsync(buf);
        await stream.WriteAsync("CONNECT {}\r\nSUB foo 1\r\nSUB bar 2\r\nSUB baz.* 3\r\n"u8.ToArray());
        await Task.Delay(200);

        var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/subz");
        var subz = await response.Content.ReadFromJsonAsync<Subsz>();
        subz.ShouldNotBeNull();
        subz.NumSubs.ShouldBeGreaterThanOrEqualTo(3u);
    }

    [Fact]
    public async Task Subz_subs_true_returns_subscription_details()
    {
        using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
        using var stream = new NetworkStream(sock);
        var buf = new byte[4096];
        _ = await stream.ReadAsync(buf);
        await stream.WriteAsync("CONNECT {}\r\nSUB foo 1\r\n"u8.ToArray());
        await Task.Delay(200);

        var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/subz?subs=true");
        var subz = await response.Content.ReadFromJsonAsync<Subsz>();
        subz.ShouldNotBeNull();
        subz.Subs.ShouldNotBeEmpty();
        subz.Subs.ShouldContain(s => s.Subject == "foo");
    }

    [Fact]
    public async Task Subz_test_subject_filters_matching_subs()
    {
        using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
        using var stream = new NetworkStream(sock);
        var buf = new byte[4096];
        _ = await stream.ReadAsync(buf);
        await stream.WriteAsync("CONNECT {}\r\nSUB foo.* 1\r\nSUB bar 2\r\n"u8.ToArray());
        await Task.Delay(200);

        var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/subz?subs=true&test=foo.hello");
        var subz = await response.Content.ReadFromJsonAsync<Subsz>();
        subz.ShouldNotBeNull();
        subz.Subs.ShouldContain(s => s.Subject == "foo.*");
        subz.Subs.ShouldNotContain(s => s.Subject == "bar");
    }

    [Fact]
    public async Task Subz_pagination()
    {
        using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
        using var stream = new NetworkStream(sock);
        var buf = new byte[4096];
        _ = await stream.ReadAsync(buf);
        await stream.WriteAsync("CONNECT {}\r\nSUB a 1\r\nSUB b 2\r\nSUB c 3\r\n"u8.ToArray());
        await Task.Delay(200);

        var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/subz?subs=true&offset=0&limit=2");
        var subz = await response.Content.ReadFromJsonAsync<Subsz>();
        subz.ShouldNotBeNull();
        subz.Subs.Length.ShouldBe(2);
        subz.Total.ShouldBeGreaterThanOrEqualTo(3);
    }

    private static int GetFreePort()
    {
        using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        sock.Bind(new IPEndPoint(IPAddress.Loopback, 0));
        return ((IPEndPoint)sock.LocalEndPoint!).Port;
    }
}