refactor: extract NATS.Server.Monitoring.Tests project

Move 39 monitoring, events, and system endpoint test files from
NATS.Server.Tests into a dedicated NATS.Server.Monitoring.Tests project.
Update namespaces, replace private GetFreePort/ReadUntilAsync with
TestUtilities shared helpers, add InternalsVisibleTo, and register
in the solution file. All 439 tests pass.
This commit is contained in:
Joseph Doherty
2026-03-12 15:44:12 -04:00
parent edf9ed770e
commit 0c086522a4
42 changed files with 131 additions and 257 deletions

View File

@@ -0,0 +1,240 @@
// Go reference: server/monitor.go — closedClients ring buffer, ClosedState tracking.
// These tests verify the fixed-size ring buffer used to track recently closed connections
// for the /connz?state=closed monitoring endpoint.
using NATS.Server.Monitoring;
namespace NATS.Server.Monitoring.Tests.Monitoring;
public class ClosedConnectionRingBufferTests
{
// -----------------------------------------------------------------------
// Helpers
// -----------------------------------------------------------------------
private static ClosedClient MakeEntry(ulong cid, string reason = "normal") => new()
{
Cid = cid,
Ip = "127.0.0.1",
Port = 4222,
Start = DateTime.UtcNow.AddSeconds(-10),
Stop = DateTime.UtcNow,
Reason = reason,
};
// -----------------------------------------------------------------------
// 1. Add_IncreasesCount
// -----------------------------------------------------------------------
/// <summary>
/// Adding entries should increase Count up to capacity.
/// </summary>
[Fact]
public void Add_IncreasesCount()
{
var buf = new ClosedConnectionRingBuffer(capacity: 10);
buf.Count.ShouldBe(0);
buf.Add(MakeEntry(1));
buf.Count.ShouldBe(1);
buf.Add(MakeEntry(2));
buf.Count.ShouldBe(2);
buf.Add(MakeEntry(3));
buf.Count.ShouldBe(3);
}
// -----------------------------------------------------------------------
// 2. Add_RingOverwrite_CapacityNotExceeded
// -----------------------------------------------------------------------
/// <summary>
/// When capacity is exceeded the count stays at capacity (oldest entry is overwritten).
/// </summary>
[Fact]
public void Add_RingOverwrite_CapacityNotExceeded()
{
const int capacity = 5;
var buf = new ClosedConnectionRingBuffer(capacity);
for (var i = 1; i <= capacity + 3; i++)
buf.Add(MakeEntry((ulong)i));
buf.Count.ShouldBe(capacity);
}
// -----------------------------------------------------------------------
// 3. GetAll_ReturnsNewestFirst
// -----------------------------------------------------------------------
/// <summary>
/// GetAll should return entries ordered newest-first.
/// </summary>
[Fact]
public void GetAll_ReturnsNewestFirst()
{
var buf = new ClosedConnectionRingBuffer(capacity: 10);
buf.Add(MakeEntry(1));
buf.Add(MakeEntry(2));
buf.Add(MakeEntry(3));
var all = buf.GetAll();
all.Count.ShouldBe(3);
all[0].Cid.ShouldBe(3UL); // newest
all[1].Cid.ShouldBe(2UL);
all[2].Cid.ShouldBe(1UL); // oldest
}
// -----------------------------------------------------------------------
// 4. GetAll_EmptyBuffer_ReturnsEmpty
// -----------------------------------------------------------------------
/// <summary>
/// GetAll on an empty buffer should return an empty list.
/// </summary>
[Fact]
public void GetAll_EmptyBuffer_ReturnsEmpty()
{
var buf = new ClosedConnectionRingBuffer(capacity: 10);
var all = buf.GetAll();
all.ShouldBeEmpty();
}
// -----------------------------------------------------------------------
// 5. GetRecent_ReturnsRequestedCount
// -----------------------------------------------------------------------
/// <summary>
/// GetRecent(n) where n &lt;= Count should return exactly n entries.
/// </summary>
[Fact]
public void GetRecent_ReturnsRequestedCount()
{
var buf = new ClosedConnectionRingBuffer(capacity: 10);
for (var i = 1; i <= 8; i++)
buf.Add(MakeEntry((ulong)i));
var recent = buf.GetRecent(3);
recent.Count.ShouldBe(3);
recent[0].Cid.ShouldBe(8UL); // newest
recent[1].Cid.ShouldBe(7UL);
recent[2].Cid.ShouldBe(6UL);
}
// -----------------------------------------------------------------------
// 6. GetRecent_LessThanAvailable_ReturnsAll
// -----------------------------------------------------------------------
/// <summary>
/// GetRecent(n) where n &gt; Count should return all available entries.
/// </summary>
[Fact]
public void GetRecent_LessThanAvailable_ReturnsAll()
{
var buf = new ClosedConnectionRingBuffer(capacity: 10);
buf.Add(MakeEntry(1));
buf.Add(MakeEntry(2));
var recent = buf.GetRecent(100);
recent.Count.ShouldBe(2);
}
// -----------------------------------------------------------------------
// 7. TotalClosed_TracksAllAdditions
// -----------------------------------------------------------------------
/// <summary>
/// TotalClosed should increment for every Add, even after the ring wraps around.
/// </summary>
[Fact]
public void TotalClosed_TracksAllAdditions()
{
const int capacity = 4;
var buf = new ClosedConnectionRingBuffer(capacity);
buf.TotalClosed.ShouldBe(0L);
for (var i = 1; i <= 10; i++)
buf.Add(MakeEntry((ulong)i));
buf.TotalClosed.ShouldBe(10L);
buf.Count.ShouldBe(capacity); // buffer is full but total reflects all 10
}
// -----------------------------------------------------------------------
// 8. Clear_ResetsCountAndBuffer
// -----------------------------------------------------------------------
/// <summary>
/// Clear should reset Count to zero and GetAll should return an empty list.
/// TotalClosed is intentionally not reset because it is a running lifetime counter.
/// </summary>
[Fact]
public void Clear_ResetsCountAndBuffer()
{
var buf = new ClosedConnectionRingBuffer(capacity: 10);
buf.Add(MakeEntry(1));
buf.Add(MakeEntry(2));
buf.Add(MakeEntry(3));
buf.TotalClosed.ShouldBe(3L);
buf.Clear();
buf.Count.ShouldBe(0);
buf.GetAll().ShouldBeEmpty();
// TotalClosed is a lifetime counter; it is not reset by Clear.
buf.TotalClosed.ShouldBe(3L);
}
// -----------------------------------------------------------------------
// 9. Capacity_ReturnsConfiguredSize
// -----------------------------------------------------------------------
/// <summary>
/// Capacity should reflect the value passed to the constructor.
/// </summary>
[Fact]
public void Capacity_ReturnsConfiguredSize()
{
var buf = new ClosedConnectionRingBuffer(capacity: 42);
buf.Capacity.ShouldBe(42);
}
// -----------------------------------------------------------------------
// 10. Add_WrapsCorrectly
// -----------------------------------------------------------------------
/// <summary>
/// After adding capacity+1 items the oldest entry (cid=1) should no longer be present,
/// and the buffer should contain the most recent 'capacity' items.
/// </summary>
[Fact]
public void Add_WrapsCorrectly()
{
const int capacity = 5;
var buf = new ClosedConnectionRingBuffer(capacity);
for (var i = 1; i <= capacity + 1; i++)
buf.Add(MakeEntry((ulong)i));
var all = buf.GetAll();
all.Count.ShouldBe(capacity);
// cid=1 (the oldest) should have been overwritten
all.Any(e => e.Cid == 1UL).ShouldBeFalse();
// The newest entry (cid=capacity+1) should be first
all[0].Cid.ShouldBe((ulong)(capacity + 1));
}
}

View File

@@ -0,0 +1,152 @@
// Go reference: server/monitor.go ClosedState.String() — reason strings emitted by
// the /connz endpoint, and server/auth.go getAuthErrClosedState — auth-related reasons.
// These tests verify the ClosedReason enum and ClosedReasonHelper helpers introduced
// in Task 89 (Gap 10.7: consistently populate closed connection reasons).
using NATS.Server.Monitoring;
namespace NATS.Server.Monitoring.Tests.Monitoring;
public class ClosedReasonTests
{
// -----------------------------------------------------------------------
// 1. ToReasonString_ClientClosed_ReturnsExpected
// -----------------------------------------------------------------------
/// <summary>
/// ClientClosed maps to the Go-compatible "Client Closed" string.
/// Go reference: server/monitor.go ClosedState.String() case ClientClosed.
/// </summary>
[Fact]
public void ToReasonString_ClientClosed_ReturnsExpected()
{
ClosedReasonHelper.ToReasonString(ClosedReason.ClientClosed).ShouldBe("Client Closed");
}
// -----------------------------------------------------------------------
// 2. ToReasonString_AllReasonsHaveStrings
// -----------------------------------------------------------------------
/// <summary>
/// Every ClosedReason enum value must produce a non-null, non-empty string.
/// Go reference: server/monitor.go ClosedState.String() — all cases covered.
/// </summary>
[Fact]
public void ToReasonString_AllReasonsHaveStrings()
{
foreach (var reason in Enum.GetValues<ClosedReason>())
{
var s = ClosedReasonHelper.ToReasonString(reason);
s.ShouldNotBeNull();
s.ShouldNotBeEmpty();
}
}
// -----------------------------------------------------------------------
// 3. FromReasonString_ValidString_ReturnsEnum
// -----------------------------------------------------------------------
/// <summary>
/// A valid Go-compatible reason string parses back to the correct enum value.
/// Go reference: server/monitor.go ClosedState.String() "Server Shutdown".
/// </summary>
[Fact]
public void FromReasonString_ValidString_ReturnsEnum()
{
ClosedReasonHelper.FromReasonString("Server Shutdown").ShouldBe(ClosedReason.ServerShutdown);
}
// -----------------------------------------------------------------------
// 4. FromReasonString_Unknown_ReturnsUnknown
// -----------------------------------------------------------------------
/// <summary>
/// An unrecognised string returns ClosedReason.Unknown.
/// </summary>
[Fact]
public void FromReasonString_Unknown_ReturnsUnknown()
{
ClosedReasonHelper.FromReasonString("Not a real reason").ShouldBe(ClosedReason.Unknown);
}
// -----------------------------------------------------------------------
// 5. FromReasonString_Null_ReturnsUnknown
// -----------------------------------------------------------------------
/// <summary>
/// A null reason string returns ClosedReason.Unknown.
/// </summary>
[Fact]
public void FromReasonString_Null_ReturnsUnknown()
{
ClosedReasonHelper.FromReasonString(null).ShouldBe(ClosedReason.Unknown);
}
// -----------------------------------------------------------------------
// 6. IsClientInitiated_ClientClosed_True
// -----------------------------------------------------------------------
/// <summary>
/// ClientClosed is the only client-initiated close reason.
/// Go reference: server/client.go closeConnection — client disconnect path.
/// </summary>
[Fact]
public void IsClientInitiated_ClientClosed_True()
{
ClosedReasonHelper.IsClientInitiated(ClosedReason.ClientClosed).ShouldBeTrue();
}
// -----------------------------------------------------------------------
// 7. IsClientInitiated_ServerShutdown_False
// -----------------------------------------------------------------------
/// <summary>
/// ServerShutdown is not a client-initiated close.
/// </summary>
[Fact]
public void IsClientInitiated_ServerShutdown_False()
{
ClosedReasonHelper.IsClientInitiated(ClosedReason.ServerShutdown).ShouldBeFalse();
}
// -----------------------------------------------------------------------
// 8. IsAuthRelated_AuthTimeout_True
// -----------------------------------------------------------------------
/// <summary>
/// AuthTimeout is auth-related.
/// Go reference: server/auth.go getAuthErrClosedState — auth timeout path.
/// </summary>
[Fact]
public void IsAuthRelated_AuthTimeout_True()
{
ClosedReasonHelper.IsAuthRelated(ClosedReason.AuthTimeout).ShouldBeTrue();
}
// -----------------------------------------------------------------------
// 9. IsAuthRelated_WriteError_False
// -----------------------------------------------------------------------
/// <summary>
/// WriteError is not auth-related.
/// </summary>
[Fact]
public void IsAuthRelated_WriteError_False()
{
ClosedReasonHelper.IsAuthRelated(ClosedReason.WriteError).ShouldBeFalse();
}
// -----------------------------------------------------------------------
// 10. IsResourceLimit_MaxConnections_True
// -----------------------------------------------------------------------
/// <summary>
/// MaxConnectionsExceeded is a resource-limit close reason.
/// Go reference: server/client.go maxConnectionsExceeded — max connections path.
/// </summary>
[Fact]
public void IsResourceLimit_MaxConnections_True()
{
ClosedReasonHelper.IsResourceLimit(ClosedReason.MaxConnectionsExceeded).ShouldBeTrue();
}
}

View File

@@ -0,0 +1,190 @@
using NATS.Server.Monitoring;
namespace NATS.Server.Monitoring.Tests.Monitoring;
/// <summary>
/// Unit tests for account-scoped /connz filtering (Gap 10.2).
/// Exercises ConnzFilterOptions.Parse, ConnzFilter.FilterByAccount, and
/// ConnzFilter.ApplyFilters in isolation — no running server required.
/// Go reference: monitor_test.go — TestConnzFilterByAccount, TestConnzWithAccount.
/// </summary>
public class ConnzAccountFilterTests
{
// -----------------------------------------------------------------------
// Helper — build a ConnzConnectionInfo with sensible defaults
// -----------------------------------------------------------------------
private static ConnzConnectionInfo MakeConn(
ulong clientId,
string? accountName,
string remoteAddress = "127.0.0.1:1234",
string? name = null,
long inMsgs = 0,
long outMsgs = 0,
long inBytes = 0,
long outBytes = 0) =>
new(clientId, remoteAddress, accountName, name, DateTime.UtcNow,
inMsgs, outMsgs, inBytes, outBytes);
// -----------------------------------------------------------------------
// FilterByAccount tests
// -----------------------------------------------------------------------
[Fact]
public void FilterByAccount_MatchingAccount_ReturnsFiltered()
{
// Go reference: monitor.go Connz() — "if opts.Account != "" { ... }" filter
var connections = new[]
{
MakeConn(1, "acctA"),
MakeConn(2, "acctB"),
MakeConn(3, "acctA"),
};
var result = ConnzFilter.FilterByAccount(connections, "acctA");
result.Count.ShouldBe(2);
result.ShouldAllBe(c => c.AccountName == "acctA");
}
[Fact]
public void FilterByAccount_NoMatch_ReturnsEmpty()
{
var connections = new[]
{
MakeConn(1, "acctA"),
MakeConn(2, "acctB"),
};
var result = ConnzFilter.FilterByAccount(connections, "acctC");
result.ShouldBeEmpty();
}
[Fact]
public void FilterByAccount_CaseInsensitive()
{
// Go reference: monitor.go — account name comparison is case-insensitive
var connections = new[]
{
MakeConn(1, "AcctA"),
MakeConn(2, "acctb"),
};
// Upper-case query against lower-case stored name
var lowerResult = ConnzFilter.FilterByAccount(connections, "accta");
lowerResult.Count.ShouldBe(1);
lowerResult[0].ClientId.ShouldBe(1UL);
// Lower-case query against mixed-case stored name
var upperResult = ConnzFilter.FilterByAccount(connections, "ACCTB");
upperResult.Count.ShouldBe(1);
upperResult[0].ClientId.ShouldBe(2UL);
}
// -----------------------------------------------------------------------
// ConnzFilterOptions.Parse tests
// -----------------------------------------------------------------------
[Fact]
public void Parse_WithAccParam_SetsAccountFilter()
{
var opts = ConnzFilterOptions.Parse("?acc=myAccount");
opts.AccountFilter.ShouldBe("myAccount");
}
[Fact]
public void Parse_WithoutParams_DefaultValues()
{
var opts = ConnzFilterOptions.Parse(null);
opts.AccountFilter.ShouldBeNull();
opts.StateFilter.ShouldBeNull();
opts.Offset.ShouldBe(0);
opts.Limit.ShouldBe(1024);
}
[Fact]
public void Parse_WithOffsetAndLimit_ParsesCorrectly()
{
var opts = ConnzFilterOptions.Parse("?acc=acctA&state=open&offset=10&limit=50");
opts.AccountFilter.ShouldBe("acctA");
opts.StateFilter.ShouldBe("open");
opts.Offset.ShouldBe(10);
opts.Limit.ShouldBe(50);
}
// -----------------------------------------------------------------------
// ApplyFilters tests
// -----------------------------------------------------------------------
[Fact]
public void ApplyFilters_WithAccountFilter_FiltersCorrectly()
{
var connections = new[]
{
MakeConn(1, "acctA"),
MakeConn(2, "acctB"),
MakeConn(3, "acctA"),
};
var opts = new ConnzFilterOptions { AccountFilter = "acctA" };
var result = ConnzFilter.ApplyFilters(connections, opts);
result.Total.ShouldBe(2);
result.Connections.Count.ShouldBe(2);
result.Connections.ShouldAllBe(c => c.AccountName == "acctA");
}
[Fact]
public void ApplyFilters_WithOffset_SkipsEntries()
{
var connections = Enumerable.Range(1, 5)
.Select(i => MakeConn((ulong)i, "acctA"))
.ToList();
var opts = new ConnzFilterOptions { Offset = 3, Limit = 10 };
var result = ConnzFilter.ApplyFilters(connections, opts);
result.Total.ShouldBe(5);
result.Connections.Count.ShouldBe(2);
result.Offset.ShouldBe(3);
// The paged items should be the last two (clientId 4 and 5)
result.Connections[0].ClientId.ShouldBe(4UL);
result.Connections[1].ClientId.ShouldBe(5UL);
}
[Fact]
public void ApplyFilters_WithLimit_CapsResults()
{
var connections = Enumerable.Range(1, 10)
.Select(i => MakeConn((ulong)i, "acctA"))
.ToList();
var opts = new ConnzFilterOptions { Limit = 3 };
var result = ConnzFilter.ApplyFilters(connections, opts);
result.Total.ShouldBe(10);
result.Connections.Count.ShouldBe(3);
result.Limit.ShouldBe(3);
}
[Fact]
public void ApplyFilters_NoFilters_ReturnsAll()
{
var connections = new[]
{
MakeConn(1, "acctA"),
MakeConn(2, "acctB"),
MakeConn(3, null),
};
var opts = new ConnzFilterOptions();
var result = ConnzFilter.ApplyFilters(connections, opts);
result.Total.ShouldBe(3);
result.Connections.Count.ShouldBe(3);
}
}

View File

@@ -0,0 +1,393 @@
using System.Net;
using System.Net.Sockets;
using System.Text;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server;
using NATS.Server.Auth;
using NATS.Server.Monitoring;
using NATS.Server.TestUtilities;
namespace NATS.Server.Monitoring.Tests.Monitoring;
/// <summary>
/// Tests for ConnzHandler filtering, sorting, pagination, and closed connection
/// ring buffer behavior.
/// Go reference: monitor_test.go — TestConnz, TestConnzSortedByCid, TestConnzSortedByBytesTo,
/// TestConnzFilter, TestConnzWithCID, TestConnzOffsetAndLimit.
/// </summary>
public class ConnzFilterTests : IAsyncLifetime
{
private readonly NatsServer _server;
private readonly NatsOptions _opts;
private readonly CancellationTokenSource _cts = new();
private readonly List<Socket> _sockets = [];
public ConnzFilterTests()
{
_opts = new NatsOptions
{
Port = TestPortAllocator.GetFreePort(),
MaxClosedClients = 100,
Users =
[
new User { Username = "alice", Password = "pw", Account = "acctA" },
new User { Username = "bob", Password = "pw", Account = "acctB" },
],
};
_server = new NatsServer(_opts, NullLoggerFactory.Instance);
}
public async Task InitializeAsync()
{
_ = _server.StartAsync(_cts.Token);
await _server.WaitForReadyAsync();
}
public async Task DisposeAsync()
{
foreach (var s in _sockets)
{
try { s.Shutdown(SocketShutdown.Both); } catch { }
s.Dispose();
}
await _cts.CancelAsync();
_server.Dispose();
}
private async Task<Socket> ConnectAsync(string user, string? subjectToSubscribe = null)
{
var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
_sockets.Add(sock);
await sock.ConnectAsync(IPAddress.Loopback, _opts.Port);
var buf = new byte[4096];
await sock.ReceiveAsync(buf, SocketFlags.None); // INFO
var connect = $"CONNECT {{\"user\":\"{user}\",\"pass\":\"pw\"}}\r\n";
await sock.SendAsync(Encoding.ASCII.GetBytes(connect));
if (subjectToSubscribe != null)
{
await sock.SendAsync(Encoding.ASCII.GetBytes($"SUB {subjectToSubscribe} sid1\r\n"));
}
await sock.SendAsync("PING\r\n"u8.ToArray());
await SocketTestHelper.ReadUntilAsync(sock, "PONG");
return sock;
}
private Connz GetConnz(string queryString = "")
{
var ctx = new DefaultHttpContext();
ctx.Request.QueryString = new QueryString(queryString);
return new ConnzHandler(_server).HandleConnz(ctx);
}
// --- Sort tests ---
[Fact]
public async Task Sort_by_cid_returns_ascending_order()
{
await ConnectAsync("alice");
await ConnectAsync("bob");
await Task.Delay(50);
var connz = GetConnz("?sort=cid");
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
for (int i = 1; i < connz.Conns.Length; i++)
{
connz.Conns[i].Cid.ShouldBeGreaterThan(connz.Conns[i - 1].Cid);
}
}
[Fact]
public async Task Sort_by_bytes_to_returns_descending_order()
{
var sock1 = await ConnectAsync("alice");
var sock2 = await ConnectAsync("bob");
await Task.Delay(50);
// Publish some data through sock1 to accumulate bytes
await sock1.SendAsync(Encoding.ASCII.GetBytes("SUB test 1\r\nPUB test 10\r\n1234567890\r\n"));
await Task.Delay(100);
var connz = GetConnz("?sort=bytes_to");
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
for (int i = 1; i < connz.Conns.Length; i++)
{
connz.Conns[i].OutBytes.ShouldBeLessThanOrEqualTo(connz.Conns[i - 1].OutBytes);
}
}
[Fact]
public async Task Sort_by_msgs_from_returns_descending_order()
{
var sock1 = await ConnectAsync("alice");
await Task.Delay(50);
// Send a PUB to increment InMsgs
await sock1.SendAsync(Encoding.ASCII.GetBytes("PUB test 3\r\nabc\r\n"));
await Task.Delay(100);
var connz = GetConnz("?sort=msgs_from");
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(1);
for (int i = 1; i < connz.Conns.Length; i++)
{
connz.Conns[i].InMsgs.ShouldBeLessThanOrEqualTo(connz.Conns[i - 1].InMsgs);
}
}
[Fact]
public async Task Sort_by_subs_returns_descending_order()
{
// Alice has 2 subs, Bob has 1
var sock1 = await ConnectAsync("alice", "test.a");
await sock1.SendAsync(Encoding.ASCII.GetBytes("SUB test.b sid2\r\n"));
var sock2 = await ConnectAsync("bob", "test.c");
await Task.Delay(100);
var connz = GetConnz("?sort=subs");
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
for (int i = 1; i < connz.Conns.Length; i++)
{
connz.Conns[i].NumSubs.ShouldBeLessThanOrEqualTo(connz.Conns[i - 1].NumSubs);
}
}
[Fact]
public async Task Sort_by_start_returns_ascending_order()
{
await ConnectAsync("alice");
await Task.Delay(20);
await ConnectAsync("bob");
await Task.Delay(50);
var connz = GetConnz("?sort=start");
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
for (int i = 1; i < connz.Conns.Length; i++)
{
connz.Conns[i].Start.ShouldBeGreaterThanOrEqualTo(connz.Conns[i - 1].Start);
}
}
// --- Filter tests ---
[Fact]
public async Task Filter_by_account_returns_only_matching_connections()
{
await ConnectAsync("alice");
await ConnectAsync("bob");
await Task.Delay(50);
var connz = GetConnz("?acc=acctA");
connz.Conns.ShouldAllBe(c => c.Account == "acctA");
connz.Conns.ShouldNotBeEmpty();
}
[Fact]
public async Task Filter_by_user_returns_only_matching_connections()
{
await ConnectAsync("alice");
await ConnectAsync("bob");
await Task.Delay(50);
var connz = GetConnz("?user=bob");
connz.Conns.ShouldAllBe(c => c.AuthorizedUser == "bob");
connz.Conns.ShouldNotBeEmpty();
}
[Fact]
public async Task Filter_by_subject_returns_matching_subscribers()
{
await ConnectAsync("alice", "orders.>");
await ConnectAsync("bob", "payments.>");
await Task.Delay(50);
var connz = GetConnz("?filter_subject=orders.new&subs=1");
connz.Conns.ShouldNotBeEmpty();
connz.Conns.ShouldAllBe(c => c.Subs.Any(s => s.Contains("orders")));
}
// --- Pagination tests ---
[Fact]
public async Task Offset_and_limit_paginates_results()
{
await ConnectAsync("alice");
await ConnectAsync("bob");
await ConnectAsync("alice");
await Task.Delay(50);
var page1 = GetConnz("?sort=cid&limit=2&offset=0");
page1.Conns.Length.ShouldBe(2);
page1.Total.ShouldBeGreaterThanOrEqualTo(3);
page1.Offset.ShouldBe(0);
page1.Limit.ShouldBe(2);
var page2 = GetConnz("?sort=cid&limit=2&offset=2");
page2.Conns.Length.ShouldBeGreaterThanOrEqualTo(1);
page2.Offset.ShouldBe(2);
// Ensure no overlap between pages
var page1Cids = page1.Conns.Select(c => c.Cid).ToHashSet();
var page2Cids = page2.Conns.Select(c => c.Cid).ToHashSet();
page1Cids.Overlaps(page2Cids).ShouldBeFalse();
}
// --- CID lookup test ---
[Fact]
public async Task Cid_lookup_returns_single_connection()
{
await ConnectAsync("alice");
await ConnectAsync("bob");
await Task.Delay(50);
// Get all connections to find a known CID
var all = GetConnz("?sort=cid");
all.Conns.ShouldNotBeEmpty();
var targetCid = all.Conns[0].Cid;
var single = GetConnz($"?cid={targetCid}");
single.Conns.Length.ShouldBe(1);
single.Conns[0].Cid.ShouldBe(targetCid);
}
[Fact]
public void Cid_lookup_nonexistent_returns_empty()
{
var result = GetConnz("?cid=99999999");
result.Conns.Length.ShouldBe(0);
result.Total.ShouldBe(0);
}
// --- Closed connection tests ---
[Fact]
public async Task Closed_state_shows_disconnected_clients()
{
var sock = await ConnectAsync("alice");
await Task.Delay(50);
// Close the connection
sock.Shutdown(SocketShutdown.Both);
sock.Close();
_sockets.Remove(sock);
await Task.Delay(200);
var connz = GetConnz("?state=closed");
connz.Conns.ShouldNotBeEmpty();
connz.Conns.ShouldAllBe(c => c.Stop != null);
connz.Conns.ShouldAllBe(c => !string.IsNullOrEmpty(c.Reason));
}
[Fact]
public async Task All_state_shows_both_open_and_closed()
{
var sock1 = await ConnectAsync("alice");
var sock2 = await ConnectAsync("bob");
await Task.Delay(50);
// Close one connection
sock1.Shutdown(SocketShutdown.Both);
sock1.Close();
_sockets.Remove(sock1);
await Task.Delay(200);
var connz = GetConnz("?state=all");
connz.Total.ShouldBeGreaterThanOrEqualTo(2);
// Should have at least one open (bob) and one closed (alice)
connz.Conns.Any(c => c.Stop == null).ShouldBeTrue("expected at least one open connection");
connz.Conns.Any(c => c.Stop != null).ShouldBeTrue("expected at least one closed connection");
}
[Fact]
public async Task Closed_ring_buffer_caps_at_max()
{
// MaxClosedClients is 100, create and close 5 connections
for (int i = 0; i < 5; i++)
{
var sock = await ConnectAsync("alice");
await Task.Delay(20);
sock.Shutdown(SocketShutdown.Both);
sock.Close();
_sockets.Remove(sock);
await Task.Delay(100);
}
var connz = GetConnz("?state=closed");
connz.Total.ShouldBeLessThanOrEqualTo(_opts.MaxClosedClients);
}
// --- Sort fallback tests ---
[Fact]
public async Task Sort_by_stop_with_open_state_falls_back_to_cid()
{
await ConnectAsync("alice");
await ConnectAsync("bob");
await Task.Delay(50);
// sort=stop with state=open should fall back to cid sorting
var connz = GetConnz("?sort=stop&state=open");
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
for (int i = 1; i < connz.Conns.Length; i++)
{
connz.Conns[i].Cid.ShouldBeGreaterThan(connz.Conns[i - 1].Cid);
}
}
// --- Combined filter + sort test ---
[Fact]
public async Task Account_filter_with_bytes_sort_and_limit()
{
// Connect multiple alice clients
for (int i = 0; i < 3; i++)
{
var sock = await ConnectAsync("alice");
// Send varying amounts of data
var data = new string('x', (i + 1) * 100);
await sock.SendAsync(Encoding.ASCII.GetBytes($"SUB test 1\r\nPUB test {data.Length}\r\n{data}\r\n"));
}
await ConnectAsync("bob");
await Task.Delay(100);
var connz = GetConnz("?acc=acctA&sort=bytes_to&limit=2");
connz.Conns.Length.ShouldBeLessThanOrEqualTo(2);
connz.Conns.ShouldAllBe(c => c.Account == "acctA");
}
[Fact]
public async Task Closed_cid_lookup_returns_from_ring_buffer()
{
var sock = await ConnectAsync("alice");
await Task.Delay(50);
// Get the CID before closing
var all = GetConnz("?sort=cid");
all.Conns.ShouldNotBeEmpty();
var targetCid = all.Conns.Last().Cid;
// Close the socket
sock.Shutdown(SocketShutdown.Both);
sock.Close();
_sockets.Remove(sock);
await Task.Delay(200);
// Look up closed connection by CID
var single = GetConnz($"?cid={targetCid}");
single.Conns.Length.ShouldBe(1);
single.Conns[0].Cid.ShouldBe(targetCid);
single.Conns[0].Stop.ShouldNotBeNull();
}
}

View File

@@ -0,0 +1,54 @@
using System.Text.Json;
using System.Text;
namespace NATS.Server.Monitoring.Tests;
public class ConnzParityFieldTests
{
[Fact]
public async Task Connz_includes_identity_tls_and_proxy_parity_fields()
{
await using var fx = await MonitoringParityFixture.StartAsync();
var jwt = BuildJwt("UISSUER", ["team:core", "tier:gold"]);
await fx.ConnectClientAsync("proxy:edge", "orders.created", jwt);
var connz = fx.GetConnz("?subs=detail&auth=true");
connz.Conns.ShouldNotBeEmpty();
var conn = connz.Conns.Single(c => c.AuthorizedUser == "proxy:edge");
conn.Proxy.ShouldNotBeNull();
conn.Proxy.Key.ShouldBe("edge");
conn.Jwt.ShouldBe(jwt);
conn.IssuerKey.ShouldBe("UISSUER");
conn.Tags.ShouldContain("team:core");
var json = JsonSerializer.Serialize(connz);
json.ShouldContain("tls_peer_cert_subject");
json.ShouldContain("tls_peer_certs");
json.ShouldContain("issuer_key");
json.ShouldContain("\"tags\"");
json.ShouldContain("proxy");
json.ShouldNotContain("jwt_issuer_key");
}
private static string BuildJwt(string issuer, string[] tags)
{
static string B64Url(string json)
{
return Convert.ToBase64String(Encoding.UTF8.GetBytes(json))
.TrimEnd('=')
.Replace('+', '-')
.Replace('/', '_');
}
var header = B64Url("{\"alg\":\"none\",\"typ\":\"JWT\"}");
var payload = B64Url(JsonSerializer.Serialize(new
{
iss = issuer,
nats = new
{
tags,
},
}));
return $"{header}.{payload}.eA";
}
}

View File

@@ -0,0 +1,119 @@
using System.Net;
using System.Net.Sockets;
using System.Text;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Auth;
using NATS.Server.Monitoring;
namespace NATS.Server.Monitoring.Tests;
public class ConnzParityFilterTests
{
[Fact]
public async Task Connz_filters_by_user_account_and_subject_and_includes_tls_peer_and_jwt_metadata()
{
await using var fx = await MonitoringParityFixture.StartAsync();
await fx.ConnectClientAsync("u", "orders.created");
await fx.ConnectClientAsync("v", "payments.created");
var connz = fx.GetConnz("?user=u&acc=A&filter_subject=orders.*&subs=detail");
connz.Conns.ShouldAllBe(c => c.Account == "A" && c.AuthorizedUser == "u");
}
}
internal sealed class MonitoringParityFixture : IAsyncDisposable
{
private readonly NatsServer _server;
private readonly CancellationTokenSource _cts;
private readonly List<TcpClient> _clients = [];
private readonly NatsOptions _options;
private MonitoringParityFixture(NatsServer server, NatsOptions options, CancellationTokenSource cts)
{
_server = server;
_options = options;
_cts = cts;
}
public static async Task<MonitoringParityFixture> StartAsync()
{
var options = new NatsOptions
{
Host = "127.0.0.1",
Port = 0,
Users =
[
new User { Username = "u", Password = "p", Account = "A" },
new User { Username = "v", Password = "p", Account = "B" },
new User { Username = "proxy:edge", Password = "p", Account = "A" },
],
};
var server = new NatsServer(options, NullLoggerFactory.Instance);
var cts = new CancellationTokenSource();
_ = server.StartAsync(cts.Token);
await server.WaitForReadyAsync();
return new MonitoringParityFixture(server, options, cts);
}
public async Task ConnectClientAsync(string username, string? subscribeSubject, string? jwt = null)
{
var client = new TcpClient();
await client.ConnectAsync(IPAddress.Loopback, _server.Port);
_clients.Add(client);
var stream = client.GetStream();
await ReadLineAsync(stream); // INFO
var connectPayload = string.IsNullOrWhiteSpace(jwt)
? $"{{\"user\":\"{username}\",\"pass\":\"p\"}}"
: $"{{\"user\":\"{username}\",\"pass\":\"p\",\"jwt\":\"{jwt}\"}}";
var connect = $"CONNECT {connectPayload}\r\n";
await stream.WriteAsync(Encoding.ASCII.GetBytes(connect));
if (!string.IsNullOrEmpty(subscribeSubject))
await stream.WriteAsync(Encoding.ASCII.GetBytes($"SUB {subscribeSubject} sid-{username}\r\n"));
await stream.FlushAsync();
await Task.Delay(30);
}
public Connz GetConnz(string queryString)
{
var ctx = new DefaultHttpContext();
ctx.Request.QueryString = new QueryString(queryString);
return new ConnzHandler(_server).HandleConnz(ctx);
}
public async Task<Varz> GetVarzAsync()
{
using var handler = new VarzHandler(_server, _options, NullLoggerFactory.Instance);
return await handler.HandleVarzAsync();
}
public async ValueTask DisposeAsync()
{
foreach (var client in _clients)
client.Dispose();
await _cts.CancelAsync();
_server.Dispose();
_cts.Dispose();
}
private static async Task<string> ReadLineAsync(NetworkStream stream)
{
var bytes = new List<byte>();
var one = new byte[1];
while (true)
{
var read = await stream.ReadAsync(one.AsMemory(0, 1));
if (read == 0)
break;
if (one[0] == (byte)'\n')
break;
if (one[0] != (byte)'\r')
bytes.Add(one[0]);
}
return Encoding.ASCII.GetString([.. bytes]);
}
}

View File

@@ -0,0 +1,171 @@
// Ported from golang/nats-server/server/monitor_test.go
// TestMonitorConnz — verify /connz lists active connections with correct fields.
// TestMonitorConnzSortedByBytesAndMsgs — verify /connz?sort=bytes_to ordering.
using System.Net;
using System.Net.Http.Json;
using System.Net.Sockets;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Monitoring;
using NATS.Server.TestUtilities;
namespace NATS.Server.Monitoring.Tests;
public class ConnzParityTests : IAsyncLifetime
{
private readonly NatsServer _server;
private readonly int _natsPort;
private readonly int _monitorPort;
private readonly CancellationTokenSource _cts = new();
private readonly HttpClient _http = new();
public ConnzParityTests()
{
_natsPort = TestPortAllocator.GetFreePort();
_monitorPort = TestPortAllocator.GetFreePort();
_server = new NatsServer(
new NatsOptions { Port = _natsPort, MonitorPort = _monitorPort },
NullLoggerFactory.Instance);
}
public async Task InitializeAsync()
{
_ = _server.StartAsync(_cts.Token);
await _server.WaitForReadyAsync();
for (var i = 0; i < 50; i++)
{
try
{
var probe = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
if (probe.IsSuccessStatusCode) break;
}
catch (HttpRequestException) { }
await Task.Delay(50);
}
}
public async Task DisposeAsync()
{
_http.Dispose();
await _cts.CancelAsync();
_server.Dispose();
}
/// <summary>
/// Corresponds to Go TestMonitorConnz.
/// Verifies /connz lists active connections and that per-connection fields
/// (ip, port, lang, version, uptime) are populated once 2 clients are connected.
/// </summary>
[Fact]
public async Task Connz_lists_active_connections()
{
var sockets = new List<Socket>();
try
{
// Connect 2 named clients
for (var i = 0; i < 2; i++)
{
var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var ns = new NetworkStream(sock);
var buf = new byte[4096];
_ = await ns.ReadAsync(buf); // consume INFO
var connect = $"CONNECT {{\"name\":\"client-{i}\",\"lang\":\"csharp\",\"version\":\"1.0\"}}\r\n";
await ns.WriteAsync(System.Text.Encoding.ASCII.GetBytes(connect));
await ns.FlushAsync();
sockets.Add(sock);
}
await Task.Delay(200);
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/connz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var connz = await response.Content.ReadFromJsonAsync<Connz>();
connz.ShouldNotBeNull();
// Both clients must appear
connz.NumConns.ShouldBeGreaterThanOrEqualTo(2);
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
// Verify per-connection identity fields on one of our named connections
var conn = connz.Conns.First(c => c.Name == "client-0");
conn.Ip.ShouldNotBeNullOrEmpty();
conn.Port.ShouldBeGreaterThan(0);
conn.Lang.ShouldBe("csharp");
conn.Version.ShouldBe("1.0");
conn.Uptime.ShouldNotBeNullOrEmpty();
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Corresponds to Go TestMonitorConnzSortedByBytesAndMsgs (bytes_to / out_bytes ordering).
/// Connects a high-traffic client that publishes 100 messages and 3 baseline clients,
/// then verifies /connz?sort=bytes_to returns connections in descending out_bytes order.
/// </summary>
[Fact]
public async Task Connz_sort_by_bytes()
{
var sockets = new List<(Socket Sock, NetworkStream Ns)>();
try
{
// Connect a subscriber first so that published messages are delivered (and counted as out_bytes)
var subSock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await subSock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var subNs = new NetworkStream(subSock);
var subBuf = new byte[4096];
_ = await subNs.ReadAsync(subBuf);
await subNs.WriteAsync("CONNECT {}\r\nSUB foo 1\r\n"u8.ToArray());
await subNs.FlushAsync();
sockets.Add((subSock, subNs));
// High-traffic publisher: publish 100 messages to "foo"
var highSock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await highSock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var highNs = new NetworkStream(highSock);
var highBuf = new byte[4096];
_ = await highNs.ReadAsync(highBuf);
await highNs.WriteAsync("CONNECT {}\r\n"u8.ToArray());
await highNs.FlushAsync();
for (var i = 0; i < 100; i++)
await highNs.WriteAsync("PUB foo 11\r\nHello World\r\n"u8.ToArray());
await highNs.FlushAsync();
sockets.Add((highSock, highNs));
// 3 baseline clients — no traffic beyond CONNECT
for (var i = 0; i < 3; i++)
{
var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var ns = new NetworkStream(sock);
var buf = new byte[4096];
_ = await ns.ReadAsync(buf);
await ns.WriteAsync("CONNECT {}\r\n"u8.ToArray());
await ns.FlushAsync();
sockets.Add((sock, ns));
}
await Task.Delay(300);
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/connz?sort=bytes_to");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var connz = await response.Content.ReadFromJsonAsync<Connz>();
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
// The first entry must have at least as many out_bytes as the second (descending order)
connz.Conns[0].OutBytes.ShouldBeGreaterThanOrEqualTo(connz.Conns[1].OutBytes);
}
finally
{
foreach (var (s, _) in sockets) s.Dispose();
}
}
}

View File

@@ -0,0 +1,179 @@
using NATS.Server.Monitoring;
namespace NATS.Server.Monitoring.Tests.Monitoring;
/// <summary>
/// Unit tests for ConnzSortOption enum, ConnzSorter.Parse, and ConnzSorter.Sort.
/// All tests exercise the types in isolation — no running server required.
/// Go reference: monitor_test.go — TestConnzSortedByCid, TestConnzSortedByBytesTo,
/// TestConnzSortedByMsgsFrom, TestConnzSortedByStart, monitor_sort_opts.go.
/// </summary>
public class ConnzSortTests
{
// -----------------------------------------------------------------------
// Helper — build a ConnInfo with specific field values
// -----------------------------------------------------------------------
private static ConnInfo MakeConn(
ulong cid = 1,
long outBytes = 0,
long inBytes = 0,
long outMsgs = 0,
long inMsgs = 0,
uint numSubs = 0,
int pending = 0,
DateTime connectedAt = default,
DateTime lastActivity = default) =>
new()
{
Cid = cid,
OutBytes = outBytes,
InBytes = inBytes,
OutMsgs = outMsgs,
InMsgs = inMsgs,
NumSubs = numSubs,
Pending = pending,
Start = connectedAt == default ? DateTime.UtcNow : connectedAt,
LastActivity = lastActivity == default ? DateTime.UtcNow : lastActivity,
};
// -----------------------------------------------------------------------
// ConnzSorter.Parse tests
// -----------------------------------------------------------------------
[Fact]
public void Parse_ConnectionId_Default()
{
// Go reference: monitor_sort_opts.go — empty/unknown value defaults to CID sort
ConnzSorter.Parse(null).ShouldBe(ConnzSortOption.ConnectionId);
ConnzSorter.Parse("").ShouldBe(ConnzSortOption.ConnectionId);
ConnzSorter.Parse(" ").ShouldBe(ConnzSortOption.ConnectionId);
ConnzSorter.Parse("cid").ShouldBe(ConnzSortOption.ConnectionId);
}
[Fact]
public void Parse_BytesTo_Parsed()
{
// Go reference: monitor_sort_opts.go — "bytes_to" => ByBytesTo
ConnzSorter.Parse("bytes_to").ShouldBe(ConnzSortOption.BytesTo);
ConnzSorter.Parse("BYTES_TO").ShouldBe(ConnzSortOption.BytesTo);
}
[Fact]
public void Parse_MsgsFrom_Parsed()
{
// Go reference: monitor_sort_opts.go — "msgs_from" => ByMsgsFrom
ConnzSorter.Parse("msgs_from").ShouldBe(ConnzSortOption.MsgsFrom);
ConnzSorter.Parse("MSGS_FROM").ShouldBe(ConnzSortOption.MsgsFrom);
}
[Fact]
public void Parse_Start_Parsed()
{
// Go reference: monitor_sort_opts.go — "start" => ByStart
ConnzSorter.Parse("start").ShouldBe(ConnzSortOption.Start);
ConnzSorter.Parse("START").ShouldBe(ConnzSortOption.Start);
}
[Fact]
public void Parse_Unknown_ReturnsDefault()
{
// Go reference: monitor_sort_opts.go — unrecognised string falls back to CID
ConnzSorter.Parse("not_a_sort_key").ShouldBe(ConnzSortOption.ConnectionId);
ConnzSorter.Parse("xyz").ShouldBe(ConnzSortOption.ConnectionId);
ConnzSorter.Parse("RANDOM").ShouldBe(ConnzSortOption.ConnectionId);
}
// -----------------------------------------------------------------------
// ConnzSorter.Sort tests
// -----------------------------------------------------------------------
[Fact]
public void Sort_ByConnectionId_Ascending()
{
// Go reference: monitor_test.go TestConnzSortedByCid — CID ascending is default
var conns = new[]
{
MakeConn(cid: 3),
MakeConn(cid: 1),
MakeConn(cid: 2),
};
var result = ConnzSorter.Sort(conns, ConnzSortOption.ConnectionId);
result.Count.ShouldBe(3);
result[0].Cid.ShouldBe(1UL);
result[1].Cid.ShouldBe(2UL);
result[2].Cid.ShouldBe(3UL);
}
[Fact]
public void Sort_ByBytesTo_Descending()
{
// Go reference: monitor_test.go TestConnzSortedByBytesTo — OutBytes descending
var conns = new[]
{
MakeConn(cid: 1, outBytes: 100),
MakeConn(cid: 2, outBytes: 300),
MakeConn(cid: 3, outBytes: 200),
};
var result = ConnzSorter.Sort(conns, ConnzSortOption.BytesTo);
result.Count.ShouldBe(3);
result[0].OutBytes.ShouldBe(300);
result[1].OutBytes.ShouldBe(200);
result[2].OutBytes.ShouldBe(100);
}
[Fact]
public void Sort_ByStart_Ascending()
{
// Go reference: monitor_test.go TestConnzSortedByStart — ConnectedAt ascending
var t0 = new DateTime(2024, 1, 1, 0, 0, 0, DateTimeKind.Utc);
var t1 = t0.AddMinutes(1);
var t2 = t0.AddMinutes(2);
var conns = new[]
{
MakeConn(cid: 1, connectedAt: t2),
MakeConn(cid: 2, connectedAt: t0),
MakeConn(cid: 3, connectedAt: t1),
};
var result = ConnzSorter.Sort(conns, ConnzSortOption.Start);
result.Count.ShouldBe(3);
result[0].Start.ShouldBe(t0);
result[1].Start.ShouldBe(t1);
result[2].Start.ShouldBe(t2);
}
[Fact]
public void Sort_ByMsgsFrom_Descending()
{
// Go reference: monitor_test.go TestConnzSortedByMsgsFrom — InMsgs descending
var conns = new[]
{
MakeConn(cid: 1, inMsgs: 50),
MakeConn(cid: 2, inMsgs: 200),
MakeConn(cid: 3, inMsgs: 10),
};
var result = ConnzSorter.Sort(conns, ConnzSortOption.MsgsFrom);
result.Count.ShouldBe(3);
result[0].InMsgs.ShouldBe(200);
result[1].InMsgs.ShouldBe(50);
result[2].InMsgs.ShouldBe(10);
}
[Fact]
public void Sort_EmptyList_ReturnsEmpty()
{
// Sorting an empty input should not throw and should return an empty list.
var result = ConnzSorter.Sort([], ConnzSortOption.BytesTo);
result.ShouldBeEmpty();
}
}

View File

@@ -0,0 +1,77 @@
// Ported from golang/nats-server/server/monitor_test.go
// TestMonitorHealthzStatusOK — verify /healthz returns HTTP 200 with status "ok".
using System.Net;
using System.Net.Sockets;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.TestUtilities;
namespace NATS.Server.Monitoring.Tests;
public class HealthzParityTests : IAsyncLifetime
{
private readonly NatsServer _server;
private readonly int _monitorPort;
private readonly CancellationTokenSource _cts = new();
private readonly HttpClient _http = new();
public HealthzParityTests()
{
_monitorPort = TestPortAllocator.GetFreePort();
_server = new NatsServer(
new NatsOptions { Port = 0, MonitorPort = _monitorPort },
NullLoggerFactory.Instance);
}
public async Task InitializeAsync()
{
_ = _server.StartAsync(_cts.Token);
await _server.WaitForReadyAsync();
for (var i = 0; i < 50; i++)
{
try
{
var probe = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
if (probe.IsSuccessStatusCode) break;
}
catch (HttpRequestException) { }
await Task.Delay(50);
}
}
public async Task DisposeAsync()
{
_http.Dispose();
await _cts.CancelAsync();
_server.Dispose();
}
/// <summary>
/// Corresponds to Go TestMonitorHealthzStatusOK.
/// Verifies GET /healthz returns HTTP 200 OK, indicating the server is healthy.
/// </summary>
[Fact]
public async Task Healthz_returns_ok()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
}
/// <summary>
/// Corresponds to Go TestMonitorHealthzStatusOK / checkHealthStatus.
/// Verifies the /healthz response body contains the "ok" status string,
/// matching the Go server's HealthStatus.Status = "ok" field.
/// </summary>
[Fact]
public async Task Healthz_returns_status_ok_json()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var body = await response.Content.ReadAsStringAsync();
// The .NET monitoring server returns Results.Ok("ok") which serializes as the JSON string "ok".
// This corresponds to the Go server's HealthStatus.Status = "ok".
body.ShouldContain("ok");
}
}

View File

@@ -0,0 +1,820 @@
// Go: TestMonitorConnz server/monitor_test.go:367
// Go: TestMonitorConnzWithSubs server/monitor_test.go:442
// Go: TestMonitorConnzWithSubsDetail server/monitor_test.go:463
// Go: TestMonitorClosedConnzWithSubsDetail server/monitor_test.go:484
// Go: TestMonitorConnzRTT server/monitor_test.go:583
// Go: TestMonitorConnzLastActivity server/monitor_test.go:638
// Go: TestMonitorConnzWithOffsetAndLimit server/monitor_test.go:732
// Go: TestMonitorConnzDefaultSorted server/monitor_test.go:806
// Go: TestMonitorConnzSortedByCid server/monitor_test.go:827
// Go: TestMonitorConnzSortedByStart server/monitor_test.go:849
// Go: TestMonitorConnzSortedByBytesAndMsgs server/monitor_test.go:871
// Go: TestMonitorConnzSortedByPending server/monitor_test.go:925
// Go: TestMonitorConnzSortedBySubs server/monitor_test.go:950
// Go: TestMonitorConnzSortedByLast server/monitor_test.go:976
// Go: TestMonitorConnzSortedByUptime server/monitor_test.go:1007
// Go: TestMonitorConnzSortedByIdle server/monitor_test.go:1202
// Go: TestMonitorConnzSortedByStopOnOpen server/monitor_test.go:1074
// Go: TestMonitorConnzSortedByReason server/monitor_test.go:1141
// Go: TestMonitorConnzWithNamedClient server/monitor_test.go:1851
// Go: TestMonitorConnzWithStateForClosedConns server/monitor_test.go:1876
// Go: TestMonitorConcurrentMonitoring server/monitor_test.go:2148
// Go: TestMonitorConnzSortByRTT server/monitor_test.go:5979
using System.Net;
using System.Net.Http.Json;
using System.Net.Sockets;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Monitoring;
using NATS.Server.TestUtilities;
namespace NATS.Server.Monitoring.Tests.Monitoring;
/// <summary>
/// Tests covering /connz endpoint behavior, ported from the Go server's monitor_test.go.
/// </summary>
public class MonitorConnzTests : IAsyncLifetime
{
private readonly NatsServer _server;
private readonly int _natsPort;
private readonly int _monitorPort;
private readonly CancellationTokenSource _cts = new();
private readonly HttpClient _http = new();
public MonitorConnzTests()
{
_natsPort = TestPortAllocator.GetFreePort();
_monitorPort = TestPortAllocator.GetFreePort();
_server = new NatsServer(
new NatsOptions { Port = _natsPort, MonitorPort = _monitorPort },
NullLoggerFactory.Instance);
}
public async Task InitializeAsync()
{
_ = _server.StartAsync(_cts.Token);
await _server.WaitForReadyAsync();
for (var i = 0; i < 50; i++)
{
try
{
var probe = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
if (probe.IsSuccessStatusCode) break;
}
catch (HttpRequestException) { }
await Task.Delay(50);
}
}
public async Task DisposeAsync()
{
_http.Dispose();
await _cts.CancelAsync();
_server.Dispose();
}
/// <summary>
/// Go: TestMonitorConnz (line 367).
/// Verifies /connz returns empty connections when no clients are connected.
/// </summary>
[Fact]
public async Task Connz_returns_empty_when_no_clients()
{
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
connz.ShouldNotBeNull();
connz.NumConns.ShouldBe(0);
connz.Total.ShouldBe(0);
connz.Conns.Length.ShouldBe(0);
}
/// <summary>
/// Go: TestMonitorConnz (line 367).
/// Verifies /connz lists active connections with populated identity fields.
/// </summary>
[Fact]
public async Task Connz_lists_active_connections_with_fields()
{
using var sock = await ConnectClientAsync("{\"name\":\"c1\",\"lang\":\"csharp\",\"version\":\"1.0\"}", "SUB foo 1\r\nPUB foo 5\r\nhello\r\n");
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
connz.ShouldNotBeNull();
connz.NumConns.ShouldBe(1);
connz.Total.ShouldBe(1);
connz.Conns.Length.ShouldBe(1);
var ci = connz.Conns[0];
// Go: ci.IP == "127.0.0.1"
ci.Ip.ShouldBe("127.0.0.1");
ci.Port.ShouldBeGreaterThan(0);
ci.Cid.ShouldBeGreaterThan(0UL);
ci.Name.ShouldBe("c1");
ci.Lang.ShouldBe("csharp");
ci.Version.ShouldBe("1.0");
ci.Start.ShouldBeGreaterThan(DateTime.MinValue);
ci.LastActivity.ShouldBeGreaterThanOrEqualTo(ci.Start);
ci.Uptime.ShouldNotBeNullOrEmpty();
ci.Idle.ShouldNotBeNullOrEmpty();
}
/// <summary>
/// Go: TestMonitorConnz (line 367).
/// Verifies /connz default limit is 1024 and offset is 0.
/// </summary>
[Fact]
public async Task Connz_default_limit_and_offset()
{
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
connz.ShouldNotBeNull();
connz.Limit.ShouldBe(1024); // Go: DefaultConnListSize
connz.Offset.ShouldBe(0);
}
/// <summary>
/// Go: TestMonitorConnzWithSubs (line 442).
/// Verifies /connz?subs=1 includes subscriptions list.
/// </summary>
[Fact]
public async Task Connz_with_subs_includes_subscription_list()
{
using var sock = await ConnectClientAsync("{}", "SUB hello.foo 1\r\n");
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?subs=1");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(1);
var ci = connz.Conns[0];
// Go: len(ci.Subs) != 1 || ci.Subs[0] != "hello.foo"
ci.Subs.ShouldContain("hello.foo");
}
/// <summary>
/// Go: TestMonitorConnzWithSubsDetail (line 463).
/// Verifies /connz?subs=detail includes subscription detail objects.
/// </summary>
[Fact]
public async Task Connz_with_subs_detail_includes_subscription_detail()
{
using var sock = await ConnectClientAsync("{}", "SUB hello.foo 1\r\n");
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?subs=detail");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(1);
var ci = connz.Conns[0];
// Go: len(ci.SubsDetail) != 1 || ci.SubsDetail[0].Subject != "hello.foo"
ci.SubsDetail.Length.ShouldBeGreaterThanOrEqualTo(1);
ci.SubsDetail.ShouldContain(sd => sd.Subject == "hello.foo");
}
/// <summary>
/// Go: TestMonitorConnzWithNamedClient (line 1851).
/// Verifies /connz exposes client name set in CONNECT options.
/// </summary>
[Fact]
public async Task Connz_shows_named_client()
{
using var sock = await ConnectClientAsync("{\"name\":\"test-client\"}");
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBe(1);
connz.Conns[0].Name.ShouldBe("test-client");
}
/// <summary>
/// Go: TestMonitorConnzWithOffsetAndLimit (line 732).
/// Verifies /connz pagination with offset and limit parameters.
/// </summary>
[Fact]
public async Task Connz_pagination_with_offset_and_limit()
{
var sockets = new List<Socket>();
try
{
for (var i = 0; i < 3; i++)
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(200);
// offset=1, limit=1 should return 1 connection with total of 3
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?offset=1&limit=1");
connz.ShouldNotBeNull();
connz.Limit.ShouldBe(1);
connz.Offset.ShouldBe(1);
connz.Conns.Length.ShouldBe(1);
connz.NumConns.ShouldBe(1);
connz.Total.ShouldBeGreaterThanOrEqualTo(3);
// offset past end should return 0
var connz2 = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?offset=10&limit=1");
connz2.ShouldNotBeNull();
connz2.Conns.Length.ShouldBe(0);
connz2.NumConns.ShouldBe(0);
connz2.Total.ShouldBeGreaterThanOrEqualTo(3);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzDefaultSorted (line 806).
/// Verifies /connz defaults to ascending CID sort order.
/// </summary>
[Fact]
public async Task Connz_default_sorted_by_cid_ascending()
{
var sockets = new List<Socket>();
try
{
for (var i = 0; i < 4; i++)
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(4);
// Go: Conns[0].Cid < Conns[1].Cid < Conns[2].Cid < Conns[3].Cid
for (var i = 1; i < connz.Conns.Length; i++)
connz.Conns[i].Cid.ShouldBeGreaterThan(connz.Conns[i - 1].Cid);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzSortedByCid (line 827).
/// Verifies /connz?sort=cid returns connections sorted by CID.
/// </summary>
[Fact]
public async Task Connz_sort_by_cid()
{
var sockets = new List<Socket>();
try
{
for (var i = 0; i < 4; i++)
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?sort=cid");
connz.ShouldNotBeNull();
for (var i = 1; i < connz.Conns.Length; i++)
connz.Conns[i].Cid.ShouldBeGreaterThan(connz.Conns[i - 1].Cid);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzSortedByStart (line 849).
/// Verifies /connz?sort=start returns connections sorted by start time.
/// </summary>
[Fact]
public async Task Connz_sort_by_start()
{
var sockets = new List<Socket>();
try
{
for (var i = 0; i < 3; i++)
{
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(10);
}
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?sort=start");
connz.ShouldNotBeNull();
for (var i = 1; i < connz.Conns.Length; i++)
connz.Conns[i].Start.ShouldBeGreaterThanOrEqualTo(connz.Conns[i - 1].Start);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzSortedByBytesAndMsgs (line 871).
/// Verifies /connz?sort=bytes_to returns connections sorted by out_bytes descending.
/// </summary>
[Fact]
public async Task Connz_sort_by_bytes_to()
{
var sockets = new List<Socket>();
try
{
// Subscriber first
sockets.Add(await ConnectClientAsync("{}", "SUB foo 1\r\n"));
// High-traffic publisher
var pub = await ConnectClientAsync("{}");
sockets.Add(pub);
using var ns = new NetworkStream(pub);
for (var i = 0; i < 50; i++)
await ns.WriteAsync("PUB foo 5\r\nhello\r\n"u8.ToArray());
await ns.FlushAsync();
// Low-traffic client
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(300);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?sort=bytes_to");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
// First entry should have >= out_bytes than second
connz.Conns[0].OutBytes.ShouldBeGreaterThanOrEqualTo(connz.Conns[1].OutBytes);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzSortedByBytesAndMsgs (line 871).
/// Verifies /connz?sort=msgs_to returns connections sorted by out_msgs descending.
/// </summary>
[Fact]
public async Task Connz_sort_by_msgs_to()
{
var sockets = new List<Socket>();
try
{
sockets.Add(await ConnectClientAsync("{}", "SUB foo 1\r\n"));
var pub = await ConnectClientAsync("{}");
sockets.Add(pub);
using var ns = new NetworkStream(pub);
for (var i = 0; i < 50; i++)
await ns.WriteAsync("PUB foo 5\r\nhello\r\n"u8.ToArray());
await ns.FlushAsync();
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(300);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?sort=msgs_to");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
connz.Conns[0].OutMsgs.ShouldBeGreaterThanOrEqualTo(connz.Conns[1].OutMsgs);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzSortedByBytesAndMsgs (line 871).
/// Verifies /connz?sort=msgs_from returns connections sorted by in_msgs descending.
/// </summary>
[Fact]
public async Task Connz_sort_by_msgs_from()
{
var sockets = new List<Socket>();
try
{
var pub = await ConnectClientAsync("{}");
sockets.Add(pub);
using var ns = new NetworkStream(pub);
for (var i = 0; i < 50; i++)
await ns.WriteAsync("PUB foo 5\r\nhello\r\n"u8.ToArray());
await ns.FlushAsync();
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(300);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?sort=msgs_from");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
connz.Conns[0].InMsgs.ShouldBeGreaterThanOrEqualTo(connz.Conns[1].InMsgs);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzSortedBySubs (line 950).
/// Verifies /connz?sort=subs returns connections sorted by subscription count descending.
/// </summary>
[Fact]
public async Task Connz_sort_by_subs()
{
var sockets = new List<Socket>();
try
{
// Client with many subs
sockets.Add(await ConnectClientAsync("{}", "SUB a 1\r\nSUB b 2\r\nSUB c 3\r\n"));
// Client with no subs
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?sort=subs");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
connz.Conns[0].NumSubs.ShouldBeGreaterThanOrEqualTo(connz.Conns[1].NumSubs);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzSortedByLast (line 976).
/// Verifies /connz?sort=last returns connections sorted by last_activity descending.
/// </summary>
[Fact]
public async Task Connz_sort_by_last_activity()
{
var sockets = new List<Socket>();
try
{
// First client connects and does something early
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(50);
// Second client connects later and does activity
sockets.Add(await ConnectClientAsync("{}", "PUB foo 2\r\nhi\r\n"));
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?sort=last");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
connz.Conns[0].LastActivity.ShouldBeGreaterThanOrEqualTo(connz.Conns[1].LastActivity);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzSortedByUptime (line 1007).
/// Verifies /connz?sort=uptime returns connections sorted by uptime descending.
/// </summary>
[Fact]
public async Task Connz_sort_by_uptime()
{
var sockets = new List<Socket>();
try
{
// First client has longer uptime
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(100);
// Second client has shorter uptime
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?sort=uptime");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
// Descending by uptime means first entry started earlier
connz.Conns[0].Start.ShouldBeLessThanOrEqualTo(connz.Conns[1].Start);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzSortedByIdle (line 1202).
/// Verifies /connz?sort=idle returns connections sorted by idle time descending.
/// </summary>
[Fact]
public async Task Connz_sort_by_idle()
{
var sockets = new List<Socket>();
try
{
// First client: older activity (more idle)
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(200);
// Second client: recent activity (less idle)
sockets.Add(await ConnectClientAsync("{}", "PUB foo 2\r\nhi\r\n"));
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?sort=idle");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
// Idle descending: first entry has older last activity
connz.Conns[0].LastActivity.ShouldBeLessThanOrEqualTo(connz.Conns[1].LastActivity);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzWithStateForClosedConns (line 1876).
/// Verifies /connz?state=closed returns recently disconnected clients.
/// </summary>
[Fact]
public async Task Connz_state_closed_returns_disconnected_clients()
{
var sock = await ConnectClientAsync("{\"name\":\"closing-client\"}");
await Task.Delay(200);
sock.Shutdown(SocketShutdown.Both);
sock.Dispose();
await Task.Delay(500);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?state=closed");
connz.ShouldNotBeNull();
connz.Conns.ShouldContain(c => c.Name == "closing-client");
var closed = connz.Conns.First(c => c.Name == "closing-client");
closed.Stop.ShouldNotBeNull();
closed.Reason.ShouldNotBeNullOrEmpty();
}
/// <summary>
/// Go: TestMonitorConnzSortedByStopOnOpen (line 1074).
/// Verifies /connz?sort=stop&state=open falls back to CID sort without error.
/// </summary>
[Fact]
public async Task Connz_sort_by_stop_with_open_state_falls_back_to_cid()
{
using var sock = await ConnectClientAsync("{}");
await Task.Delay(200);
// Go: sort by stop on open state should fallback
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/connz?sort=stop&state=open");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
}
/// <summary>
/// Go: TestMonitorConnzSortedByReason (line 1141).
/// Verifies /connz?sort=reason&state=closed sorts by close reason.
/// </summary>
[Fact]
public async Task Connz_sort_by_reason_on_closed()
{
var sock = await ConnectClientAsync("{}");
await Task.Delay(100);
sock.Shutdown(SocketShutdown.Both);
sock.Dispose();
await Task.Delay(500);
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/connz?sort=reason&state=closed");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
}
/// <summary>
/// Go: TestMonitorConnzSortedByReasonOnOpen (line 1180).
/// Verifies /connz?sort=reason&state=open falls back to CID sort without error.
/// </summary>
[Fact]
public async Task Connz_sort_by_reason_with_open_state_falls_back()
{
using var sock = await ConnectClientAsync("{}");
await Task.Delay(200);
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/connz?sort=reason&state=open");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
}
/// <summary>
/// Go: TestMonitorConnzSortByRTT (line 5979).
/// Verifies /connz?sort=rtt does not error.
/// </summary>
[Fact]
public async Task Connz_sort_by_rtt_succeeds()
{
using var sock = await ConnectClientAsync("{}");
await Task.Delay(200);
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/connz?sort=rtt");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
}
/// <summary>
/// Go: TestMonitorConnz (line 367).
/// Verifies /connz per-connection message stats are populated after pub/sub.
/// </summary>
[Fact]
public async Task Connz_per_connection_message_stats()
{
using var sock = await ConnectClientAsync("{}", "SUB foo 1\r\nPUB foo 5\r\nhello\r\n");
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBe(1);
var ci = connz.Conns[0];
// Go: ci.InMsgs == 1, ci.InBytes == 5
ci.InMsgs.ShouldBeGreaterThanOrEqualTo(1L);
ci.InBytes.ShouldBeGreaterThanOrEqualTo(5L);
}
/// <summary>
/// Go: TestMonitorConnzRTT (line 583).
/// Verifies /connz includes RTT field for connected clients.
/// </summary>
[Fact]
public async Task Connz_includes_rtt_field()
{
using var sock = await ConnectClientAsync("{}");
// Send a PING to trigger RTT measurement
using var ns = new NetworkStream(sock);
await ns.WriteAsync("PING\r\n"u8.ToArray());
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(1);
// RTT may or may not be populated depending on implementation, but field must exist
connz.Conns[0].Rtt.ShouldNotBeNull();
}
/// <summary>
/// Go: TestMonitorConnzLastActivity (line 638).
/// Verifies /connz last_activity is updated after message activity.
/// </summary>
[Fact]
public async Task Connz_last_activity_updates_after_message()
{
using var sock = await ConnectClientAsync("{}");
await Task.Delay(100);
// Record initial last activity
var connz1 = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
var initial = connz1!.Conns[0].LastActivity;
// Do more activity
using var ns = new NetworkStream(sock);
await ns.WriteAsync("PUB foo 5\r\nhello\r\n"u8.ToArray());
await ns.FlushAsync();
await Task.Delay(200);
var connz2 = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
var updated = connz2!.Conns[0].LastActivity;
// Activity should have updated
updated.ShouldBeGreaterThanOrEqualTo(initial);
}
/// <summary>
/// Go: TestMonitorConcurrentMonitoring (line 2148).
/// Verifies concurrent /connz requests do not cause errors.
/// </summary>
[Fact]
public async Task Connz_handles_concurrent_requests()
{
using var sock = await ConnectClientAsync("{}");
await Task.Delay(200);
var tasks = Enumerable.Range(0, 10).Select(async _ =>
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/connz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
});
await Task.WhenAll(tasks);
}
/// <summary>
/// Go: TestMonitorConnz (line 367).
/// Verifies /connz JSON uses correct Go-compatible field names.
/// </summary>
[Fact]
public async Task Connz_json_uses_go_field_names()
{
using var sock = await ConnectClientAsync("{}");
await Task.Delay(200);
var body = await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/connz");
body.ShouldContain("\"server_id\"");
body.ShouldContain("\"num_connections\"");
body.ShouldContain("\"connections\"");
}
/// <summary>
/// Go: TestMonitorConnzWithStateForClosedConns (line 1876).
/// Verifies /connz?state=all returns both open and closed connections.
/// </summary>
[Fact]
public async Task Connz_state_all_returns_both_open_and_closed()
{
// Connect and disconnect one client
var sock = await ConnectClientAsync("{\"name\":\"will-close\"}");
await Task.Delay(100);
sock.Shutdown(SocketShutdown.Both);
sock.Dispose();
await Task.Delay(300);
// Connect another client that stays open
using var sock2 = await ConnectClientAsync("{\"name\":\"stays-open\"}");
await Task.Delay(200);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?state=all");
connz.ShouldNotBeNull();
connz.Total.ShouldBeGreaterThanOrEqualTo(2);
}
/// <summary>
/// Go: TestMonitorConnz (line 367).
/// Verifies /connz server_id matches the server's ID.
/// </summary>
[Fact]
public async Task Connz_server_id_matches_server()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
connz!.Id.ShouldBe(varz!.Id);
}
/// <summary>
/// Go: TestMonitorConnzSortedByPending (line 925).
/// Verifies /connz?sort=pending returns connections sorted by pending bytes descending.
/// </summary>
[Fact]
public async Task Connz_sort_by_pending()
{
var sockets = new List<Socket>();
try
{
sockets.Add(await ConnectClientAsync("{}"));
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(200);
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/connz?sort=pending");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Go: TestMonitorConnzSortedByBytesAndMsgs (line 871).
/// Verifies /connz?sort=bytes_from returns connections sorted by in_bytes descending.
/// </summary>
[Fact]
public async Task Connz_sort_by_bytes_from()
{
var sockets = new List<Socket>();
try
{
// High-traffic publisher
var pub = await ConnectClientAsync("{}");
sockets.Add(pub);
using var ns = new NetworkStream(pub);
for (var i = 0; i < 50; i++)
await ns.WriteAsync("PUB foo 5\r\nhello\r\n"u8.ToArray());
await ns.FlushAsync();
// Low-traffic client
sockets.Add(await ConnectClientAsync("{}"));
await Task.Delay(300);
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz?sort=bytes_from");
connz.ShouldNotBeNull();
connz.Conns.Length.ShouldBeGreaterThanOrEqualTo(2);
connz.Conns[0].InBytes.ShouldBeGreaterThanOrEqualTo(connz.Conns[1].InBytes);
}
finally
{
foreach (var s in sockets) s.Dispose();
}
}
/// <summary>
/// Helper to connect a raw TCP client to the NATS server, send CONNECT and optional commands,
/// and return the socket. The caller is responsible for disposing the socket.
/// </summary>
private async Task<Socket> ConnectClientAsync(string connectJson, string? extraCommands = null)
{
var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None); // consume INFO
var cmd = $"CONNECT {connectJson}\r\n";
if (extraCommands is not null)
cmd += extraCommands;
await sock.SendAsync(System.Text.Encoding.ASCII.GetBytes(cmd), SocketFlags.None);
return sock;
}
}

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,263 @@
// Go: TestMonitorConnzWithRoutes server/monitor_test.go:1405
// Go: TestMonitorRoutezRace server/monitor_test.go:2210
// Go: TestMonitorRoutezRTT server/monitor_test.go:3919
// Go: TestMonitorRoutezPoolSize server/monitor_test.go:5705
// Go: TestMonitorClusterEmptyWhenNotDefined server/monitor_test.go:2456
using System.Net;
using System.Net.Http.Json;
using System.Net.Sockets;
using System.Text.Json;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Configuration;
using NATS.Server.Monitoring;
using NATS.Server.TestUtilities;
namespace NATS.Server.Monitoring.Tests.Monitoring;
/// <summary>
/// Tests covering /routez endpoint behavior, ported from the Go server's monitor_test.go.
/// </summary>
public class MonitorRoutezTests
{
/// <summary>
/// Go: TestMonitorConnzWithRoutes (line 1405).
/// Verifies that /routez returns valid JSON with routes and num_routes fields.
/// </summary>
[Fact]
public async Task Routez_returns_routes_and_num_routes()
{
await using var fx = await RoutezFixture.StartAsync();
var body = await fx.GetStringAsync("/routez");
body.ShouldContain("routes");
body.ShouldContain("num_routes");
}
/// <summary>
/// Go: TestMonitorConnzWithRoutes (line 1405).
/// Verifies /routez num_routes is 0 when no cluster routes are configured.
/// </summary>
[Fact]
public async Task Routez_num_routes_is_zero_without_cluster()
{
await using var fx = await RoutezFixture.StartAsync();
var doc = await fx.GetJsonDocumentAsync("/routez");
doc.RootElement.GetProperty("num_routes").GetInt32().ShouldBe(0);
}
/// <summary>
/// Go: TestMonitorConnzWithRoutes (line 1405).
/// Verifies /connz does not include route connections (they appear under /routez only).
/// </summary>
[Fact]
public async Task Connz_does_not_include_route_connections()
{
await using var fx = await RoutezFixture.StartAsync();
var connz = await fx.GetFromJsonAsync<Connz>("/connz");
connz.ShouldNotBeNull();
// Without any clients, connz should be empty
connz.NumConns.ShouldBe(0);
}
/// <summary>
/// Go: TestMonitorRoutezRace (line 2210).
/// Verifies concurrent /routez requests do not cause errors or data corruption.
/// </summary>
[Fact]
public async Task Routez_handles_concurrent_requests()
{
await using var fx = await RoutezFixture.StartAsync();
var tasks = Enumerable.Range(0, 10).Select(async _ =>
{
var response = await fx.GetAsync("/routez");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
});
await Task.WhenAll(tasks);
}
/// <summary>
/// Go: TestMonitorClusterEmptyWhenNotDefined (line 2456).
/// Verifies /varz cluster section has empty name when no cluster is configured.
/// </summary>
[Fact]
public async Task Varz_cluster_empty_when_not_defined()
{
await using var fx = await RoutezFixture.StartAsync();
var varz = await fx.GetFromJsonAsync<Varz>("/varz");
varz.ShouldNotBeNull();
varz.Cluster.ShouldNotBeNull();
varz.Cluster.Name.ShouldBe("");
}
/// <summary>
/// Go: TestMonitorConnzWithRoutes (line 1405).
/// Verifies /routez JSON field naming matches Go server format.
/// </summary>
[Fact]
public async Task Routez_json_uses_expected_field_names()
{
await using var fx = await RoutezFixture.StartAsync();
var body = await fx.GetStringAsync("/routez");
body.ShouldContain("\"routes\"");
body.ShouldContain("\"num_routes\"");
}
/// <summary>
/// Go: TestMonitorCluster (line 2724).
/// Verifies /varz includes cluster section even when cluster is enabled.
/// Note: The .NET server currently initializes the cluster section with defaults;
/// the Go server populates it with cluster config. This test verifies the section exists.
/// </summary>
[Fact]
public async Task Varz_includes_cluster_section_when_cluster_enabled()
{
await using var fx = await RoutezFixture.StartWithClusterAsync();
var varz = await fx.GetFromJsonAsync<Varz>("/varz");
varz.ShouldNotBeNull();
varz.Cluster.ShouldNotBeNull();
}
/// <summary>
/// Go: TestMonitorConnzWithRoutes (line 1405).
/// Verifies /routez response includes routes field even when num_routes is 0.
/// </summary>
[Fact]
public async Task Routez_includes_routes_field_even_when_empty()
{
await using var fx = await RoutezFixture.StartAsync();
var doc = await fx.GetJsonDocumentAsync("/routez");
doc.RootElement.TryGetProperty("routes", out _).ShouldBeTrue();
}
/// <summary>
/// Go: TestMonitorConnzWithRoutes (line 1405).
/// Verifies /routez returns HTTP 200 OK.
/// </summary>
[Fact]
public async Task Routez_returns_http_200()
{
await using var fx = await RoutezFixture.StartAsync();
var response = await fx.GetAsync("/routez");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
}
/// <summary>
/// Go: TestMonitorCluster (line 2724).
/// Verifies /routez endpoint is accessible when cluster is configured.
/// </summary>
[Fact]
public async Task Routez_accessible_with_cluster_config()
{
await using var fx = await RoutezFixture.StartWithClusterAsync();
var response = await fx.GetAsync("/routez");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var body = await response.Content.ReadAsStringAsync();
body.ShouldContain("routes");
}
}
internal sealed class RoutezFixture : IAsyncDisposable
{
private readonly NatsServer _server;
private readonly CancellationTokenSource _cts;
private readonly HttpClient _http;
private readonly int _monitorPort;
private RoutezFixture(NatsServer server, CancellationTokenSource cts, HttpClient http, int monitorPort)
{
_server = server;
_cts = cts;
_http = http;
_monitorPort = monitorPort;
}
public static async Task<RoutezFixture> StartAsync()
{
var monitorPort = TestPortAllocator.GetFreePort();
var options = new NatsOptions
{
Host = "127.0.0.1",
Port = 0,
MonitorPort = monitorPort,
};
return await CreateAndStartAsync(options, monitorPort);
}
public static async Task<RoutezFixture> StartWithClusterAsync()
{
var monitorPort = TestPortAllocator.GetFreePort();
var options = new NatsOptions
{
Host = "127.0.0.1",
Port = 0,
MonitorPort = monitorPort,
Cluster = new ClusterOptions
{
Host = "127.0.0.1",
Port = 0,
Name = "test-cluster",
},
};
return await CreateAndStartAsync(options, monitorPort);
}
private static async Task<RoutezFixture> CreateAndStartAsync(NatsOptions options, int monitorPort)
{
var server = new NatsServer(options, NullLoggerFactory.Instance);
var cts = new CancellationTokenSource();
_ = server.StartAsync(cts.Token);
await server.WaitForReadyAsync();
var http = new HttpClient();
for (var i = 0; i < 50; i++)
{
try
{
var response = await http.GetAsync($"http://127.0.0.1:{monitorPort}/healthz");
if (response.IsSuccessStatusCode) break;
}
catch { }
await Task.Delay(50);
}
return new RoutezFixture(server, cts, http, monitorPort);
}
public Task<string> GetStringAsync(string path)
=> _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}{path}");
public Task<HttpResponseMessage> GetAsync(string path)
=> _http.GetAsync($"http://127.0.0.1:{_monitorPort}{path}");
public Task<T?> GetFromJsonAsync<T>(string path)
=> _http.GetFromJsonAsync<T>($"http://127.0.0.1:{_monitorPort}{path}");
public async Task<JsonDocument> GetJsonDocumentAsync(string path)
{
var body = await GetStringAsync(path);
return JsonDocument.Parse(body);
}
public async ValueTask DisposeAsync()
{
_http.Dispose();
await _cts.CancelAsync();
_server.Dispose();
_cts.Dispose();
}
}

View File

@@ -0,0 +1,350 @@
// Go: TestMonitorStacksz server/monitor_test.go:2135
// Go: TestMonitorConcurrentMonitoring server/monitor_test.go:2148
// Go: TestMonitorHandleRoot server/monitor_test.go:1819
// Go: TestMonitorHTTPBasePath server/monitor_test.go:220
// Go: TestMonitorAccountz server/monitor_test.go:4300
// Go: TestMonitorAccountStatz server/monitor_test.go:4330
using System.Net;
using System.Net.Http.Json;
using System.Net.Sockets;
using System.Text.Json;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Monitoring;
using NATS.Server.TestUtilities;
namespace NATS.Server.Monitoring.Tests.Monitoring;
/// <summary>
/// Tests covering miscellaneous monitoring endpoints: root, accountz, accstatz,
/// gatewayz, leafz, and concurrent monitoring safety.
/// Ported from the Go server's monitor_test.go.
/// </summary>
public class MonitorStackszTests : IAsyncLifetime
{
private readonly NatsServer _server;
private readonly int _natsPort;
private readonly int _monitorPort;
private readonly CancellationTokenSource _cts = new();
private readonly HttpClient _http = new();
public MonitorStackszTests()
{
_natsPort = TestPortAllocator.GetFreePort();
_monitorPort = TestPortAllocator.GetFreePort();
_server = new NatsServer(
new NatsOptions { Port = _natsPort, MonitorPort = _monitorPort },
NullLoggerFactory.Instance);
}
public async Task InitializeAsync()
{
_ = _server.StartAsync(_cts.Token);
await _server.WaitForReadyAsync();
for (var i = 0; i < 50; i++)
{
try
{
var probe = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
if (probe.IsSuccessStatusCode) break;
}
catch (HttpRequestException) { }
await Task.Delay(50);
}
}
public async Task DisposeAsync()
{
_http.Dispose();
await _cts.CancelAsync();
_server.Dispose();
}
/// <summary>
/// Go: TestMonitorHandleRoot (line 1819).
/// Verifies GET / returns HTTP 200 with endpoint listing.
/// </summary>
[Fact]
public async Task Root_returns_endpoint_listing()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var body = await response.Content.ReadAsStringAsync();
body.ShouldContain("varz");
body.ShouldContain("connz");
body.ShouldContain("routez");
body.ShouldContain("healthz");
}
/// <summary>
/// Go: TestMonitorHandleRoot (line 1819).
/// Verifies GET / response includes subsz endpoint.
/// </summary>
[Fact]
public async Task Root_includes_subz_endpoint()
{
var body = await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/");
body.ShouldContain("subz");
}
/// <summary>
/// Go: TestMonitorAccountz (line 4300).
/// Verifies /accountz returns valid JSON with accounts list.
/// </summary>
[Fact]
public async Task Accountz_returns_accounts_list()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/accountz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var body = await response.Content.ReadAsStringAsync();
body.ShouldContain("accounts");
body.ShouldContain("num_accounts");
}
/// <summary>
/// Go: TestMonitorAccountz (line 4300).
/// Verifies /accountz num_accounts is at least 1 (global account).
/// </summary>
[Fact]
public async Task Accountz_num_accounts_at_least_one()
{
var doc = JsonDocument.Parse(await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/accountz"));
doc.RootElement.GetProperty("num_accounts").GetInt32().ShouldBeGreaterThanOrEqualTo(1);
}
/// <summary>
/// Go: TestMonitorAccountStatz (line 4330).
/// Verifies /accstatz returns aggregate account statistics.
/// </summary>
[Fact]
public async Task Accstatz_returns_aggregate_stats()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/accstatz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var body = await response.Content.ReadAsStringAsync();
body.ShouldContain("total_accounts");
body.ShouldContain("total_connections");
body.ShouldContain("total_subscriptions");
}
/// <summary>
/// Go: TestMonitorAccountStatz (line 4330).
/// Verifies /accstatz total_accounts is at least 1.
/// </summary>
[Fact]
public async Task Accstatz_total_accounts_at_least_one()
{
var doc = JsonDocument.Parse(await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/accstatz"));
doc.RootElement.GetProperty("total_accounts").GetInt32().ShouldBeGreaterThanOrEqualTo(1);
}
/// <summary>
/// Go: TestMonitorGateway (line 2880).
/// Verifies /gatewayz returns valid JSON.
/// </summary>
[Fact]
public async Task Gatewayz_returns_valid_json()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/gatewayz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var body = await response.Content.ReadAsStringAsync();
body.ShouldContain("gateways");
}
/// <summary>
/// Go: TestMonitorLeafNode (line 3112).
/// Verifies /leafz returns valid JSON.
/// </summary>
[Fact]
public async Task Leafz_returns_valid_json()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/leafz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var body = await response.Content.ReadAsStringAsync();
body.ShouldContain("leafs");
}
/// <summary>
/// Go: TestMonitorConcurrentMonitoring (line 2148).
/// Verifies concurrent requests across multiple endpoint types do not fail.
/// </summary>
[Fact]
public async Task Concurrent_requests_across_endpoints_succeed()
{
var endpoints = new[] { "varz", "varz", "connz", "connz", "subz", "subz", "routez", "routez" };
var tasks = endpoints.Select(async endpoint =>
{
for (var i = 0; i < 10; i++)
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/{endpoint}");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
}
});
await Task.WhenAll(tasks);
}
/// <summary>
/// Go: TestMonitorConcurrentMonitoring (line 2148).
/// Verifies concurrent /healthz requests do not fail.
/// </summary>
[Fact]
public async Task Concurrent_healthz_requests_succeed()
{
var tasks = Enumerable.Range(0, 20).Select(async _ =>
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
});
await Task.WhenAll(tasks);
}
/// <summary>
/// Go: TestMonitorHttpStatsNoUpdatedWhenUsingServerFuncs (line 2435).
/// Verifies /varz http_req_stats keys include all endpoints that were accessed.
/// </summary>
[Fact]
public async Task Http_req_stats_tracks_accessed_endpoints()
{
// Access multiple endpoints
await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/connz");
await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/subz");
await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/routez");
await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/varz");
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.HttpReqStats.ShouldContainKey("/connz");
varz.HttpReqStats.ShouldContainKey("/subz");
varz.HttpReqStats.ShouldContainKey("/routez");
varz.HttpReqStats.ShouldContainKey("/varz");
}
/// <summary>
/// Go: TestMonitorHandleRoot (line 1819).
/// Verifies GET / includes jsz endpoint in listing.
/// </summary>
[Fact]
public async Task Root_includes_jsz_endpoint()
{
var body = await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/");
body.ShouldContain("jsz");
}
/// <summary>
/// Go: TestMonitorHandleRoot (line 1819).
/// Verifies GET / includes accountz endpoint in listing.
/// </summary>
[Fact]
public async Task Root_includes_accountz_endpoint()
{
var body = await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/");
body.ShouldContain("accountz");
}
/// <summary>
/// Go: TestMonitorServerIDs (line 2410).
/// Verifies multiple monitoring endpoints return the same server_id.
/// </summary>
[Fact]
public async Task All_endpoints_return_consistent_server_id()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz");
varz.ShouldNotBeNull();
connz.ShouldNotBeNull();
subsz.ShouldNotBeNull();
var serverId = varz.Id;
serverId.ShouldNotBeNullOrEmpty();
connz.Id.ShouldBe(serverId);
subsz.Id.ShouldBe(serverId);
}
/// <summary>
/// Go: TestMonitorAccountStatz (line 4330).
/// Verifies /accstatz total_connections updates after a client connects.
/// </summary>
[Fact]
public async Task Accstatz_total_connections_updates_after_connect()
{
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None);
await sock.SendAsync("CONNECT {}\r\n"u8.ToArray(), SocketFlags.None);
await Task.Delay(200);
var doc = JsonDocument.Parse(await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/accstatz"));
doc.RootElement.GetProperty("total_connections").GetInt32().ShouldBeGreaterThanOrEqualTo(1);
}
/// <summary>
/// Go: TestMonitorAccountStatz (line 4330).
/// Verifies /accstatz total_subscriptions updates after a client subscribes.
/// </summary>
[Fact]
public async Task Accstatz_total_subscriptions_updates_after_subscribe()
{
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None);
await sock.SendAsync("CONNECT {}\r\nSUB test 1\r\n"u8.ToArray(), SocketFlags.None);
await Task.Delay(200);
var doc = JsonDocument.Parse(await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/accstatz"));
doc.RootElement.GetProperty("total_subscriptions").GetInt32().ShouldBeGreaterThanOrEqualTo(1);
}
/// <summary>
/// Go: TestMonitorAccountz (line 4300).
/// Verifies /accountz includes per-account fields: name, connections, subscriptions.
/// </summary>
[Fact]
public async Task Accountz_includes_per_account_fields()
{
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None);
await sock.SendAsync("CONNECT {}\r\nSUB test 1\r\n"u8.ToArray(), SocketFlags.None);
await Task.Delay(200);
var body = await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/accountz");
body.ShouldContain("\"name\"");
body.ShouldContain("\"connections\"");
body.ShouldContain("\"subscriptions\"");
}
/// <summary>
/// Go: TestMonitorGateway (line 2880).
/// Verifies /gatewayz includes num_gateways field.
/// </summary>
[Fact]
public async Task Gatewayz_includes_num_gateways()
{
var body = await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/gatewayz");
body.ShouldContain("gateways");
}
/// <summary>
/// Go: TestMonitorLeafNode (line 3112).
/// Verifies /leafz includes num_leafs field.
/// </summary>
[Fact]
public async Task Leafz_includes_num_leafs()
{
var body = await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/leafz");
body.ShouldContain("leafs");
}
}

View File

@@ -0,0 +1,354 @@
// Go: TestSubsz server/monitor_test.go:1538
// Go: TestMonitorSubszDetails server/monitor_test.go:1609
// Go: TestMonitorSubszWithOffsetAndLimit server/monitor_test.go:1642
// Go: TestMonitorSubszTestPubSubject server/monitor_test.go:1675
// Go: TestMonitorSubszMultiAccount server/monitor_test.go:1709
// Go: TestMonitorSubszMultiAccountWithOffsetAndLimit server/monitor_test.go:1777
using System.Net;
using System.Net.Http.Json;
using System.Net.Sockets;
using System.Text.Json;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Monitoring;
using NATS.Server.TestUtilities;
namespace NATS.Server.Monitoring.Tests.Monitoring;
/// <summary>
/// Tests covering /subz (subscriptionsz) endpoint behavior,
/// ported from the Go server's monitor_test.go.
/// </summary>
public class MonitorSubszTests : IAsyncLifetime
{
private readonly NatsServer _server;
private readonly int _natsPort;
private readonly int _monitorPort;
private readonly CancellationTokenSource _cts = new();
private readonly HttpClient _http = new();
public MonitorSubszTests()
{
_natsPort = TestPortAllocator.GetFreePort();
_monitorPort = TestPortAllocator.GetFreePort();
_server = new NatsServer(
new NatsOptions { Port = _natsPort, MonitorPort = _monitorPort },
NullLoggerFactory.Instance);
}
public async Task InitializeAsync()
{
_ = _server.StartAsync(_cts.Token);
await _server.WaitForReadyAsync();
for (var i = 0; i < 50; i++)
{
try
{
var probe = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
if (probe.IsSuccessStatusCode) break;
}
catch (HttpRequestException) { }
await Task.Delay(50);
}
}
public async Task DisposeAsync()
{
_http.Dispose();
await _cts.CancelAsync();
_server.Dispose();
}
/// <summary>
/// Go: TestSubsz (line 1538).
/// Verifies /subz returns valid JSON with server_id, num_subscriptions fields.
/// </summary>
[Fact]
public async Task Subz_returns_valid_json_with_server_id()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/subz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var subsz = await response.Content.ReadFromJsonAsync<Subsz>();
subsz.ShouldNotBeNull();
subsz.Id.ShouldNotBeNullOrEmpty();
}
/// <summary>
/// Go: TestSubsz (line 1538).
/// Verifies /subz reports num_subscriptions after clients subscribe.
/// </summary>
[Fact]
public async Task Subz_reports_subscription_count()
{
using var sock = await ConnectClientAsync("SUB foo 1\r\n");
await Task.Delay(200);
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz");
subsz.ShouldNotBeNull();
subsz.NumSubs.ShouldBeGreaterThanOrEqualTo(1u);
}
/// <summary>
/// Go: TestMonitorSubszDetails (line 1609).
/// Verifies /subz?subs=1 returns subscription details with subject info.
/// </summary>
[Fact]
public async Task Subz_with_subs_returns_subscription_details()
{
using var sock = await ConnectClientAsync("SUB foo.* 1\r\nSUB foo.bar 2\r\nSUB foo.foo 3\r\n");
await Task.Delay(200);
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz?subs=1");
subsz.ShouldNotBeNull();
// Go: sl.NumSubs != 3, sl.Total != 3, len(sl.Subs) != 3
subsz.NumSubs.ShouldBeGreaterThanOrEqualTo(3u);
subsz.Total.ShouldBeGreaterThanOrEqualTo(3);
subsz.Subs.Length.ShouldBeGreaterThanOrEqualTo(3);
}
/// <summary>
/// Go: TestMonitorSubszDetails (line 1609).
/// Verifies subscription detail entries contain the correct subject names.
/// </summary>
[Fact]
public async Task Subz_detail_entries_contain_subject_names()
{
using var sock = await ConnectClientAsync("SUB foo.bar 1\r\nSUB foo.baz 2\r\n");
await Task.Delay(200);
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz?subs=1");
subsz.ShouldNotBeNull();
subsz.Subs.ShouldContain(s => s.Subject == "foo.bar");
subsz.Subs.ShouldContain(s => s.Subject == "foo.baz");
}
/// <summary>
/// Go: TestMonitorSubszWithOffsetAndLimit (line 1642).
/// Verifies /subz pagination with offset and limit parameters.
/// </summary>
[Fact]
public async Task Subz_pagination_with_offset_and_limit()
{
// Create many subscriptions
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None);
await sock.SendAsync("CONNECT {}\r\n"u8.ToArray(), SocketFlags.None);
for (var i = 0; i < 200; i++)
await sock.SendAsync(System.Text.Encoding.ASCII.GetBytes($"SUB foo.{i} {i + 1}\r\n"), SocketFlags.None);
await Task.Delay(300);
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz?subs=1&offset=10&limit=100");
subsz.ShouldNotBeNull();
// Go: sl.NumSubs != 200, sl.Total != 200, sl.Offset != 10, sl.Limit != 100, len(sl.Subs) != 100
subsz.NumSubs.ShouldBeGreaterThanOrEqualTo(200u);
subsz.Total.ShouldBeGreaterThanOrEqualTo(200);
subsz.Offset.ShouldBe(10);
subsz.Limit.ShouldBe(100);
subsz.Subs.Length.ShouldBe(100);
}
/// <summary>
/// Go: TestMonitorSubszTestPubSubject (line 1675).
/// Verifies /subz?test=foo.foo filters subscriptions matching a concrete subject.
/// </summary>
[Fact]
public async Task Subz_test_subject_filters_matching_subscriptions()
{
using var sock = await ConnectClientAsync("SUB foo.* 1\r\nSUB foo.bar 2\r\nSUB foo.foo 3\r\n");
await Task.Delay(200);
// foo.foo matches "foo.*" and "foo.foo" but not "foo.bar"
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz?subs=1&test=foo.foo");
subsz.ShouldNotBeNull();
// Go: sl.Total != 2, len(sl.Subs) != 2
subsz.Total.ShouldBe(2);
subsz.Subs.Length.ShouldBe(2);
}
/// <summary>
/// Go: TestMonitorSubszTestPubSubject (line 1675).
/// Verifies /subz?test=foo returns no matches when no subscription matches exactly.
/// </summary>
[Fact]
public async Task Subz_test_subject_no_match_returns_empty()
{
using var sock = await ConnectClientAsync("SUB foo.* 1\r\nSUB foo.bar 2\r\n");
await Task.Delay(200);
// "foo" alone does not match "foo.*" or "foo.bar"
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz?subs=1&test=foo");
subsz.ShouldNotBeNull();
subsz.Subs.Length.ShouldBe(0);
}
/// <summary>
/// Go: TestSubsz (line 1538).
/// Verifies /subz default has no subscription details (subs not requested).
/// </summary>
[Fact]
public async Task Subz_default_does_not_include_details()
{
using var sock = await ConnectClientAsync("SUB foo 1\r\n");
await Task.Delay(200);
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz");
subsz.ShouldNotBeNull();
subsz.Subs.Length.ShouldBe(0);
}
/// <summary>
/// Go: TestSubsz (line 1538).
/// Verifies /subscriptionsz works as an alias for /subz.
/// </summary>
[Fact]
public async Task Subscriptionsz_is_alias_for_subz()
{
using var sock = await ConnectClientAsync("SUB foo 1\r\n");
await Task.Delay(200);
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subscriptionsz");
subsz.ShouldNotBeNull();
subsz.Id.ShouldNotBeNullOrEmpty();
subsz.NumSubs.ShouldBeGreaterThanOrEqualTo(1u);
}
/// <summary>
/// Go: TestSubsz (line 1538).
/// Verifies /subz JSON uses correct Go-compatible field names.
/// </summary>
[Fact]
public async Task Subz_json_uses_go_field_names()
{
var body = await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/subz");
body.ShouldContain("\"server_id\"");
body.ShouldContain("\"num_subscriptions\"");
}
/// <summary>
/// Go: TestMonitorSubszDetails (line 1609).
/// Verifies subscription details include sid and cid fields.
/// </summary>
[Fact]
public async Task Subz_details_include_sid_and_cid()
{
using var sock = await ConnectClientAsync("SUB foo 99\r\n");
await Task.Delay(200);
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz?subs=1");
subsz.ShouldNotBeNull();
subsz.Subs.Length.ShouldBeGreaterThanOrEqualTo(1);
var sub = subsz.Subs.First(s => s.Subject == "foo");
sub.Sid.ShouldBe("99");
sub.Cid.ShouldBeGreaterThan(0UL);
}
/// <summary>
/// Go: TestSubsz (line 1538).
/// Verifies /subz returns HTTP 200 OK.
/// </summary>
[Fact]
public async Task Subz_returns_http_200()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/subz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
}
/// <summary>
/// Go: TestSubsz (line 1538).
/// Verifies /subz num_cache reflects the cache state of the subscription trie.
/// </summary>
[Fact]
public async Task Subz_includes_num_cache()
{
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz");
subsz.ShouldNotBeNull();
// num_cache should be >= 0
subsz.NumCache.ShouldBeGreaterThanOrEqualTo(0);
}
/// <summary>
/// Go: TestMonitorSubszWithOffsetAndLimit (line 1642).
/// Verifies /subz with offset=0 and limit=0 uses defaults.
/// </summary>
[Fact]
public async Task Subz_offset_zero_uses_default_limit()
{
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz?offset=0");
subsz.ShouldNotBeNull();
subsz.Offset.ShouldBe(0);
subsz.Limit.ShouldBe(1024); // default limit
}
/// <summary>
/// Go: TestMonitorConcurrentMonitoring (line 2148).
/// Verifies concurrent /subz requests do not cause errors.
/// </summary>
[Fact]
public async Task Subz_handles_concurrent_requests()
{
using var sock = await ConnectClientAsync("SUB foo 1\r\n");
await Task.Delay(200);
var tasks = Enumerable.Range(0, 10).Select(async _ =>
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/subz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
});
await Task.WhenAll(tasks);
}
/// <summary>
/// Go: TestMonitorSubszTestPubSubject (line 1675).
/// Verifies /subz?test with wildcard subject foo.* matches foo.bar and foo.baz.
/// </summary>
[Fact]
public async Task Subz_test_wildcard_match()
{
using var sock = await ConnectClientAsync("SUB foo.bar 1\r\nSUB foo.baz 2\r\nSUB bar.x 3\r\n");
await Task.Delay(200);
// test=foo.bar should match foo.bar literal
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz?subs=1&test=foo.bar");
subsz.ShouldNotBeNull();
subsz.Total.ShouldBe(1);
subsz.Subs.Length.ShouldBe(1);
subsz.Subs[0].Subject.ShouldBe("foo.bar");
}
/// <summary>
/// Go: TestMonitorSubszMultiAccount (line 1709).
/// Verifies /subz now timestamp is plausible.
/// </summary>
[Fact]
public async Task Subz_now_is_plausible_timestamp()
{
var before = DateTime.UtcNow;
var subsz = await _http.GetFromJsonAsync<Subsz>($"http://127.0.0.1:{_monitorPort}/subz");
var after = DateTime.UtcNow;
subsz.ShouldNotBeNull();
subsz.Now.ShouldBeGreaterThanOrEqualTo(before.AddSeconds(-1));
subsz.Now.ShouldBeLessThanOrEqualTo(after.AddSeconds(1));
}
private async Task<Socket> ConnectClientAsync(string extraCommands)
{
var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None);
await sock.SendAsync(System.Text.Encoding.ASCII.GetBytes($"CONNECT {{}}\r\n{extraCommands}"), SocketFlags.None);
return sock;
}
}

View File

@@ -0,0 +1,521 @@
// Go: TestMonitorHandleVarz server/monitor_test.go:275
// Go: TestMyUptime server/monitor_test.go:135
// Go: TestMonitorVarzSubscriptionsResetProperly server/monitor_test.go:257
// Go: TestMonitorNoPort server/monitor_test.go:168
// Go: TestMonitorHTTPBasePath server/monitor_test.go:220
// Go: TestMonitorHandleRoot server/monitor_test.go:1819
// Go: TestMonitorServerIDs server/monitor_test.go:2410
// Go: TestMonitorHttpStatsNoUpdatedWhenUsingServerFuncs server/monitor_test.go:2435
// Go: TestMonitorVarzRaces server/monitor_test.go:2641
using System.Net;
using System.Net.Http.Json;
using System.Net.Sockets;
using System.Text.Json;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Monitoring;
using NATS.Server.TestUtilities;
namespace NATS.Server.Monitoring.Tests.Monitoring;
/// <summary>
/// Tests covering /varz endpoint behavior, ported from the Go server's monitor_test.go.
/// </summary>
public class MonitorVarzTests : IAsyncLifetime
{
private readonly NatsServer _server;
private readonly int _natsPort;
private readonly int _monitorPort;
private readonly CancellationTokenSource _cts = new();
private readonly HttpClient _http = new();
public MonitorVarzTests()
{
_natsPort = TestPortAllocator.GetFreePort();
_monitorPort = TestPortAllocator.GetFreePort();
_server = new NatsServer(
new NatsOptions { Port = _natsPort, MonitorPort = _monitorPort },
NullLoggerFactory.Instance);
}
public async Task InitializeAsync()
{
_ = _server.StartAsync(_cts.Token);
await _server.WaitForReadyAsync();
for (var i = 0; i < 50; i++)
{
try
{
var probe = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
if (probe.IsSuccessStatusCode) break;
}
catch (HttpRequestException) { }
await Task.Delay(50);
}
}
public async Task DisposeAsync()
{
_http.Dispose();
await _cts.CancelAsync();
_server.Dispose();
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275), mode=0.
/// Verifies /varz returns valid JSON with server identity fields including
/// server_id, version, start time within 10s, host, port, max_payload.
/// </summary>
[Fact]
public async Task Varz_returns_server_identity_and_start_within_10_seconds()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/varz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var varz = await response.Content.ReadFromJsonAsync<Varz>();
varz.ShouldNotBeNull();
varz.Id.ShouldNotBeNullOrEmpty();
varz.Version.ShouldNotBeNullOrEmpty();
// Go: if time.Since(v.Start) > 10*time.Second { t.Fatal(...) }
(DateTime.UtcNow - varz.Start).ShouldBeLessThan(TimeSpan.FromSeconds(10));
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275), after connecting client.
/// Verifies /varz tracks connections, in_msgs, out_msgs, in_bytes, out_bytes
/// after a client connects, subscribes, and publishes.
/// </summary>
[Fact]
public async Task Varz_tracks_connection_stats_after_client_pubsub()
{
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None);
// Subscribe, publish 5-byte payload "hello", then flush
await sock.SendAsync("CONNECT {}\r\nSUB foo 1\r\nPUB foo 5\r\nhello\r\n"u8.ToArray(), SocketFlags.None);
await Task.Delay(200);
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
// Go: v.Connections != 1
varz.Connections.ShouldBeGreaterThanOrEqualTo(1);
// Go: v.TotalConnections < 1
varz.TotalConnections.ShouldBeGreaterThanOrEqualTo(1UL);
// Go: v.InMsgs != 1
varz.InMsgs.ShouldBeGreaterThanOrEqualTo(1L);
// Go: v.InBytes != 5
varz.InBytes.ShouldBeGreaterThanOrEqualTo(5L);
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies that /varz reports subscriptions count after a client subscribes.
/// </summary>
[Fact]
public async Task Varz_reports_subscription_count()
{
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None);
await sock.SendAsync("CONNECT {}\r\nSUB test 1\r\nSUB test2 2\r\n"u8.ToArray(), SocketFlags.None);
await Task.Delay(200);
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.Subscriptions.ShouldBeGreaterThanOrEqualTo(2u);
}
/// <summary>
/// Go: TestMonitorVarzSubscriptionsResetProperly (line 257).
/// Verifies /varz subscriptions count remains stable across multiple calls,
/// and does not double on each request.
/// </summary>
[Fact]
public async Task Varz_subscriptions_do_not_double_across_repeated_calls()
{
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None);
await sock.SendAsync("CONNECT {}\r\nSUB test 1\r\n"u8.ToArray(), SocketFlags.None);
await Task.Delay(200);
var varz1 = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
var subs1 = varz1!.Subscriptions;
var varz2 = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
var subs2 = varz2!.Subscriptions;
// Go: check that we get same number back (not doubled)
subs2.ShouldBe(subs1);
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz exposes JetStream config and stats sections.
/// </summary>
[Fact]
public async Task Varz_includes_jetstream_section()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.JetStream.ShouldNotBeNull();
varz.JetStream.Config.ShouldNotBeNull();
varz.JetStream.Stats.ShouldNotBeNull();
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz includes runtime metrics: mem > 0, cores > 0.
/// </summary>
[Fact]
public async Task Varz_includes_runtime_metrics()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.Mem.ShouldBeGreaterThan(0L);
varz.Cores.ShouldBeGreaterThan(0);
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz uptime string is non-empty and matches expected format (e.g. "0s", "1m2s").
/// </summary>
[Fact]
public async Task Varz_uptime_is_formatted_string()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.Uptime.ShouldNotBeNullOrEmpty();
// Uptime should end with 's' (seconds), matching Go format like "0s", "1m0s"
varz.Uptime.ShouldEndWith("s");
}
/// <summary>
/// Go: TestMyUptime (line 135).
/// Verifies the uptime formatting logic produces correct duration strings.
/// Tests: 22s, 4m22s, 4h4m22s, 32d4h4m22s.
/// </summary>
[Theory]
[InlineData(22, "22s")]
[InlineData(22 + 4 * 60, "4m22s")]
[InlineData(22 + 4 * 60 + 4 * 3600, "4h4m22s")]
[InlineData(22 + 4 * 60 + 4 * 3600 + 32 * 86400, "32d4h4m22s")]
public void Uptime_format_matches_go_myUptime(int totalSeconds, string expected)
{
var ts = TimeSpan.FromSeconds(totalSeconds);
var result = FormatUptime(ts);
result.ShouldBe(expected);
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz serializes with correct Go JSON field names.
/// </summary>
[Fact]
public async Task Varz_json_uses_go_field_names()
{
var response = await _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}/varz");
response.ShouldContain("\"server_id\"");
response.ShouldContain("\"server_name\"");
response.ShouldContain("\"in_msgs\"");
response.ShouldContain("\"out_msgs\"");
response.ShouldContain("\"in_bytes\"");
response.ShouldContain("\"out_bytes\"");
response.ShouldContain("\"max_payload\"");
response.ShouldContain("\"total_connections\"");
response.ShouldContain("\"slow_consumers\"");
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz includes nested configuration sections for cluster, gateway, leaf.
/// </summary>
[Fact]
public async Task Varz_includes_cluster_gateway_leaf_sections()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.Cluster.ShouldNotBeNull();
varz.Gateway.ShouldNotBeNull();
varz.Leaf.ShouldNotBeNull();
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz max_payload defaults to 1MB.
/// </summary>
[Fact]
public async Task Varz_max_payload_defaults_to_1MB()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.MaxPayload.ShouldBe(1024 * 1024);
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz host and port match the configured values.
/// </summary>
[Fact]
public async Task Varz_host_and_port_match_configuration()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.Port.ShouldBe(_natsPort);
varz.Host.ShouldNotBeNullOrEmpty();
}
/// <summary>
/// Go: TestMonitorServerIDs (line 2410).
/// Verifies /varz and /connz both expose the same server_id.
/// </summary>
[Fact]
public async Task Varz_and_connz_report_matching_server_id()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
var connz = await _http.GetFromJsonAsync<Connz>($"http://127.0.0.1:{_monitorPort}/connz");
varz.ShouldNotBeNull();
connz.ShouldNotBeNull();
varz.Id.ShouldNotBeNullOrEmpty();
connz.Id.ShouldBe(varz.Id);
}
/// <summary>
/// Go: TestMonitorHttpStatsNoUpdatedWhenUsingServerFuncs (line 2435).
/// Verifies /varz http_req_stats tracks endpoint hit counts and increments on each call.
/// </summary>
[Fact]
public async Task Varz_http_req_stats_increment_on_each_request()
{
// First request establishes baseline
await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/varz");
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.HttpReqStats.ShouldContainKey("/varz");
var count = varz.HttpReqStats["/varz"];
count.ShouldBeGreaterThanOrEqualTo(2UL);
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz includes slow_consumer_stats section with breakdown fields.
/// </summary>
[Fact]
public async Task Varz_includes_slow_consumer_stats_breakdown()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.SlowConsumerStats.ShouldNotBeNull();
varz.SlowConsumerStats.Clients.ShouldBeGreaterThanOrEqualTo(0UL);
varz.SlowConsumerStats.Routes.ShouldBeGreaterThanOrEqualTo(0UL);
varz.SlowConsumerStats.Gateways.ShouldBeGreaterThanOrEqualTo(0UL);
varz.SlowConsumerStats.Leafs.ShouldBeGreaterThanOrEqualTo(0UL);
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz includes proto version field.
/// </summary>
[Fact]
public async Task Varz_includes_proto_version()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.Proto.ShouldBeGreaterThanOrEqualTo(0);
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz config_load_time is set.
/// </summary>
[Fact]
public async Task Varz_config_load_time_is_set()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.ConfigLoadTime.ShouldBeGreaterThan(DateTime.MinValue);
}
/// <summary>
/// Go: TestMonitorVarzRaces (line 2641).
/// Verifies concurrent /varz requests do not cause errors or data corruption.
/// </summary>
[Fact]
public async Task Varz_handles_concurrent_requests_without_errors()
{
var tasks = Enumerable.Range(0, 10).Select(async _ =>
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/varz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var v = await response.Content.ReadFromJsonAsync<Varz>();
v.ShouldNotBeNull();
v.Id.ShouldNotBeNullOrEmpty();
});
await Task.WhenAll(tasks);
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz out_msgs increments when messages are delivered to subscribers.
/// </summary>
[Fact]
public async Task Varz_out_msgs_increments_on_delivery()
{
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None);
// Subscribe then publish to matched subject
await sock.SendAsync("CONNECT {}\r\nSUB foo 1\r\nPUB foo 5\r\nhello\r\n"u8.ToArray(), SocketFlags.None);
await Task.Delay(200);
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
// Message was published and delivered to the subscriber, so out_msgs >= 1
varz.OutMsgs.ShouldBeGreaterThanOrEqualTo(1L);
varz.OutBytes.ShouldBeGreaterThanOrEqualTo(5L);
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz includes MQTT section in response.
/// </summary>
[Fact]
public async Task Varz_includes_mqtt_section()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.Mqtt.ShouldNotBeNull();
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz includes websocket section.
/// </summary>
[Fact]
public async Task Varz_includes_websocket_section()
{
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
varz.Websocket.ShouldNotBeNull();
}
/// <summary>
/// Go: TestMonitorHandleRoot (line 1819).
/// Verifies GET / returns a listing of available monitoring endpoints.
/// </summary>
[Fact]
public async Task Root_endpoint_returns_endpoint_listing()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var body = await response.Content.ReadAsStringAsync();
body.ShouldContain("varz");
body.ShouldContain("connz");
body.ShouldContain("healthz");
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz total_connections tracks cumulative connections, not just active.
/// </summary>
[Fact]
public async Task Varz_total_connections_tracks_cumulative_count()
{
// Connect and disconnect a client
var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None);
await sock.SendAsync("CONNECT {}\r\n"u8.ToArray(), SocketFlags.None);
await Task.Delay(100);
sock.Shutdown(SocketShutdown.Both);
sock.Dispose();
await Task.Delay(300);
// Connect a second client (still active)
using var sock2 = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock2.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
buf = new byte[4096];
_ = await sock2.ReceiveAsync(buf, SocketFlags.None);
await sock2.SendAsync("CONNECT {}\r\n"u8.ToArray(), SocketFlags.None);
await Task.Delay(200);
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
varz.ShouldNotBeNull();
// Total should be >= 2 (both connections counted), active should be 1
varz.TotalConnections.ShouldBeGreaterThanOrEqualTo(2UL);
varz.Connections.ShouldBeGreaterThanOrEqualTo(1);
}
/// <summary>
/// Go: TestMonitorNoPort (line 168).
/// Verifies that when no monitor port is configured, monitoring endpoints are not accessible.
/// This is a standalone test since it uses a different server configuration.
/// </summary>
[Fact]
public async Task Monitor_not_accessible_when_port_not_configured()
{
var natsPort = TestPortAllocator.GetFreePort();
var server = new NatsServer(
new NatsOptions { Port = natsPort, MonitorPort = 0 },
NullLoggerFactory.Instance);
var cts = new CancellationTokenSource();
_ = server.StartAsync(cts.Token);
await server.WaitForReadyAsync();
try
{
using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(2) };
// Try a random port where no monitor should be running
var act = async () => await http.GetAsync("http://127.0.0.1:11245/varz");
await act.ShouldThrowAsync<Exception>();
}
finally
{
await cts.CancelAsync();
server.Dispose();
}
}
/// <summary>
/// Go: TestMonitorHandleVarz (line 275).
/// Verifies /varz now field returns a plausible UTC timestamp.
/// </summary>
[Fact]
public async Task Varz_now_is_plausible_utc_timestamp()
{
var before = DateTime.UtcNow;
var varz = await _http.GetFromJsonAsync<Varz>($"http://127.0.0.1:{_monitorPort}/varz");
var after = DateTime.UtcNow;
varz.ShouldNotBeNull();
varz.Now.ShouldBeGreaterThanOrEqualTo(before.AddSeconds(-1));
varz.Now.ShouldBeLessThanOrEqualTo(after.AddSeconds(1));
}
// Helper: matches Go server myUptime() format
private static string FormatUptime(TimeSpan ts)
{
if (ts.TotalDays >= 1)
return $"{(int)ts.TotalDays}d{ts.Hours}h{ts.Minutes}m{ts.Seconds}s";
if (ts.TotalHours >= 1)
return $"{(int)ts.TotalHours}h{ts.Minutes}m{ts.Seconds}s";
if (ts.TotalMinutes >= 1)
return $"{(int)ts.TotalMinutes}m{ts.Seconds}s";
return $"{(int)ts.TotalSeconds}s";
}
}

View File

@@ -0,0 +1,39 @@
using System.Text.Json;
using NATS.Server.Monitoring;
namespace NATS.Server.Monitoring.Tests.Monitoring;
public class MonitoringHealthAndSortParityBatch1Tests
{
[Fact]
public void SortOpt_IsValid_matches_defined_values()
{
foreach (var value in Enum.GetValues<SortOpt>())
value.IsValid().ShouldBeTrue();
((SortOpt)999).IsValid().ShouldBeFalse();
}
[Fact]
public void HealthStatus_ok_serializes_with_go_shape_fields()
{
var json = JsonSerializer.Serialize(HealthStatus.Ok());
json.ShouldContain("\"status\":\"ok\"");
json.ShouldContain("\"status_code\":200");
json.ShouldContain("\"errors\":[]");
}
[Fact]
public void HealthzError_serializes_enum_as_string()
{
var json = JsonSerializer.Serialize(new HealthzError
{
Type = HealthzErrorType.JetStream,
Error = "jetstream unavailable",
});
json.ShouldContain("\"type\":\"JetStream\"");
json.ShouldContain("\"error\":\"jetstream unavailable\"");
}
}

View File

@@ -0,0 +1,87 @@
using System.Net;
using System.Net.Sockets;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.TestUtilities;
namespace NATS.Server.Monitoring.Tests;
public class PprofEndpointTests
{
[Fact]
public async Task Debug_pprof_endpoint_returns_profile_index_when_profport_enabled()
{
await using var fx = await PprofMonitorFixture.StartWithProfilingAsync();
var body = await fx.GetStringAsync("/debug/pprof");
body.ShouldContain("profiles");
}
}
internal sealed class PprofMonitorFixture : IAsyncDisposable
{
private readonly NatsServer _server;
private readonly CancellationTokenSource _cts;
private readonly HttpClient _http;
private readonly int _monitorPort;
private PprofMonitorFixture(NatsServer server, CancellationTokenSource cts, HttpClient http, int monitorPort)
{
_server = server;
_cts = cts;
_http = http;
_monitorPort = monitorPort;
}
public static async Task<PprofMonitorFixture> StartWithProfilingAsync()
{
var monitorPort = TestPortAllocator.GetFreePort();
var options = new NatsOptions
{
Host = "127.0.0.1",
Port = 0,
MonitorPort = monitorPort,
ProfPort = monitorPort,
};
var server = new NatsServer(options, NullLoggerFactory.Instance);
var cts = new CancellationTokenSource();
_ = server.StartAsync(cts.Token);
await server.WaitForReadyAsync();
var http = new HttpClient();
for (var i = 0; i < 50; i++)
{
try
{
var response = await http.GetAsync($"http://127.0.0.1:{monitorPort}/healthz");
if (response.IsSuccessStatusCode)
break;
}
catch
{
}
await Task.Delay(50);
}
return new PprofMonitorFixture(server, cts, http, monitorPort);
}
public Task<string> GetStringAsync(string path)
{
return _http.GetStringAsync($"http://127.0.0.1:{_monitorPort}{path}");
}
public Task<byte[]> GetBytesAsync(string path)
{
return _http.GetByteArrayAsync($"http://127.0.0.1:{_monitorPort}{path}");
}
public async ValueTask DisposeAsync()
{
_http.Dispose();
await _cts.CancelAsync();
_server.Dispose();
_cts.Dispose();
}
}

View File

@@ -0,0 +1,18 @@
using System.Text.Json;
namespace NATS.Server.Monitoring.Tests.Monitoring;
public class PprofRuntimeParityTests
{
[Fact]
public async Task Profiling_endpoint_returns_runtime_profile_artifacts_and_config_options_map_to_runtime_behavior()
{
await using var fx = await PprofMonitorFixture.StartWithProfilingAsync();
var payload = await fx.GetBytesAsync("/debug/pprof/profile?seconds=2");
var doc = JsonDocument.Parse(payload);
doc.RootElement.GetProperty("profile").GetString().ShouldBe("cpu");
doc.RootElement.GetProperty("seconds").GetInt32().ShouldBe(2);
doc.RootElement.GetProperty("thread_count").GetInt32().ShouldBeGreaterThan(0);
}
}

View File

@@ -0,0 +1,65 @@
using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
using System.Text.Json;
using NATS.Server.Monitoring;
namespace NATS.Server.Monitoring.Tests.Monitoring;
public class TlsPeerCertParityTests
{
[Fact]
public void TLSPeerCert_serializes_go_shape_fields()
{
var cert = new TLSPeerCert
{
Subject = "CN=peer",
SubjectPKISha256 = new string('a', 64),
CertSha256 = new string('b', 64),
};
var json = JsonSerializer.Serialize(cert);
json.ShouldContain("\"subject\":\"CN=peer\"");
json.ShouldContain("\"subject_pk_sha256\":");
json.ShouldContain("\"cert_sha256\":");
}
[Fact]
public void TlsPeerCertMapper_produces_subject_and_sha256_values_from_certificate()
{
using var rsa = RSA.Create(2048);
var req = new CertificateRequest("CN=peer", rsa, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1);
using var cert = req.CreateSelfSigned(DateTimeOffset.UtcNow.AddDays(-1), DateTimeOffset.UtcNow.AddDays(1));
var mapped = TlsPeerCertMapper.FromCertificate(cert);
mapped.Length.ShouldBe(1);
mapped[0].Subject.ShouldContain("CN=peer");
mapped[0].SubjectPKISha256.Length.ShouldBe(64);
mapped[0].CertSha256.Length.ShouldBe(64);
}
[Fact]
public void ConnInfo_json_includes_tls_peer_certs_array()
{
var info = new ConnInfo
{
Cid = 1,
TlsPeerCertSubject = "CN=peer",
TlsPeerCerts =
[
new TLSPeerCert
{
Subject = "CN=peer",
SubjectPKISha256 = new string('c', 64),
CertSha256 = new string('d', 64),
},
],
};
var json = JsonSerializer.Serialize(info);
json.ShouldContain("\"tls_peer_certs\":[");
json.ShouldContain("\"subject_pk_sha256\":");
json.ShouldContain("\"cert_sha256\":");
}
}

View File

@@ -0,0 +1,132 @@
// Ported from golang/nats-server/server/monitor_test.go
// TestMonitorHandleVarz — verify /varz returns valid server identity fields and tracks message stats.
using System.Net;
using System.Net.Http.Json;
using System.Net.Sockets;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Server.Monitoring;
using NATS.Server.TestUtilities;
namespace NATS.Server.Monitoring.Tests;
public class VarzParityTests : IAsyncLifetime
{
private readonly NatsServer _server;
private readonly int _natsPort;
private readonly int _monitorPort;
private readonly CancellationTokenSource _cts = new();
private readonly HttpClient _http = new();
public VarzParityTests()
{
_natsPort = TestPortAllocator.GetFreePort();
_monitorPort = TestPortAllocator.GetFreePort();
_server = new NatsServer(
new NatsOptions { Port = _natsPort, MonitorPort = _monitorPort },
NullLoggerFactory.Instance);
}
public async Task InitializeAsync()
{
_ = _server.StartAsync(_cts.Token);
await _server.WaitForReadyAsync();
for (var i = 0; i < 50; i++)
{
try
{
var probe = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/healthz");
if (probe.IsSuccessStatusCode) break;
}
catch (HttpRequestException) { }
await Task.Delay(50);
}
}
public async Task DisposeAsync()
{
_http.Dispose();
await _cts.CancelAsync();
_server.Dispose();
}
/// <summary>
/// Corresponds to Go TestMonitorHandleVarz (first block, mode=0).
/// Verifies the /varz endpoint returns valid JSON containing required server identity fields:
/// server_id, version, now, start, host, port, max_payload, mem, cores.
/// </summary>
[Fact]
public async Task Varz_returns_valid_json_with_server_info()
{
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/varz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var varz = await response.Content.ReadFromJsonAsync<Varz>();
varz.ShouldNotBeNull();
// server_id must be present and non-empty
varz.Id.ShouldNotBeNullOrEmpty();
// version must be present
varz.Version.ShouldNotBeNullOrEmpty();
// now must be a plausible timestamp (not default DateTime.MinValue)
varz.Now.ShouldBeGreaterThan(DateTime.MinValue);
// start must be within a reasonable window of now
(DateTime.UtcNow - varz.Start).ShouldBeLessThan(TimeSpan.FromSeconds(30));
// host and port must reflect server configuration
varz.Host.ShouldNotBeNullOrEmpty();
varz.Port.ShouldBe(_natsPort);
// max_payload is 1 MB by default (Go reference: defaultMaxPayload = 1MB)
varz.MaxPayload.ShouldBe(1024 * 1024);
// uptime must be non-empty
varz.Uptime.ShouldNotBeNullOrEmpty();
// runtime metrics must be populated
varz.Mem.ShouldBeGreaterThan(0L);
varz.Cores.ShouldBeGreaterThan(0);
}
/// <summary>
/// Corresponds to Go TestMonitorHandleVarz (second block after connecting a client).
/// Verifies /varz correctly tracks connections, total_connections, in_msgs, in_bytes
/// after a client connects, subscribes, and publishes a message.
/// </summary>
[Fact]
public async Task Varz_tracks_connections_and_messages()
{
using var sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
await sock.ConnectAsync(new IPEndPoint(IPAddress.Loopback, _natsPort));
var buf = new byte[4096];
_ = await sock.ReceiveAsync(buf, SocketFlags.None); // consume INFO
// CONNECT + SUB + PUB "hello" (5 bytes) to "test"
var cmd = "CONNECT {}\r\nSUB test 1\r\nPUB test 5\r\nhello\r\n"u8.ToArray();
await sock.SendAsync(cmd, SocketFlags.None);
await Task.Delay(200);
var response = await _http.GetAsync($"http://127.0.0.1:{_monitorPort}/varz");
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var varz = await response.Content.ReadFromJsonAsync<Varz>();
varz.ShouldNotBeNull();
// At least 1 active connection
varz.Connections.ShouldBeGreaterThanOrEqualTo(1);
// Total connections must have been counted
varz.TotalConnections.ShouldBeGreaterThanOrEqualTo(1UL);
// in_msgs: at least the 1 PUB we sent
varz.InMsgs.ShouldBeGreaterThanOrEqualTo(1L);
// in_bytes: at least 5 bytes ("hello")
varz.InBytes.ShouldBeGreaterThanOrEqualTo(5L);
}
}

View File

@@ -0,0 +1,17 @@
namespace NATS.Server.Monitoring.Tests;
public class VarzSlowConsumerBreakdownTests
{
[Fact]
public async Task Varz_contains_slow_consumer_breakdown_fields()
{
await using var fx = await MonitoringParityFixture.StartAsync();
var varz = await fx.GetVarzAsync();
varz.SlowConsumerStats.ShouldNotBeNull();
varz.SlowConsumerStats.Clients.ShouldBeGreaterThanOrEqualTo((ulong)0);
varz.SlowConsumerStats.Routes.ShouldBeGreaterThanOrEqualTo((ulong)0);
varz.SlowConsumerStats.Gateways.ShouldBeGreaterThanOrEqualTo((ulong)0);
varz.SlowConsumerStats.Leafs.ShouldBeGreaterThanOrEqualTo((ulong)0);
}
}