From 0fb03f67eaf5f01e3c514e0e6929573641edaa7f Mon Sep 17 00:00:00 2001 From: sengjea Date: Tue, 3 Nov 2026 10:58:44 +0000 Subject: [PATCH] Add taskgraphtest.RunWithOverrides, 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. taskgraphtest.RunWithOverrides runs a graph with the tasks providing the keys of some bindings replaced by them. The replaced tasks do not run. Replacing tasks is for tests, so it is not part of the Graph API: the engine's support is reached through an internal package, which nothing outside this module can import, and taskgraphtest is the only package which exposes it. Graph.Run and Graph.Check are unchanged, and an input for a key which a task provides is still a duplicate binding, 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, as is overriding a key twice or a key which is also an input. An absent override replaces the task with a failure. Overrides are part of the Binder returned, and inputs only a replaced task needed are no longer required. CheckWithOverrides validates them without running the graph. Skipped tasks are not signalled by the tasks they depend on, which would otherwise block on a channel with no receiver. Overrider holds one set of overrides for code which runs several graphs, such as an operator with a graph to create and one to delete: each graph takes the overrides for the keys it provides, and Check finds any which are for none of them. taskgraphtest's Test and Suite take Overrides, those of a Test replacing those of its Suite for the same key. --- README.md | 43 +++ graph.go | 78 +++-- internal/overridehook/hook.go | 26 ++ override.go | 208 +++++++++++++ override_test.go | 465 ++++++++++++++++++++++++++++ taskgraphtest/overrides.go | 120 +++++++ taskgraphtest/taskgraphtest.go | 35 ++- taskgraphtest/taskgraphtest_test.go | 148 +++++++++ 8 files changed, 1095 insertions(+), 28 deletions(-) create mode 100644 internal/overridehook/hook.go create mode 100644 override.go create mode 100644 override_test.go create mode 100644 taskgraphtest/overrides.go create mode 100644 taskgraphtest/taskgraphtest_test.go diff --git a/README.md b/README.md index f3fe750..b3fe66f 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,49 @@ 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. `taskgraphtest` can replace +the task which provides a key: + +```go +taskgraphtest.RunWithOverrides(ctx, graph, inputs, []tg.Binding{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. + +Replacing tasks is only available from `taskgraphtest`. `Graph.Run` has no way to do it, and the +engine's support is in an `internal` package, which nothing outside this module can import. Code which +runs a graph and needs to be tested can take a function for running graphs, and be handed +`Overrider.Run` by its tests. + +* An 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 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. `CheckWithOverrides` + finds these without running the graph. +* The override may be absent: `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, 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 given. 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. +* An `Overrider` holds one set of overrides for something which runs several graphs, such as an + operator with a graph to create and a graph to delete. Each graph is run with the overrides for the + keys it provides, and `Overrider.Check` makes sure that every override is for some graph. + +`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..10806e1 100644 --- a/graph.go +++ b/graph.go @@ -90,9 +90,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 +134,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,34 +271,24 @@ 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, nil, false) return err } // Run is Graph.Run. -func (g *graph) Run(ctx context.Context, inputs ...Binding) (b Binder, err error) { +func (g *graph) Run(ctx context.Context, inputs ...Binding) (Binder, error) { + return g.runWithOverrides(ctx, inputs, nil, false) +} + +// runWithOverrides runs the graph with some of its tasks replaced by the overrides, which are for +// tests (see taskgraphtest). +func (g *graph) runWithOverrides( + ctx context.Context, + inputs, overrides []Binding, + ignoreUnprovided bool, +) (b Binder, err error) { startTime := time.Now() defer func() { result := "success" @@ -299,20 +298,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, overrides, ignoreUnprovided) 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 +334,31 @@ 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 +394,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/internal/overridehook/hook.go b/internal/overridehook/hook.go new file mode 100644 index 0000000..b506399 --- /dev/null +++ b/internal/overridehook/hook.go @@ -0,0 +1,26 @@ +// Package overridehook lets taskgraphtest reach the part of the graph engine which replaces tasks. +// +// Replacing tasks is meant for tests, so it is not part of the taskgraph package's API. Go only lets +// packages within this module import an internal package, so this is how taskgraphtest, and nothing +// outside the module, can get at it. The taskgraph package sets the variables when it is +// initialised. +// +// The functions take the arguments typed as any, as this package can't import taskgraph, which +// imports it; taskgraphtest knows the types and converts. +package overridehook + +import "context" + +var ( + // RunWithOverrides runs a graph made by taskgraph.New, as Graph.Run does, with some of its tasks + // replaced by the override bindings, and returns a taskgraph.Binder. If ignoreUnprovided is set, + // overrides for keys which the graph doesn't provide are left out, rather than being an error. + RunWithOverrides func(g any, ctx context.Context, inputs, overrides []any, ignoreUnprovided bool) (any, error) + + // CheckWithOverrides is Graph.Check, with overrides. + CheckWithOverrides func(g any, inputs, overrides []any) error + + // Provides reports whether some task in a graph made by taskgraph.New provides the key (a + // taskgraph.ID). + Provides func(g any, id any) bool +) diff --git a/override.go b/override.go new file mode 100644 index 0000000..1a6ad46 --- /dev/null +++ b/override.go @@ -0,0 +1,208 @@ +package taskgraph + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + + set "github.com/deckarep/golang-set/v2" + "github.com/thought-machine/taskgraph/internal/overridehook" +) + +var ( + // ErrUnknownOverride is returned by taskgraphtest's RunWithOverrides and CheckWithOverrides 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 by taskgraphtest's RunWithOverrides and CheckWithOverrides 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") +) + +func init() { + overridehook.RunWithOverrides = func(g any, ctx context.Context, inputs, overrides []any, ignoreUnprovided bool) (any, error) { + graph, err := asGraph(g) + if err != nil { + return nil, err + } + return graph.runWithOverrides( + ctx, + toBindings(inputs), + toBindings(overrides), + ignoreUnprovided, + ) + } + overridehook.CheckWithOverrides = func(g any, inputs, overrides []any) error { + graph, err := asGraph(g) + if err != nil { + return err + } + _, err = graph.buildInputs(toBindings(inputs), toBindings(overrides), false) + return err + } + overridehook.Provides = func(g any, id any) bool { + graph, err := asGraph(g) + if err != nil { + return false + } + return graph.allProvided.Contains(id.(ID)) + } +} + +func asGraph(g any) (*graph, error) { + graph, ok := g.(*graph) + if !ok { + return nil, fmt.Errorf( + "overrides can only be applied to a Graph made by taskgraph.New, not %T", + g, + ) + } + return graph, nil +} + +func toBindings(in []any) []Binding { + out := make([]Binding, 0, len(in)) + for _, b := range in { + out = append(out, b.(Binding)) + } + return out +} + +// 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 and overrides against the graph, working out which tasks, if +// any, the overrides replace. If ignoreUnprovided is set, overrides for keys which no task in the +// graph provides are left out, instead of being an error. +func (g *graph) buildInputs( + inputs, overrides []Binding, + ignoreUnprovided bool, +) (*runInputs, error) { + b := NewBinder() + if err := b.Store(inputs...); err != nil { + return nil, wrapStackErrorf("duplicate input: %w", err) + } + + result := &runInputs{inputs: b, overrides: overrides, skipped: map[*graphNode]struct{}{}} + if ignoreUnprovided { + result.overrides = nil + for _, override := range overrides { + if g.allProvided.Contains(override.ID()) { + result.overrides = append(result.overrides, override) + } + } + } + if len(result.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..b56dbad --- /dev/null +++ b/override_test.go @@ -0,0 +1,465 @@ +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 := runWith(testContext(t), f.graph, 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 +} + +// override marks an input as an override, for runWith and checkWith to separate from the inputs. +type override struct{ tg.Binding } + +func ov(b tg.Binding) tg.Binding { return override{b} } + +func splitOverrides(inputs []tg.Binding) (ordinary, overrides []tg.Binding) { + for _, input := range inputs { + if o, ok := input.(override); ok { + overrides = append(overrides, o.Binding) + } else { + ordinary = append(ordinary, input) + } + } + return ordinary, overrides +} + +func runWith(ctx context.Context, g tg.Graph, inputs ...tg.Binding) (tg.Binder, error) { + ordinary, overrides := splitOverrides(inputs) + return tgt.RunWithOverrides(ctx, g, ordinary, overrides) +} + +func checkWith(g tg.Graph, inputs ...tg.Binding) error { + ordinary, overrides := splitOverrides(inputs) + return tgt.CheckWithOverrides(g, ordinary, overrides) +} + +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), ov(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), ov(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), ov(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 := runWith(testContext(t), g, keyIn.Bind(1), ov(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 := runWith(testContext(t), g, keyFlag.Bind(flag), ov(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 := runWith(testContext(t), g, f.keyIn.Bind(1), ov(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 := runWith(testContext(t), g); !errors.Is(err, tg.ErrMissingInputs) { + t.Errorf("without the override: got %v; want ErrMissingInputs", err) + } + result, err := runWith(testContext(t), g, ov(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 := runWith(testContext(t), g, keyIn.Bind(1), ov(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), ov(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), ov(keyA.Bind(1)), ov(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), ov(keyA.Bind(2))}, + wantError: tg.ErrDuplicateBinding, + foundBeforeRunning: true, + }, + { + description: "only some of the keys which a task provides", + inputs: []tg.Binding{keyIn.Bind(1), ov(keyM1.Bind(1))}, + wantError: tg.ErrPartialOverride, + foundBeforeRunning: true, + }, + } { + t.Run(tc.description, func(t *testing.T) { + runsM.Store(0) + if _, err := runWith(testContext(t), g, 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 := checkWith(g, 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), ov(keyM1.Bind(10)), ov(keyM2.Bind(20))} + if err := checkWith(g, inputs...); err != nil { + t.Fatalf("Check(): %v", err) + } + result, err := runWith(testContext(t), g, 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 := runWith( + testContext(t), + outer, + keyIn.Bind(1), + ov(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 := runWith(testContext(t), outer, keyIn.Bind(1), ov(keyX.Bind(9)), ov(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/overrides.go b/taskgraphtest/overrides.go new file mode 100644 index 0000000..c12632f --- /dev/null +++ b/taskgraphtest/overrides.go @@ -0,0 +1,120 @@ +package taskgraphtest + +import ( + "context" + "fmt" + "sort" + "strings" + + tg "github.com/thought-machine/taskgraph" + "github.com/thought-machine/taskgraph/internal/overridehook" +) + +func toAny(bindings []tg.Binding) []any { + out := make([]any, 0, len(bindings)) + for _, b := range bindings { + out = append(out, b) + } + return out +} + +// RunWithOverrides runs the graph, as Graph.Run does with the inputs, with the tasks which provide +// the keys of the overrides replaced by them. It is the way to substitute what the tasks of a graph +// would otherwise create for themselves, such as clients of other services, in a test: +// +// taskgraphtest.RunWithOverrides(ctx, graph, inputs, []tg.Binding{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 (tg.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 overrides are included in the Binder returned, as the +// replaced tasks' results would have been, and inputs which only a replaced task needed are not +// required. +// +// An override may be absent, which replaces the task with a failure: +// +// keyClient.BindError(errUnreachable) +// +// Overriding a key which no task provides is an error (tg.ErrUnknownOverride), as is overriding the +// same key twice, or overriding a key which is also given as an input. An input for a key which a task +// provides is still an error (tg.ErrDuplicateBinding), so a key can only ever be replaced on purpose. +// +// Overrides apply to the graph's own tasks. A graph run as a task with Graph.AsTask is replaced as a +// whole, by overriding every key it exposes, but its tasks can't be overridden individually. +func RunWithOverrides( + ctx context.Context, + g tg.Graph, + inputs, overrides []tg.Binding, +) (tg.Binder, error) { + return runWithOverrides(ctx, g, inputs, overrides, false) +} + +func runWithOverrides( + ctx context.Context, + g tg.Graph, + inputs, overrides []tg.Binding, + ignoreUnprovided bool, +) (tg.Binder, error) { + result, err := overridehook.RunWithOverrides( + g, + ctx, + toAny(inputs), + toAny(overrides), + ignoreUnprovided, + ) + if err != nil { + return nil, err + } + return result.(tg.Binder), nil +} + +// CheckWithOverrides is Graph.Check, with the overrides RunWithOverrides would be given: it finds +// unknown, duplicated and partial overrides, and which inputs are still needed, without running +// anything. +func CheckWithOverrides(g tg.Graph, inputs, overrides []tg.Binding) error { + return overridehook.CheckWithOverrides(g, toAny(inputs), toAny(overrides)) +} + +// An Overrider runs graphs with one set of overrides. It is for something which runs several graphs, +// such as an operator with a graph to create and one to delete, and which can only be given one set: +// each graph is run with those of the overrides which are for keys it provides, and Check makes sure +// that every override is for some graph. +type Overrider struct { + overrides []tg.Binding +} + +// NewOverrider returns an Overrider with the overrides. +func NewOverrider(overrides ...tg.Binding) *Overrider { + return &Overrider{overrides: overrides} +} + +// Run runs the graph with the inputs, and those of the Overrider's overrides which are for keys the +// graph provides. Its signature is that of a function for running a graph, so it can be handed to +// code which runs graphs. +func (o *Overrider) Run(ctx context.Context, g tg.Graph, inputs ...tg.Binding) (tg.Binder, error) { + return runWithOverrides(ctx, g, inputs, o.overrides, true) +} + +// Check returns an error if any of the overrides is for a key which none of the graphs provide, which +// is nearly always one left behind by a task which has been renamed or removed. +func (o *Overrider) Check(graphs ...tg.Graph) error { + var unknown []string + for _, override := range o.overrides { + provided := false + for _, g := range graphs { + if overridehook.Provides(g, override.ID()) { + provided = true + break + } + } + if !provided { + unknown = append(unknown, override.ID().String()) + } + } + if len(unknown) > 0 { + sort.Strings(unknown) + return fmt.Errorf("%w: %s", tg.ErrUnknownOverride, strings.Join(unknown, ", ")) + } + return nil +} diff --git a/taskgraphtest/taskgraphtest.go b/taskgraphtest/taskgraphtest.go index b783131..ebf1791 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 with + // RunWithOverrides; 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,7 @@ 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...) + result, err := RunWithOverrides(ctx, test.Graph, test.Inputs, test.Overrides) if !errors.Is(err, test.WantError) { t.Fatalf("Difference in error from Graph.Run(): got %v; want %v", err, test.WantError) @@ -354,6 +362,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 +395,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..379c52e --- /dev/null +++ b/taskgraphtest/taskgraphtest_test.go @@ -0,0 +1,148 @@ +package taskgraphtest_test + +import ( + "context" + "errors" + "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) +} + +// An operator which has a graph to create and one to delete is given one set of overrides. +func TestOverrider(t *testing.T) { + keyIn, keyA, keyB := tg.NewKey[int]("in"), tg.NewKey[int]("a"), tg.NewKey[int]("b") + keyD, keyUnused := tg.NewKey[int]("d"), tg.NewKey[int]("unused") + create := tgt.Must[tg.Graph](t)(tg.New("create", tg.WithTasks( + tg.SimpleTask1[int, int]( + "A", + keyA, + func(_ context.Context, in int) (int, error) { return in * 2, nil }, + keyIn, + ), + tg.SimpleTask1[int, int]( + "B", + keyB, + func(_ context.Context, a int) (int, error) { return a + 1, nil }, + keyA, + ), + ))) + del := tgt.Must[tg.Graph](t)(tg.New("delete", tg.WithTasks( + tg.SimpleTask1[int, int]( + "D", + keyD, + func(_ context.Context, in int) (int, error) { return in * 3, nil }, + keyIn, + ), + ))) + + overrider := tgt.NewOverrider(keyA.Bind(10), keyD.Bind(20)) + if err := overrider.Check(create, del); err != nil { + t.Fatalf("Check(): %v", err) + } + + // Each graph takes the overrides which are for it, and ignores the others. + result, err := overrider.Run(context.Background(), create, keyIn.Bind(1)) + if err != nil { + t.Fatalf("running the graph to create: %v", err) + } + if got, err := keyB.Get(result); err != nil || got != 11 { + t.Errorf("b = %d, %v; want 11, from the override of a", got, err) + } + result, err = overrider.Run(context.Background(), del, keyIn.Bind(1)) + if err != nil { + t.Fatalf("running the graph to delete: %v", err) + } + if got, err := keyD.Get(result); err != nil || got != 20 { + t.Errorf("d = %d, %v; want 20, from the override", got, err) + } + + // An override for neither graph is found by Check, though each Run leaves it out. + stale := tgt.NewOverrider(keyA.Bind(10), keyUnused.Bind(1)) + if err := stale.Check(create, del); !errors.Is(err, tg.ErrUnknownOverride) { + t.Errorf("Check() with a stale override: got %v; want ErrUnknownOverride", err) + } + if _, err := stale.Run(context.Background(), create, keyIn.Bind(1)); err != nil { + t.Errorf("Run() with an override for another graph: %v", err) + } +} + +// fakeGraph is a Graph which taskgraph.New didn't make. +type fakeGraph struct{ tg.Graph } + +func TestOverridesNeedAGraphMadeByTaskgraph(t *testing.T) { + if _, err := tgt.RunWithOverrides(context.Background(), fakeGraph{}, nil, nil); err == nil { + t.Error( + "RunWithOverrides() on a Graph which taskgraph.New didn't make: got nil; want an error", + ) + } +}