[-] fix integration tests

Author: pashagolub

.github/workflows/ci.yml renamed to .github/workflows/build.yml

@@ -1,10 +1,21 @@
-name: CI
+name: Build and Test
+
+permissions:
+ contents: write
+ packages: write
+ pages: write
 
 on:
  push:
- branches: [ main, 'feature/*' ]
+ branches:
+ - master
  pull_request:
- branches: [ main ]
+ paths-ignore: ['docs/**', '*.md']
+ workflow_dispatch:
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.ref }}
+ cancel-in-progress: true
 
 jobs:
  test:

cmd/etcd_fdw/integration_test.go renamed to internal/sync/integration_test.go

@@ -1,8 +1,7 @@
-package main
+package sync
 
 import (
 "context"
-"database/sql"
 "fmt"
 "testing"
 "time"
@@ -13,8 +12,6 @@ import (
 "github.com/testcontainers/testcontainers-go"
 "github.com/testcontainers/testcontainers-go/modules/postgres"
 "github.com/testcontainers/testcontainers-go/wait"
-
-"github.com/cybertec-postgresql/etcd_fdw/internal/sync"
 )
 
 func setupPostgreSQLContainer(ctx context.Context, t *testing.T) (*pgxpool.Pool, testcontainers.Container) {
@@ -40,7 +37,7 @@ func setupPostgreSQLContainer(ctx context.Context, t *testing.T) (*pgxpool.Pool,
 CREATE TABLE etcd (
 ts timestamp with time zone NOT NULL DEFAULT now(),
 key text NOT NULL,
-value text,
+value text NOT NULL,
 revision bigint NOT NULL,
 tombstone boolean NOT NULL DEFAULT false,
 PRIMARY KEY(key, revision)
@@ -53,7 +50,7 @@ func setupPostgreSQLContainer(ctx context.Context, t *testing.T) (*pgxpool.Pool,
 return pool, pgContainer
 }
 
-func setupEtcdContainer(ctx context.Context, t *testing.T) (*sync.EtcdClient, testcontainers.Container) {
+func setupEtcdContainer(ctx context.Context, t *testing.T) (*EtcdClient, testcontainers.Container) {
 etcdContainer, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
 ContainerRequest: testcontainers.ContainerRequest{
 Image: "quay.io/coreos/etcd:v3.5.9",
@@ -76,13 +73,13 @@ func setupEtcdContainer(ctx context.Context, t *testing.T) (*sync.EtcdClient, te
 require.NoError(t, err)
 
 dsn := "etcd://" + endpoint + "/test"
-etcdClient, err := sync.NewEtcdClient(dsn)
+etcdClient, err := NewEtcdClient(dsn)
 require.NoError(t, err)
 
 return etcdClient, etcdContainer
 }
 
-func setupTestContainers(t *testing.T) (*pgxpool.Pool, *sync.EtcdClient, func()) {
+func setupTestContainers(t *testing.T) (*pgxpool.Pool, *EtcdClient, func()) {
 ctx := context.Background()
 
 pool, pgContainer := setupPostgreSQLContainer(ctx, t)
@@ -117,18 +114,18 @@ func TestPollingMechanism(t *testing.T) {
 require.NoError(t, err)
 
 // Test GetPendingRecords function
-pendingRecords, err := sync.GetPendingRecords(ctx, pool)
+pendingRecords, err := GetPendingRecords(ctx, pool)
 require.NoError(t, err)
 assert.Len(t, pendingRecords, 1)
 assert.Equal(t, "test/polling/key1", pendingRecords[0].Key)
 assert.Equal(t, "value1", pendingRecords[0].Value)
 
 // Test UpdateRevision function
-err = sync.UpdateRevision(ctx, pool, "test/polling/key1", 123)
+err = UpdateRevision(ctx, pool, "test/polling/key1", 123)
 require.NoError(t, err)
 
 // Verify record was updated
-pendingAfterUpdate, err := sync.GetPendingRecords(ctx, pool)
+pendingAfterUpdate, err := GetPendingRecords(ctx, pool)
 require.NoError(t, err)
 assert.Len(t, pendingAfterUpdate, 0, "No pending records should remain after update")
 
@@ -154,7 +151,7 @@ func TestBulkInsert(t *testing.T) {
 defer cancel()
 
 // Prepare test records
-records := []sync.KeyValueRecord{
+records := []KeyValueRecord{
 {
 Key: "test/bulk/key1",
 Value: ("value1"),
@@ -179,7 +176,7 @@ func TestBulkInsert(t *testing.T) {
 }
 
 // Test BulkInsert function
-err := sync.BulkInsert(ctx, pool, records)
+err := BulkInsert(ctx, pool, records)
 require.NoError(t, err)
 
 // Verify records were inserted correctly
@@ -189,17 +186,17 @@ func TestBulkInsert(t *testing.T) {
 assert.Equal(t, 3, count)
 
 // Verify specific record details
-var key, value sql.NullString
+var key, value string
 var revision int64
 var tombstone bool
 err = pool.QueryRow(ctx, `
 SELECT key, value, revision, tombstone 
 FROM etcd WHERE key = 'test/bulk/key1'
 `).Scan(&key, &value, &revision, &tombstone)
 require.NoError(t, err)
-assert.Equal(t, "test/bulk/key1", key.String)
-assert.True(t, value.Valid)
-assert.Equal(t, "value1", value.String)
+assert.Equal(t, "test/bulk/key1", key)
+assert.NotEmpty(t, value)
+assert.Equal(t, "value1", value)
 assert.Equal(t, int64(100), revision)
 assert.False(t, tombstone)
 
@@ -209,8 +206,8 @@ func TestBulkInsert(t *testing.T) {
 FROM etcd WHERE key = 'test/bulk/key3'
 `).Scan(&key, &value, &revision, &tombstone)
 require.NoError(t, err)
-assert.Equal(t, "test/bulk/key3", key.String)
-assert.False(t, value.Valid) // NULL value
+assert.Equal(t, "test/bulk/key3", key)
+assert.Empty(t, value) // NULL value
 assert.Equal(t, int64(102), revision)
 assert.True(t, tombstone)
 }
@@ -227,23 +224,23 @@ func TestInsertPendingRecord(t *testing.T) {
 defer cancel()
 
 // Test inserting a new pending record
-err := sync.InsertPendingRecord(ctx, pool, "test/pending/key1", ("value1"), false)
+err := InsertPendingRecord(ctx, pool, "test/pending/key1", ("value1"), false)
 require.NoError(t, err)
 
 // Verify record was inserted with revision = -1
 var revision int64
-var value sql.NullString
+var value string
 err = pool.QueryRow(ctx, `
 SELECT revision, value FROM etcd 
 WHERE key = 'test/pending/key1'
 `).Scan(&revision, &value)
 require.NoError(t, err)
 assert.Equal(t, int64(-1), revision)
-assert.True(t, value.Valid)
-assert.Equal(t, "value1", value.String)
+assert.NotEmpty(t, value)
+assert.Equal(t, "value1", value)
 
 // Test inserting second record with same key (should create new record with different timestamp)
-err = sync.InsertPendingRecord(ctx, pool, "test/pending/key1", ("updated_value"), false)
+err = InsertPendingRecord(ctx, pool, "test/pending/key1", ("updated_value"), false)
 require.NoError(t, err)
 
 // Verify both records exist (different timestamps, both with revision = -1)
@@ -256,7 +253,7 @@ func TestInsertPendingRecord(t *testing.T) {
 assert.Equal(t, 1, count, "Should have 1 pending records for the same key with latest value")
 
 // Test inserting tombstone record
-err = sync.InsertPendingRecord(ctx, pool, "test/pending/key2", "", true)
+err = InsertPendingRecord(ctx, pool, "test/pending/key2", "", true)
 require.NoError(t, err)
 
 // Verify tombstone record
@@ -267,7 +264,7 @@ func TestInsertPendingRecord(t *testing.T) {
 `).Scan(&revision, &value, &tombstone)
 require.NoError(t, err)
 assert.Equal(t, int64(-1), revision)
-assert.False(t, value.Valid) // NULL value
+assert.Empty(t, value)
 assert.True(t, tombstone)
 }
 
@@ -283,7 +280,7 @@ func TestGetLatestRevision(t *testing.T) {
 defer cancel()
 
 // Test with empty table
-latestRevision, err := sync.GetLatestRevision(ctx, pool)
+latestRevision, err := GetLatestRevision(ctx, pool)
 require.NoError(t, err)
 assert.Equal(t, int64(0), latestRevision)
 
@@ -298,7 +295,7 @@ func TestGetLatestRevision(t *testing.T) {
 require.NoError(t, err)
 
 // Test latest revision (should ignore -1 pending records)
-latestRevision, err = sync.GetLatestRevision(ctx, pool)
+latestRevision, err = GetLatestRevision(ctx, pool)
 require.NoError(t, err)
 assert.Equal(t, int64(150), latestRevision)
 }
@@ -320,13 +317,13 @@ func TestPendingRecordFiltering(t *testing.T) {
 ('test/filter/synced1', 'value1', 100, false),
 ('test/filter/pending1', 'value2', -1, false),
 ('test/filter/synced2', 'value3', 200, false),
-('test/filter/pending2', NULL, -1, true),
+('test/filter/pending2', '', -1, true),
 ('test/filter/pending3', 'value4', -1, false)
 `)
 require.NoError(t, err)
 
 // Test GetPendingRecords only returns revision = -1
-pendingRecords, err := sync.GetPendingRecords(ctx, pool)
+pendingRecords, err := GetPendingRecords(ctx, pool)
 require.NoError(t, err)
 assert.Len(t, pendingRecords, 3)
 
@@ -361,22 +358,22 @@ func TestConflictResolution(t *testing.T) {
 defer cancel()
 
 // Insert a pending record
-err := sync.InsertPendingRecord(ctx, pool, "test/conflict/key1", "pending_value", false)
+err := InsertPendingRecord(ctx, pool, "test/conflict/key1", "pending_value", false)
 require.NoError(t, err)
 
 // Verify it's pending
-pendingRecords, err := sync.GetPendingRecords(ctx, pool)
+pendingRecords, err := GetPendingRecords(ctx, pool)
 require.NoError(t, err)
 assert.Len(t, pendingRecords, 1)
 assert.Equal(t, "test/conflict/key1", pendingRecords[0].Key)
 assert.Equal(t, int64(-1), pendingRecords[0].Revision)
 
 // Simulate etcd sync by updating revision
-err = sync.UpdateRevision(ctx, pool, "test/conflict/key1", 300)
+err = UpdateRevision(ctx, pool, "test/conflict/key1", 300)
 require.NoError(t, err)
 
 // Verify record is no longer pending
-pendingAfterUpdate, err := sync.GetPendingRecords(ctx, pool)
+pendingAfterUpdate, err := GetPendingRecords(ctx, pool)
 require.NoError(t, err)
 assert.Len(t, pendingAfterUpdate, 0)
 
@@ -390,7 +387,7 @@ func TestConflictResolution(t *testing.T) {
 assert.Equal(t, int64(300), revision)
 
 // Test updating non-existent pending record (should fail gracefully)
-err = sync.UpdateRevision(ctx, pool, "test/conflict/nonexistent", 400)
+err = UpdateRevision(ctx, pool, "test/conflict/nonexistent", 400)
 assert.Error(t, err)
 assert.Contains(t, err.Error(), "no pending record found")
 }
@@ -410,10 +407,10 @@ func TestPerformanceOpsPerSecond(t *testing.T) {
 recordCount := 1000
 start := time.Now()
 
-records := make([]sync.KeyValueRecord, recordCount)
+records := make([]KeyValueRecord, recordCount)
 for i := 0; i < recordCount; i++ {
 value := fmt.Sprintf("test_value_%d", i)
-records[i] = sync.KeyValueRecord{
+records[i] = KeyValueRecord{
 Key: fmt.Sprintf("test/perf/key%d", i),
 Value: value,
 Revision: int64(i + 1),
@@ -422,7 +419,7 @@ func TestPerformanceOpsPerSecond(t *testing.T) {
 }
 }
 
-err := sync.BulkInsert(ctx, pool, records)
+err := BulkInsert(ctx, pool, records)
 require.NoError(t, err)
 
 elapsed := time.Since(start)
@@ -453,11 +450,11 @@ func TestPerformanceSyncLatency(t *testing.T) {
 // Insert pending record
 key := fmt.Sprintf("test/latency/key%d", i)
 value := fmt.Sprintf("test_value_%d", i)
-err := sync.InsertPendingRecord(ctx, pool, key, value, false)
+err := InsertPendingRecord(ctx, pool, key, value, false)
 require.NoError(t, err)
 
 // Update revision (simulating sync completion)
-err = sync.UpdateRevision(ctx, pool, key, int64(i+1))
+err = UpdateRevision(ctx, pool, key, int64(i+1))
 require.NoError(t, err)
 
 latency := time.Since(start)

internal/sync/postgresql.go

@@ -210,15 +210,10 @@ func InsertPendingRecord(ctx context.Context, pool PgxIface, key string, value s
 ON CONFLICT (key, revision) DO UPDATE 
 SET value = EXCLUDED.value, ts = CURRENT_TIMESTAMP, tombstone = EXCLUDED.tombstone;
 `
-
-var valueParam any
 if tombstone {
-valueParam = nil // Insert NULL for tombstones
-} else {
-valueParam = value
+value = "" // Use empty string for tombstone records
 }
-
-_, err := pool.Exec(ctx, query, key, valueParam, tombstone)
+_, err := pool.Exec(ctx, query, key, value, tombstone)
 if err != nil {
 return fmt.Errorf("failed to insert pending record: %w", err)
 }

internal/sync/postgresql_test.go

@@ -11,7 +11,7 @@ import (
 )
 
 // TestBulkInsert tests bulk insert operation with pgxmock (simplified)
-func TestBulkInsert(t *testing.T) {
+func TestBulkInsertMock(t *testing.T) {
 mock, err := pgxmock.NewPool()
 require.NoError(t, err)
 defer mock.Close()
@@ -106,7 +106,7 @@ func TestUpdateRevisionNotFound(t *testing.T) {
 }
 
 // TestGetLatestRevision tests getting latest revision with pgxmock
-func TestGetLatestRevision(t *testing.T) {
+func TestGetLatestRevisionMock(t *testing.T) {
 mock, err := pgxmock.NewPool()
 require.NoError(t, err)
 defer mock.Close()
@@ -149,7 +149,7 @@ func TestGetLatestRevisionEmpty(t *testing.T) {
 }
 
 // TestInsertPendingRecord tests inserting pending record with pgxmock
-func TestInsertPendingRecord(t *testing.T) {
+func TestInsertPendingRecordMock(t *testing.T) {
 mock, err := pgxmock.NewPool()
 require.NoError(t, err)
 defer mock.Close()
@@ -176,9 +176,9 @@ func TestInsertPendingRecordTombstone(t *testing.T) {
 
 ctx := context.Background()
 
-// Test tombstone record insert (value should be nil)
+// Test tombstone record insert (value should be empty string)
 mock.ExpectExec(`INSERT INTO etcd \(key, value, revision, tombstone\)`).
-WithArgs("test-key", nil, true).
+WithArgs("test-key", "", true).
 WillReturnResult(pgxmock.NewResult("INSERT", 1))
 
 err = InsertPendingRecord(ctx, mock, "test-key", "test-value", true)