103 lines
3.2 KiB
C#
103 lines
3.2 KiB
C#
using NSubstitute;
|
|
using Shouldly;
|
|
using ZB.MOM.NatsNet.Server;
|
|
|
|
namespace ZB.MOM.NatsNet.Server.Tests.JetStream;
|
|
|
|
public sealed class NatsStreamSnapshotMonitorTests
|
|
{
|
|
[Fact]
|
|
public void RegisterPreAck_ClearPreAck_UpdatesState()
|
|
{
|
|
var stream = CreateStream(RetentionPolicy.InterestPolicy);
|
|
var consumer = new NatsConsumer("S", new ConsumerConfig { Name = "c1" }, DateTime.UtcNow);
|
|
|
|
stream.RegisterPreAck(consumer, 2);
|
|
stream.HasPreAck(consumer, 2).ShouldBeTrue();
|
|
|
|
stream.ClearPreAck(consumer, 2);
|
|
stream.HasPreAck(consumer, 2).ShouldBeFalse();
|
|
}
|
|
|
|
[Fact]
|
|
public void AckMsg_WhenSequenceAhead_ReturnsTrueAndRegistersPreAck()
|
|
{
|
|
var stream = CreateStream(RetentionPolicy.InterestPolicy);
|
|
stream.SetupStore(null).ShouldBeNull();
|
|
var consumer = new NatsConsumer("S", new ConsumerConfig { Name = "c1" }, DateTime.UtcNow);
|
|
|
|
var removed = stream.AckMsg(consumer, 50);
|
|
|
|
removed.ShouldBeTrue();
|
|
stream.HasPreAck(consumer, 50).ShouldBeTrue();
|
|
}
|
|
|
|
[Fact]
|
|
public void Snapshot_WithStore_ReturnsSnapshotResult()
|
|
{
|
|
var stream = CreateStream(RetentionPolicy.InterestPolicy);
|
|
stream.SetupStore(null).ShouldBeNull();
|
|
stream.Store!.StoreMsg("events", null, [1], ttl: 0);
|
|
|
|
var (result, error) = stream.Snapshot(TimeSpan.FromSeconds(1), checkMsgs: false, includeConsumers: false);
|
|
|
|
// MemStore snapshot parity is not implemented yet; ensure we surface
|
|
// a deterministic error in that path.
|
|
if (stream.Store.Type() == StorageType.MemoryStorage)
|
|
{
|
|
error.ShouldNotBeNull();
|
|
result.ShouldBeNull();
|
|
}
|
|
else
|
|
{
|
|
error.ShouldBeNull();
|
|
result.ShouldNotBeNull();
|
|
}
|
|
}
|
|
|
|
[Fact]
|
|
public void CheckInMonitor_ClearMonitorRunning_TogglesState()
|
|
{
|
|
var stream = CreateStream(RetentionPolicy.InterestPolicy);
|
|
|
|
stream.CheckInMonitor().ShouldBeFalse();
|
|
stream.IsMonitorRunning().ShouldBeTrue();
|
|
|
|
stream.ClearMonitorRunning();
|
|
stream.IsMonitorRunning().ShouldBeFalse();
|
|
}
|
|
|
|
[Fact]
|
|
public void CheckConsumerReplication_Mismatch_Throws()
|
|
{
|
|
var stream = CreateStream(RetentionPolicy.InterestPolicy);
|
|
stream.Config.Replicas = 3;
|
|
stream.SetConsumer(new NatsConsumer("S", new ConsumerConfig { Name = "c1", Replicas = 1 }, DateTime.UtcNow));
|
|
|
|
Should.Throw<InvalidOperationException>(stream.CheckConsumerReplication);
|
|
}
|
|
|
|
[Fact]
|
|
public void TrackReplicationTraffic_SystemAccountNode_DoesNotThrow()
|
|
{
|
|
var stream = CreateStream(RetentionPolicy.InterestPolicy);
|
|
var node = Substitute.For<IRaftNode>();
|
|
node.IsSystemAccount().Returns(true);
|
|
|
|
Should.NotThrow(() => stream.TrackReplicationTraffic(node, size: 256, replicas: 3));
|
|
}
|
|
|
|
private static NatsStream CreateStream(RetentionPolicy retention)
|
|
{
|
|
var account = new Account { Name = "A" };
|
|
var config = new StreamConfig
|
|
{
|
|
Name = "S",
|
|
Storage = StorageType.MemoryStorage,
|
|
Subjects = ["events.>"],
|
|
Retention = retention,
|
|
};
|
|
return new NatsStream(account, config, DateTime.UtcNow);
|
|
}
|
|
}
|