dohertj2/natsnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add batch 32 JS cluster meta backlog test mappings

Browse Source
This commit is contained in:
Joseph Doherty
2026-02-28 22:30:41 -05:00
parent 7e8b81b648
commit 34bc4e1137
12 changed files with 616 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

191
dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/JetStreamClusterTests1.Impltests.cs Normal file
View File

@@ -0,0 +1,191 @@
using NSubstitute;
using Shouldly;
using ZB.MOM.NatsNet.Server;
namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;
public sealed class JetStreamClusterTests1
{
[Fact] // T:772
public void JetStreamClusterConsumerState_ShouldSucceed()
{
var assignment = new ConsumerAssignment
{
Name = "C1",
Stream = "S1",
Created = DateTime.UtcNow,
Config = new ConsumerConfig { Name = "C1", Metadata = new Dictionary<string, string>() },
};
var unsupported = JetStreamCluster.NewUnsupportedConsumerAssignment(
assignment,
new InvalidOperationException("json: bad consumer config"));
unsupported.Reason.ShouldContain("unsupported - config error");
unsupported.Info.Name.ShouldBe("C1");
unsupported.Info.Stream.ShouldBe("S1");
}
[Fact] // T:774
public void JetStreamClusterMetaSnapshotsAndCatchup_ShouldSucceed()
{
var (server, error) = NatsServer.NewServer(new ServerOptions());
error.ShouldBeNull();
server.ShouldNotBeNull();
var unsupported = new UnsupportedStreamAssignment
{
Reason = "stopped",
Info = new StreamInfo { Config = new StreamConfig { Name = "ORDERS" } },
};
unsupported.SetupInfoSub(server!, new StreamAssignment { Config = new StreamConfig { Name = "ORDERS" } });
unsupported.InfoSub.ShouldNotBeNull();
var response = unsupported.HandleClusterStreamInfoRequest();
response.OfflineReason.ShouldBe("stopped");
response.StreamInfo!.Config.Name.ShouldBe("ORDERS");
unsupported.CloseInfoSub();
unsupported.InfoSub.ShouldBeNull();
}
[Fact] // T:775
public void JetStreamClusterMetaSnapshotsMultiChange_ShouldSucceed()
{
var meta = Substitute.For<IRaftNode>();
meta.Leader().Returns(true);
var cluster = new JetStreamCluster { Meta = meta };
cluster.IsLeader().ShouldBeTrue();
}
[Fact] // T:791
public void JetStreamClusterStreamSnapshotCatchupWithPurge_ShouldSucceed()
{
var node = Substitute.For<IRaftNode>();
node.Current().Returns(true);
var cluster = new JetStreamCluster
{
Meta = Substitute.For<IRaftNode>(),
Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
{
["A"] = new Dictionary<string, StreamAssignment>
{
["S"] = new() { Config = new StreamConfig { Name = "S" }, Group = new RaftGroup { Node = node } },
},
},
};
cluster.IsStreamCurrent("A", "S").ShouldBeTrue();
}
[Fact] // T:809
public void JetStreamClusterPeerRemovalAPI_ShouldSucceed()
{
var cluster = new JetStreamCluster();
var assignment = new StreamAssignment { Config = new StreamConfig { Name = "S" } };
cluster.TrackInflightStreamProposal("A", assignment, deleted: false);
cluster.TrackInflightStreamProposal("A", assignment, deleted: true);
cluster.InflightStreams["A"]["S"].Ops.ShouldBe(2UL);
cluster.InflightStreams["A"]["S"].Deleted.ShouldBeTrue();
cluster.RemoveInflightStreamProposal("A", "S");
cluster.InflightStreams["A"]["S"].Ops.ShouldBe(1UL);
cluster.RemoveInflightStreamProposal("A", "S");
cluster.InflightStreams.ContainsKey("A").ShouldBeFalse();
}
[Fact] // T:810
public void JetStreamClusterPeerRemovalAndStreamReassignment_ShouldSucceed()
{
var cluster = new JetStreamCluster();
var assignment = new ConsumerAssignment { Name = "C1" };
cluster.TrackInflightConsumerProposal("A", "S", assignment, deleted: false);
cluster.TrackInflightConsumerProposal("A", "S", assignment, deleted: true);
cluster.InflightConsumers["A"]["S"]["C1"].Ops.ShouldBe(2UL);
cluster.InflightConsumers["A"]["S"]["C1"].Deleted.ShouldBeTrue();
cluster.RemoveInflightConsumerProposal("A", "S", "C1");
cluster.InflightConsumers["A"]["S"]["C1"].Ops.ShouldBe(1UL);
cluster.RemoveInflightConsumerProposal("A", "S", "C1");
cluster.InflightConsumers.ContainsKey("A").ShouldBeFalse();
}
[Fact] // T:811
public void JetStreamClusterPeerRemovalAndStreamReassignmentWithoutSpace_ShouldSucceed()
{
var meta = Substitute.For<IRaftNode>();
meta.ID().Returns("N1");
var cluster = new JetStreamCluster
{
Meta = meta,
Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
{
["A"] = new Dictionary<string, StreamAssignment>
{
["S"] = new() { Group = new RaftGroup { Peers = ["N1", "N2"] } },
},
},
};
var account = new Account { Name = "A" };
cluster.IsStreamAssigned(account, "S").ShouldBeTrue();
}
[Fact] // T:817
public void JetStreamClusterPeerOffline_ShouldSucceed()
{
var meta = Substitute.For<IRaftNode>();
meta.ID().Returns("N9");
var cluster = new JetStreamCluster
{
Meta = meta,
Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
{
["A"] = new Dictionary<string, StreamAssignment>
{
["S"] = new() { Group = new RaftGroup { Peers = ["N1", "N2"] } },
},
},
};
cluster.IsStreamLeader("A", "S").ShouldBeFalse();
}
[Fact] // T:853
public void JetStreamClusterConsumerInfoAfterCreate_ShouldSucceed()
{
var meta = Substitute.For<IRaftNode>();
meta.ID().Returns("N1");
var cluster = new JetStreamCluster
{
Meta = meta,
Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>
{
["A"] = new Dictionary<string, StreamAssignment>
{
["S"] = new()
{
Consumers = new Dictionary<string, ConsumerAssignment>
{
["C1"] = new() { Name = "C1", Group = new RaftGroup { Peers = ["N1"] } },
},
},
},
},
};
cluster.IsConsumerLeader("A", "S", "C1").ShouldBeTrue();
}
}