Function TestWorker_Work

rivertest/worker_test.go:73–481

(t *testing.T)

}

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

	ctx := context.Background()

	type testBundle struct {
		config *river.Config
		driver *riverpgxv5.Driver
		tx     pgx.Tx
	}

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

		var (
			config = &river.Config{ID: "rivertest-worker"}
			driver = riverpgxv5.New(nil)
			tx     = riverdbtest.TestTxPgx(ctx, t)
		)

		return &testBundle{
			config: config,
			driver: driver,
			tx:     tx,
		}
	}

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

		bundle := setup(t)

		worker := river.WorkFunc(func(ctx context.Context, job *river.Job[testArgs]) error {
			require.Equal(t, testArgs{Value: "test"}, job.Args)
			require.Equal(t, 1, job.Attempt)
			require.NotNil(t, job.AttemptedAt)
			require.WithinDuration(t, time.Now(), *job.AttemptedAt, 5*time.Second)
			require.Equal(t, []string{"rivertest-worker"}, job.AttemptedBy)
			require.WithinDuration(t, time.Now(), job.CreatedAt, 5*time.Second)
			require.JSONEq(t, `{"value": "test"}`, string(job.EncodedArgs))
			require.Empty(t, job.Errors)
			require.Nil(t, job.FinalizedAt)
			require.Positive(t, job.ID)
			require.Equal(t, "rivertest_work_test", job.Kind)
			require.Equal(t, river.MaxAttemptsDefault, job.MaxAttempts)
			require.Equal(t, []byte(`{}`), job.Metadata)
			require.Equal(t, river.PriorityDefault, job.Priority)
			require.Equal(t, river.QueueDefault, job.Queue)
			require.WithinDuration(t, time.Now(), job.ScheduledAt, 2*time.Second)
			require.Equal(t, rivertype.JobStateRunning, job.State)
			require.Equal(t, []string{}, job.Tags)
			require.Nil(t, job.UniqueKey)

			_, hasContextKeyInsideTestWorker := ctx.Value(execution.ContextKeyInsideTestWorker{}).(bool)
			require.True(t, hasContextKeyInsideTestWorker)

			return nil
		})