Update tests to use embedded NATS

liftbridge-io · Dec 29, 2020 · f5403db · f5403db
1 parent 1147822
commit f5403db
Show file tree

Hide file tree

Showing 8 changed files with 87 additions and 331 deletions.
diff --git a/server/activity_test.go b/server/activity_test.go
@@ -7,18 +7,13 @@ import (
 
 	lift "github.com/liftbridge-io/go-liftbridge/v2"
 	liftApi "github.com/liftbridge-io/liftbridge-api/go"
-	natsdTest "github.com/nats-io/nats-server/v2/test"
 	"github.com/stretchr/testify/require"
 )
 
 // Ensure activity stream creation event occurs.
 func TestActivityStreamCreateStream(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.ActivityStream.Enabled = true
@@ -64,10 +59,6 @@ func TestActivityStreamCreateStream(t *testing.T) {
 func TestActivityStreamDeleteStream(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.ActivityStream.Enabled = true
@@ -117,10 +108,6 @@ func TestActivityStreamDeleteStream(t *testing.T) {
 func TestActivityStreamPauseStream(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.ActivityStream.Enabled = true
@@ -172,10 +159,6 @@ func TestActivityStreamPauseStream(t *testing.T) {
 func TestActivityStreamResumeStream(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.ActivityStream.Enabled = true
@@ -230,10 +213,6 @@ func TestActivityStreamResumeStream(t *testing.T) {
 func TestActivityStreamSetStreamReadonly(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.ActivityStream.Enabled = true
@@ -285,10 +264,6 @@ func TestActivityStreamSetStreamReadonly(t *testing.T) {
 func TestActivityStreamSetStreamReadwrite(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.ActivityStream.Enabled = true

diff --git a/server/api_test.go b/server/api_test.go
@@ -10,15 +10,14 @@ import (
 	"time"
 
 	lift "github.com/liftbridge-io/go-liftbridge/v2"
+	proto "github.com/liftbridge-io/liftbridge-api/go"
 	natsdTest "github.com/nats-io/nats-server/v2/test"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
-	proto "github.com/liftbridge-io/liftbridge-api/go"
-
 	"github.com/liftbridge-io/liftbridge/server/protocol"
 )
 
@@ -39,10 +38,6 @@ func assertMsg(t *testing.T, expected *message, msg *lift.Message) {
 func TestCreateStream(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
@@ -67,10 +62,6 @@ func TestCreateStream(t *testing.T) {
 func TestCreateStreamPropagate(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure first server.
 	s1Config := getTestConfig("a", true, 0)
 	s1 := runServerWithConfig(t, s1Config)
@@ -100,12 +91,13 @@ func TestCreateStreamPropagate(t *testing.T) {
 func TestCreateStreamNoMetadataLeader(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
+	// Use an external NATS server.
 	ns := natsdTest.RunDefaultServer()
 	defer ns.Shutdown()
 
 	// Configure first server.
 	s1Config := getTestConfig("a", true, 0)
+	s1Config.EmbeddedNATS = false
 	s1 := runServerWithConfig(t, s1Config)
 	defer s1.Stop()
 
@@ -137,10 +129,6 @@ func TestCreateStreamNoMetadataLeader(t *testing.T) {
 func TestCreateStreamInsufficientReplicas(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
@@ -162,10 +150,6 @@ func TestCreateStreamInsufficientReplicas(t *testing.T) {
 func TestCreateStreamPartitioned(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
@@ -189,10 +173,6 @@ func TestCreateStreamPartitioned(t *testing.T) {
 func TestSubscribeStreamNoSuchStream(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
@@ -217,10 +197,6 @@ func TestSubscribeStreamNoSuchStream(t *testing.T) {
 func TestDeleteStream(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
@@ -262,10 +238,6 @@ func TestDeleteStream(t *testing.T) {
 func TestDeleteStreamPropagate(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure first server.
 	s1Config := getTestConfig("a", true, 0)
 	s1 := runServerWithConfig(t, s1Config)
@@ -295,10 +267,6 @@ func TestDeleteStreamPropagate(t *testing.T) {
 func TestSubscribeStreamNotLeader(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure first server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
@@ -354,10 +322,6 @@ func TestSubscribeStreamNotLeader(t *testing.T) {
 func TestSubscribeStreamNotLeaderDefaultBehavior(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure first server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
@@ -414,14 +378,11 @@ func TestSubscribeStreamNotLeaderDefaultBehavior(t *testing.T) {
 func TestStreamReceiveMsgFromReplica(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
 	defer s1.Stop()
+
 	// Configure second server.
 	s2Config := getTestConfig("b", false, 5051)
 	s2 := runServerWithConfig(t, s2Config)
@@ -543,10 +504,6 @@ func TestStreamReceiveMsgFromReplica(t *testing.T) {
 func TestStreamPublishSubscribe(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.Clustering.ReplicationMaxBytes = 1024
@@ -661,10 +618,6 @@ func TestStreamPublishSubscribe(t *testing.T) {
 func TestLegacyPublish(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.Clustering.ReplicationMaxBytes = 1024
@@ -734,10 +687,6 @@ func TestLegacyPublish(t *testing.T) {
 func TestPublishToSubject(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
@@ -873,10 +822,6 @@ func TestSubscribePartitionClosed(t *testing.T) {
 func TestSubscribeStopPosition(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
@@ -1005,10 +950,6 @@ func TestGetStreamConfig(t *testing.T) {
 func TestSetFetchCursor(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.CursorsStream.Partitions = 5
@@ -1063,10 +1004,6 @@ func TestSetFetchCursor(t *testing.T) {
 func TestSetFetchCursorNoCache(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.CursorsStream.Partitions = 5
@@ -1144,10 +1081,6 @@ func publishAndReceive(t *testing.T, client lift.Client, stream string) {
 func TestFetchPartitionMetadataMessagesReceivedTimestamps(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
@@ -1200,10 +1133,6 @@ func TestFetchPartitionMetadataMessagesReceivedTimestamps(t *testing.T) {
 func TestFetchPartitionMetadataPauseTimestamps(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)
@@ -1256,10 +1185,6 @@ func TestFetchPartitionMetadataPauseTimestamps(t *testing.T) {
 func TestFetchPartitionMetadataReadonlyTimestamps(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)

diff --git a/server/configs/tls.yaml b/server/configs/tls.yaml
@@ -4,3 +4,4 @@ tls:
   cert: ./configs/certs/server.crt
 logging.level: error
 clustering.raft.bootstrap.seed: true
+nats.embedded: true
diff --git a/server/fsm_test.go b/server/fsm_test.go
@@ -6,18 +6,13 @@ import (
 	"time"
 
 	lift "github.com/liftbridge-io/go-liftbridge/v2"
-	natsdTest "github.com/nats-io/nats-server/v2/test"
 	"github.com/stretchr/testify/require"
 )
 
 // Ensure Raft FSM properly snapshots and restores state.
 func TestFSMSnapshotRestore(t *testing.T) {
 	defer cleanupStorage(t)
 
-	// Use a central NATS server.
-	ns := natsdTest.RunDefaultServer()
-	defer ns.Shutdown()
-
 	// Configure the server as a seed.
 	s1Config := getTestConfig("a", true, 5050)
 	s1 := runServerWithConfig(t, s1Config)