Function exerciseQueue

riverdriver/riverdrivertest/queue.go:18–528

(ctx context.Context, t *testing.T, executorWithTx func(ctx context.Context, t *testing.T) (riverdriver.Executor, riverdriver.Driver[TTx]))

)

func exerciseQueue[TTx any](ctx context.Context, t *testing.T, executorWithTx func(ctx context.Context, t *testing.T) (riverdriver.Executor, riverdriver.Driver[TTx])) {
	t.Helper()

	type testBundle struct {
		driver riverdriver.Driver[TTx]
	}

	setup := func(ctx context.Context, t *testing.T) (riverdriver.Executor, *testBundle) {
		t.Helper()

		exec, driver := executorWithTx(ctx, t)

		return exec, &testBundle{
			driver: driver,
		}
	}

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

		mustUnmarshalJSON := func(t *testing.T, data []byte) map[string]any {
			t.Helper()

			var dataMap map[string]any
			require.NoError(t, json.Unmarshal(data, &dataMap))
			return dataMap
		}

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

			exec, bundle := setup(ctx, t)

			metadata := []byte(`{"foo": "bar"}`)
			now := time.Now().UTC()
			queue, err := exec.QueueCreateOrSetUpdatedAt(ctx, &riverdriver.QueueCreateOrSetUpdatedAtParams{
				Metadata: metadata,
				Name:     "new-queue",
				Now:      &now,
			})
			require.NoError(t, err)
			require.WithinDuration(t, now, queue.CreatedAt, bundle.driver.TimePrecision())
			require.JSONEq(t, string(metadata), string(queue.Metadata))
			require.Equal(t, "new-queue", queue.Name)
			require.Nil(t, queue.PausedAt)
			require.WithinDuration(t, now, queue.UpdatedAt, bundle.driver.TimePrecision())
		})

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

			exec, bundle := setup(ctx, t)

			now := time.Now().UTC().Add(-5 * time.Minute)
			queue, err := exec.QueueCreateOrSetUpdatedAt(ctx, &riverdriver.QueueCreateOrSetUpdatedAtParams{
				Name:     "new-queue",
				PausedAt: ptrutil.Ptr(now),
			})