Function Test_Client_InsertManyTx

client_test.go:4375–4693

(t *testing.T)

}

func Test_Client_InsertManyTx(t *testing.T) {
	t.Parallel()

	ctx := context.Background()

	type testBundle struct {
		schema string
		tx     pgx.Tx
	}

	setup := func(t *testing.T) (*Client[pgx.Tx], *testBundle) {
		t.Helper()

		var (
			dbPool = riversharedtest.DBPool(ctx, t)
			driver = riverpgxv5.New(dbPool)
			schema = riverdbtest.TestSchema(ctx, t, driver, nil)
			config = newTestConfig(t, schema)
			client = newTestClient(t, dbPool, config)
		)

		tx, err := dbPool.Begin(ctx)
		require.NoError(t, err)
		t.Cleanup(func() { tx.Rollback(ctx) })

		return client, &testBundle{
			schema: schema,
			tx:     tx,
		}
	}

	t.Run("SucceedsWithMultipleJobs", func(t *testing.T) {
		t.Parallel()

		client, bundle := setup(t)

		now := time.Now().UTC()

		results, err := client.InsertManyTx(ctx, bundle.tx, []InsertManyParams{
			{Args: noOpArgs{Name: "Foo"}, InsertOpts: &InsertOpts{Metadata: []byte(`{"a": "b"}`), Queue: "foo", Priority: 2}},
			{Args: noOpArgs{}, InsertOpts: &InsertOpts{ScheduledAt: now.Add(time.Minute)}},
		})
		require.NoError(t, err)
		require.Len(t, results, 2)

		require.False(t, results[0].UniqueSkippedAsDuplicate)
		require.Equal(t, 0, results[0].Job.Attempt)
		require.Nil(t, results[0].Job.AttemptedAt)
		require.WithinDuration(t, now, results[0].Job.CreatedAt, 2*time.Second)
		require.Empty(t, results[0].Job.AttemptedBy)
		require.Positive(t, results[0].Job.ID)
		require.JSONEq(t, `{"name": "Foo"}`, string(results[0].Job.EncodedArgs))
		require.Empty(t, results[0].Job.Errors)
		require.Nil(t, results[0].Job.FinalizedAt)
		require.Equal(t, "noOp", results[0].Job.Kind)
		require.Equal(t, 25, results[0].Job.MaxAttempts)
		require.JSONEq(t, `{"a": "b"}`, string(results[0].Job.Metadata))
		require.Equal(t, 2, results[0].Job.Priority)