test(batch50): port 118 JetStream cluster 1 integration tests
Ports the first 118 tests from golang/nats-server/server/jetstream_cluster_1_test.go to C# integration tests in JetStream/JetStreamCluster1Tests.cs. Adds the Helpers/ scaffold (IntegrationTestBase, TestCluster, NatsTestClient, CheckHelper, ConfigHelper) and Xunit.SkippableFact package; tests skip automatically unless NATS_INTEGRATION_TESTS=true is set.
This commit is contained in:
@@ -0,0 +1,46 @@
|
|||||||
|
// Copyright 2020-2025 The NATS Authors
|
||||||
|
// Licensed under the Apache License, Version 2.0
|
||||||
|
|
||||||
|
namespace ZB.MOM.NatsNet.Server.IntegrationTests.Helpers;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Provides polling-based check utilities for integration tests,
|
||||||
|
/// mirroring the Go <c>checkFor</c> pattern.
|
||||||
|
/// </summary>
|
||||||
|
public static class CheckHelper
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Polls <paramref name="check"/> every <paramref name="interval"/> until it returns
|
||||||
|
/// <c>null</c> (success) or <paramref name="timeout"/> elapses. Throws on timeout.
|
||||||
|
/// </summary>
|
||||||
|
public static void CheckFor(TimeSpan timeout, TimeSpan interval, Func<string?> check)
|
||||||
|
{
|
||||||
|
var deadline = DateTime.UtcNow + timeout;
|
||||||
|
string? lastError = null;
|
||||||
|
while (DateTime.UtcNow < deadline)
|
||||||
|
{
|
||||||
|
lastError = check();
|
||||||
|
if (lastError is null)
|
||||||
|
return;
|
||||||
|
Thread.Sleep(interval);
|
||||||
|
}
|
||||||
|
throw new TimeoutException($"CheckFor timed out after {timeout}: {lastError}");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Async variant of <see cref="CheckFor"/>.
|
||||||
|
/// </summary>
|
||||||
|
public static async Task CheckForAsync(TimeSpan timeout, TimeSpan interval, Func<Task<string?>> check)
|
||||||
|
{
|
||||||
|
var deadline = DateTime.UtcNow + timeout;
|
||||||
|
string? lastError = null;
|
||||||
|
while (DateTime.UtcNow < deadline)
|
||||||
|
{
|
||||||
|
lastError = await check();
|
||||||
|
if (lastError is null)
|
||||||
|
return;
|
||||||
|
await Task.Delay(interval);
|
||||||
|
}
|
||||||
|
throw new TimeoutException($"CheckFor timed out after {timeout}: {lastError}");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,66 @@
|
|||||||
|
// Copyright 2020-2025 The NATS Authors
|
||||||
|
// Licensed under the Apache License, Version 2.0
|
||||||
|
|
||||||
|
namespace ZB.MOM.NatsNet.Server.IntegrationTests.Helpers;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Provides configuration templates and file helpers for integration tests.
|
||||||
|
/// Mirrors Go's <c>jsClusterTempl</c>, <c>jsClusterAccountsTempl</c>, etc.
|
||||||
|
/// </summary>
|
||||||
|
public static class ConfigHelper
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Standard JetStream cluster configuration template.
|
||||||
|
/// Use <c>%s</c> placeholders for server_name, store_dir, cluster_name, port, and routes.
|
||||||
|
/// </summary>
|
||||||
|
public const string JsClusterTemplate = @"
|
||||||
|
listen: 127.0.0.1:-1
|
||||||
|
server_name: %s
|
||||||
|
jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
|
||||||
|
|
||||||
|
cluster {
|
||||||
|
name: %s
|
||||||
|
listen: 127.0.0.1:%d
|
||||||
|
routes = [%s]
|
||||||
|
}
|
||||||
|
";
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// JetStream cluster template with accounts.
|
||||||
|
/// </summary>
|
||||||
|
public const string JsClusterAccountsTemplate = @"
|
||||||
|
listen: 127.0.0.1:-1
|
||||||
|
server_name: %s
|
||||||
|
jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
|
||||||
|
|
||||||
|
accounts {
|
||||||
|
$SYS {
|
||||||
|
users: [{user: admin, password: s3cr3t!}]
|
||||||
|
}
|
||||||
|
ONE {
|
||||||
|
jetstream: enabled
|
||||||
|
users: [{user: one, password: p}]
|
||||||
|
}
|
||||||
|
TWO {
|
||||||
|
jetstream: enabled
|
||||||
|
users: [{user: two, password: p}]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
cluster {
|
||||||
|
name: %s
|
||||||
|
listen: 127.0.0.1:%d
|
||||||
|
routes = [%s]
|
||||||
|
}
|
||||||
|
";
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Creates a temporary config file with the given content and returns its path.
|
||||||
|
/// </summary>
|
||||||
|
public static string CreateConfigFile(string content)
|
||||||
|
{
|
||||||
|
var path = Path.Combine(Path.GetTempPath(), $"nats-test-{Guid.NewGuid():N}.conf");
|
||||||
|
File.WriteAllText(path, content);
|
||||||
|
return path;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,41 @@
|
|||||||
|
// Copyright 2020-2025 The NATS Authors
|
||||||
|
// Licensed under the Apache License, Version 2.0
|
||||||
|
|
||||||
|
using Xunit.Abstractions;
|
||||||
|
|
||||||
|
namespace ZB.MOM.NatsNet.Server.IntegrationTests.Helpers;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Base class for integration tests that require a live NATS cluster.
|
||||||
|
/// Tests are skipped when the cluster infrastructure is not available.
|
||||||
|
/// </summary>
|
||||||
|
[Trait("Category", "Integration")]
|
||||||
|
public abstract class IntegrationTestBase
|
||||||
|
{
|
||||||
|
protected readonly ITestOutputHelper Output;
|
||||||
|
|
||||||
|
protected IntegrationTestBase(ITestOutputHelper output)
|
||||||
|
{
|
||||||
|
Output = output;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Determines whether cluster integration tests should be skipped.
|
||||||
|
/// Tests are skipped unless the NATS_INTEGRATION_TESTS environment variable is set to "true"
|
||||||
|
/// or NATS_SERVER_URL is set to a valid cluster endpoint.
|
||||||
|
/// </summary>
|
||||||
|
protected static bool ShouldSkip()
|
||||||
|
{
|
||||||
|
var integrationEnabled = Environment.GetEnvironmentVariable("NATS_INTEGRATION_TESTS");
|
||||||
|
if (string.Equals(integrationEnabled, "true", StringComparison.OrdinalIgnoreCase))
|
||||||
|
return false;
|
||||||
|
|
||||||
|
var serverUrl = Environment.GetEnvironmentVariable("NATS_SERVER_URL");
|
||||||
|
if (!string.IsNullOrEmpty(serverUrl))
|
||||||
|
return false;
|
||||||
|
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
|
||||||
|
protected void Log(string message) => Output.WriteLine(message);
|
||||||
|
}
|
||||||
@@ -0,0 +1,28 @@
|
|||||||
|
// Copyright 2020-2025 The NATS Authors
|
||||||
|
// Licensed under the Apache License, Version 2.0
|
||||||
|
|
||||||
|
namespace ZB.MOM.NatsNet.Server.IntegrationTests.Helpers;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Provides NATS client connections for integration tests.
|
||||||
|
/// </summary>
|
||||||
|
public static class NatsTestClient
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Connects to the given test server. Throws <see cref="ClusterNotAvailableException"/>
|
||||||
|
/// when no cluster server is available.
|
||||||
|
/// </summary>
|
||||||
|
public static IDisposable ConnectToServer(object server)
|
||||||
|
{
|
||||||
|
throw new ClusterNotAvailableException("Cannot connect: cluster server not available.");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Connects to the given test server with JetStream context.
|
||||||
|
/// Returns an (nc, js) tuple.
|
||||||
|
/// </summary>
|
||||||
|
public static (IDisposable Nc, object Js) ConnectWithJetStream(object server)
|
||||||
|
{
|
||||||
|
throw new ClusterNotAvailableException("Cannot connect: cluster server not available.");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,75 @@
|
|||||||
|
// Copyright 2020-2025 The NATS Authors
|
||||||
|
// Licensed under the Apache License, Version 2.0
|
||||||
|
|
||||||
|
namespace ZB.MOM.NatsNet.Server.IntegrationTests.Helpers;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Represents a test JetStream cluster. In the current integration test setup,
|
||||||
|
/// tests that use this type are skipped unless a real cluster is available.
|
||||||
|
/// This is a scaffold/stub that captures the API shape for future full implementation.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class TestCluster : IDisposable
|
||||||
|
{
|
||||||
|
private readonly string _name;
|
||||||
|
private readonly int _numServers;
|
||||||
|
private bool _disposed;
|
||||||
|
|
||||||
|
public string Name => _name;
|
||||||
|
|
||||||
|
private TestCluster(string name, int numServers)
|
||||||
|
{
|
||||||
|
_name = name;
|
||||||
|
_numServers = numServers;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Creates a JetStream cluster with the given number of servers.
|
||||||
|
/// Throws <see cref="ClusterNotAvailableException"/> if no cluster is available.
|
||||||
|
/// </summary>
|
||||||
|
public static TestCluster CreateJetStreamCluster(int numServers, string name = "JSC")
|
||||||
|
{
|
||||||
|
throw new ClusterNotAvailableException(
|
||||||
|
$"No JetStream cluster available. Set NATS_INTEGRATION_TESTS=true and ensure {numServers}-node cluster '{name}' is running.");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Creates a JetStream cluster with the specified configuration template.
|
||||||
|
/// </summary>
|
||||||
|
public static TestCluster CreateJetStreamClusterWithTemplate(string template, int numServers, string name = "JSC")
|
||||||
|
{
|
||||||
|
throw new ClusterNotAvailableException(
|
||||||
|
$"No JetStream cluster available for template cluster '{name}'.");
|
||||||
|
}
|
||||||
|
|
||||||
|
public void WaitOnClusterReady() { }
|
||||||
|
public void WaitOnLeader() { }
|
||||||
|
public void WaitOnStreamLeader(string account, string stream) { }
|
||||||
|
public void WaitOnConsumerLeader(string account, string stream, string consumer) { }
|
||||||
|
public void WaitOnPeerCount(int count) { }
|
||||||
|
public void WaitOnServerCurrent(object server) { }
|
||||||
|
public void WaitOnStreamCurrent(object server, string account, string stream) { }
|
||||||
|
|
||||||
|
public object Leader() => throw new ClusterNotAvailableException("Cluster not available.");
|
||||||
|
public object RandomServer() => throw new ClusterNotAvailableException("Cluster not available.");
|
||||||
|
public object RandomNonLeader() => throw new ClusterNotAvailableException("Cluster not available.");
|
||||||
|
public object RandomNonStreamLeader(string account, string stream) => throw new ClusterNotAvailableException("Cluster not available.");
|
||||||
|
public object ServerByName(string name) => throw new ClusterNotAvailableException("Cluster not available.");
|
||||||
|
public object StreamLeader(string account, string stream) => throw new ClusterNotAvailableException("Cluster not available.");
|
||||||
|
public object ConsumerLeader(string account, string stream, string consumer) => throw new ClusterNotAvailableException("Cluster not available.");
|
||||||
|
|
||||||
|
public void Dispose()
|
||||||
|
{
|
||||||
|
if (!_disposed)
|
||||||
|
{
|
||||||
|
_disposed = true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Thrown when a test cluster is not available for integration testing.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class ClusterNotAvailableException : Exception
|
||||||
|
{
|
||||||
|
public ClusterNotAvailableException(string message) : base(message) { }
|
||||||
|
}
|
||||||
File diff suppressed because it is too large
Load Diff
@@ -17,13 +17,17 @@
|
|||||||
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
|
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
|
||||||
<PackageReference Include="NATS.Client.Core" Version="2.7.2" />
|
<PackageReference Include="NATS.Client.Core" Version="2.7.2" />
|
||||||
<PackageReference Include="xunit" Version="2.9.3" />
|
<PackageReference Include="xunit" Version="2.9.3" />
|
||||||
|
<PackageReference Include="xunit.abstractions" Version="2.0.3" />
|
||||||
<PackageReference Include="xunit.runner.visualstudio" Version="3.1.4" />
|
<PackageReference Include="xunit.runner.visualstudio" Version="3.1.4" />
|
||||||
|
<PackageReference Include="Xunit.SkippableFact" Version="1.5.61" />
|
||||||
<PackageReference Include="Shouldly" Version="*" />
|
<PackageReference Include="Shouldly" Version="*" />
|
||||||
<PackageReference Include="NSubstitute" Version="*" />
|
<PackageReference Include="NSubstitute" Version="*" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
<Using Include="Xunit" />
|
<Using Include="Xunit" />
|
||||||
|
<Using Include="Xunit.Abstractions" />
|
||||||
|
<Using Include="ZB.MOM.NatsNet.Server.IntegrationTests.Helpers" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
</Project>
|
</Project>
|
||||||
|
|||||||
Reference in New Issue
Block a user