Function TestQueueCleaner

internal/maintenance/queue_cleaner_test.go:22–305

(t *testing.T)

)

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

	ctx := context.Background()

	type testBundle struct {
		deleteHorizon time.Time
		exec          riverdriver.Executor
	}

	setup := func(t *testing.T) (*QueueCleaner, *testBundle) {
		t.Helper()

		tx := riverdbtest.TestTxPgx(ctx, t)
		bundle := &testBundle{
			deleteHorizon: time.Now().Add(-QueueRetentionPeriodDefault),
			exec:          riverpgxv5.New(nil).UnwrapExecutor(tx),
		}

		cleaner := NewQueueCleaner(
			riversharedtest.BaseServiceArchetype(t),
			&QueueCleanerConfig{
				Interval:        queueCleanerIntervalDefault,
				RetentionPeriod: QueueRetentionPeriodDefault,
			},
			bundle.exec)
		cleaner.StaggerStartupDisable(true)
		cleaner.TestSignals.Init(t)
		t.Cleanup(cleaner.Stop)

		return cleaner, bundle
	}

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

		cleaner := NewQueueCleaner(riversharedtest.BaseServiceArchetype(t), &QueueCleanerConfig{}, nil)

		require.Equal(t, QueueRetentionPeriodDefault, cleaner.Config.RetentionPeriod)
		require.Equal(t, queueCleanerIntervalDefault, cleaner.Config.Interval)
	})

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

		cleaner, _ := setup(t)
		cleaner.Logger = riversharedtest.LoggerWarn(t)  // loop started/stop log is very noisy; suppress
		cleaner.TestSignals = QueueCleanerTestSignals{} // deinit so channels don't fill

		startstoptest.Stress(ctx, t, cleaner)
	})

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

		cleaner, bundle := setup(t)

		now := time.Now()