[mysql] Add integration test scaffolding

.github/workflows/gha-integration-test.yaml → .github/workflows/gha-integration-tests.yaml

@@ -2,7 +2,7 @@ name: Integration tests
 run-name: Running integration tests
 on: [push]
 jobs:
-  container-job:
+  postgres-integration-test:
     runs-on: ubuntu-latest
     container: golang:1.22
     services:
@@ -21,5 +21,16 @@ jobs:
         uses: actions/setup-go@v5
         with:
           go-version: 1.22
-      - name: Run integration test
-        run: PG_HOST=postgres make itest
+      - name: Run Postgres integration test
+        run: PG_HOST=postgres make postgres-itest
+  mysql-integration-test:
+    runs-on: ubuntu-latest
+    container: golang:1.22
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: 1.22
+      - name: Run MySQL integration test
+        run: MYSQL_HOST=postgres make mysql-itest

Makefile

@@ -6,9 +6,15 @@ static:
 test:
 	go test ./...
 
-.PHONY: integration-test
-itest:
-	go run sources/postgres/integration_test/main.go
+.PHONY: mysql-itest
+mysql-itest:
+	go run integration_tests/mysql/main.go
+
+
+.PHONY: postgres-itest
+postgres-itest:
+	go run integration_tests/postgres/main.go
+
 
 .PHONY: race
 race:

integration_tests/mysql/main.go

@@ -0,0 +1,14 @@
+package main
+
+import (
+	"log/slog"
+	"os"
+
+	"github.com/lmittmann/tint"
+)
+
+func main() {
+	os.Setenv("TZ", "UTC")
+	slog.SetDefault(slog.New(tint.NewHandler(os.Stderr, &tint.Options{Level: slog.LevelInfo})))
+
+}

sources/postgres/integration_test/main.go → integration_tests/postgres/main.go

@@ -8,13 +8,12 @@ import (
 	"maps"
 	"math/rand/v2"
 	"os"
-	"strings"
 
-	"github.com/artie-labs/transfer/lib/cdc/util"
 	_ "github.com/jackc/pgx/v5/stdlib"
 	"github.com/lmittmann/tint"
 
 	"github.com/artie-labs/reader/config"
+	"github.com/artie-labs/reader/integration_tests/utils"
 	"github.com/artie-labs/reader/lib"
 	"github.com/artie-labs/reader/lib/debezium"
 	"github.com/artie-labs/reader/lib/logger"
@@ -55,43 +54,6 @@ func main() {
 	}
 }
 
-func getPayload(message lib.RawMessage) util.SchemaEventPayload {
-	payloadTyped, ok := message.GetPayload().(util.SchemaEventPayload)
-	if !ok {
-		panic("payload is not of type util.SchemaEventPayload")
-	}
-	return payloadTyped
-}
-
-func checkDifference(name, expected, actual string) bool {
-	if expected == actual {
-		return false
-	}
-	expectedLines := strings.Split(expected, "\n")
-	actualLines := strings.Split(actual, "\n")
-	fmt.Printf("Expected %s:\n", name)
-	fmt.Println("--------------------------------------------------------------------------------")
-	for i, line := range expectedLines {
-		prefix := " "
-		if i >= len(actualLines) || line != actualLines[i] {
-			prefix = ">"
-		}
-		fmt.Println(prefix + line)
-	}
-	fmt.Println("--------------------------------------------------------------------------------")
-	fmt.Printf("Actual %s:\n", name)
-	fmt.Println("--------------------------------------------------------------------------------")
-	for i, line := range actualLines {
-		prefix := " "
-		if i >= len(expectedLines) || line != expectedLines[i] {
-			prefix = ">"
-		}
-		fmt.Println(prefix + line)
-	}
-	fmt.Println("--------------------------------------------------------------------------------")
-	return true
-}
-
 func readTable(db *sql.DB, tableName string, batchSize int) ([]lib.RawMessage, error) {
 	tableCfg := config.PostgreSQLTable{
 		Schema:    "public",
@@ -725,12 +687,12 @@ func testTypes(db *sql.DB) error {
 		return fmt.Errorf("failed to marshal payload")
 	}
 
-	if checkDifference("partition key", `{"pk":1}`, string(keyBytes)) {
+	if utils.checkDifference("partition key", `{"pk":1}`, string(keyBytes)) {
 		return fmt.Errorf("partition key does not match")
 	}
 
-	expectedPayload := fmt.Sprintf(expectedPayloadTemplate, getPayload(row).Payload.Source.TsMs, tempTableName)
-	if checkDifference("payload", expectedPayload, string(valueBytes)) {
+	expectedPayload := fmt.Sprintf(expectedPayloadTemplate, utils.getPayload(row).Payload.Source.TsMs, tempTableName)
+	if utils.checkDifference("payload", expectedPayload, string(valueBytes)) {
 		return fmt.Errorf("payload does not match")
 	}
 
@@ -864,7 +826,7 @@ func testScan(db *sql.DB) error {
 			if !maps.Equal(row.PartitionKey, expectedPartitionKeys[i]) {
 				return fmt.Errorf("partition keys are different for row %d, batch size %d, %T != %T", i, batchSize, row.PartitionKey, expectedPartitionKeys[i])
 			}
-			textValue := getPayload(row).Payload.After["c_text_value"]
+			textValue := utils.GetPayload(row).Payload.After["c_text_value"]
 			if textValue != expectedValues[i] {
 				return fmt.Errorf("row values are different for row %d, batch size %d, %T != %T", i, batchSize, textValue, expectedPartitionKeys[i])
 			}

integration_tests/utils/utils.go

@@ -0,0 +1,46 @@
+package utils
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/artie-labs/reader/lib"
+	"github.com/artie-labs/transfer/lib/cdc/util"
+)
+
+func GetPayload(message lib.RawMessage) util.SchemaEventPayload {
+	payloadTyped, ok := message.GetPayload().(util.SchemaEventPayload)
+	if !ok {
+		panic("payload is not of type util.SchemaEventPayload")
+	}
+	return payloadTyped
+}
+
+func CheckDifference(name, expected, actual string) bool {
+	if expected == actual {
+		return false
+	}
+	expectedLines := strings.Split(expected, "\n")
+	actualLines := strings.Split(actual, "\n")
+	fmt.Printf("Expected %s:\n", name)
+	fmt.Println("--------------------------------------------------------------------------------")
+	for i, line := range expectedLines {
+		prefix := " "
+		if i >= len(actualLines) || line != actualLines[i] {
+			prefix = ">"
+		}
+		fmt.Println(prefix + line)
+	}
+	fmt.Println("--------------------------------------------------------------------------------")
+	fmt.Printf("Actual %s:\n", name)
+	fmt.Println("--------------------------------------------------------------------------------")
+	for i, line := range actualLines {
+		prefix := " "
+		if i >= len(expectedLines) || line != expectedLines[i] {
+			prefix = ">"
+		}
+		fmt.Println(prefix + line)
+	}
+	fmt.Println("--------------------------------------------------------------------------------")
+	return true
+}