Add CLI sync mysql integration test (#2858)

backend/internal/neosyncdb/integration_test.go

@@ -2,25 +2,21 @@ package neosyncdb
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"log/slog"
-	"os"
 	"testing"
 	"time"
 
 	"github.com/jackc/pgx/v5/pgtype"
-	"github.com/jackc/pgx/v5/pgxpool"
 	db_queries "github.com/nucleuscloud/neosync/backend/gen/go/db"
 	mgmtv1alpha1 "github.com/nucleuscloud/neosync/backend/gen/go/protos/mgmt/v1alpha1"
 	nucleuserrors "github.com/nucleuscloud/neosync/backend/internal/errors"
 	pg_models "github.com/nucleuscloud/neosync/backend/sql/postgresql/models"
 	neomigrate "github.com/nucleuscloud/neosync/internal/migrate"
+	"github.com/nucleuscloud/neosync/internal/testutil"
+	tcpostgres "github.com/nucleuscloud/neosync/internal/testutil/testcontainers/postgres"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
-	"github.com/testcontainers/testcontainers-go"
-	testpg "github.com/testcontainers/testcontainers-go/modules/postgres"
-	"github.com/testcontainers/testcontainers-go/wait"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -31,12 +27,9 @@ var (
 type IntegrationTestSuite struct {
 	suite.Suite
 
-	pgpool *pgxpool.Pool
-
 	ctx context.Context
 
-	pgcontainer   *testpg.PostgresContainer
-	connstr       string
+	pgcontainer   *tcpostgres.PostgresTestContainer
 	migrationsDir string
 
 	db *NeosyncDb
@@ -45,38 +38,21 @@ type IntegrationTestSuite struct {
 func (s *IntegrationTestSuite) SetupSuite() {
 	s.ctx = context.Background()
 
-	pgcontainer, err := testpg.Run(
-		s.ctx,
-		"postgres:15",
-		testcontainers.WithWaitStrategy(
-			wait.ForLog("database system is ready to accept connections").
-				WithOccurrence(2).WithStartupTimeout(5*time.Second),
-		),
-	)
+	pgcontainer, err := tcpostgres.NewPostgresTestContainer(s.ctx)
 	if err != nil {
 		panic(err)
 	}
 	s.pgcontainer = pgcontainer
-	connstr, err := pgcontainer.ConnectionString(s.ctx, "sslmode=disable")
-	if err != nil {
-		panic(err)
-	}
-	s.connstr = connstr
 
-	pool, err := pgxpool.New(s.ctx, connstr)
-	if err != nil {
-		panic(err)
-	}
-	s.pgpool = pool
 	s.migrationsDir = "../../sql/postgresql/schema"
 
-	s.db = New(pool, db_queries.New())
+	s.db = New(s.pgcontainer.DB, db_queries.New())
 }
 
 // Runs before each test
 func (s *IntegrationTestSuite) SetupTest() {
 	discardLogger := slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{}))
-	err := neomigrate.Up(s.ctx, s.connstr, s.migrationsDir, discardLogger)
+	err := neomigrate.Up(s.ctx, s.pgcontainer.URL, s.migrationsDir, discardLogger)
 	if err != nil {
 		panic(err)
 	}
@@ -86,33 +62,28 @@ func (s *IntegrationTestSuite) TearDownTest() {
 	// Dropping here because 1) more efficient and 2) we have a bad down migration
 	// _jobs-connection-id-null.down that breaks due to having a null connection_id column.
 	// we should do something about that at some point. Running this single drop is easier though
-	_, err := s.pgpool.Exec(s.ctx, "DROP SCHEMA IF EXISTS neosync_api CASCADE")
+	_, err := s.pgcontainer.DB.Exec(s.ctx, "DROP SCHEMA IF EXISTS neosync_api CASCADE")
 	if err != nil {
 		panic(err)
 	}
-	_, err = s.pgpool.Exec(s.ctx, "DROP TABLE IF EXISTS public.schema_migrations")
+	_, err = s.pgcontainer.DB.Exec(s.ctx, "DROP TABLE IF EXISTS public.schema_migrations")
 	if err != nil {
 		panic(err)
 	}
 }
 
 func (s *IntegrationTestSuite) TearDownSuite() {
-	if s.pgpool != nil {
-		s.pgpool.Close()
-	}
 	if s.pgcontainer != nil {
-		err := s.pgcontainer.Terminate(s.ctx)
+		err := s.pgcontainer.TearDown(s.ctx)
 		if err != nil {
 			panic(err)
 		}
 	}
 }
 
 func TestIntegrationTestSuite(t *testing.T) {
-	evkey := "INTEGRATION_TESTS_ENABLED"
-	shouldRun := os.Getenv(evkey)
-	if shouldRun != "1" {
-		slog.Warn(fmt.Sprintf("skipping integration tests, set %s=1 to enable", evkey))
+	ok := testutil.ShouldRunIntegrationTest()
+	if !ok {
 		return
 	}
 	suite.Run(t, new(IntegrationTestSuite))

backend/pkg/integration-test/integration-test-util.go

@@ -48,6 +48,34 @@ func CreatePostgresConnection(
 	return resp.Msg.GetConnection()
 }
 
+func CreateMysqlConnection(
+	ctx context.Context,
+	t *testing.T,
+	connclient mgmtv1alpha1connect.ConnectionServiceClient,
+	accountId string,
+	name string,
+	mysqlurl string,
+) *mgmtv1alpha1.Connection {
+	resp, err := connclient.CreateConnection(
+		ctx,
+		connect.NewRequest(&mgmtv1alpha1.CreateConnectionRequest{
+			AccountId: accountId,
+			Name:      name,
+			ConnectionConfig: &mgmtv1alpha1.ConnectionConfig{
+				Config: &mgmtv1alpha1.ConnectionConfig_MysqlConfig{
+					MysqlConfig: &mgmtv1alpha1.MysqlConnectionConfig{
+						ConnectionConfig: &mgmtv1alpha1.MysqlConnectionConfig_Url{
+							Url: mysqlurl,
+						},
+					},
+				},
+			},
+		}),
+	)
+	RequireNoErrResp(t, resp, err)
+	return resp.Msg.GetConnection()
+}
+
 func SetUser(ctx context.Context, t *testing.T, client mgmtv1alpha1connect.UserAccountServiceClient) string {
 	resp, err := client.SetUser(ctx, connect.NewRequest(&mgmtv1alpha1.SetUserRequest{}))
 	RequireNoErrResp(t, resp, err)

backend/pkg/sqlmanager/mssql/integration_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"log/slog"
 	"net/url"
 	"os"
 	"path/filepath"
@@ -17,6 +16,7 @@ import (
 
 	mssql_queries "github.com/nucleuscloud/neosync/backend/pkg/mssql-querier"
 	sqlmanager_shared "github.com/nucleuscloud/neosync/backend/pkg/sqlmanager/shared"
+	"github.com/nucleuscloud/neosync/internal/testutil"
 	"github.com/stretchr/testify/suite"
 	testmssql "github.com/testcontainers/testcontainers-go/modules/mssql"
 )
@@ -298,10 +298,8 @@ func (s *IntegrationTestSuite) TearDownSuite() {
 }
 
 func TestIntegrationTestSuite(t *testing.T) {
-	evkey := "INTEGRATION_TESTS_ENABLED"
-	shouldRun := os.Getenv(evkey)
-	if shouldRun != "1" {
-		slog.Warn(fmt.Sprintf("skipping integration tests, set %s=1 to enable", evkey))
+	ok := testutil.ShouldRunIntegrationTest()
+	if !ok {
 		return
 	}
 	suite.Run(t, new(IntegrationTestSuite))

backend/pkg/sqlmanager/mssql_sql-manager_integration_test.go

@@ -5,14 +5,14 @@ import (
 	"fmt"
 	slog "log/slog"
 	"net/url"
-	"os"
 	"sync"
 	"testing"
 
 	"github.com/google/uuid"
 	mgmtv1alpha1 "github.com/nucleuscloud/neosync/backend/gen/go/protos/mgmt/v1alpha1"
 	mssql_queries "github.com/nucleuscloud/neosync/backend/pkg/mssql-querier"
 	"github.com/nucleuscloud/neosync/backend/pkg/sqlconnect"
+	"github.com/nucleuscloud/neosync/internal/testutil"
 	"github.com/stretchr/testify/suite"
 
 	testmssql "github.com/testcontainers/testcontainers-go/modules/mssql"
@@ -90,10 +90,8 @@ func (s *MssqlIntegrationTestSuite) TearDownSuite() {
 }
 
 func TestMssqlIntegrationTestSuite(t *testing.T) {
-	evkey := "INTEGRATION_TESTS_ENABLED"
-	shouldRun := os.Getenv(evkey)
-	if shouldRun != "1" {
-		slog.Warn(fmt.Sprintf("skipping integration tests, set %s=1 to enable", evkey))
+	ok := testutil.ShouldRunIntegrationTest()
+	if !ok {
 		return
 	}
 	suite.Run(t, new(MssqlIntegrationTestSuite))

backend/pkg/sqlmanager/mysql/integration_test.go

@@ -3,13 +3,12 @@ package sqlmanager_mysql
 import (
 	"context"
 	"fmt"
-	"log/slog"
-	"os"
 	"testing"
 
 	_ "github.com/go-sql-driver/mysql"
 
 	mysql_queries "github.com/nucleuscloud/neosync/backend/gen/go/db/dbschemas/mysql"
+	"github.com/nucleuscloud/neosync/internal/testutil"
 	tcmysql "github.com/nucleuscloud/neosync/internal/testutil/testcontainers/mysql"
 	"github.com/stretchr/testify/suite"
 )
@@ -75,10 +74,8 @@ func (s *IntegrationTestSuite) TearDownSuite() {
 }
 
 func TestIntegrationTestSuite(t *testing.T) {
-	evkey := "INTEGRATION_TESTS_ENABLED"
-	shouldRun := os.Getenv(evkey)
-	if shouldRun != "1" {
-		slog.Warn(fmt.Sprintf("skipping integration tests, set %s=1 to enable", evkey))
+	ok := testutil.ShouldRunIntegrationTest()
+	if !ok {
 		return
 	}
 	suite.Run(t, new(IntegrationTestSuite))

backend/pkg/sqlmanager/mysql_sql-manager_integration_test.go

@@ -2,9 +2,7 @@ package sqlmanager
 
 import (
 	context "context"
-	"fmt"
 	slog "log/slog"
-	"os"
 	"sync"
 	"testing"
 
@@ -15,6 +13,7 @@ import (
 	"github.com/stretchr/testify/suite"
 
 	_ "github.com/go-sql-driver/mysql"
+	"github.com/nucleuscloud/neosync/internal/testutil"
 	tcmysql "github.com/nucleuscloud/neosync/internal/testutil/testcontainers/mysql"
 )
 
@@ -77,10 +76,8 @@ func (s *MysqlIntegrationTestSuite) TearDownSuite() {
 }
 
 func TestMysqlIntegrationTestSuite(t *testing.T) {
-	evkey := "INTEGRATION_TESTS_ENABLED"
-	shouldRun := os.Getenv(evkey)
-	if shouldRun != "1" {
-		slog.Warn(fmt.Sprintf("skipping integration tests, set %s=1 to enable", evkey))
+	ok := testutil.ShouldRunIntegrationTest()
+	if !ok {
 		return
 	}
 	suite.Run(t, new(MysqlIntegrationTestSuite))

backend/pkg/sqlmanager/postgres/integration_test.go

@@ -4,12 +4,11 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"log/slog"
-	"os"
 	"testing"
 
 	pg_queries "github.com/nucleuscloud/neosync/backend/gen/go/db/dbschemas/postgresql"
 	sqlmanager_shared "github.com/nucleuscloud/neosync/backend/pkg/sqlmanager/shared"
+	"github.com/nucleuscloud/neosync/internal/testutil"
 	tcpostgres "github.com/nucleuscloud/neosync/internal/testutil/testcontainers/postgres"
 	"github.com/stretchr/testify/suite"
 )
@@ -79,10 +78,8 @@ func (s *IntegrationTestSuite) TearDownSuite() {
 }
 
 func TestIntegrationTestSuite(t *testing.T) {
-	evkey := "INTEGRATION_TESTS_ENABLED"
-	shouldRun := os.Getenv(evkey)
-	if shouldRun != "1" {
-		slog.Warn(fmt.Sprintf("skipping integration tests, set %s=1 to enable", evkey))
+	ok := testutil.ShouldRunIntegrationTest()
+	if !ok {
 		return
 	}
 	suite.Run(t, new(IntegrationTestSuite))

backend/pkg/sqlmanager/postgres_sql-manager_integration_test.go

@@ -2,12 +2,9 @@ package sqlmanager
 
 import (
 	context "context"
-	"fmt"
 	slog "log/slog"
-	"os"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/google/uuid"
 	pg_queries "github.com/nucleuscloud/neosync/backend/gen/go/db/dbschemas/postgresql"
@@ -17,15 +14,14 @@ import (
 	"github.com/stretchr/testify/suite"
 
 	_ "github.com/jackc/pgx/v5/stdlib"
-	"github.com/testcontainers/testcontainers-go"
-	testpg "github.com/testcontainers/testcontainers-go/modules/postgres"
-	"github.com/testcontainers/testcontainers-go/wait"
+	"github.com/nucleuscloud/neosync/internal/testutil"
+	tcpostgres "github.com/nucleuscloud/neosync/internal/testutil/testcontainers/postgres"
 )
 
 type PostgresIntegrationTestSuite struct {
 	suite.Suite
 
-	pgcontainer *testpg.PostgresContainer
+	pgcontainer *tcpostgres.PostgresTestContainer
 
 	ctx context.Context
 
@@ -40,27 +36,15 @@ type PostgresIntegrationTestSuite struct {
 func (s *PostgresIntegrationTestSuite) SetupSuite() {
 	s.ctx = context.Background()
 
-	pgcontainer, err := testpg.Run(
-		s.ctx,
-		"postgres:15",
-		testcontainers.WithWaitStrategy(
-			wait.ForLog("database system is ready to accept connections").
-				WithOccurrence(2).WithStartupTimeout(5*time.Second),
-		),
-	)
+	pgcontainer, err := tcpostgres.NewPostgresTestContainer(s.ctx)
 	if err != nil {
 		panic(err)
 	}
 	s.pgcontainer = pgcontainer
 
-	connstr, err := pgcontainer.ConnectionString(s.ctx, "sslmode=disable")
-	if err != nil {
-		panic(err)
-	}
-
 	s.pgcfg = &mgmtv1alpha1.PostgresConnectionConfig{
 		ConnectionConfig: &mgmtv1alpha1.PostgresConnectionConfig_Url{
-			Url: connstr,
+			Url: pgcontainer.URL,
 		},
 	}
 	s.mgmtconn = &mgmtv1alpha1.Connection{
@@ -85,18 +69,16 @@ func (s *PostgresIntegrationTestSuite) TearDownTest() {
 
 func (s *PostgresIntegrationTestSuite) TearDownSuite() {
 	if s.pgcontainer != nil {
-		err := s.pgcontainer.Terminate(s.ctx)
+		err := s.pgcontainer.TearDown(s.ctx)
 		if err != nil {
 			panic(err)
 		}
 	}
 }
 
 func TestPostgresIntegrationTestSuite(t *testing.T) {
-	evkey := "INTEGRATION_TESTS_ENABLED"
-	shouldRun := os.Getenv(evkey)
-	if shouldRun != "1" {
-		slog.Warn(fmt.Sprintf("skipping integration tests, set %s=1 to enable", evkey))
+	ok := testutil.ShouldRunIntegrationTest()
+	if !ok {
 		return
 	}
 	suite.Run(t, new(PostgresIntegrationTestSuite))