package runner
import (
"context"
"encoding/json"
"net/http/httptest"
"strings"
"testing"
"time"
)
// A STEP THAT DEPENDS ON ANOTHER STILL KNOWS ITS CHECK-IN.
//
// dagnats hands a step with one dependency that step's output as its
// input, and a step with several a map of their outputs. pikchr-studio's
// deploy, the first quarry pipeline with depends_on, failed with "the
// task names no check-in" because the runner's output was "ok". The
// output is now the task's subject, so it flows down the graph.
func TestAStepAfterAnotherGetsTheCheckin(t *testing.T) {
subjectIn := func(checkin string) json.RawMessage {
b, _ := json.Marshal(map[string]string{"repo": "alpha", "checkin": checkin, "branch": "trunk"})
return b
}
b := &fakeBridge{queue: map[string][]polled{
"t-test": {{TaskID: "test", RunID: "r", StepID: "test", Input: subjectIn("c1")}},
"t-deploy": nil,
}, resolved: map[string]map[string]any{}, logs: map[string]string{}}
srv := httptest.NewServer(b.handler())
defer srv.Close()
manifest := `{"tasks":{"t-test":["true"],"t-deploy":["sh","-c","echo deployed"]}}`
r, err := New(Config{
Bridge: srv.URL, Token: "WT", Tasks: []string{"t-test", "t-deploy"}, MaxTasks: 1,
WorkDir: t.TempDir(), Checkout: fakeCheckout{map[string]string{"c1": manifest}},
Heartbeat: 50 * time.Millisecond, Logf: t.Logf,
})
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { r.Run(ctx); close(done) }()
wait := func(id string) map[string]any {
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(20 * time.Millisecond) {
b.mu.Lock()
got, ok := b.resolved[id]
b.mu.Unlock()
if ok {
return got
}
}
t.Fatalf("FAIL: task %s never resolved", id)
return nil
}
// The engine's hand-off, done here by hand: the test's output becomes
// the deploy's input (one dependency), then a fan-in of two.
out := wait("test")
if out["action"] != "complete" {
t.Fatalf("FAIL: the test task resolved %v", out)
}
carried, _ := json.Marshal(out["output"])
fanIn, _ := json.Marshal(map[string]json.RawMessage{"test": carried, "worker": subjectIn("c1")})
split, _ := json.Marshal(map[string]json.RawMessage{"test": carried, "worker": subjectIn("c2")})
b.mu.Lock()
b.queue["t-deploy"] = []polled{
{TaskID: "one", RunID: "r", StepID: "deploy", Input: carried},
{TaskID: "fan", RunID: "r", StepID: "deploy", Input: fanIn},
{TaskID: "split", RunID: "r", StepID: "deploy", Input: split},
}
b.mu.Unlock()
one, fan, bad := wait("one"), wait("fan"), wait("split")
cancel()
<-done
if one["action"] != "complete" || !strings.Contains(b.logs["one"], "deployed") {
t.Errorf("FAIL: a step after one dependency resolved %v, log %q; want it run at c1", one, b.logs["one"])
}
if fan["action"] != "complete" {
t.Errorf("FAIL: a step after two dependencies on c1 resolved %v; want it run", fan)
}
if bad["action"] != "fail" || !strings.Contains(bad["error"].(string), "names no check-in") {
t.Errorf("FAIL: dependencies naming two check-ins resolved %v; want a refusal", bad)
}
t.Log("ok: the check-in flows through one dependency and a fan-in, and a disagreement is refused")
}