Function TestRedisQueue

preconf-share/simqueue/queue_test.go:89–292

TestRedisQueue tests that the redis queue works as expected it's implemented in one method to test possible interactions between different tests

(t *testing.T)

// TestRedisQueue tests that the redis queue works as expected
// it's implemented in one method to test possible interactions between different tests
func TestRedisQueue(t *testing.T) {
	ctx := context.Background()
	r := newQueueRunner("test_queue")
	worker := newTestWorker()

	//

	// test that queue can be cancelled
	t.Run("empty queue cancel", func(t *testing.T) {
		r.startProcessLoop(ctx, []ProcessFunc{worker.processOk})

		// wait so code gets to the blocking pop opearation
		time.Sleep(10 * time.Millisecond)

		r.stopProcessLoop()
	})

	// test that normal processing works
	t.Run("normal processing", func(t *testing.T) {
		r.startProcessLoop(ctx, []ProcessFunc{worker.processOk})

		err := r.queue.UpdateBlock(1)
		require.NoError(t, err)

		err = r.queue.Push(context.Background(), []byte("test"), false, 2, 2)
		require.NoError(t, err)

		require.Equal(t, "test", string(worker.nextProcessed(processTimeout)))

		require.Nil(t, worker.nextProcessed(processTimeout))

		r.stopProcessLoop()
	})

	// test multiple workers
	t.Run("multiple workers", func(t *testing.T) {
		workers := MultipleWorkers(worker.processOk, 10, rate.Inf, 1)
		r.startProcessLoop(ctx, workers)

		err := r.queue.UpdateBlock(1)
		require.NoError(t, err)

		for i := 0; i < 10; i++ {
			err = r.queue.Push(ctx, []byte("test-multiple"), false, 2, 2)
			require.NoError(t, err)
		}

		for i := 0; i < 10; i++ {
			require.Equal(t, "test-multiple", string(worker.nextProcessed(processTimeout)))
		}

		require.Nil(t, worker.nextProcessed(processTimeout))
		r.stopProcessLoop()
	})

	// test that stale items are not processed
	t.Run("test queue cleanup", func(t *testing.T) {
		err := r.queue.Push(context.Background(), []byte("test-stale"), false, 2, 2)