refactor integration test, add note about delete test

Author: hariso

source_integration_test.go

@@ -38,16 +38,68 @@ import (
 	"github.com/shopspring/decimal"
 )
 
-// todo test updates and deletes, for not null and nullable values
-func TestSource_ReadN_Snapshot_CDC(t *testing.T) {
+type readTestCase struct {
+	name        string
+	notNullOnly bool
+	snapshot    bool
+	cdc         bool
+	opDelete    bool
+}
+
+func TestSource_ReadN(t *testing.T) {
+	testCases := []readTestCase{
+		{
+			name:        "snapshot not only only",
+			notNullOnly: true,
+			snapshot:    true,
+		},
+		{
+			name:        "snapshot with nullable values",
+			notNullOnly: false,
+			snapshot:    true,
+		},
+		{
+			name:        "cdc not only only",
+			notNullOnly: true,
+			cdc:         true,
+		},
+		{
+			name:        "cdc with nullable values",
+			notNullOnly: false,
+			cdc:         true,
+		},
+
+		{
+			name:        "delete cdc data not only only",
+			notNullOnly: true,
+			cdc:         true,
+			opDelete:    true,
+		},
+		{
+			name:        "delete cdc data with nullable values",
+			notNullOnly: false,
+			cdc:         true,
+			opDelete:    true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			runReadTest(t, tc)
+		})
+	}
+}
+
+func runReadTest(t *testing.T, tc readTestCase) {
 	is := is.New(t)
 	ctx := test.Context(t)
-	notNullOnly := false
 	conn := test.ConnectSimple(ctx, t, test.RepmgrConnString)
 
 	tableName := createTableWithManyTypes(ctx, t)
-	// Write the snapshot data
-	insertRow(ctx, is, conn, tableName, 1, notNullOnly)
+
+	if tc.snapshot {
+		insertRow(ctx, is, conn, tableName, 1, tc.notNullOnly)
+	}
 
 	slotName := "conduitslot1"
 	publicationName := "conduitpub1"
@@ -79,20 +131,27 @@ func TestSource_ReadN_Snapshot_CDC(t *testing.T) {
 		is.NoErr(s.Teardown(ctx))
 	})
 
-	// Read, ack, and assert the snapshot record is OK
-	rec := readAndAck(ctx, is, s)
-	assertRecordOK(is, tableName, rec, 1, notNullOnly)
+	if tc.snapshot {
+		// Read, ack, and assert the snapshot record is OK
+		rec := readAndAck(ctx, is, s)
+		assertRecordOK(is, tableName, rec, 1, tc.notNullOnly)
+	}
 
-	// Write the CDC data
-	insertRow(ctx, is, conn, tableName, 2, notNullOnly)
+	if tc.cdc {
+		insertRow(ctx, is, conn, tableName, 1, tc.notNullOnly)
 
-	// Read, ack, and verify the CDC record
-	rec = readAndAck(ctx, is, s)
-	assertRecordOK(is, tableName, rec, 2, notNullOnly)
+		// Read, ack, and verify the CDC record
+		rec := readAndAck(ctx, is, s)
+		assertRecordOK(is, tableName, rec, 1, tc.notNullOnly)
+	}
 
-	deleteRow(ctx, is, conn, tableName, 2)
-	rec = readAndAck(ctx, is, s)
-	is.Equal(opencdc.OperationDelete, rec.Operation)
+	if tc.opDelete {
+		// https://github.com/ConduitIO/conduit-connector-postgres/issues/301
+		t.Skip("Skipping delete test, see GitHub issue ")
+		deleteRow(ctx, is, conn, tableName, 1)
+		rec := readAndAck(ctx, is, s)
+		is.Equal(opencdc.OperationDelete, rec.Operation)
+	}
 }
 
 func readAndAck(ctx context.Context, is *is.I, s sdk.Source) opencdc.Record {