test writer with stdlib tx

Author: nikolayk812

errors.go

@@ -7,4 +7,6 @@ var (
 	ErrTxUnsupportedType = errors.New("tx has unsupported type")
 
 	ErrTableEmpty = errors.New("table is empty")
+
+	ErrPoolNil = errors.New("pool is nil")
 )

reader.go

@@ -2,7 +2,6 @@ package outbox
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"time"
 
@@ -37,7 +36,7 @@ type reader struct {
 
 func NewReader(table string, pool *pgxpool.Pool, opts ...ReadOption) (Reader, error) {
 	if pool == nil {
-		return nil, errors.New("pool is nil")
+		return nil, ErrPoolNil
 	}
 	if table == "" {
 		return nil, ErrTableEmpty

writer_reader_test.go

@@ -2,15 +2,16 @@ package outbox_test
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"fmt"
-	"log/slog"
 	"os"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
+	_ "github.com/jackc/pgx/v5/stdlib"
 	outbox "github.com/nikolayk812/pgx-outbox"
 	"github.com/nikolayk812/pgx-outbox/internal/containers"
 	"github.com/nikolayk812/pgx-outbox/internal/fakes"
@@ -29,6 +30,7 @@ var ctx = context.Background()
 type WriterReaderTestSuite struct {
 	suite.Suite
 	pool      *pgxpool.Pool
+	db        *sql.DB
 	container testcontainers.Container
 
 	writer outbox.Writer
@@ -50,6 +52,9 @@ func (suite *WriterReaderTestSuite) SetupSuite() {
 	suite.pool, err = pgxpool.New(ctx, connStr)
 	suite.noError(err)
 
+	suite.db, err = sql.Open("pgx", connStr)
+	suite.noError(err)
+
 	suite.writer, err = outbox.NewWriter(outboxTable)
 	suite.noError(err)
 
@@ -61,55 +66,82 @@ func (suite *WriterReaderTestSuite) TearDownSuite() {
 	if suite.pool != nil {
 		suite.pool.Close()
 	}
+	if suite.db != nil {
+		suite.NoError(suite.db.Close())
+	}
 	if suite.container != nil {
-		if err := suite.container.Terminate(ctx); err != nil {
-			slog.Error("suite.container.Terminate", slog.Any("error", err))
-		}
+		suite.NoError(suite.container.Terminate(ctx))
 	}
 
 	goleak.VerifyNone(suite.T())
 }
 
-func (suite *WriterReaderTestSuite) TestWriter_New() {
+//nolint:dupl
+func (suite *WriterReaderTestSuite) TestWriter_WriteMessage() {
+	invalidMessage := fakes.FakeMessage()
+	invalidMessage.Broker = ""
+
 	tests := []struct {
 		name    string
-		table   string
-		options []outbox.WriteOption
-		wantErr error
+		in      []types.Message
+		wantErr bool
 	}{
 		{
-			name:    "empty table",
-			table:   "",
-			wantErr: outbox.ErrTableEmpty,
+			name: "no messages",
+			in:   []types.Message{},
 		},
 		{
-			name:  "non-empty table",
-			table: "outbox_messages",
+			name: "single message",
+			in: []types.Message{
+				fakes.FakeMessage(),
+			},
 		},
 		{
-			name:    "with options",
-			table:   "outbox_messages",
-			options: []outbox.WriteOption{outbox.WithDisablePreparedBatch()},
+			name: "multiple in",
+			in: []types.Message{
+				fakes.FakeMessage(),
+				fakes.FakeMessage(),
+			},
 		},
+		{
+			name: "invalid message",
+			in: []types.Message{
+				invalidMessage,
+			},
+			wantErr: true,
+		},
+		// Add more test cases as needed
 	}
 
 	for _, tt := range tests {
 		suite.Run(tt.name, func() {
 			t := suite.T()
 
-			writer, err := outbox.NewWriter(tt.table, tt.options...)
-			if tt.wantErr != nil {
-				require.ErrorIs(t, err, tt.wantErr)
-				return
+			// GIVEN
+			for _, message := range tt.in {
+				id, err := suite.write(message)
+				if tt.wantErr {
+					require.Error(t, err)
+					return
+				}
+				require.NoError(t, err)
+				assert.Positive(t, id)
 			}
 
+			limit := maxInt(1, len(tt.in))
+
+			// THEN
+			actual, err := suite.reader.Read(ctx, limit)
 			require.NoError(t, err)
-			assert.NotNil(t, writer)
+			assertEqualMessages(t, tt.in, actual)
+
+			suite.markAll()
 		})
 	}
 }
 
-func (suite *WriterReaderTestSuite) TestWriter_WriteMessage() {
+//nolint:dupl
+func (suite *WriterReaderTestSuite) TestWriter_WriteMessageStdLib() {
 	invalidMessage := fakes.FakeMessage()
 	invalidMessage.Broker = ""
 
@@ -151,7 +183,7 @@ func (suite *WriterReaderTestSuite) TestWriter_WriteMessage() {
 
 			// GIVEN
 			for _, message := range tt.in {
-				id, err := suite.write(message)
+				id, err := suite.writeStdLib(message)
 				if tt.wantErr {
 					require.Error(t, err)
 					return
@@ -487,6 +519,34 @@ func (suite *WriterReaderTestSuite) beginTx(ctx context.Context) (pgx.Tx, func(e
 	return tx, commitFunc, nil
 }
 
+func (suite *WriterReaderTestSuite) beginTxStdLib() (*sql.Tx, func(err error) error, error) {
+	emptyFunc := func(_ error) error { return nil }
+
+	tx, err := suite.db.Begin()
+	if err != nil {
+		return nil, emptyFunc, fmt.Errorf("db.Begin: %w", err)
+	}
+
+	commitFunc := func(execErr error) error {
+		if execErr != nil {
+			rbErr := tx.Rollback()
+			if rbErr != nil {
+				return fmt.Errorf("tx.Rollback %v: %w", execErr, rbErr) //nolint:errorlint
+			}
+			return execErr
+		}
+
+		txErr := tx.Commit()
+		if txErr != nil {
+			return fmt.Errorf("tx.Commit: %w", txErr)
+		}
+
+		return nil
+	}
+
+	return tx, commitFunc, nil
+}
+
 func (suite *WriterReaderTestSuite) write(message types.Message) (_ int64, txErr error) {
 	tx, commitFunc, err := suite.beginTx(ctx)
 	if err != nil {
@@ -506,6 +566,25 @@ func (suite *WriterReaderTestSuite) write(message types.Message) (_ int64, txErr
 	return id, nil
 }
 
+func (suite *WriterReaderTestSuite) writeStdLib(message types.Message) (_ int64, txErr error) {
+	tx, commitFunc, err := suite.beginTxStdLib()
+	if err != nil {
+		return 0, fmt.Errorf("beginTx: %w", err)
+	}
+	defer func() {
+		if err := commitFunc(txErr); err != nil {
+			txErr = fmt.Errorf("commitFunc: %w", err)
+		}
+	}()
+
+	id, err := suite.writer.Write(ctx, tx, message)
+	if err != nil {
+		return 0, fmt.Errorf("writer.Write: %w", err)
+	}
+
+	return id, nil
+}
+
 func (suite *WriterReaderTestSuite) writeBatch(messages []types.Message) (_ []int64, txErr error) {
 	tx, commitFunc, err := suite.beginTx(ctx)
 	if err != nil {
@@ -549,13 +628,54 @@ func (suite *WriterReaderTestSuite) noError(err error) {
 	suite.Require().NoError(err)
 }
 
+//nolint:unparam
 func maxInt(x, y int) int {
 	if x > y {
 		return x
 	}
 	return y
 }
 
+// TestWriter_New is just to increase coverage.
+func (suite *WriterReaderTestSuite) TestWriter_New() {
+	tests := []struct {
+		name    string
+		table   string
+		options []outbox.WriteOption
+		wantErr error
+	}{
+		{
+			name:    "empty table",
+			table:   "",
+			wantErr: outbox.ErrTableEmpty,
+		},
+		{
+			name:  "non-empty table",
+			table: "outbox_messages",
+		},
+		{
+			name:    "with options",
+			table:   "outbox_messages",
+			options: []outbox.WriteOption{outbox.WithDisablePreparedBatch()},
+		},
+	}
+
+	for _, tt := range tests {
+		suite.Run(tt.name, func() {
+			t := suite.T()
+
+			writer, err := outbox.NewWriter(tt.table, tt.options...)
+			if tt.wantErr != nil {
+				require.ErrorIs(t, err, tt.wantErr)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.NotNil(t, writer)
+		})
+	}
+}
+
 func (suite *WriterReaderTestSuite) TestWriter_WriteWithNilTx() {
 	message := fakes.FakeMessage()
 
@@ -599,3 +719,48 @@ func (suite *WriterReaderTestSuite) TestWriter_WriteWithNilTx() {
 		})
 	}
 }
+
+// TestReader_New is just to increase coverage.
+func (suite *WriterReaderTestSuite) TestReader_New() {
+	tests := []struct {
+		name    string
+		table   string
+		pool    *pgxpool.Pool
+		option  outbox.ReadOption
+		wantErr error
+	}{
+		{
+			name:    "empty table",
+			table:   "",
+			pool:    suite.pool,
+			wantErr: outbox.ErrTableEmpty,
+		},
+		{
+			name:    "nil pool",
+			table:   "outbox_messages",
+			pool:    nil,
+			wantErr: outbox.ErrPoolNil,
+		},
+		{
+			name:   "with option",
+			table:  "outbox_messages",
+			pool:   suite.pool,
+			option: outbox.WithReadFilter(types.MessageFilter{Brokers: []string{"broker1"}}),
+		},
+	}
+
+	for _, tt := range tests {
+		suite.Run(tt.name, func() {
+			t := suite.T()
+
+			reader, err := outbox.NewReader(tt.table, tt.pool, tt.option)
+			if tt.wantErr != nil {
+				require.ErrorIs(t, err, tt.wantErr)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.NotNil(t, reader)
+		})
+	}
+}