Function Test_Client_QueueList

client_test.go:6425–6496

(t *testing.T)

}

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

	ctx := context.Background()

	type testBundle struct {
		schema string
	}

	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)
		)

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

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

		client, bundle := setup(t)

		requireQueuesEqual := func(t *testing.T, target, actual *rivertype.Queue) {
			t.Helper()
			require.WithinDuration(t, target.CreatedAt, actual.CreatedAt, time.Millisecond)
			require.Equal(t, target.Metadata, actual.Metadata)
			require.Equal(t, target.Name, actual.Name)
			if target.PausedAt == nil {
				require.Nil(t, actual.PausedAt)
			} else {
				require.NotNil(t, actual.PausedAt)
				require.WithinDuration(t, *target.PausedAt, *actual.PausedAt, time.Millisecond)
			}
		}

		listRes, err := client.QueueList(ctx, NewQueueListParams().First(2))
		require.NoError(t, err)
		require.Empty(t, listRes.Queues)

		// Make queue1, pause it, refetch:
		queue1 := testfactory.Queue(ctx, t, client.driver.GetExecutor(), &testfactory.QueueOpts{Metadata: []byte(`{"foo": "bar"}`), Schema: bundle.schema})
		require.NoError(t, client.QueuePause(ctx, queue1.Name, nil))
		queue1, err = client.QueueGet(ctx, queue1.Name)
		require.NoError(t, err)

		queue2 := testfactory.Queue(ctx, t, client.driver.GetExecutor(), &testfactory.QueueOpts{Schema: bundle.schema})
		queue3 := testfactory.Queue(ctx, t, client.driver.GetExecutor(), &testfactory.QueueOpts{Schema: bundle.schema})

		listRes, err = client.QueueList(ctx, NewQueueListParams().First(2))
		require.NoError(t, err)