Tilt local env: Expand the set of tests that can run locally by adding Toxyproxy service and separating PG connections in tests

Tiltfile

@@ -6,7 +6,7 @@ docker_build('flow-api', '.',
     dockerfile='stacks/flow.Dockerfile',
     target='flow-api',
     only=['flow/', 'stacks/flow.Dockerfile'],
-    build_args={'PEERDB_VERSION_SHA_SHORT': os.getenv('PEERDB_VERSION_SHA_SHORT', '')},
+    build_args={'PEERDB_VERSION_SHA_SHORT': os.getenv('PEERDB_VERSION_SHA_SHORT', 'unknown')},
 )
 
 docker_build('flow-worker', '.',
@@ -149,6 +149,11 @@ dc_resource('dozzle', labels=['Monitoring'], links=[
     link('http://localhost:8118', 'Dozzle Container Monitor'),
 ])
 
+# Test services: Services supporting test execution that are not data stores, like proxies, mock servers, etc.
+
+dc_resource('toxiproxy', labels=['TestInfra'], auto_init=False)
+dc_resource('openssh', labels=['TestInfra'], auto_init=False)
+
 # Cleanup
 
 load('ext://uibutton', 'cmd_button', 'location')
@@ -178,6 +183,8 @@ def e2e_test(name, test_run, extra_deps=[], vars_overrides={}):
         resource_deps=['flow-api', 'flow-worker', 'catalog', 'provision-clickhouse'] + extra_deps,
     )
 
+# Generic e2e tests
+
 # Postgres to ClickHouse generic tests
 e2e_test('postgres', 'TestGenericCH_PG', ['provision-postgres'])
 
@@ -191,4 +198,8 @@ e2e_test('mysql-pos', 'TestGenericCH_MySQL', ['provision-mysql-pos'], vars_overr
 e2e_test('mariadb', 'TestGenericCH_MySQL', ['provision-mariadb'], vars_overrides={'CI_MYSQL_PORT': '$CI_MARIADB_PORT'})
 
 # MongoDB to ClickHouse test suite
-e2e_test('mongodb', 'TestMongoClickhouseSuite', ['provision-mongodb'])
+e2e_test('mongodb', 'TestMongoClickhouseSuite', ['provision-mongodb'])
+
+# API e2e tests
+
+e2e_test('api-postgres', 'TestApiPg', ['provision-postgres'])

ancillary-docker-compose.yml

@@ -129,6 +129,53 @@ services:
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock
 
+  toxiproxy:
+    container_name: peerdb-test-toxiproxy
+    # Same image as CI flow
+    image: ghcr.io/shopify/toxiproxy:2.12.0@sha256:9378ed52a28bc50edc1350f936f518f31fa95f0d15917d6eb40b8e376d1a214e
+    restart: unless-stopped
+    ports:
+      - 18474:8474
+      - 9902:9902
+      - 9904:9904
+      - 9903:9903
+      - 42001:42001
+      - 42002:42002
+      - 42003:42003
+      - 49001:49001
+      - 49002:49002
+      - 49003:49003
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    healthcheck:
+      test: ["CMD", "/toxiproxy-cli", "list"]
+      interval: 2s
+      timeout: 10s
+      retries: 5
+
+  openssh:
+    container_name: peerdb-test-ssh
+    # Same image as CI flow
+    image: linuxserver/openssh-server:latest@sha256:57c13941ef385d735cca18ee969eeba18e42ab95480c97af2182505aacdfe307
+    restart: unless-stopped
+    ports:
+      - "2222:2222"
+    environment:
+      - PUID=1000
+      - PGID=1000
+      - TZ=Etc/UTC
+      - USER_NAME=testuser
+      - USER_PASSWORD=testpass
+      - PASSWORD_ACCESS=true
+      - DOCKER_MODS=linuxserver/mods:openssh-server-ssh-tunnel
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    healthcheck:
+      test: ["CMD", "nc", "-z", "localhost", "2222"]
+      interval: 2s
+      timeout: 10s
+      retries: 5
+
 volumes:
   postgres_data:
   clickhouse_data:
@@ -140,3 +187,4 @@ volumes:
 networks:
   default:
     name: peerdb_network
+    external: true

flow/connectors/clickhouse/s3_iam_role_test.go

@@ -3,6 +3,7 @@ package connclickhouse
 import (
 	"fmt"
 	"os"
+	"strconv"
 	"testing"
 	"time"
 
@@ -13,6 +14,11 @@ import (
 )
 
 func TestIAMRoleCanIssueSelectFromS3(t *testing.T) {
+	const bucketNameEnvVar = "FLOW_TESTS_AWS_S3_BUCKET_NAME"
+	if os.Getenv(bucketNameEnvVar) == "" {
+		t.Skipf("skipping test since %s is not set", bucketNameEnvVar)
+	}
+
 	for _, envVar := range []string{
 		"AWS_ACCESS_KEY_ID",
 		"AWS_SECRET_ACCESS_KEY",
@@ -34,13 +40,26 @@ func TestIAMRoleCanIssueSelectFromS3(t *testing.T) {
 	} {
 		t.Setenv(envVar, os.Getenv("FLOW_TESTS_"+envVar))
 	}
-	t.Setenv("PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME", os.Getenv("FLOW_TESTS_AWS_S3_BUCKET_NAME"))
+	t.Setenv("PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME", os.Getenv(bucketNameEnvVar))
 	ctx := t.Context()
 
+	chHost := "localhost"
+	if host := os.Getenv("CI_CLICKHOUSE_HOST"); host != "" {
+		chHost = host
+	}
+	var chPort uint32 = 9000
+	if port := os.Getenv("CI_CLICKHOUSE_NATIVE_PORT"); port != "" {
+		p, err := strconv.ParseUint(port, 10, 32)
+		if err != nil {
+			t.Fatalf("invalid CI_CLICKHOUSE_NATIVE_PORT %q: %v", port, err)
+		}
+		chPort = uint32(p)
+	}
+
 	conn, err := NewClickHouseConnector(ctx, nil,
 		&protos.ClickhouseConfig{
-			Host:       "localhost",
-			Port:       9000,
+			Host:       chHost,
+			Port:       chPort,
 			Database:   "default",
 			DisableTls: true,
 		})

flow/connectors/mysql/ssh_keepalive_test.go

@@ -3,6 +3,8 @@ package connmysql
 import (
 	"context"
 	"os"
+	"strconv"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -24,11 +26,28 @@ func setupMySQLConnectorWithSSH(ctx context.Context, t *testing.T, proxyName str
 	toxiproxyClient := utils.NewToxiproxyClient(t)
 	sshProxy := utils.CreateSSHProxy(t, toxiproxyClient, proxyName, proxyPort)
 
+	mysqlHost := "mysql"
+	if envHost := os.Getenv("CI_MYSQL_HOST"); envHost != "" {
+		mysqlHost = envHost
+	}
+
+	var mysqlPort uint32 = 3306
+	if envPortStr := os.Getenv("CI_MYSQL_PORT"); envPortStr != "" {
+		envPort, err := strconv.ParseUint(strings.TrimSpace(strings.Split(envPortStr, "#")[0]), 10, 32)
+		require.NoError(t, err, "Failed to parse CI_MYSQL_PORT")
+		mysqlPort = uint32(envPort)
+	}
+
+	mysqlRootPass := "cipass"
+	if envPass := os.Getenv("CI_MYSQL_ROOT_PASSWORD"); envPass != "" {
+		mysqlRootPass = envPass
+	}
+
 	connector, err := NewMySqlConnector(ctx, &protos.MySqlConfig{
-		Host:     "mysql",
-		Port:     3306,
+		Host:     mysqlHost,
+		Port:     mysqlPort,
 		User:     "root",
-		Password: "cipass",
+		Password: mysqlRootPass,
 		Database: "mysql",
 		SshConfig: &protos.SSHConfig{
 			Host:     "localhost",

flow/connectors/postgres/ssh_keepalive_test.go

@@ -2,6 +2,9 @@ package connpostgres
 
 import (
 	"context"
+	"os"
+	"strconv"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -23,9 +26,30 @@ func setupPostgresConnectorWithSSH(ctx context.Context, t *testing.T, proxyName
 	toxiproxyClient := utils.NewToxiproxyClient(t)
 	sshProxy := utils.CreateSSHProxy(t, toxiproxyClient, proxyName, proxyPort)
 
+	pgHost := "localhost" // Default for local environments
+
+	// In local set-up environments like Tilt, when a non catalog
+	// instance of Postgres is present, we use it instead of the catalog.
+	if envHost := os.Getenv("PG_HOST"); envHost != "" {
+		pgHost = envHost
+	}
+
+	// In CI, TOXIPROXY_POSTGRES_HOST is set, usually pointing to 'catalog'
+	// network.
+	if toxiproxyPgHost := os.Getenv("TOXIPROXY_POSTGRES_HOST"); toxiproxyPgHost != "" {
+		pgHost = toxiproxyPgHost
+	}
+
+	var pgPort uint32 = 5432
+	if envPortStr := os.Getenv("PG_PORT"); envPortStr != "" {
+		envPort, err := strconv.ParseUint(strings.TrimSpace(strings.Split(envPortStr, "#")[0]), 10, 32)
+		require.NoError(t, err, "Failed to parse PG_PORT")
+		pgPort = uint32(envPort)
+	}
+
 	connector, err := NewPostgresConnector(ctx, nil, &protos.PostgresConfig{
-		Host:     "catalog",
-		Port:     5432,
+		Host:     pgHost,
+		Port:     pgPort,
 		User:     "postgres",
 		Password: "postgres",
 		Database: "postgres",