Worker integration tests (#2211)

nucleuscloud · Jun 26, 2024 · 93fe68d · 93fe68d
1 parent 4a85b1b
commit 93fe68d
Show file tree

Hide file tree

Showing 18 changed files with 1,362 additions and 880 deletions.
diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml
@@ -56,30 +56,6 @@ jobs:
   test:
     name: test
     runs-on: ubuntu-latest
-    services:
-      postgres:
-        image: postgres:15
-        ports:
-          - 5432:5432
-        env:
-          POSTGRES_DB: postgres
-          POSTGRES_PASSWORD: postgres
-          PGUSER: postgres
-        options: >-
-          --health-cmd pg_isready
-          --health-interval 10s
-          --health-timeout 5s
-          --health-retries 5
-      redis:
-        image: redis
-        ports:
-          - 6379:6379
-        options: >-
-          --health-cmd "redis-cli ping"
-          --health-interval 10s
-          --health-timeout 5s
-          --health-retries 5
-
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -95,8 +71,6 @@ jobs:
           go test -race -coverprofile=coverage.out -covermode=atomic ./...
         env:
           INTEGRATION_TESTS_ENABLED: 1
-          TEST_DB_URL: postgres://postgres:postgres@localhost:5432/postgres
-          TEST_REDIS_URL: redis://localhost:6379
 
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v4

diff --git a/backend/pkg/sqlmanager/postgres/integration_test.go b/backend/pkg/sqlmanager/postgres/integration_test.go
@@ -40,24 +40,20 @@ func (s *IntegrationTestSuite) SetupSuite() {
 	s.ctx = context.Background()
 	s.schema = "sqlmanagerpostgres"
 
-	dburl := os.Getenv("TEST_DB_URL")
-	if dburl == "" {
-		pgcontainer, err := testpg.RunContainer(s.ctx,
-			testcontainers.WithImage("postgres:15"),
-			testcontainers.WithWaitStrategy(
-				wait.ForLog("database system is ready to accept connections").
-					WithOccurrence(2).WithStartupTimeout(5*time.Second),
-			),
-		)
-		if err != nil {
-			panic(err)
-		}
-		s.pgcontainer = pgcontainer
-		connstr, err := pgcontainer.ConnectionString(s.ctx)
-		if err != nil {
-			panic(err)
-		}
-		dburl = connstr
+	pgcontainer, err := testpg.RunContainer(s.ctx,
+		testcontainers.WithImage("postgres:15"),
+		testcontainers.WithWaitStrategy(
+			wait.ForLog("database system is ready to accept connections").
+				WithOccurrence(2).WithStartupTimeout(5*time.Second),
+		),
+	)
+	if err != nil {
+		panic(err)
+	}
+	s.pgcontainer = pgcontainer
+	connstr, err := pgcontainer.ConnectionString(s.ctx)
+	if err != nil {
+		panic(err)
 	}
 
 	setupSql, err := os.ReadFile("./testdata/setup.sql")
@@ -72,7 +68,7 @@ func (s *IntegrationTestSuite) SetupSuite() {
 	}
 	s.teardownSql = string(teardownSql)
 
-	pool, err := pgxpool.New(s.ctx, dburl)
+	pool, err := pgxpool.New(s.ctx, connstr)
 	if err != nil {
 		panic(err)
 	}

diff --git a/worker/pkg/query-builder/integration_test.go b/worker/pkg/query-builder/integration_test.go
@@ -36,24 +36,20 @@ func (s *IntegrationTestSuite) SetupSuite() {
 	s.ctx = context.Background()
 	s.schema = "genbenthosconfigs_querybuilder"
 
-	dburl := os.Getenv("TEST_DB_URL")
-	if dburl == "" {
-		pgcontainer, err := testpg.RunContainer(s.ctx,
-			testcontainers.WithImage("postgres:15"),
-			testcontainers.WithWaitStrategy(
-				wait.ForLog("database system is ready to accept connections").
-					WithOccurrence(2).WithStartupTimeout(5*time.Second),
-			),
-		)
-		if err != nil {
-			panic(err)
-		}
-		s.pgcontainer = pgcontainer
-		connstr, err := pgcontainer.ConnectionString(s.ctx)
-		if err != nil {
-			panic(err)
-		}
-		dburl = connstr
+	pgcontainer, err := testpg.RunContainer(s.ctx,
+		testcontainers.WithImage("postgres:15"),
+		testcontainers.WithWaitStrategy(
+			wait.ForLog("database system is ready to accept connections").
+				WithOccurrence(2).WithStartupTimeout(5*time.Second),
+		),
+	)
+	if err != nil {
+		panic(err)
+	}
+	s.pgcontainer = pgcontainer
+	connstr, err := pgcontainer.ConnectionString(s.ctx)
+	if err != nil {
+		panic(err)
 	}
 
 	setupSql, err := os.ReadFile("./testdata/setup.sql")
@@ -68,7 +64,7 @@ func (s *IntegrationTestSuite) SetupSuite() {
 	}
 	s.teardownSql = string(teardownSql)
 
-	pool, err := pgxpool.New(s.ctx, dburl)
+	pool, err := pgxpool.New(s.ctx, connstr)
 	if err != nil {
 		panic(err)
 	}

diff --git a/worker/pkg/workflows/datasync/workflow/integration_test.go b/worker/pkg/workflows/datasync/workflow/integration_test.go
@@ -44,30 +44,25 @@ type IntegrationTestSuite struct {
 func (s *IntegrationTestSuite) SetupSuite() {
 	s.ctx = context.Background()
 
-	testDbUrl := os.Getenv("TEST_DB_URL")
-	dburl := fmt.Sprintf("%s?sslmode=disable", testDbUrl)
-	if testDbUrl == "" {
-		pgcontainer, err := testpg.RunContainer(s.ctx,
-			testcontainers.WithImage("postgres:15"),
-			postgres.WithDatabase("postgres"),
-			testcontainers.WithWaitStrategy(
-				wait.ForLog("database system is ready to accept connections").
-					WithOccurrence(2).WithStartupTimeout(5*time.Second),
-			),
-		)
-		if err != nil {
-			panic(err)
-		}
-		s.pgcontainer = pgcontainer
-		connstr, err := pgcontainer.ConnectionString(s.ctx, "sslmode=disable")
-		if err != nil {
-			panic(err)
-		}
-		dburl = connstr
+	pgcontainer, err := testpg.RunContainer(s.ctx,
+		testcontainers.WithImage("postgres:15"),
+		postgres.WithDatabase("postgres"),
+		testcontainers.WithWaitStrategy(
+			wait.ForLog("database system is ready to accept connections").
+				WithOccurrence(2).WithStartupTimeout(5*time.Second),
+		),
+	)
+	if err != nil {
+		panic(err)
+	}
+	s.pgcontainer = pgcontainer
+	connstr, err := pgcontainer.ConnectionString(s.ctx, "sslmode=disable")
+	if err != nil {
+		panic(err)
 	}
 
 	s.databases = []string{"datasync_source", "datasync_target"}
-	pool, err := pgxpool.New(s.ctx, dburl)
+	pool, err := pgxpool.New(s.ctx, connstr)
 	if err != nil {
 		panic(err)
 	}
@@ -81,7 +76,7 @@ func (s *IntegrationTestSuite) SetupSuite() {
 		}
 	}
 
-	srcUrl, err := getDbPgUrl(dburl, "datasync_source", "disable")
+	srcUrl, err := getDbPgUrl(connstr, "datasync_source", "disable")
 	if err != nil {
 		panic(err)
 	}
@@ -92,7 +87,7 @@ func (s *IntegrationTestSuite) SetupSuite() {
 	}
 	s.sourcePgPool = sourceConn
 
-	targetUrl, err := getDbPgUrl(dburl, "datasync_target", "disable")
+	targetUrl, err := getDbPgUrl(connstr, "datasync_target", "disable")
 	if err != nil {
 		panic(err)
 	}
@@ -106,19 +101,43 @@ func (s *IntegrationTestSuite) SetupSuite() {
 	s.querier = pg_queries.New()
 
 	// redis
-	redisUrl := os.Getenv("TEST_REDIS_URL")
-	s.redisUrl = redisUrl
-	if redisUrl == "" {
-		redisContainer, err := redis.RunContainer(s.ctx,
-			testcontainers.WithImage("docker.io/redis:7"),
-			redis.WithSnapshotting(10, 1),
-			redis.WithLogLevel(redis.LogLevelVerbose),
-		)
+	redisContainer, err := redis.RunContainer(s.ctx,
+		testcontainers.WithImage("docker.io/redis:7"),
+		redis.WithSnapshotting(10, 1),
+		redis.WithLogLevel(redis.LogLevelVerbose),
+	)
+	if err != nil {
+		panic(err)
+	}
+	s.rediscontainer = redisContainer
+	s.redisUrl, err = redisContainer.ConnectionString(s.ctx)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func (s *IntegrationTestSuite) SetupSourceDb(testFolder string, files []string) {
+	s.T().Logf("setting up source db. folder: %s \n", testFolder)
+	for _, file := range files {
+		setupSourceSql, err := os.ReadFile(fmt.Sprintf("./testdata/%s/%s", testFolder, file))
+		if err != nil {
+			panic(err)
+		}
+		_, err = s.sourcePgPool.Exec(s.ctx, string(setupSourceSql))
+		if err != nil {
+			panic(err)
+		}
+	}
+}
+
+func (s *IntegrationTestSuite) SetupTargetDb(testFolder string, files []string) {
+	s.T().Logf("setting up target db. folder: %s \n", testFolder)
+	for _, file := range files {
+		setupTargetSql, err := os.ReadFile(fmt.Sprintf("./testdata/%s/%s", testFolder, file))
 		if err != nil {
 			panic(err)
 		}
-		s.rediscontainer = redisContainer
-		s.redisUrl, err = redisContainer.ConnectionString(s.ctx)
+		_, err = s.targetPgPool.Exec(s.ctx, string(setupTargetSql))
 		if err != nil {
 			panic(err)
 		}

diff --git a/worker/pkg/workflows/datasync/workflow/testdata/double-reference/insert.sql b/worker/pkg/workflows/datasync/workflow/testdata/double-reference/insert.sql
@@ -0,0 +1,29 @@
+SET search_path TO double_reference;
+-- COMPANY DATA
+INSERT INTO company (name, url, employee_count, uuid)
+VALUES
+  ('Acme Corporation', 'www.acme.com', 500, uuid_generate_v4()),
+  ('Global Enterprises', 'globalenterprises.net', 1200, uuid_generate_v4()),
+  ('Tech Innovations', 'www.techinnovations.io', 250, uuid_generate_v4());
+
+-- DEPARTMENT DATA 
+INSERT INTO department (name, url, company_id, uuid)
+VALUES
+  ('Marketing', 'marketing.acme.com', 1, uuid_generate_v4()),  -- Acme Corporation
+  ('Sales', 'sales.acme.com', 1, uuid_generate_v4()), 
+  ('Finance', null, 2, uuid_generate_v4()), -- Global Enterprises
+  ('R&D', 'rnd.techinnovations.io', 3, uuid_generate_v4()); -- Tech Innovations
+
+-- TRANSACTION DATA
+INSERT INTO transaction (id, amount, created, updated, department_id, date, currency, description, uuid)
+VALUES
+  (1, 250.50, now() - interval '2 weeks', now(), 1, '2024-05-01', 'USD', 'Office Supplies', uuid_generate_v4()),
+  (2, 1250.00, now() - interval '5 days', now(), 2, '2024-05-06', 'GBP', 'Travel Expenses', uuid_generate_v4()),
+  (3, 87.25, now() - interval '1 month', now(), 3, '2024-04-20', 'EUR', 'Lunch Meeting', uuid_generate_v4());
+  -- Repeat with varied data ...
+
+-- EXPENSE REPORT DATA
+INSERT INTO expense_report (id, invoice_id, date, amount, department_source_id, department_destination_id, created, updated, currency, transaction_type, paid, adjustment_amount, transaction_id)
+VALUES
+  (1, 'INV-1234', '2024-05-03', 500.00, 1, 2, now() - interval '15 days', now(), 'USD', 1, true, null, 1),
+  (2,'INV-5678', '2024-04-28', 128.75, 3, 1, now() - interval '20 days', now() - interval '1 day', 'CAD', 2, false, 12.50, 3);