Function TestWorker_WorkJob

rivertest/worker_test.go:483–621

(t *testing.T)

}

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

	ctx := context.Background()

	type testBundle struct {
		client   *river.Client[pgx.Tx]
		config   *river.Config
		driver   *riverpgxv5.Driver
		tx       pgx.Tx
		workFunc func(ctx context.Context, job *river.Job[testArgs]) error
	}

	setup := func(t *testing.T) (*Worker[testArgs, pgx.Tx], *testBundle) {
		t.Helper()

		var (
			config = &river.Config{ID: "rivertest-workjob"}
			driver = riverpgxv5.New(nil)
		)

		client, err := river.NewClient(driver, config)
		require.NoError(t, err)

		bundle := &testBundle{
			client:   client,
			config:   config,
			driver:   driver,
			tx:       riverdbtest.TestTxPgx(ctx, t),
			workFunc: func(ctx context.Context, job *river.Job[testArgs]) error { return nil },
		}

		worker := river.WorkFunc(func(ctx context.Context, job *river.Job[testArgs]) error {
			return bundle.workFunc(ctx, job)
		})

		return NewWorker(t, driver, config, worker), bundle
	}

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

		testWorker, bundle := setup(t)

		bundle.workFunc = func(ctx context.Context, job *river.Job[testArgs]) error {
			require.WithinDuration(t, time.Now(), *job.AttemptedAt, 5*time.Second)
			require.Equal(t, []string{"rivertest-workjob"}, job.AttemptedBy)
			require.Equal(t, rivertype.JobStateRunning, job.State)
			return nil
		}

		insertRes, err := bundle.client.InsertTx(ctx, bundle.tx, testArgs{}, nil)
		require.NoError(t, err)

		res, err := testWorker.WorkJob(ctx, t, bundle.tx, insertRes.Job)
		require.NoError(t, err)
		require.Equal(t, river.EventKindJobCompleted, res.EventKind)
	})