natsnet/dotnet/tests/ZB.MOM.NatsNet.Server.IntegrationTests/NatsServerBehaviorTests.cs

// Copyright 2012-2025 The NATS Authors
// Licensed under the Apache License, Version 2.0

using NATS.Client.Core;
using Shouldly;

namespace ZB.MOM.NatsNet.Server.IntegrationTests;

/// <summary>
/// Behavioral baseline tests against the reference Go NATS server.
/// These tests require a running Go NATS server on localhost:4222.
/// Start with: cd golang/nats-server && go run . -p 4222
/// </summary>
[Trait("Category", "Integration")]
public class NatsServerBehaviorTests : IAsyncLifetime
{
    private NatsConnection? _nats;

    public async Task InitializeAsync()
    {
        _nats = new NatsConnection(new NatsOpts { Url = "nats://localhost:4222" });
        await _nats.ConnectAsync();
    }

    public async Task DisposeAsync()
    {
        if (_nats is not null)
            await _nats.DisposeAsync();
    }

    [Fact]
    public async Task BasicPubSub_ShouldDeliverMessage()
    {
        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        var received = new TaskCompletionSource<string>();

        _ = Task.Run(async () =>
        {
            await foreach (var msg in _nats!.SubscribeAsync<string>("test.hello", cancellationToken: cts.Token))
            {
                received.TrySetResult(msg.Data ?? "");
                break;
            }
        }, cts.Token);

        // Give subscriber a moment to register
        await Task.Delay(100, cts.Token);
        await _nats!.PublishAsync("test.hello", "world");
        var result = await received.Task.WaitAsync(cts.Token);
        result.ShouldBe("world");
    }

    [Fact]
    public async Task WildcardSubscription_DotStar_ShouldMatch()
    {
        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        var received = new TaskCompletionSource<string>();

        _ = Task.Run(async () =>
        {
            await foreach (var msg in _nats!.SubscribeAsync<string>("foo.*", cancellationToken: cts.Token))
            {
                received.TrySetResult(msg.Subject);
                break;
            }
        }, cts.Token);

        await Task.Delay(100, cts.Token);
        await _nats!.PublishAsync("foo.bar", "payload");
        var subject = await received.Task.WaitAsync(cts.Token);
        subject.ShouldBe("foo.bar");
    }

    [Fact]
    public async Task WildcardSubscription_GreaterThan_ShouldMatchMultiLevel()
    {
        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        var received = new TaskCompletionSource<string>();

        _ = Task.Run(async () =>
        {
            await foreach (var msg in _nats!.SubscribeAsync<string>("foo.>", cancellationToken: cts.Token))
            {
                received.TrySetResult(msg.Subject);
                break;
            }
        }, cts.Token);

        await Task.Delay(100, cts.Token);
        await _nats!.PublishAsync("foo.bar.baz", "payload");
        var subject = await received.Task.WaitAsync(cts.Token);
        subject.ShouldBe("foo.bar.baz");
    }

    [Fact]
    public async Task QueueGroup_ShouldDeliverToOnlyOneSubscriber()
    {
        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        var count1 = 0;
        var count2 = 0;

        _ = Task.Run(async () =>
        {
            await foreach (var _ in _nats!.SubscribeAsync<string>("qg.test", queueGroup: "workers", cancellationToken: cts.Token))
                Interlocked.Increment(ref count1);
        }, cts.Token);

        _ = Task.Run(async () =>
        {
            await foreach (var _ in _nats!.SubscribeAsync<string>("qg.test", queueGroup: "workers", cancellationToken: cts.Token))
                Interlocked.Increment(ref count2);
        }, cts.Token);

        await Task.Delay(200, cts.Token);

        for (var i = 0; i < 10; i++)
            await _nats!.PublishAsync("qg.test", $"msg{i}");

        await Task.Delay(500, cts.Token);
        (count1 + count2).ShouldBe(10);
        count1.ShouldBeGreaterThan(0);
        count2.ShouldBeGreaterThan(0);
    }

    [Fact]
    public async Task ConnectDisconnect_ShouldNotThrow()
    {
        var nats2 = new NatsConnection(new NatsOpts { Url = "nats://localhost:4222" });
        await Should.NotThrowAsync(async () =>
        {
            await nats2.ConnectAsync();
            await nats2.DisposeAsync();
        });
    }
}