feat: add ftltest.WithDatabase(…) to get a clean test db (#1436)

```
ftltest.Context(
  ftltest.WithDatabase(dbHandle),
)
```
This new option does the following:
- reads the existing DSN but appends `_test` to the table name
- clears out all tables in the db's public schema

Integration test `TestDatabase()` has been updated to have real and test
db set up, with a deployment writing to the real db and unit tests
running on a test db.

go-runtime/ftl/ftltest/ftltest.go

@@ -3,8 +3,10 @@ package ftltest
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"fmt"
+	"net/url"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -16,6 +18,7 @@ import (
 	"github.com/TBD54566975/ftl/internal/log"
 	"github.com/TBD54566975/ftl/internal/modulecontext"
 	"github.com/TBD54566975/ftl/internal/slices"
+	_ "github.com/jackc/pgx/v5/stdlib" // SQL driver
 )
 
 type OptionsState struct {
@@ -168,6 +171,64 @@ func WithSecret[T ftl.SecretType](secret ftl.SecretValue[T], value T) Option {
 	}
 }
 
+// WithDatabase sets up a database for testing by appending "_test" to the DSN and emptying all tables
+//
+// To be used when setting up a context for a test:
+// ctx := ftltest.Context(
+//
+//	ftltest.WithDatabase(db),
+//	... other options
+//
+// )
+func WithDatabase(dbHandle ftl.Database) Option {
+	return func(ctx context.Context, state *OptionsState) error {
+		envarName := fmt.Sprintf("FTL_POSTGRES_DSN_%s_%s", strings.ToUpper(ftl.Module()), strings.ToUpper(dbHandle.Name))
+		originalDSN, ok := os.LookupEnv(envarName)
+		if !ok {
+			return fmt.Errorf("missing DSN for database %s: expected to find it at the environment variable %s", dbHandle.Name, envarName)
+		}
+
+		// convert DSN by appending "_test" to table name
+		// postgres DSN format: postgresql://[user[:password]@][netloc][:port][/dbname][?param1=value1&...]
+		// source: https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING
+		dsnURL, err := url.Parse(originalDSN)
+		if err != nil {
+			return fmt.Errorf("could not parse DSN: %w", err)
+		}
+		if dsnURL.Path == "" {
+			return fmt.Errorf("DSN for %s must include table name: %s", dbHandle.Name, originalDSN)
+		}
+		dsnURL.Path += "_test"
+		dsn := dsnURL.String()
+
+		// connect to db and clear out the contents of each table
+		sqlDB, err := sql.Open("pgx", dsn)
+		if err != nil {
+			return fmt.Errorf("could not create database %q with DSN %q: %w", dbHandle.Name, dsn, err)
+		}
+		_, err = sqlDB.ExecContext(ctx, `DO $$ 
+		DECLARE 
+		   table_name text; 
+		BEGIN 
+		   FOR table_name IN (SELECT tablename FROM pg_tables WHERE schemaname = 'public') 
+		   LOOP 
+			  EXECUTE 'DELETE FROM ' || table_name; 
+		   END LOOP; 
+		END $$;`)
+		if err != nil {
+			return fmt.Errorf("could not clear tables in database %q: %w", dbHandle.Name, err)
+		}
+
+		// replace original database with test database
+		replacementDB, err := modulecontext.NewDatabase(modulecontext.DBTypePostgres, dsn)
+		if err != nil {
+			return fmt.Errorf("could not create database %q with DSN %q: %w", dbHandle.Name, dsn, err)
+		}
+		state.databases[dbHandle.Name] = replacementDB
+		return nil
+	}
+}
+
 // WhenVerb replaces an implementation for a verb
 //
 // To be used when setting up a context for a test:

integration/actions_test.go

@@ -262,9 +262,22 @@ func queryRow(database string, query string, expected ...interface{}) action {
 }
 
 // Create a database for use by a module.
-func createDB(t testing.TB, module, dbName string) {
+func createDBAction(module, dbName string, isTest bool) action {
+	return func(t testing.TB, ic testContext) error {
+		createDB(t, module, dbName, isTest)
+		return nil
+	}
+}
+
+func createDB(t testing.TB, module, dbName string, isTestDb bool) {
+	// envars do not include test suffix
 	t.Setenv(fmt.Sprintf("FTL_POSTGRES_DSN_%s_%s", strings.ToUpper(module), strings.ToUpper(dbName)),
 		fmt.Sprintf("postgres://postgres:secret@localhost:54320/%s?sslmode=disable", dbName))
+
+	// insert test suffix if needed when actually setting up db
+	if isTestDb {
+		dbName += "_test"
+	}
 	infof("Creating database %s", dbName)
 	db, err := sql.Open("pgx", "postgres://postgres:secret@localhost:54320/ftl?sslmode=disable")
 	assert.NoError(t, err, "failed to open database connection")
@@ -276,6 +289,9 @@ func createDB(t testing.TB, module, dbName string) {
 	err = db.Ping()
 	assert.NoError(t, err, "failed to ping database")
 
+	_, err = db.Exec("DROP DATABASE IF EXISTS " + dbName)
+	assert.NoError(t, err, "failed to delete existing database")
+
 	_, err = db.Exec("CREATE DATABASE " + dbName)
 	assert.NoError(t, err, "failed to create database")
 

integration/integration_test.go

@@ -84,12 +84,18 @@ func TestNonExportedDecls(t *testing.T) {
 }
 
 func TestDatabase(t *testing.T) {
-	createDB(t, "database", "testdb")
+	createDB(t, "database", "testdb", false)
 	run(t,
+		// deploy real module against "testdb"
 		copyModule("database"),
 		deploy("database"),
 		call("database", "insert", obj{"data": "hello"}, func(response obj) error { return nil }),
 		queryRow("testdb", "SELECT data FROM requests", "hello"),
+
+		// run tests which should only affect "testdb_test"
+		createDBAction("database", "testdb", true),
+		testModule("database"),
+		queryRow("testdb", "SELECT data FROM requests", "hello"),
 	)
 }
 

integration/testdata/go/database/database_test.go

@@ -0,0 +1,50 @@
+package database
+
+import (
+	"context"
+	"testing"
+
+	"github.com/TBD54566975/ftl/go-runtime/ftl/ftltest"
+	"github.com/alecthomas/assert/v2"
+)
+
+func TestDatabase(t *testing.T) {
+	ctx := ftltest.Context(
+		ftltest.WithDatabase(db),
+	)
+
+	Insert(ctx, InsertRequest{Data: "unit test 1"})
+	list, err := getAll(ctx)
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(list))
+	assert.Equal(t, "unit test 1", list[0])
+
+	ctx = ftltest.Context(
+		ftltest.WithDatabase(db),
+	)
+
+	Insert(ctx, InsertRequest{Data: "unit test 2"})
+	list, err = getAll(ctx)
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(list))
+	assert.Equal(t, "unit test 2", list[0])
+}
+
+func getAll(ctx context.Context) ([]string, error) {
+	rows, err := db.Get(ctx).Query("SELECT data FROM requests ORDER BY created_at;")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	list := []string{}
+	for rows.Next() {
+		var data string
+		err := rows.Scan(&data)
+		if err != nil {
+			return nil, err
+		}
+		list = append(list, data)
+	}
+	return list, nil
+}

integration/testdata/go/database/go.mod

@@ -2,20 +2,28 @@ module ftl/database
 
 go 1.22.2
 
-require github.com/TBD54566975/ftl v0.189.0
+require (
+	github.com/TBD54566975/ftl v0.189.0
+	github.com/alecthomas/assert/v2 v2.9.0
+)
 
 require (
 	connectrpc.com/connect v1.16.1 // indirect
 	connectrpc.com/grpcreflect v1.2.0 // indirect
 	connectrpc.com/otelconnect v0.7.0 // indirect
+	github.com/BurntSushi/toml v1.3.2 // indirect
+	github.com/TBD54566975/scaffolder v0.8.0 // indirect
 	github.com/alecthomas/concurrency v0.0.2 // indirect
+	github.com/alecthomas/kong v0.9.0 // indirect
 	github.com/alecthomas/participle/v2 v2.1.1 // indirect
+	github.com/alecthomas/repr v0.4.0 // indirect
 	github.com/alecthomas/types v0.14.0 // indirect
 	github.com/alessio/shellescape v1.4.2 // indirect
 	github.com/danieljoos/wincred v1.2.0 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
+	github.com/hexops/gotextdiff v1.0.3 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
 	github.com/jackc/pgx/v5 v5.5.5 // indirect