attempt to run tests in a tester sidecar

Signed-off-by: R.I.Pienaar <[email protected]>

Author: ripienaar

.github/workflows/go.yaml

@@ -4,40 +4,62 @@ on: [push, pull_request]
 jobs:
   lint_and_test:
     runs-on: ubuntu-latest
+    container: golang:1.24-alpine # required to run in the docker network and avoid badly performing
+                                  # port range expose from docker
 
     strategy:
       matrix:
         go: [ "1.23", "1.24" ]
 
+    services:
+      nats:
+        image: ripienaar/testing.go
+        options: --name jetstream   # only needed to facilitate debugging with docker logs etc
+
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
 
       - name: Setup Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version: ${{ matrix.go }}
 
       - name: Install deps
-        shell: bash --noprofile --norc -x -eo pipefail {0}
         run: |
+          set -e
+
           export GOPATH="$RUNNER_WORKSPACE"
           go version
           go install honnef.co/go/tools/cmd/staticcheck@latest
           go install github.com/client9/misspell/cmd/misspell@latest
 
       - name: Lint
-        shell: bash --noprofile --norc -x -eo pipefail {0}
         run: |
+          set -e
+
           export GOPATH="$RUNNER_WORKSPACE"
           PATH=$PATH:$GOPATH/bin
           $(exit $(go fmt $(go list ./...) | wc -l))
           find . -type f -name "*.go" | xargs misspell -error -locale US
-          staticcheck -f stylish $(go list ./... | grep -v governor)
+          staticcheck -f stylish $(go list ./...)
           go vet ./...
 
       - name: Run tests
-        shell: bash --noprofile --norc -x -eo pipefail {0}
         run: |
+          set -e
+
           export GOPATH="$RUNNER_WORKSPACE"
-          go test -v -race -p 1 ./...
+          export TESTER_URL="nats://nats:4222" # name here matches the service name
+          
+          go test -v  -p 1 -failfast ./...
+
+      # debugging only
+      - name: docker logs
+        if: always()
+        run: |
+          set -e
+
+          apk add docker    
+          docker ps -a
+          docker logs jetstream

balancer/balancer_test.go

@@ -1,23 +1,25 @@
 package balancer
 
 import (
-	"context"
 	"fmt"
-	"net/url"
 	"os"
-	"path/filepath"
 	"testing"
 	"time"
 
 	"github.com/nats-io/jsm.go"
 	"github.com/nats-io/jsm.go/api"
-	"github.com/nats-io/nats-server/v2/server"
+	testapi "github.com/nats-io/jsm.go/test/testing_client/api"
+	"github.com/nats-io/jsm.go/test/testing_client/srvtest"
 	"github.com/nats-io/nats.go"
 )
 
 func TestBalancer(t *testing.T) {
-	withJSCluster(t, func(t *testing.T, servers []*server.Server, nc *nats.Conn, mgr *jsm.Manager) error {
+	withTesterJetStreamCluster(t, func(t *testing.T, mgr *jsm.Manager, _ *srvtest.Client, servers []*testapi.ManagedServer) {
 		var err error
+
+		nc := mgr.NatsConn()
+		firstServer := servers[0].Name
+
 		waitTime := 100 * time.Millisecond
 		streams := []*jsm.Stream{}
 		for i := 1; i <= 3; i++ {
@@ -28,8 +30,8 @@ func TestBalancer(t *testing.T) {
 				t.Fatalf("could not create stream %s", err)
 			}
 			info, _ := s.ClusterInfo()
-			if info.Leader != "s1" {
-				placement := api.Placement{Preferred: "s1"}
+			if info.Leader != firstServer {
+				placement := api.Placement{Preferred: firstServer}
 				err = s.LeaderStepDown(&placement)
 				if err != nil {
 					t.Fatalf("could not move stream %s", err)
@@ -59,29 +61,29 @@ func TestBalancer(t *testing.T) {
 
 		b, err := New(nc, api.NewDefaultLogger(api.DebugLevel))
 		if err != nil {
-			return err
+			t.Fatalf("create failed: %v", err.Error())
 		}
 
 		count, err := b.BalanceStreams(streams)
 		if err != nil {
-			return err
+			t.Fatalf("Balance failed: %v", err.Error())
 		}
 
 		if count == 0 {
-			return err
+			t.Fatal("Balanceed 0 streams")
 		}
 
 		consumers := []*jsm.Consumer{}
 		for i := 1; i <= 3; i++ {
 			consumerName := fmt.Sprintf("testc%d", i)
 			c, err := mgr.NewConsumer("tests1", jsm.DurableName(consumerName), jsm.ConsumerOverrideReplicas(3))
 			if err != nil {
-				return err
+				t.Fatalf("could not create consumer %s", err)
 			}
 
 			info, _ := c.ClusterInfo()
-			if info.Leader != "s1" {
-				placement := api.Placement{Preferred: "s1"}
+			if info.Leader != firstServer {
+				placement := api.Placement{Preferred: firstServer}
 				err = c.LeaderStepDown(&placement)
 				if err != nil {
 					t.Fatalf("could not move consumer %s", err)
@@ -111,102 +113,34 @@ func TestBalancer(t *testing.T) {
 
 		count, err = b.BalanceConsumers(consumers)
 		if err != nil {
-			return err
+			t.Fatalf("Balance failed: %v", err)
 		}
 
 		if count == 0 {
-			return err
+			t.Fatal("Balanced 0 consumers")
 		}
-
-		return nil
 	})
 }
 
-func withJSCluster(t *testing.T, cb func(*testing.T, []*server.Server, *nats.Conn, *jsm.Manager) error) {
+func withTesterJetStreamCluster(t *testing.T, fn func(*testing.T, *jsm.Manager, *srvtest.Client, []*testapi.ManagedServer)) {
 	t.Helper()
 
-	d, err := os.MkdirTemp("", "jstest")
-	if err != nil {
-		t.Fatalf("temp dir could not be made: %s", err)
+	url := os.Getenv("TESTER_URL")
+	if url == "" {
+		url = "nats://localhost:4222"
 	}
-	defer os.RemoveAll(d)
-
-	var (
-		servers []*server.Server
-	)
-
-	for i := 1; i <= 3; i++ {
-		opts := &server.Options{
-			JetStream:  true,
-			StoreDir:   filepath.Join(d, fmt.Sprintf("s%d", i)),
-			Port:       -1,
-			Host:       "localhost",
-			ServerName: fmt.Sprintf("s%d", i),
-			LogFile:    "/dev/null",
-			Cluster: server.ClusterOpts{
-				Name: "TEST",
-				Port: 12000 + i,
-			},
-			Routes: []*url.URL{
-				{Host: "localhost:12001"},
-				{Host: "localhost:12002"},
-				{Host: "localhost:12003"},
-			},
-		}
 
-		s, err := server.NewServer(opts)
+	client := srvtest.New(t, url)
+	t.Cleanup(func() {
+		client.Reset(t)
+	})
+
+	client.WithJetStreamCluster(t, 3, func(t *testing.T, nc *nats.Conn, servers []*testapi.ManagedServer) {
+		mgr, err := jsm.New(nc)
 		if err != nil {
-			t.Fatalf("server %d start failed: %v", i, err)
-		}
-		s.ConfigureLogger()
-		go s.Start()
-		if !s.ReadyForConnections(10 * time.Second) {
-			t.Errorf("nats server %d did not start", i)
+			t.Fatal(err)
 		}
-		defer func() {
-			s.Shutdown()
-		}()
-
-		servers = append(servers, s)
-	}
-
-	if len(servers) != 3 {
-		t.Fatalf("servers did not start")
-	}
-
-	nc, err := nats.Connect(servers[0].ClientURL(), nats.UseOldRequestStyle())
-	if err != nil {
-		t.Fatalf("client start failed: %s", err)
-	}
-	defer nc.Close()
 
-	mgr, err := jsm.New(nc, jsm.WithTimeout(5*time.Second))
-	if err != nil {
-		t.Fatalf("manager creation failed: %s", err)
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
-	ticker := time.NewTicker(250 * time.Millisecond)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ticker.C:
-			_, err := mgr.JetStreamAccountInfo()
-			if err != nil {
-				continue
-			}
-
-			err = cb(t, servers, nc, mgr)
-
-			if err != nil {
-				t.Fatal(err)
-			}
-
-			return
-		case <-ctx.Done():
-			t.Fatalf("jetstream did not become available")
-		}
-	}
+		fn(t, mgr, client, servers)
+	})
 }