fixed repl tests nuking all slots and publications

PeerDB-io · Sep 5, 2023 · 71a9670 · 71a9670
1 parent efd7b1b
commit 71a9670
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 3 deletions.
diff --git a/flow/connectors/postgres/postgres.go b/flow/connectors/postgres/postgres.go
@@ -692,7 +692,7 @@ func (c *PostgresConnector) ReplayTableSchemaDelta(flowJobName string, schemaDel
 	for _, addedColumn := range schemaDelta.AddedColumns {
 		_, err = tableSchemaModifyTx.Exec(c.ctx, fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s",
 			schemaDelta.DstTableName, addedColumn.ColumnName,
-			qValueKindToPostgresType(string(addedColumn.ColumnType))))
+			qValueKindToPostgresType(addedColumn.ColumnType)))
 		if err != nil {
 			return fmt.Errorf("failed to add column %s for table %s: %w", addedColumn.ColumnName,
 				schemaDelta.SrcTableName, err)

diff --git a/flow/connectors/postgres/postgres_repl_test.go b/flow/connectors/postgres/postgres_repl_test.go
@@ -72,7 +72,7 @@ func (suite *PostgresReplicationSnapshotTestSuite) TearDownSuite() {
 
 	// Fetch all the publications
 	rows, err := teardownTx.Query(context.Background(),
-		"SELECT pubname FROM pg_publication WHERE pubname LIKE 'peerflow_pub%'")
+		"SELECT pubname FROM pg_publication WHERE pubname LIKE $1", fmt.Sprintf("%%%s", "test_simple_slot_creation"))
 	require.NoError(suite.T(), err)
 
 	// Iterate over the publications and drop them
@@ -87,7 +87,8 @@ func (suite *PostgresReplicationSnapshotTestSuite) TearDownSuite() {
 	}
 
 	_, err = teardownTx.Exec(context.Background(),
-		"SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots")
+		"SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots WHERE slot_name LIKE $1",
+		fmt.Sprintf("%%%s", "test_simple_slot_creation"))
 	require.NoError(suite.T(), err)
 
 	err = teardownTx.Commit(context.Background())