Function Test_RecordedOutput

recorded_output_test.go:19–171

(t *testing.T)

)

func Test_RecordedOutput(t *testing.T) {
	t.Parallel()
	ctx := context.Background()

	type JobArgs struct {
		testutil.JobArgsReflectKind[JobArgs]
	}

	type myOutput struct {
		Message string `json:"message"`
	}

	type testBundle struct {
		dbPool *pgxpool.Pool
		schema string
	}

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

		var (
			dbPool = riversharedtest.DBPool(ctx, t)
			driver = riverpgxv5.New(dbPool)
			schema = riverdbtest.TestSchema(ctx, t, driver, nil)
			config = newTestConfig(t, schema)
			client = newTestClient(t, dbPool, config)
		)

		t.Cleanup(func() { require.NoError(t, client.Stop(ctx)) })

		return client, &testBundle{
			dbPool: dbPool,
			schema: schema,
		}
	}

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

		client, _ := setup(t)

		validOutput := myOutput{Message: "it worked"}
		expectedOutput := `{"output":{"message":"it worked"}}`
		AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
			return RecordOutput(ctx, validOutput)
		}))

		subChan := subscribe(t, client)
		startClient(ctx, t, client)

		insertRes, err := client.Insert(ctx, JobArgs{}, nil)
		require.NoError(t, err)

		event := riversharedtest.WaitOrTimeout(t, subChan)
		require.Equal(t, EventKindJobCompleted, event.Kind)
		require.JSONEq(t, expectedOutput, string(event.Job.Metadata))

		jobFromDB, err := client.JobGet(ctx, insertRes.Job.ID)