-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add unit tests covering logical replication and conn slot errors
- Loading branch information
Showing
1,633 changed files
with
266,363 additions
and
20,936 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,101 @@ | ||
| package test | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "strings" | ||
| "testing" | ||
|
|
||
| "github.com/jackc/pgx/v5" | ||
| "github.com/jackc/pgx/v5/pgconn" | ||
| "github.com/stretchr/testify/require" | ||
| tc "github.com/testcontainers/testcontainers-go" | ||
| pgtc "github.com/testcontainers/testcontainers-go/modules/postgres" | ||
| ) | ||
|
|
||
| type StartPGOpts struct { | ||
| Version int | ||
| DisableLogicalReplication bool | ||
| DisableCreateRoles bool | ||
| DisableCreateSlot bool | ||
| } | ||
|
|
||
| func StartPG(t *testing.T, ctx context.Context, opts StartPGOpts) (tc.Container, pgx.ConnConfig) { | ||
| t.Helper() | ||
| args := []tc.ContainerCustomizer{ | ||
| pgtc.WithDatabase("db"), | ||
| pgtc.WithUsername("postgres"), | ||
| pgtc.WithPassword("password"), | ||
| pgtc.BasicWaitStrategies(), | ||
| } | ||
| if !opts.DisableLogicalReplication { | ||
| args = append(args, tc.CustomizeRequest(tc.GenericContainerRequest{ | ||
| ContainerRequest: tc.ContainerRequest{ | ||
| Cmd: []string{"-c", "wal_level=logical"}, | ||
| }, | ||
| })) | ||
| } | ||
| c, err := pgtc.Run(ctx, | ||
| fmt.Sprintf("docker.io/postgres:%d-alpine", opts.Version), | ||
| args..., | ||
| ) | ||
|
|
||
| conn, err := pgconn.Connect(ctx, connString(t, c)) | ||
| if err != nil { | ||
| require.NoError(t, err) | ||
| } | ||
|
|
||
| if !opts.DisableCreateRoles { | ||
| // Create the replication slot. | ||
| err := prepareRoles(ctx, conn) | ||
| require.NoError(t, err) | ||
| } | ||
| if !opts.DisableCreateSlot { | ||
| // Create the replication slot. | ||
| err := createReplicationSlot(ctx, conn) | ||
| require.NoError(t, err) | ||
| } | ||
|
|
||
| require.NoError(t, err) | ||
| return c, connOpts(t, c) | ||
| } | ||
|
|
||
| func prepareRoles(ctx context.Context, c *pgconn.PgConn) error { | ||
| stmt := ` | ||
| CREATE USER inngest WITH REPLICATION PASSWORD 'password'; | ||
| GRANT USAGE ON SCHEMA public TO inngest; | ||
| GRANT SELECT ON ALL TABLES IN SCHEMA public TO inngest; | ||
| ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO inngest; | ||
| CREATE PUBLICATION inngest FOR ALL TABLES; | ||
| ` | ||
| res := c.Exec(ctx, stmt) | ||
| if err := res.Close(); err != nil { | ||
| return err | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func createReplicationSlot(ctx context.Context, c *pgconn.PgConn) error { | ||
| stmt := ` | ||
| -- pgoutput logical repl plugin | ||
| SELECT pg_create_logical_replication_slot('inngest_cdc', 'pgoutput'); | ||
| ` | ||
| res := c.Exec(ctx, stmt) | ||
| if err := res.Close(); err != nil { | ||
| return err | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func connString(t *testing.T, c tc.Container) string { | ||
| p, err := c.MappedPort(context.TODO(), "5432") | ||
| require.NoError(t, err) | ||
| port := strings.ReplaceAll(string(p), "/tcp", "") | ||
| return fmt.Sprintf("postgres://postgres:password@localhost:%s/db", port) | ||
| } | ||
|
|
||
| func connOpts(t *testing.T, c tc.Container) pgx.ConnConfig { | ||
| cfg, err := pgx.ParseConfig(connString(t, c)) | ||
| require.NoError(t, err) | ||
| return *cfg | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| package replicator | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
| "testing" | ||
|
|
||
| "github.com/inngest/pgcap/internal/test" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestConnectingWithoutLogicalReplicationFails(t *testing.T) { | ||
| ctx := context.Background() | ||
| versions := []int{12, 13, 14, 15, 16} | ||
|
|
||
| for _, v := range versions { | ||
| c, conn := test.StartPG(t, ctx, test.StartPGOpts{ | ||
| Version: v, | ||
| DisableLogicalReplication: true, | ||
| DisableCreateSlot: true, | ||
| }) | ||
|
|
||
| opts := PostgresOpts{Config: conn} | ||
| r, err := Postgres(ctx, opts) | ||
| require.NoError(t, err) | ||
|
|
||
| err = r.Pull(ctx, nil) | ||
| require.ErrorIs(t, err, ErrLogicalReplicationNotSetUp) | ||
|
|
||
| c.Stop(ctx, nil) | ||
| } | ||
| } | ||
|
|
||
| func TestConnectingWithoutReplicationSlotFails(t *testing.T) { | ||
| ctx := context.Background() | ||
| versions := []int{12, 13, 14, 15, 16} | ||
|
|
||
| for _, v := range versions { | ||
| c, conn := test.StartPG(t, ctx, test.StartPGOpts{ | ||
| Version: v, | ||
| DisableCreateSlot: true, | ||
| }) | ||
|
|
||
| opts := PostgresOpts{Config: conn} | ||
| r, err := Postgres(ctx, opts) | ||
| require.NoError(t, err) | ||
|
|
||
| err = r.Pull(ctx, nil) | ||
| require.ErrorIs(t, err, ErrReplicationSlotNotFound) | ||
|
|
||
| c.Stop(ctx, nil) | ||
| } | ||
| } | ||
|
|
||
| func TestMultipleConectionsFail(t *testing.T) { | ||
| // versions := []int{12, 13, 14, 15, 16} | ||
| versions := []int{14} | ||
|
|
||
| for _, v := range versions { | ||
| ctx := context.Background() | ||
| c, conn := test.StartPG(t, ctx, test.StartPGOpts{ | ||
| Version: v, | ||
| }) | ||
|
|
||
| // The first time we connect things should succeed. | ||
| opts := PostgresOpts{Config: conn} | ||
| r1, err := Postgres(ctx, opts) | ||
| require.NoError(t, err) | ||
|
|
||
| wg := sync.WaitGroup{} | ||
|
|
||
| wg.Add(1) | ||
| go func() { | ||
| defer wg.Done() | ||
| err := r1.Pull(ctx, nil) | ||
| require.NoError(t, err) | ||
| }() | ||
|
|
||
| r2, err := Postgres(ctx, opts) | ||
| err = r2.Pull(ctx, nil) | ||
| require.Error(t, err) | ||
|
|
||
| c.Stop(ctx, nil) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| version = 1 | ||
|
|
||
| test_patterns = [ | ||
| "*_test.go" | ||
| ] | ||
|
|
||
| [[analyzers]] | ||
| name = "go" | ||
| enabled = true | ||
|
|
||
| [analyzers.meta] | ||
| import_path = "dario.cat/mergo" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| #### joe made this: http://goel.io/joe | ||
|
|
||
| #### go #### | ||
| # Binaries for programs and plugins | ||
| *.exe | ||
| *.dll | ||
| *.so | ||
| *.dylib | ||
|
|
||
| # Test binary, build with `go test -c` | ||
| *.test | ||
|
|
||
| # Output of the go coverage tool, specifically when used with LiteIDE | ||
| *.out | ||
|
|
||
| # Project-local glide cache, RE: https://github.com/Masterminds/glide/issues/736 | ||
| .glide/ | ||
|
|
||
| #### vim #### | ||
| # Swap | ||
| [._]*.s[a-v][a-z] | ||
| [._]*.sw[a-p] | ||
| [._]s[a-v][a-z] | ||
| [._]sw[a-p] | ||
|
|
||
| # Session | ||
| Session.vim | ||
|
|
||
| # Temporary | ||
| .netrwhist | ||
| *~ | ||
| # Auto-generated tag files | ||
| tags |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| language: go | ||
| arch: | ||
| - amd64 | ||
| - ppc64le | ||
| install: | ||
| - go get -t | ||
| - go get golang.org/x/tools/cmd/cover | ||
| - go get github.com/mattn/goveralls | ||
| script: | ||
| - go test -race -v ./... | ||
| after_script: | ||
| - $HOME/gopath/bin/goveralls -service=travis-ci -repotoken $COVERALLS_TOKEN |
Oops, something went wrong.