Skip to content

Commit 08a640d

Browse files
brandurbrandur
committed
Allow explicit schema injection to rivertest.Require* test functions
Here, resolve #907 by letting an explicit schema be injected into `rivertest.Require*` assertions in a similar way that one can be used in a client. This approach adds a schema in `RequireInsertedOpts`. This comment does a good job of highlight all the potential approaches for adding a schema [1], and unfortunately none of them are all that great. I implemented one other version of this (a variant of option 2 in that list), which as some advantages, but in the end it just ended up ballooning the API out to an uncomfortable degree. The worst part about adding schema to `RequireInsertedOpts` is its interact with the `RequireMany*` functions, where each expectation can set its own schema, and it's not clear what would happen if different expectations set different schemas. I resolved this ambiguity by making it an error to mix and match schemas. Assertions are allowed to send a schema in only the first position like: jobs := requireManyInserted(ctx, bundle.mockT, bundle.driver, []ExpectedJob{ {Args: &Job1Args{String: "foo"}, Opts: bundle.schemaOpts}, {Args: &Job1Args{String: "bar"}}, }) Or send the same schema in all positions: jobs := requireManyInserted(ctx, bundle.mockT, bundle.driver, []ExpectedJob{ {Args: &Job1Args{String: "foo"}, Opts: bundle.schemaOpts}, {Args: &Job1Args{String: "bar"}, Opts: bundle.schemaOpts}, }) But they aren't allowed to set a schema only in position other than the first, or mix and match schemas between expectations. Fixes #907. [1] #907 (comment)
1 parent 5e4c094 commit 08a640d

4 files changed

Lines changed: 251 additions & 144 deletions

File tree

‎rivertest/example_require_inserted_test.go‎

Lines changed: 12 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,10 @@ func Example_requireInserted() {
4343
workers := river.NewWorkers()
4444
river.AddWorker(workers, &RequiredWorker{})
4545

46-
schema := riverdbtest.TestSchema(ctx, testutil.PanicTB(), riverpgxv5.New(dbPool), nil)
46+
var (
47+
schema = riverdbtest.TestSchema(ctx, testutil.PanicTB(), riverpgxv5.New(dbPool), nil)
48+
schemaOpts = &rivertest.RequireInsertedOpts{Schema: schema}
49+
)
4750

4851
riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
4952
Logger: slog.New(&slogutil.SlogMessageOnlyHandler{Level: slog.LevelWarn}),
@@ -72,32 +75,23 @@ func Example_requireInserted() {
7275
// *testing.T that comes from a test's argument.
7376
t := &testing.T{}
7477

75-
// This is needed because rivertest does not yet support an injected schema.
76-
if _, err := tx.Exec(ctx, "SET search_path TO "+schema); err != nil {
77-
panic(err)
78-
}
79-
80-
job := rivertest.RequireInsertedTx[*riverpgxv5.Driver](ctx, t, tx, &RequiredArgs{}, nil)
78+
job := rivertest.RequireInsertedTx[*riverpgxv5.Driver](ctx, t, tx, &RequiredArgs{}, schemaOpts)
8179
fmt.Printf("Test passed with message: %s\n", job.Args.Message)
8280

8381
// Verify the same job again, and this time that it was inserted at the
8482
// default priority and default queue.
8583
_ = rivertest.RequireInsertedTx[*riverpgxv5.Driver](ctx, t, tx, &RequiredArgs{}, &rivertest.RequireInsertedOpts{
8684
Priority: 1,
8785
Queue: river.QueueDefault,
86+
Schema: schema,
8887
})
8988

90-
// Due to some refactoring to make schemas injectable, we don't yet have a
91-
// way of injecting a schema at the pool level. The rivertest API will need
92-
// to be expanded to allow it.
93-
/*
94-
// Insert and verify one on a pool instead of transaction.
95-
_, err = riverClient.Insert(ctx, &RequiredArgs{Message: "Hello from pool."}, nil)
96-
if err != nil {
97-
panic(err)
98-
}
99-
_ = rivertest.RequireInserted(ctx, t, riverpgxv5.New(dbPool), &RequiredArgs{}, nil)
100-
*/
89+
// Insert and verify one on a pool instead of transaction.
90+
_, err = riverClient.Insert(ctx, &RequiredArgs{Message: "Hello from pool."}, nil)
91+
if err != nil {
92+
panic(err)
93+
}
94+
_ = rivertest.RequireInserted(ctx, t, riverpgxv5.New(dbPool), &RequiredArgs{}, schemaOpts)
10195

10296
// Output:
10397
// Test passed with message: Hello.

‎rivertest/example_require_many_inserted_test.go‎

Lines changed: 14 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,10 @@ func Example_requireManyInserted() {
6161
river.AddWorker(workers, &FirstRequiredWorker{})
6262
river.AddWorker(workers, &SecondRequiredWorker{})
6363

64-
schema := riverdbtest.TestSchema(ctx, testutil.PanicTB(), riverpgxv5.New(dbPool), nil)
64+
var (
65+
schema = riverdbtest.TestSchema(ctx, testutil.PanicTB(), riverpgxv5.New(dbPool), nil)
66+
schemaOpts = &rivertest.RequireInsertedOpts{Schema: schema}
67+
)
6568

6669
riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
6770
Logger: slog.New(&slogutil.SlogMessageOnlyHandler{Level: slog.LevelWarn}),
@@ -97,13 +100,8 @@ func Example_requireManyInserted() {
97100
// *testing.T that comes from a test's argument.
98101
t := &testing.T{}
99102

100-
// This is needed because rivertest does not yet support an injected schema.
101-
if _, err := tx.Exec(ctx, "SET search_path TO "+schema); err != nil {
102-
panic(err)
103-
}
104-
105103
jobs := rivertest.RequireManyInsertedTx[*riverpgxv5.Driver](ctx, t, tx, []rivertest.ExpectedJob{
106-
{Args: &FirstRequiredArgs{}},
104+
{Args: &FirstRequiredArgs{}, Opts: schemaOpts},
107105
{Args: &SecondRequiredArgs{}},
108106
{Args: &FirstRequiredArgs{}},
109107
})
@@ -117,22 +115,18 @@ func Example_requireManyInserted() {
117115
{Args: &SecondRequiredArgs{}, Opts: &rivertest.RequireInsertedOpts{
118116
Priority: 1,
119117
Queue: river.QueueDefault,
118+
Schema: schema,
120119
}},
121120
})
122121

123-
// Due to some refactoring to make schemas injectable, we don't yet have a
124-
// way of injecting a schema at the pool level. The rivertest API will need
125-
// to be expanded to allow it.
126-
/*
127-
// Insert and verify one on a pool instead of transaction.
128-
_, err = riverClient.Insert(ctx, &FirstRequiredArgs{Message: "Hello from pool."}, nil)
129-
if err != nil {
130-
panic(err)
131-
}
132-
_ = rivertest.RequireManyInserted(ctx, t, riverpgxv5.New(dbPool), []rivertest.ExpectedJob{
133-
{Args: &FirstRequiredArgs{}},
134-
})
135-
*/
122+
// Insert and verify one on a pool instead of transaction.
123+
_, err = riverClient.Insert(ctx, &FirstRequiredArgs{Message: "Hello from pool."}, nil)
124+
if err != nil {
125+
panic(err)
126+
}
127+
_ = rivertest.RequireManyInserted(ctx, t, riverpgxv5.New(dbPool), []rivertest.ExpectedJob{
128+
{Args: &FirstRequiredArgs{}, Opts: schemaOpts},
129+
})
136130

137131
// Output:
138132
// Job 0 args: {"message": "Hello from first."}

‎rivertest/rivertest.go‎

Lines changed: 74 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,6 @@ import (
1818
"github.com/riverqueue/river/rivertype"
1919
)
2020

21-
// A placeholder for empty schema placeholders that'll need to be fixed to
22-
// something better at some point.
23-
const emptySchema = ""
24-
2521
// testingT is an interface wrapper around *testing.T that's implemented by all
2622
// of *testing.T, *testing.F, and *testing.B.
2723
//
@@ -72,6 +68,13 @@ type RequireInsertedOpts struct {
7268
// No assertion is made if left the zero value.
7369
ScheduledAt time.Time
7470

71+
// Schema is a non-standard Schema where River tables are located. All table
72+
// references in assertion queries will use this value as a prefix.
73+
//
74+
// Defaults to empty, which causes the client to look for tables using the
75+
// setting of Postgres `search_path`.
76+
Schema string
77+
7578
// State is the expected state of the inserted job.
7679
//
7780
// No assertion is made if left the zero value.
@@ -101,12 +104,12 @@ type RequireInsertedOpts struct {
101104
// to cover that case instead.
102105
func RequireInserted[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, tb testing.TB, driver TDriver, expectedJob TArgs, opts *RequireInsertedOpts) *river.Job[TArgs] {
103106
tb.Helper()
104-
return requireInserted(ctx, tb, driver, emptySchema, expectedJob, opts)
107+
return requireInserted(ctx, tb, driver, expectedJob, opts)
105108
}
106109

107-
func requireInserted[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, driver TDriver, schema string, expectedJob TArgs, opts *RequireInsertedOpts) *river.Job[TArgs] {
110+
func requireInserted[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, driver TDriver, expectedJob TArgs, opts *RequireInsertedOpts) *river.Job[TArgs] {
108111
t.Helper()
109-
actualArgs, err := requireInsertedErr[TDriver](ctx, t, driver.GetExecutor(), schema, expectedJob, opts)
112+
actualArgs, err := requireInsertedErr[TDriver](ctx, t, driver.GetExecutor(), expectedJob, opts)
110113
if err != nil {
111114
failure(t, "Internal failure: %s", err)
112115
}
@@ -131,27 +134,32 @@ func requireInserted[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobAr
131134
// to cover that case instead.
132135
func RequireInsertedTx[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, tb testing.TB, tx TTx, expectedJob TArgs, opts *RequireInsertedOpts) *river.Job[TArgs] {
133136
tb.Helper()
134-
return requireInsertedTx[TDriver](ctx, tb, tx, emptySchema, expectedJob, opts)
137+
return requireInsertedTx[TDriver](ctx, tb, tx, expectedJob, opts)
135138
}
136139

137140
// Internal function used by the tests so that the exported version can take
138141
// `testing.TB` instead of `testing.T`.
139142
//
140143
// Also takes a schema for testing purposes, which I haven't quite figured out
141144
// how to get into the public API yet.
142-
func requireInsertedTx[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, tx TTx, schema string, expectedJob TArgs, opts *RequireInsertedOpts) *river.Job[TArgs] {
145+
func requireInsertedTx[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, tx TTx, expectedJob TArgs, opts *RequireInsertedOpts) *river.Job[TArgs] {
143146
t.Helper()
144147
var driver TDriver
145-
actualArgs, err := requireInsertedErr[TDriver](ctx, t, driver.UnwrapExecutor(tx), schema, expectedJob, opts)
148+
actualArgs, err := requireInsertedErr[TDriver](ctx, t, driver.UnwrapExecutor(tx), expectedJob, opts)
146149
if err != nil {
147150
failure(t, "Internal failure: %s", err)
148151
}
149152
return actualArgs
150153
}
151154

152-
func requireInsertedErr[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, exec riverdriver.Executor, schema string, expectedJob TArgs, opts *RequireInsertedOpts) (*river.Job[TArgs], error) {
155+
func requireInsertedErr[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, exec riverdriver.Executor, expectedJob TArgs, opts *RequireInsertedOpts) (*river.Job[TArgs], error) {
153156
t.Helper()
154157

158+
var schema string
159+
if opts != nil {
160+
schema = opts.Schema
161+
}
162+
155163
// Returned ordered by ID.
156164
jobRows, err := exec.JobGetByKindMany(ctx, &riverdriver.JobGetByKindManyParams{
157165
Kind: []string{expectedJob.Kind()},
@@ -205,12 +213,12 @@ func requireInsertedErr[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.Jo
205213
// the given opts.
206214
func RequireNotInserted[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, tb testing.TB, driver TDriver, expectedJob TArgs, opts *RequireInsertedOpts) {
207215
tb.Helper()
208-
requireNotInserted(ctx, tb, driver, emptySchema, expectedJob, opts)
216+
requireNotInserted(ctx, tb, driver, expectedJob, opts)
209217
}
210218

211-
func requireNotInserted[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, driver TDriver, schema string, expectedJob TArgs, opts *RequireInsertedOpts) {
219+
func requireNotInserted[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, driver TDriver, expectedJob TArgs, opts *RequireInsertedOpts) {
212220
t.Helper()
213-
err := requireNotInsertedErr[TDriver](ctx, t, driver.GetExecutor(), schema, expectedJob, opts)
221+
err := requireNotInsertedErr[TDriver](ctx, t, driver.GetExecutor(), expectedJob, opts)
214222
if err != nil {
215223
failure(t, "Internal failure: %s", err)
216224
}
@@ -234,26 +242,31 @@ func requireNotInserted[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.Jo
234242
// the given opts.
235243
func RequireNotInsertedTx[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, tb testing.TB, tx TTx, expectedJob TArgs, opts *RequireInsertedOpts) {
236244
tb.Helper()
237-
requireNotInsertedTx[TDriver](ctx, tb, tx, emptySchema, expectedJob, opts)
245+
requireNotInsertedTx[TDriver](ctx, tb, tx, expectedJob, opts)
238246
}
239247

240248
// Internal function used by the tests so that the exported version can take
241249
// `testing.TB` instead of `testing.T`.
242250
//
243251
// Also takes a schema for testing purposes, which I haven't quite figured out
244252
// how to get into the public API yet.
245-
func requireNotInsertedTx[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, tx TTx, schema string, expectedJob TArgs, opts *RequireInsertedOpts) {
253+
func requireNotInsertedTx[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, tx TTx, expectedJob TArgs, opts *RequireInsertedOpts) {
246254
t.Helper()
247255
var driver TDriver
248-
err := requireNotInsertedErr[TDriver](ctx, t, driver.UnwrapExecutor(tx), schema, expectedJob, opts)
256+
err := requireNotInsertedErr[TDriver](ctx, t, driver.UnwrapExecutor(tx), expectedJob, opts)
249257
if err != nil {
250258
failure(t, "Internal failure: %s", err)
251259
}
252260
}
253261

254-
func requireNotInsertedErr[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, exec riverdriver.Executor, schema string, expectedJob TArgs, opts *RequireInsertedOpts) error {
262+
func requireNotInsertedErr[TDriver riverdriver.Driver[TTx], TTx any, TArgs river.JobArgs](ctx context.Context, t testingT, exec riverdriver.Executor, expectedJob TArgs, opts *RequireInsertedOpts) error {
255263
t.Helper()
256264

265+
var schema string
266+
if opts != nil {
267+
schema = opts.Schema
268+
}
269+
257270
// Returned ordered by ID.
258271
jobRows, err := exec.JobGetByKindMany(ctx, &riverdriver.JobGetByKindManyParams{
259272
Kind: []string{expectedJob.Kind()},
@@ -322,14 +335,20 @@ type ExpectedJob struct {
322335
// the number specified, and will fail in case this expectation isn't met. So if
323336
// a job of a certain kind is emitted multiple times, it must be expected
324337
// multiple times.
338+
//
339+
// If RequireInsertedOpts.Schema is used, it may be set only in the first
340+
// expectation's options (and all expectations will use that schema), or the
341+
// same schema may be set in every expectation. Setting a non-empty schema after
342+
// the first expectation if the first's was empty is not allowed, and neither is
343+
// mixing and matching schemas between options.
325344
func RequireManyInserted[TDriver riverdriver.Driver[TTx], TTx any](ctx context.Context, tb testing.TB, driver TDriver, expectedJobs []ExpectedJob) []*rivertype.JobRow {
326345
tb.Helper()
327-
return requireManyInserted(ctx, tb, driver, string(emptySchema), expectedJobs)
346+
return requireManyInserted(ctx, tb, driver, expectedJobs)
328347
}
329348

330-
func requireManyInserted[TDriver riverdriver.Driver[TTx], TTx any](ctx context.Context, t testingT, driver TDriver, schema string, expectedJobs []ExpectedJob) []*rivertype.JobRow {
349+
func requireManyInserted[TDriver riverdriver.Driver[TTx], TTx any](ctx context.Context, t testingT, driver TDriver, expectedJobs []ExpectedJob) []*rivertype.JobRow {
331350
t.Helper()
332-
actualArgs, err := requireManyInsertedErr[TDriver](ctx, t, driver.GetExecutor(), schema, expectedJobs)
351+
actualArgs, err := requireManyInsertedErr[TDriver](ctx, t, driver.GetExecutor(), expectedJobs)
333352
if err != nil {
334353
failure(t, "Internal failure: %s", err)
335354
}
@@ -356,31 +375,61 @@ func requireManyInserted[TDriver riverdriver.Driver[TTx], TTx any](ctx context.C
356375
// the number specified, and will fail in case this expectation isn't met. So if
357376
// a job of a certain kind is emitted multiple times, it must be expected
358377
// multiple times.
378+
//
379+
// If RequireInsertedOpts.Schema is used, it may be set only in the first
380+
// expectation's options (and all expectations will use that schema), or the
381+
// same schema may be set in every expectation. Setting a non-empty schema after
382+
// the first expectation if the first's was empty is not allowed, and neither is
383+
// mixing and matching schemas between options.
359384
func RequireManyInsertedTx[TDriver riverdriver.Driver[TTx], TTx any](ctx context.Context, tb testing.TB, tx TTx, expectedJobs []ExpectedJob) []*rivertype.JobRow {
360385
tb.Helper()
361-
return requireManyInsertedTx[TDriver](ctx, tb, tx, emptySchema, expectedJobs)
386+
return requireManyInsertedTx[TDriver](ctx, tb, tx, expectedJobs)
362387
}
363388

364389
// Internal function used by the tests so that the exported version can take
365390
// `testing.TB` instead of `testing.T`.
366391
//
367392
// Also takes a schema for testing purposes, which I haven't quite figured out
368393
// how to get into the public API yet.
369-
func requireManyInsertedTx[TDriver riverdriver.Driver[TTx], TTx any](ctx context.Context, t testingT, tx TTx, schema string, expectedJobs []ExpectedJob) []*rivertype.JobRow {
394+
func requireManyInsertedTx[TDriver riverdriver.Driver[TTx], TTx any](ctx context.Context, t testingT, tx TTx, expectedJobs []ExpectedJob) []*rivertype.JobRow {
370395
t.Helper()
371396
var driver TDriver
372-
actualArgs, err := requireManyInsertedErr[TDriver](ctx, t, driver.UnwrapExecutor(tx), schema, expectedJobs)
397+
actualArgs, err := requireManyInsertedErr[TDriver](ctx, t, driver.UnwrapExecutor(tx), expectedJobs)
373398
if err != nil {
374399
failure(t, "Internal failure: %s", err)
375400
}
376401
return actualArgs
377402
}
378403

379-
func requireManyInsertedErr[TDriver riverdriver.Driver[TTx], TTx any](ctx context.Context, t testingT, exec riverdriver.Executor, schema string, expectedJobs []ExpectedJob) ([]*rivertype.JobRow, error) {
404+
func requireManyInsertedErr[TDriver riverdriver.Driver[TTx], TTx any](ctx context.Context, t testingT, exec riverdriver.Executor, expectedJobs []ExpectedJob) ([]*rivertype.JobRow, error) {
380405
t.Helper()
381406

382407
expectedArgsKinds := sliceutil.Map(expectedJobs, func(j ExpectedJob) string { return j.Args.Kind() })
383408

409+
var schema string
410+
if len(expectedJobs) > 0 && expectedJobs[0].Opts != nil {
411+
schema = expectedJobs[0].Opts.Schema
412+
}
413+
414+
// For simplicity (and because I can't think of any reason anyone would need
415+
// to do otherwise), require that if an explicit schema is being set that
416+
// it's the same explicit schema for all options. Callers may specify the
417+
// schema only once in the first expectation's options, or specify the same
418+
// schema for all expectations' options, but they're not allowed to set a
419+
// schema after the first expectation's options if it wasn't set in the
420+
// first, and not allowed to mix and match schemas between options.
421+
for i, expectedJob := range expectedJobs {
422+
if opts := expectedJob.Opts; opts != nil {
423+
if schema == "" && opts.Schema != "" ||
424+
schema != "" && opts.Schema != "" && schema != opts.Schema {
425+
return nil, fmt.Errorf(
426+
"when setting RequireInsertedOpts.Schema with RequireMany schema should be set only at index 0 or the same schema set for all options; "+
427+
"expectedJobs[0].Opts.Schema = %q, expectedJobs[%d].Opts.Schema = %q",
428+
schema, i, opts.Schema)
429+
}
430+
}
431+
}
432+
384433
// Returned ordered by ID.
385434
jobRows, err := exec.JobGetByKindMany(ctx, &riverdriver.JobGetByKindManyParams{
386435
Kind: expectedArgsKinds,

0 commit comments

Comments
 (0)