fix: Add back testing for all types (#1015)

There is still more work to do to make the testing function "production" ready because they become quite vital and some of them are public APIs so we need to think how to generalize them correctly so they are testable, expand etc.. but this should be a good start just so we can start releasing some destinations

Author: yevgenypats

plugin/testing_upsert.go

@@ -12,8 +12,8 @@ import (
 	"github.com/cloudquery/plugin-sdk/v4/schema"
 )
 
-func (s *WriterTestSuite) testUpsert(ctx context.Context) error {
-	tableName := fmt.Sprintf("cq_test_upsert_%d", time.Now().Unix())
+func (s *WriterTestSuite) testUpsertBasic(ctx context.Context) error {
+	tableName := fmt.Sprintf("cq_upsert_basic_%d", time.Now().Unix())
 	table := &schema.Table{
 		Name: tableName,
 		Columns: []schema.Column{
@@ -63,3 +63,50 @@ func (s *WriterTestSuite) testUpsert(ctx context.Context) error {
 
 	return nil
 }
+
+func (s *WriterTestSuite) testUpsertAll(ctx context.Context) error {
+	tableName := fmt.Sprintf("cq_upsert_all_%d", time.Now().Unix())
+	table := schema.TestTable(tableName, s.genDatOptions)
+	table.Columns = append(table.Columns, schema.Column{Name: "name", Type: arrow.BinaryTypes.String, PrimaryKey: true})
+	if err := s.plugin.writeOne(ctx, &message.WriteMigrateTable{
+		Table: table,
+	}); err != nil {
+		return fmt.Errorf("failed to create table: %w", err)
+	}
+
+	records := schema.GenTestData(table, schema.GenTestDataOptions{MaxRows: 2})
+	record := records[0]
+
+	if err := s.plugin.writeOne(ctx, &message.WriteInsert{
+		Record: record,
+	}); err != nil {
+		return fmt.Errorf("failed to insert record: %w", err)
+	}
+
+	records, err := s.plugin.readAll(ctx, table)
+	if err != nil {
+		return fmt.Errorf("failed to readAll: %w", err)
+	}
+	totalItems := TotalRows(records)
+	if totalItems != 1 {
+		return fmt.Errorf("expected 1 item, got %d", totalItems)
+	}
+
+	if err := s.plugin.writeOne(ctx, &message.WriteInsert{
+		Record: record,
+	}); err != nil {
+		return fmt.Errorf("failed to insert record: %w", err)
+	}
+
+	records, err = s.plugin.readAll(ctx, table)
+	if err != nil {
+		return fmt.Errorf("failed to sync: %w", err)
+	}
+
+	totalItems = TotalRows(records)
+	if totalItems != 1 {
+		return fmt.Errorf("expected 1 item, got %d", totalItems)
+	}
+
+	return nil
+}

plugin/testing_write.go

@@ -90,18 +90,32 @@ func TestWriterSuiteRunner(t *testing.T, p *Plugin, tests WriterTestSuiteTests,
 		if suite.tests.SkipUpsert {
 			t.Skip("skipping " + t.Name())
 		}
-		if err := suite.testUpsert(ctx); err != nil {
-			t.Fatal(err)
-		}
+		t.Run("Basic", func(t *testing.T) {
+			if err := suite.testUpsertBasic(ctx); err != nil {
+				t.Fatal(err)
+			}
+		})
+		t.Run("All", func(t *testing.T) {
+			if err := suite.testUpsertAll(ctx); err != nil {
+				t.Fatal(err)
+			}
+		})
 	})
 
 	t.Run("TestInsert", func(t *testing.T) {
 		if suite.tests.SkipInsert {
 			t.Skip("skipping " + t.Name())
 		}
-		if err := suite.testInsert(ctx); err != nil {
-			t.Fatal(err)
-		}
+		t.Run("Basic", func(t *testing.T) {
+			if err := suite.testInsertBasic(ctx); err != nil {
+				t.Fatal(err)
+			}
+		})
+		t.Run("All", func(t *testing.T) {
+			if err := suite.testInsertAll(ctx); err != nil {
+				t.Fatal(err)
+			}
+		})
 	})
 
 	t.Run("TestDeleteStale", func(t *testing.T) {

plugin/testing_write_insert.go

@@ -20,8 +20,8 @@ func TotalRows(records []arrow.Record) int64 {
 	return totalRows
 }
 
-func (s *WriterTestSuite) testInsert(ctx context.Context) error {
-	tableName := fmt.Sprintf("cq_test_insert_%d", time.Now().Unix())
+func (s *WriterTestSuite) testInsertBasic(ctx context.Context) error {
+	tableName := fmt.Sprintf("cq_insert_basic_%d", time.Now().Unix())
 	table := &schema.Table{
 		Name: tableName,
 		Columns: []schema.Column{
@@ -71,3 +71,50 @@ func (s *WriterTestSuite) testInsert(ctx context.Context) error {
 
 	return nil
 }
+
+func (s *WriterTestSuite) testInsertAll(ctx context.Context) error {
+	tableName := fmt.Sprintf("cq_insert_all_%d", time.Now().Unix())
+	table := schema.TestTable(tableName, s.genDatOptions)
+	if err := s.plugin.writeOne(ctx, &message.WriteMigrateTable{
+		Table: table,
+	}); err != nil {
+		return fmt.Errorf("failed to create table: %w", err)
+	}
+	records := schema.GenTestData(table, schema.GenTestDataOptions{
+		MaxRows: 2,
+	})
+	record := records[0]
+
+	if err := s.plugin.writeOne(ctx, &message.WriteInsert{
+		Record: record,
+	}); err != nil {
+		return fmt.Errorf("failed to insert record: %w", err)
+	}
+	readRecords, err := s.plugin.readAll(ctx, table)
+	if err != nil {
+		return fmt.Errorf("failed to sync: %w", err)
+	}
+
+	totalItems := TotalRows(readRecords)
+	if totalItems != 1 {
+		return fmt.Errorf("expected 1 item, got %d", totalItems)
+	}
+
+	if err := s.plugin.writeOne(ctx, &message.WriteInsert{
+		Record: record,
+	}); err != nil {
+		return fmt.Errorf("failed to insert record: %w", err)
+	}
+
+	readRecords, err = s.plugin.readAll(ctx, table)
+	if err != nil {
+		return fmt.Errorf("failed to sync: %w", err)
+	}
+
+	totalItems = TotalRows(readRecords)
+	if totalItems != 2 {
+		return fmt.Errorf("expected 2 items, got %d", totalItems)
+	}
+
+	return nil
+}