Function Test_Client_QueueUpdateTx

client_test.go:6695–6754

(t *testing.T)

}

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

	ctx := context.Background()

	type testBundle struct {
		executorTx riverdriver.ExecutorTx
		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{
			executorTx: client.driver.UnwrapExecutor(tx),
			schema:     schema,
			tx:         tx,
		}
	}

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

		client, bundle := setup(t)

		queue := testfactory.Queue(ctx, t, bundle.executorTx, &testfactory.QueueOpts{Schema: bundle.schema})
		require.Equal(t, []byte(`{}`), queue.Metadata)

		queue, err := client.QueueUpdateTx(ctx, bundle.tx, queue.Name, &QueueUpdateParams{
			Metadata: []byte(`{"foo":"bar"}`),
		})
		require.NoError(t, err)
		require.JSONEq(t, `{"foo":"bar"}`, string(queue.Metadata))

		queue, err = client.QueueUpdateTx(ctx, bundle.tx, queue.Name, &QueueUpdateParams{
			Metadata: nil,
		})
		require.NoError(t, err)
		require.JSONEq(t, `{"foo":"bar"}`, string(queue.Metadata))

		queue, err = client.QueueUpdateTx(ctx, bundle.tx, queue.Name, &QueueUpdateParams{
			Metadata: []byte(`{}`),
		})
		require.NoError(t, err)
		require.JSONEq(t, `{}`, string(queue.Metadata))