From 0e44be236c489ed07d2a0ad1fb209c1e686a9e9f Mon Sep 17 00:00:00 2001 From: sengjea Date: Thu, 29 Oct 2026 10:37:07 +0000 Subject: [PATCH] Add Override, to replace the task which provides a key Tasks often create things for themselves, such as clients of other services, which leaves a graph with no way to give it fakes in a test. An input wrapped in Override replaces the task providing its key, which does not run, instead of being a duplicate binding. A plain input for a provided key is still an error, so a key is only ever replaced on purpose. Overriding a key no task provides, or only some of the keys a task provides, is an error too, as is overriding a key twice. An absent override replaces the task with a failure. Overrides are part of the Binder returned from Run, and inputs only a replaced task needed are no longer required. Graph.Check validates overrides. Skipped tasks are not signalled by the tasks they depend on, which would otherwise block on a channel with no receiver. taskgraphtest's Test and Suite take Overrides, those of a Test replacing those of its Suite for the same key. --- README.md | 36 +++ graph.go | 69 +++--- override.go | 185 ++++++++++++++ override_test.go | 357 ++++++++++++++++++++++++++++ taskgraphtest/taskgraphtest.go | 39 ++- taskgraphtest/taskgraphtest_test.go | 72 ++++++ 6 files changed, 729 insertions(+), 29 deletions(-) create mode 100644 override.go create mode 100644 override_test.go create mode 100644 taskgraphtest/taskgraphtest_test.go diff --git a/README.md b/README.md index f3fe750..64fe5ba 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,42 @@ result, execution of the graph is performed as a parallelised breadth first sear This package contains some creative APIs which work around limitations in Golang generics; chief of these is the `ID` type used in place of slices of keys because Golang slices are invariant. +## Overriding tasks in tests + +A graph's tasks often create things for themselves: clients of other services, say. That makes a +whole graph hard to test, as there is no way to give it a fake client. `Override` marks an input as +replacing the task which provides its key: + +```go +graph.Run(ctx, inputs..., tg.Override(keyClient.Bind(fakeClient))) +``` + +The task providing `keyClient` is not run, and the tasks which depend on it see the fake. This makes +a graph's dependencies a matter of which keys its tasks provide: anything which a test may want to +replace should be a key of its own. + +* An ordinary input for a key which a task provides is still an error, so a key is only ever + replaced on purpose. The same goes for overriding one key twice, or overriding a key which is + also an ordinary input. +* If a task provides several keys they must all be overridden, or none (`ErrPartialOverride`). + Otherwise the task would still have to run, with all of its effects, with some of its results + thrown away. +* Overriding a key which no task provides is an error (`ErrUnknownOverride`), which catches + overrides left behind by a task which has since been renamed or removed. `Graph.Check` finds these + without running the graph. +* The override may be absent: `tg.Override(keyClient.BindError(err))` replaces the task with a + failure, to see how the tasks which depend on it cope. +* The overrides are included in the `Binder` returned from `Run`, as the results of the tasks they + replace would have been. +* Inputs which only a replaced task needed are not required. +* Tasks which only produced inputs for a replaced task still run, as the graph can't know that its + caller doesn't want their results. +* Overrides apply to the graph they are passed to. A graph run as a task with `AsTask` is replaced + as a whole, by overriding every key it exposes, but its tasks can't be overridden individually. + +`taskgraphtest.Test` and `taskgraphtest.Suite` take `Overrides`, those of a `Test` replacing those +of its `Suite` for the same key. + ## Limitations and sharp edges * Taskgraph has no deadlock detection. In theory the fact that the graph is acyclic should prevent diff --git a/graph.go b/graph.go index efec4e9..16c3ba4 100644 --- a/graph.go +++ b/graph.go @@ -54,11 +54,14 @@ type Graph interface { // // This is intended to be run by a genrule at build time to assert that all keys required by tasks // in the graph are provided either as an input or by some other task in the graph. It also checks - // that there are no duplicate inputs. + // that there are no duplicate inputs, and that any Overrides are valid (see Override). Check(inputs ...Binding) error // Run executes the task graph with the given inputs, returning a Binder containing the bound - // values from all tasks (but not any of the input bindings). + // values from all tasks (but not any of the input bindings, other than those given as Overrides, + // which stand in for the results of the tasks they replace). + // + // Inputs given using Override replace the tasks which provide their keys, which are not run. // // It is advisable to set a timeout on the passed context, although it is up to the individual // tasks to listen for context cancellation. @@ -90,9 +93,16 @@ type Graph interface { type runState struct { Binder signals map[string]chan struct{} + // skipped holds the IDs of the nodes which are not run because their tasks are overridden. They + // have nothing listening for signals, and nothing which depends on them waits for them. + skipped map[string]struct{} } func (rs *runState) signal(ctx context.Context, childID string) (err error) { + if _, skipped := rs.skipped[childID]; skipped { + return nil + } + // Capture (and ignore) panics caused by sends to closed channels, which may occur if a task // starts slowly and sees that all of its dependencies are available before we can signal it. defer func() { @@ -127,6 +137,8 @@ type graphNode struct { const ( traceTaskgraphAbsentKeysPrefix = "taskgraph.absent_keys." + traceTaskgraphOverriddenKeys = "taskgraph.overridden_keys" + traceTaskgraphSkippedTasks = "taskgraph.skipped_tasks" ) // Execute the task against the binder provided in the runState. @@ -262,29 +274,9 @@ type graph struct { logger Logger } -func (g *graph) buildInputBinder(inputs ...Binding) (Binder, error) { - b := NewBinder() - - if err := b.Store(inputs...); err != nil { - return nil, wrapStackErrorf("duplicate input: %w", err) - } - - var missingInputs []string - for requiredInput := range g.allDependencies.Difference(g.allProvided).Iter() { - if !b.Has(requiredInput) { - missingInputs = append(missingInputs, requiredInput.String()) - } - } - if len(missingInputs) > 0 { - return nil, wrapStackErrorf("%w: %s", ErrMissingInputs, strings.Join(missingInputs, ", ")) - } - - return b, nil -} - // Check is Graph.Check. func (g *graph) Check(inputs ...Binding) error { - _, err := g.buildInputBinder(inputs...) + _, err := g.buildInputs(inputs...) return err } @@ -299,20 +291,33 @@ func (g *graph) Run(ctx context.Context, inputs ...Binding) (b Binder, err error executionLatency.WithLabelValues(g.name, result). Observe(float64(time.Since(startTime) / time.Millisecond)) }() - base, err := g.buildInputBinder(inputs...) + in, err := g.buildInputs(inputs...) if err != nil { return nil, err } + // Overrides stand in for the results of the tasks they replace, so they are stored with the + // results, and are returned along with them. outputs := NewBinder() + if err := outputs.Store(in.overrides...); err != nil { + return nil, wrapStackErrorf("storing overrides: %w", err) + } overlay := &overlayBinder{ - base: base, + base: in.inputs, overlay: outputs, } tCtx, span := g.tracer.Start(ctx, g.name) defer span.End() - if err := g.runWithBinder(tCtx, overlay); err != nil { + if len(in.overrides) > 0 { + span.SetAttributes( + attribute.StringSlice(traceTaskgraphOverriddenKeys, in.overriddenKeys()), + attribute.StringSlice(traceTaskgraphSkippedTasks, in.skippedTaskNames()), + ) + g.logger.Debugf("graph %s overriding keys %v, so not running tasks %v", + g.name, in.overriddenKeys(), in.skippedTaskNames()) + } + if err := g.runWithBinder(tCtx, overlay, in.skipped); err != nil { span.RecordError(err) return nil, err } @@ -322,20 +327,27 @@ func (g *graph) Run(ctx context.Context, inputs ...Binding) (b Binder, err error // Sets up the per-run state of the graph, and runs all of the tasks in their own goroutines until // all have terminated. If any task returns an error, the entire graph run is cancelled. -func (g *graph) runWithBinder(ctx context.Context, binder Binder) error { +func (g *graph) runWithBinder(ctx context.Context, binder Binder, skipped map[*graphNode]struct{}) error { rs := &runState{ Binder: binder, signals: map[string]chan struct{}{}, + skipped: map[string]struct{}{}, } for _, gn := range g.nodes { rs.signals[gn.id] = make(chan struct{}) } + for gn := range skipped { + rs.skipped[gn.id] = struct{}{} + } // errgroup always cancels the derived context before returning from Wait(), so the select below // must listen to the parent context's Done() channel. eg, egCtx := errgroup.WithContext(ctx) for _, gn := range g.nodes { + if _, isSkipped := skipped[gn]; isSkipped { + continue + } eg.Go(gn.runFunc(egCtx, rs)) } @@ -371,7 +383,8 @@ func (g *graph) AsTask(exposeKeys ...ID) (Task, error) { exposeKeys: exposeSet, } - if err := g.runWithBinder(ctx, gtb); err != nil { + // Overrides apply only to the graph they are given to, not to any graphs it runs as tasks. + if err := g.runWithBinder(ctx, gtb, nil); err != nil { return nil, err } diff --git a/override.go b/override.go new file mode 100644 index 0000000..4565eef --- /dev/null +++ b/override.go @@ -0,0 +1,185 @@ +package taskgraph + +import ( + "errors" + "fmt" + "sort" + "strings" + + set "github.com/deckarep/golang-set/v2" +) + +var ( + // ErrUnknownOverride is returned from Graph.Run() and Graph.Check() if an Override is given for a + // key which no task in the graph provides. This is almost always a stale override, left over from + // a task which has since been renamed or removed. + ErrUnknownOverride = errors.New("override of a key which no task provides") + + // ErrPartialOverride is returned from Graph.Run() and Graph.Check() if Overrides cover some, but + // not all, of the keys which a task provides. The task would still have to run, with all of its + // effects, only to have some of its results discarded, which is very unlikely to be what was + // intended. Override all of a task's keys, or none of them. + ErrPartialOverride = errors.New("override of only some of the keys which a task provides") +) + +// An overrideBinding is a Binding which has been marked, using Override, as replacing the task which +// provides its key. +type overrideBinding struct { + Binding +} + +// Override marks an input Binding as replacing whichever task provides its key, rather than being +// an ordinary input. It is intended for tests, to substitute the dependencies of a graph which its +// tasks would otherwise create for themselves: +// +// graph.Run(ctx, inputs..., tg.Override(keyClient.Bind(fakeClient))) +// +// The task providing the key is not run. If the task provides other keys as well, they must all be +// overridden too (see ErrPartialOverride), so that the real task's effects can never happen +// unnoticed. Tasks which have no other purpose than to produce the inputs of a task which is not run +// are still run, as are tasks which depend on the overridden key; they see the override as if the +// task which it replaces had produced it. The override is included in the Binder returned from +// Run, as the replaced task's results would have been. +// +// The Binding may be absent, which replaces the task with a failure: +// +// tg.Override(keyClient.BindError(errUnreachable)) +// +// Overriding a key which no task provides is an error (ErrUnknownOverride), as is overriding the +// same key twice, or overriding a key which is also given as an ordinary input. An ordinary input +// for a key which a task provides remains an error (ErrDuplicateBinding), so a key can only ever be +// replaced on purpose. +func Override(b Binding) Binding { + if _, ok := b.(overrideBinding); ok { + return b + } + return overrideBinding{Binding: b} +} + +// runInputs is everything a graph is given to run with, once checked against the graph. +type runInputs struct { + // inputs holds the ordinary inputs. + inputs Binder + // overrides holds the overriding bindings, unwrapped. + overrides []Binding + // skipped holds the nodes of the tasks which the overrides replace. + skipped map[*graphNode]struct{} +} + +// overriddenKeys returns the IDs of the overridden keys, sorted, for reporting. +func (ri *runInputs) overriddenKeys() []string { + var ids []string + for _, o := range ri.overrides { + ids = append(ids, o.ID().String()) + } + sort.Strings(ids) + return ids +} + +// skippedTaskNames returns the names of the tasks which are not run, sorted, for reporting. +func (ri *runInputs) skippedTaskNames() []string { + var names []string + for node := range ri.skipped { + names = append(names, node.task.Name()) + } + sort.Strings(names) + return names +} + +// buildInputs checks the given inputs against the graph, working out which tasks, if any, they +// replace. +func (g *graph) buildInputs(inputs ...Binding) (*runInputs, error) { + var ordinary, overrides []Binding + for _, input := range inputs { + if o, ok := input.(overrideBinding); ok { + overrides = append(overrides, o.Binding) + } else { + ordinary = append(ordinary, input) + } + } + + b := NewBinder() + if err := b.Store(ordinary...); err != nil { + return nil, wrapStackErrorf("duplicate input: %w", err) + } + + result := &runInputs{inputs: b, overrides: overrides, skipped: map[*graphNode]struct{}{}} + if len(overrides) > 0 { + if err := g.checkOverrides(b, result); err != nil { + return nil, err + } + } + + // The dependencies which have to be given as inputs are those of the tasks which will run, which + // are fewer if some are replaced. + required := g.allDependencies.Difference(g.allProvided) + if len(result.skipped) > 0 { + required = set.NewSet[ID]() + for _, node := range g.nodes { + if _, skipped := result.skipped[node]; !skipped { + required.Append(node.task.Depends()...) + } + } + required = required.Difference(g.allProvided) + } + var missingInputs []string + for requiredInput := range required.Iter() { + if !b.Has(requiredInput) { + missingInputs = append(missingInputs, requiredInput.String()) + } + } + if len(missingInputs) > 0 { + sort.Strings(missingInputs) + return nil, wrapStackErrorf("%w: %s", ErrMissingInputs, strings.Join(missingInputs, ", ")) + } + + return result, nil +} + +// checkOverrides checks the overrides, filling in which nodes they replace. +func (g *graph) checkOverrides(ordinary Binder, result *runInputs) error { + overridden := set.NewSet[ID]() + for _, o := range result.overrides { + if ordinary.Has(o.ID()) || overridden.Contains(o.ID()) { + return wrapStackErrorf("duplicate input: %w: %q", ErrDuplicateBinding, o.ID()) + } + overridden.Add(o.ID()) + } + + var unknown []string + for id := range overridden.Difference(g.allProvided).Iter() { + unknown = append(unknown, id.String()) + } + if len(unknown) > 0 { + sort.Strings(unknown) + return wrapStackErrorf("%w: %s", ErrUnknownOverride, strings.Join(unknown, ", ")) + } + + var partial []string + for _, node := range g.nodes { + provides := node.task.Provides() + var covered, uncovered []string + for _, id := range provides { + if overridden.Contains(id) { + covered = append(covered, id.String()) + } else { + uncovered = append(uncovered, id.String()) + } + } + switch { + case len(covered) == 0: + case len(uncovered) == 0: + result.skipped[node] = struct{}{} + default: + partial = append(partial, fmt.Sprintf( + "task %s provides [%s, %s], of which only [%s] are overridden", + node.task.Name(), strings.Join(covered, ", "), strings.Join(uncovered, ", "), strings.Join(covered, ", "), + )) + } + } + if len(partial) > 0 { + sort.Strings(partial) + return wrapStackErrorf("%w: %s", ErrPartialOverride, strings.Join(partial, "; ")) + } + return nil +} diff --git a/override_test.go b/override_test.go new file mode 100644 index 0000000..05b464f --- /dev/null +++ b/override_test.go @@ -0,0 +1,357 @@ +package taskgraph_test + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + tg "github.com/thought-machine/taskgraph" + tgt "github.com/thought-machine/taskgraph/taskgraphtest" +) + +// overrideFixture is a small graph, in -> A -> B, whose tasks count how often they run: +// +// A: a = in * 2 +// B: b = a + 1 +type overrideFixture struct { + keyIn, keyA, keyB tg.Key[int] + runsA, runsB atomic.Int32 + graph tg.Graph +} + +func newOverrideFixture(t *testing.T) *overrideFixture { + t.Helper() + f := &overrideFixture{ + keyIn: tg.NewKey[int]("in"), + keyA: tg.NewKey[int]("a"), + keyB: tg.NewKey[int]("b"), + } + f.graph = tgt.Must[tg.Graph](t)(tg.New("override_graph", tg.WithTasks( + tg.SimpleTask1[int, int]("A", f.keyA, func(_ context.Context, in int) (int, error) { + f.runsA.Add(1) + return in * 2, nil + }, f.keyIn), + tg.SimpleTask1[int, int]("B", f.keyB, func(_ context.Context, a int) (int, error) { + f.runsB.Add(1) + return a + 1, nil + }, f.keyA), + ))) + return f +} + +func (f *overrideFixture) run(t *testing.T, inputs ...tg.Binding) tg.Binder { + t.Helper() + result, err := f.graph.Run(testContext(t), inputs...) + if err != nil { + t.Fatalf("Run() failed: %v", err) + } + return result +} + +// testContext returns a context which ends soon, so that a graph which wrongly waits for something +// which will never happen fails the test, rather than hanging it. +func testContext(t *testing.T) context.Context { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + t.Cleanup(cancel) + return ctx +} + +func mustGet[T any](t *testing.T, key tg.ReadOnlyKey[T], b tg.Binder) T { + t.Helper() + got, err := key.Get(b) + if err != nil { + t.Fatalf("getting %s: %v", key.ID(), err) + } + return got +} + +func TestOverride_WithoutOverrideTasksRun(t *testing.T) { + f := newOverrideFixture(t) + result := f.run(t, f.keyIn.Bind(1)) + + if got, want := mustGet(t, f.keyB, result), 3; got != want { + t.Errorf("b = %d; want %d", got, want) + } + if f.runsA.Load() != 1 || f.runsB.Load() != 1 { + t.Errorf("tasks ran A %d times, B %d times; want once each", f.runsA.Load(), f.runsB.Load()) + } +} + +func TestOverride_ReplacesTheTaskWhichProvidesTheKey(t *testing.T) { + f := newOverrideFixture(t) + result := f.run(t, f.keyIn.Bind(1), tg.Override(f.keyA.Bind(100))) + + if f.runsA.Load() != 0 { + t.Errorf("A ran %d times; want it replaced", f.runsA.Load()) + } + if got, want := mustGet(t, f.keyB, result), 101; got != want { + t.Errorf("b = %d; want %d, from the override", got, want) + } + // The override takes the place of the result the task would have produced. + if got, want := mustGet(t, f.keyA, result), 100; got != want { + t.Errorf("a in the result = %d; want the override, %d", got, want) + } +} + +func TestOverride_GraphCanBeRunAgainWithDifferentOverrides(t *testing.T) { + f := newOverrideFixture(t) + + if got, want := mustGet(t, f.keyB, f.run(t, f.keyIn.Bind(1), tg.Override(f.keyA.Bind(10)))), 11; got != want { + t.Errorf("first run: b = %d; want %d", got, want) + } + if got, want := mustGet(t, f.keyB, f.run(t, f.keyIn.Bind(1))), 3; got != want { + t.Errorf("second run, without an override: b = %d; want %d", got, want) + } + if got, want := mustGet(t, f.keyB, f.run(t, f.keyIn.Bind(1), tg.Override(f.keyA.Bind(20)))), 21; got != want { + t.Errorf("third run: b = %d; want %d", got, want) + } + if f.runsA.Load() != 1 { + t.Errorf("A ran %d times over the three runs; want once, in the one without an override", f.runsA.Load()) + } +} + +func TestOverride_AbsentOverrideFailsTheTask(t *testing.T) { + errBoom := errors.New("boom") + keyIn, keyA, keyB := tg.NewKey[int]("in"), tg.NewKey[int]("a"), tg.NewKey[int]("b") + var runsA atomic.Int32 + g := tgt.Must[tg.Graph](t)(tg.New("absent_override", tg.WithTasks( + tg.SimpleTask1[int, int]("A", keyA, func(_ context.Context, in int) (int, error) { + runsA.Add(1) + return in, nil + }, keyIn), + // Takes the absence of a in its stride, as a task which can cope with a missing dependency does. + tg.NewTask("B", func(_ context.Context, b tg.Binder) ([]tg.Binding, error) { + if _, err := keyA.Get(b); err != nil { + return []tg.Binding{keyB.Bind(-1)}, nil + } + return []tg.Binding{keyB.Bind(1)}, nil + }, []tg.ID{keyA.ID()}, []tg.ID{keyB.ID()}), + ))) + + result, err := g.Run(testContext(t), keyIn.Bind(1), tg.Override(keyA.BindError(errBoom))) + if err != nil { + t.Fatalf("Run() failed: %v", err) + } + if runsA.Load() != 0 { + t.Errorf("A ran %d times; want it replaced by the failure", runsA.Load()) + } + if got := mustGet(t, keyB, result); got != -1 { + t.Errorf("b = %d; want -1, as B saw a absent", got) + } + tgt.ExpectAbsentError(t, result, keyA, errBoom) +} + +func TestOverride_ConditionalTaskIsReplacedWhateverItsCondition(t *testing.T) { + for _, flag := range []bool{true, false} { + keyFlag, keyT, keyOut := tg.NewKey[bool]("flag"), tg.NewKey[int]("t"), tg.NewKey[int]("out") + var runsT atomic.Int32 + g := tgt.Must[tg.Graph](t)(tg.New("conditional_override", tg.WithTasks( + tg.Conditional{ + Wrapped: tg.SimpleTask("T", keyT, func(context.Context, tg.Binder) (int, error) { + runsT.Add(1) + return 1, nil + }), + Condition: tg.ConditionAnd{keyFlag}, + }.Locate(), + tg.SimpleTask1[int, int]("Out", keyOut, func(_ context.Context, v int) (int, error) { + return v * 10, nil + }, keyT), + ))) + + result, err := g.Run(testContext(t), keyFlag.Bind(flag), tg.Override(keyT.Bind(7))) + if err != nil { + t.Fatalf("flag=%v: Run() failed: %v", flag, err) + } + if runsT.Load() != 0 { + t.Errorf("flag=%v: T ran %d times; want it replaced", flag, runsT.Load()) + } + if got, want := mustGet(t, keyOut, result), 70; got != want { + t.Errorf("flag=%v: out = %d; want %d", flag, got, want) + } + } +} + +func TestOverride_TasksWhichFedAReplacedTaskStillRunAndDoNotBlock(t *testing.T) { + // This documents a limitation: the graph can't know which results its caller wants, so it can't + // tell that nothing needs what only the replaced task consumed. + f := newOverrideFixture(t) + keyU, runsU := tg.NewKey[int]("u"), atomic.Int32{} + g := tgt.Must[tg.Graph](t)(tg.New("upstream_still_runs", tg.WithTasks( + tg.SimpleTask1[int, int]("U", keyU, func(_ context.Context, in int) (int, error) { + runsU.Add(1) + return in, nil + }, f.keyIn), + tg.SimpleTask1[int, int]("A", f.keyA, func(_ context.Context, u int) (int, error) { + return u * 2, nil + }, keyU), + ))) + + if _, err := g.Run(testContext(t), f.keyIn.Bind(1), tg.Override(f.keyA.Bind(5))); err != nil { + t.Fatalf("Run() failed: %v", err) + } + if runsU.Load() != 1 { + t.Errorf("U ran %d times; want once", runsU.Load()) + } +} + +func TestOverride_InputsOnlyAReplacedTaskNeededAreNotRequired(t *testing.T) { + keyOnlyForA, keyA, keyOut := tg.NewKey[int]("only_for_a"), tg.NewKey[int]("a"), tg.NewKey[int]("out") + g := tgt.Must[tg.Graph](t)(tg.New("needed_inputs", tg.WithTasks( + tg.SimpleTask1[int, int]("A", keyA, func(_ context.Context, v int) (int, error) { return v, nil }, keyOnlyForA), + tg.SimpleTask1[int, int]("Out", keyOut, func(_ context.Context, a int) (int, error) { return a + 1, nil }, keyA), + ))) + + if _, err := g.Run(testContext(t)); !errors.Is(err, tg.ErrMissingInputs) { + t.Errorf("without the override: got %v; want ErrMissingInputs", err) + } + result, err := g.Run(testContext(t), tg.Override(keyA.Bind(4))) + if err != nil { + t.Fatalf("with the override: Run() failed: %v", err) + } + if got, want := mustGet(t, keyOut, result), 5; got != want { + t.Errorf("out = %d; want %d", got, want) + } +} + +func TestOverride_TasksDependingOnReplacedAndOrdinaryKeysDoNotWaitForeverOrDeadlock(t *testing.T) { + keyIn, keyA, keyP, keyBoth, keyOnlyA := tg.NewKey[int]("in"), tg.NewKey[int]("a"), tg.NewKey[int]("p"), + tg.NewKey[int]("both"), tg.NewKey[int]("only_a") + g := tgt.Must[tg.Graph](t)(tg.New("mixed_dependents", tg.WithTasks( + tg.SimpleTask1[int, int]("A", keyA, func(_ context.Context, v int) (int, error) { return v, nil }, keyIn), + tg.SimpleTask1[int, int]("P", keyP, func(_ context.Context, v int) (int, error) { + time.Sleep(20 * time.Millisecond) + return v * 100, nil + }, keyIn), + tg.SimpleTask2[int, int, int]("Both", keyBoth, func(_ context.Context, a, p int) (int, error) { + return a + p, nil + }, keyA, keyP), + tg.SimpleTask1[int, int]("OnlyA", keyOnlyA, func(_ context.Context, a int) (int, error) { return a, nil }, keyA), + ))) + + result, err := g.Run(testContext(t), keyIn.Bind(1), tg.Override(keyA.Bind(7))) + if err != nil { + t.Fatalf("Run() failed: %v", err) + } + if got, want := mustGet(t, keyBoth, result), 107; got != want { + t.Errorf("both = %d; want %d", got, want) + } + if got, want := mustGet(t, keyOnlyA, result), 7; got != want { + t.Errorf("only_a = %d; want %d", got, want) + } +} + +func TestOverride_Errors(t *testing.T) { + keyIn, keyA, keyM1, keyM2, keyZ := tg.NewKey[int]("in"), tg.NewKey[int]("a"), tg.NewKey[int]("m1"), + tg.NewKey[int]("m2"), tg.NewKey[int]("z") + var runsM atomic.Int32 + g := tgt.Must[tg.Graph](t)(tg.New("override_errors", tg.WithTasks( + tg.SimpleTask1[int, int]("A", keyA, func(_ context.Context, v int) (int, error) { return v, nil }, keyIn), + tg.NewTask("M", func(_ context.Context, _ tg.Binder) ([]tg.Binding, error) { + runsM.Add(1) + return []tg.Binding{keyM1.Bind(1), keyM2.Bind(2)}, nil + }, nil, []tg.ID{keyM1.ID(), keyM2.ID()}), + ))) + + for _, tc := range []struct { + description string + inputs []tg.Binding + wantError error + // Whether the problem is found before anything runs, which is true of everything but a plain + // input for a key which a task provides, which is only found when that task stores its result. + foundBeforeRunning bool + }{ + { + description: "a key which no task provides", + inputs: []tg.Binding{keyIn.Bind(1), tg.Override(keyZ.Bind(1))}, + wantError: tg.ErrUnknownOverride, + foundBeforeRunning: true, + }, + { + description: "an input, not an override, for a key which a task provides", + inputs: []tg.Binding{keyIn.Bind(1), keyA.Bind(5)}, + wantError: tg.ErrDuplicateBinding, + }, + { + description: "the same key overridden twice", + inputs: []tg.Binding{keyIn.Bind(1), tg.Override(keyA.Bind(1)), tg.Override(keyA.Bind(2))}, + wantError: tg.ErrDuplicateBinding, + foundBeforeRunning: true, + }, + { + description: "a key which is also an ordinary input", + inputs: []tg.Binding{keyIn.Bind(1), keyA.Bind(1), tg.Override(keyA.Bind(2))}, + wantError: tg.ErrDuplicateBinding, + foundBeforeRunning: true, + }, + { + description: "only some of the keys which a task provides", + inputs: []tg.Binding{keyIn.Bind(1), tg.Override(keyM1.Bind(1))}, + wantError: tg.ErrPartialOverride, + foundBeforeRunning: true, + }, + } { + t.Run(tc.description, func(t *testing.T) { + runsM.Store(0) + if _, err := g.Run(testContext(t), tc.inputs...); !errors.Is(err, tc.wantError) { + t.Errorf("Run(): got %v; want %v", err, tc.wantError) + } + if tc.foundBeforeRunning { + if runsM.Load() != 0 { + t.Errorf("M ran %d times; want nothing to run, as this is rejected first", runsM.Load()) + } + if err := g.Check(tc.inputs...); !errors.Is(err, tc.wantError) { + t.Errorf("Check(): got %v; want %v", err, tc.wantError) + } + } + }) + } + + t.Run("all of the keys which a task provides", func(t *testing.T) { + runsM.Store(0) + inputs := []tg.Binding{keyIn.Bind(1), tg.Override(keyM1.Bind(10)), tg.Override(keyM2.Bind(20))} + if err := g.Check(inputs...); err != nil { + t.Fatalf("Check(): %v", err) + } + result, err := g.Run(testContext(t), inputs...) + if err != nil { + t.Fatalf("Run(): %v", err) + } + if runsM.Load() != 0 { + t.Errorf("M ran %d times; want it replaced", runsM.Load()) + } + if got := mustGet(t, keyM2, result); got != 20 { + t.Errorf("m2 = %d; want 20", got) + } + }) +} + +func TestOverride_GraphRunAsATaskIsReplacedAsAWhole(t *testing.T) { + keyIn, keyX, keyY := tg.NewKey[int]("in"), tg.NewKey[int]("x"), tg.NewKey[int]("y") + var runsInner atomic.Int32 + inner := tgt.Must[tg.Graph](t)(tg.New("inner", tg.WithTasks( + tg.NewTask("XY", func(_ context.Context, b tg.Binder) ([]tg.Binding, error) { + runsInner.Add(1) + in := mustGet(t, keyIn, b) + return []tg.Binding{keyX.Bind(in), keyY.Bind(in * 2)}, nil + }, []tg.ID{keyIn.ID()}, []tg.ID{keyX.ID(), keyY.ID()}), + ))) + asTask := tgt.Must[tg.Task](t)(inner.AsTask(keyX.ID(), keyY.ID())) + outer := tgt.Must[tg.Graph](t)(tg.New("outer", tg.WithTasks(asTask))) + + if _, err := outer.Run(testContext(t), keyIn.Bind(1), tg.Override(keyX.Bind(9))); !errors.Is(err, tg.ErrPartialOverride) { + t.Errorf("overriding one of the keys the task exposes: got %v; want ErrPartialOverride", err) + } + result, err := outer.Run(testContext(t), keyIn.Bind(1), tg.Override(keyX.Bind(9)), tg.Override(keyY.Bind(8))) + if err != nil { + t.Fatalf("overriding both: Run() failed: %v", err) + } + if runsInner.Load() != 0 { + t.Errorf("the inner graph ran %d times; want it replaced", runsInner.Load()) + } + if got := mustGet(t, keyY, result); got != 8 { + t.Errorf("y = %d; want 8", got) + } +} diff --git a/taskgraphtest/taskgraphtest.go b/taskgraphtest/taskgraphtest.go index b783131..2fd9eee 100644 --- a/taskgraphtest/taskgraphtest.go +++ b/taskgraphtest/taskgraphtest.go @@ -263,6 +263,14 @@ type Test struct { // Inputs defines the inputs passed to the graph for the test. Inputs []tg.Binding + // Overrides defines bindings which replace the tasks providing their keys, as if each had been + // passed in Inputs using tg.Override; the replaced tasks are not run. Overrides given by the Suite + // are applied as well, those given here replacing them for the same key. + // + // The overrides are part of the result, as the replaced tasks' results would have been, so + // CheckExcessBindings counts them among the bindings to be wanted. + Overrides []tg.Binding + // WantError defines the expected error returned from the Graph's execution (note that the test // will always fail if Graph setup/configuration fails). Compared using errors.Is. WantError error @@ -314,7 +322,11 @@ func (test Test) Run(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), test.Timeout) defer cancel() - result, err := test.Graph.Run(ctx, test.Inputs...) + inputs := append([]tg.Binding{}, test.Inputs...) + for _, override := range test.Overrides { + inputs = append(inputs, tg.Override(override)) + } + result, err := test.Graph.Run(ctx, inputs...) if !errors.Is(err, test.WantError) { t.Fatalf("Difference in error from Graph.Run(): got %v; want %v", err, test.WantError) @@ -354,6 +366,11 @@ type Suite struct { // the set is expected to contain a single element. Task tg.TaskSet + // Overrides defines bindings which replace the tasks providing their keys in every test of the + // suite (see Test.Overrides). A test which overrides the same key itself uses its own binding in + // place of the suite's. + Overrides []tg.Binding + // Tests defines the individual tests to be run. Tests []Test @@ -382,10 +399,30 @@ func (s Suite) Run(t *testing.T) { if test.Timeout <= 0 { test.Timeout = s.Timeout } + test.Overrides = mergeOverrides(s.Overrides, test.Overrides) t.Run(test.Description, test.Run) } } +// mergeOverrides returns the overrides of a suite and of one of its tests together, those of the test +// replacing those of the suite which are for the same key. +func mergeOverrides(suite, test []tg.Binding) []tg.Binding { + if len(suite) == 0 { + return test + } + overridden := set.NewSet[tg.ID]() + for _, b := range test { + overridden.Add(b.ID()) + } + merged := append([]tg.Binding{}, test...) + for _, b := range suite { + if !overridden.Contains(b.ID()) { + merged = append(merged, b) + } + } + return merged +} + // DummyTaskFunc returns a function suitable for passing to taskgraph.NewTask which returns the // given bindings. func DummyTaskFunc(bindings ...tg.Binding) func(context.Context, tg.Binder) ([]tg.Binding, error) { diff --git a/taskgraphtest/taskgraphtest_test.go b/taskgraphtest/taskgraphtest_test.go new file mode 100644 index 0000000..806fb43 --- /dev/null +++ b/taskgraphtest/taskgraphtest_test.go @@ -0,0 +1,72 @@ +package taskgraphtest_test + +import ( + "context" + "sync/atomic" + "testing" + + tg "github.com/thought-machine/taskgraph" + tgt "github.com/thought-machine/taskgraph/taskgraphtest" +) + +func TestOverrides(t *testing.T) { + keyIn, keyA, keyB := tg.NewKey[int]("in"), tg.NewKey[int]("a"), tg.NewKey[int]("b") + var runsA atomic.Int32 + graph := tgt.Must[tg.Graph](t)(tg.New("overrides", tg.WithTasks( + tg.SimpleTask1[int, int]("A", keyA, func(_ context.Context, in int) (int, error) { + runsA.Add(1) + return in * 2, nil + }, keyIn), + tg.SimpleTask1[int, int]("B", keyB, func(_ context.Context, a int) (int, error) { + return a + 1, nil + }, keyA), + ))) + + tgt.Suite{ + Graph: graph, + Overrides: []tg.Binding{keyA.Bind(10)}, + Tests: []tgt.Test{ + { + Description: "a suite's overrides apply to its tests", + Inputs: []tg.Binding{keyIn.Bind(1)}, + WantBindings: []tgt.BindingMatcher{tgt.Match(keyB.Bind(11))}, + }, + { + Description: "a test's override replaces the suite's for the same key", + Inputs: []tg.Binding{keyIn.Bind(1)}, + Overrides: []tg.Binding{keyA.Bind(20)}, + WantBindings: []tgt.BindingMatcher{tgt.Match(keyB.Bind(21))}, + }, + { + Description: "overrides are part of the result", + Inputs: []tg.Binding{keyIn.Bind(1)}, + WantBindings: []tgt.BindingMatcher{tgt.Match(keyA.Bind(10)), tgt.Match(keyB.Bind(11))}, + }, + { + Description: "excess bindings includes them", + Inputs: []tg.Binding{keyIn.Bind(1)}, + WantBindings: []tgt.BindingMatcher{tgt.Match(keyA.Bind(10)), tgt.Match(keyB.Bind(11))}, + CheckExcessBindings: true, + }, + { + Description: "an absent override makes the task fail", + Inputs: []tg.Binding{keyIn.Bind(1)}, + Overrides: []tg.Binding{keyA.BindAbsent()}, + WantBindings: []tgt.BindingMatcher{tgt.Match(keyA.BindAbsent())}, + WantError: tg.ErrIsAbsent, + }, + }, + }.Run(t) + + if runsA.Load() != 0 { + t.Errorf("A ran %d times; want it replaced in every test", runsA.Load()) + } + + // A test which isn't part of a suite can override too. + tgt.Test{ + Graph: graph, + Inputs: []tg.Binding{keyIn.Bind(1)}, + Overrides: []tg.Binding{keyA.Bind(5)}, + WantBindings: []tgt.BindingMatcher{tgt.Match(keyB.Bind(6))}, + }.Run(t) +}