package runner
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
)
// fakeCheckout writes a tree for each check-in: a manifest, and nothing
// else the tests need.
type fakeCheckout struct{ manifests map[string]string }
func (f fakeCheckout) Checkout(_ context.Context, dir, checkin string) error {
if err := os.MkdirAll(filepath.Join(dir, "ci"), 0o755); err != nil {
return err
}
return os.WriteFile(filepath.Join(dir, ManifestPath), []byte(f.manifests[checkin]), 0o644)
}
// fakeBridge hands out the queued tasks, one per poll of the matching
// type, and records resolutions and logs.
type fakeBridge struct {
mu sync.Mutex
queue map[string][]polled // task type -> tasks
resolved map[string]map[string]any
logs map[string]string
connects int
auth []string
// secrets answers the forge's secrets route, as the forge would for
// the task named; nil answers 404.
secrets func(taskID string, names []string) (map[string]string, int)
// cancelled answers 409 to every resolve of these tasks, as the
// engine does once their run is cancelled (dagnats#737).
cancelled map[string]bool
actions []string // every resolve action seen, "id action"
}
func (b *fakeBridge) handler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("POST /v1/workers/connect", func(w http.ResponseWriter, r *http.Request) {
b.mu.Lock()
b.connects++
b.auth = append(b.auth, r.Header.Get("Authorization"))
b.mu.Unlock()
// Read the body so the server watches the connection, then
// hold the stream open the way the engine does.
io.Copy(io.Discard, r.Body)
w.Header().Set("Content-Type", "text/event-stream")
w.(http.Flusher).Flush()
<-r.Context().Done()
})
mux.HandleFunc("POST /v1/tasks/poll", func(w http.ResponseWriter, r *http.Request) {
var req struct {
TaskTypes []string `json:"task_types"`
}
json.NewDecoder(r.Body).Decode(&req)
b.mu.Lock()
var out []polled
if len(req.TaskTypes) == 1 {
q := b.queue[req.TaskTypes[0]]
if len(q) > 0 {
out, b.queue[req.TaskTypes[0]] = q[:1], q[1:]
}
}
b.mu.Unlock()
if out == nil {
time.Sleep(20 * time.Millisecond)
out = []polled{}
}
json.NewEncoder(w).Encode(out)
})
mux.HandleFunc("POST /v1/tasks/{id}/resolve", func(w http.ResponseWriter, r *http.Request) {
var body map[string]any
json.NewDecoder(r.Body).Decode(&body)
b.mu.Lock()
b.actions = append(b.actions, r.PathValue("id")+" "+fmt.Sprint(body["action"]))
if b.cancelled[r.PathValue("id")] {
b.mu.Unlock()
http.Error(w, "run cancelled", http.StatusConflict)
return
}
if body["action"] != "heartbeat" {
b.resolved[r.PathValue("id")] = body
}
b.mu.Unlock()
})
mux.HandleFunc("POST /v1/tasks/{id}/secrets", func(w http.ResponseWriter, r *http.Request) {
if b.secrets == nil {
http.NotFound(w, r)
return
}
var body struct{ Names []string }
json.NewDecoder(r.Body).Decode(&body)
vals, code := b.secrets(r.PathValue("id"), body.Names)
if code != 200 {
http.Error(w, "secrets refused: this check-in is on \"feat\"", code)
return
}
json.NewEncoder(w).Encode(map[string]any{"secrets": vals})
})
mux.HandleFunc("POST /v1/tasks/{id}/logs", func(w http.ResponseWriter, r *http.Request) {
var body struct {
Chunks []struct{ Stream, Data string } `json:"chunks"`
}
json.NewDecoder(r.Body).Decode(&body)
// The engine names exactly two streams and refuses the rest
// with a 400; the fake does the same, or a runner sending
// "stdout" passes here and shows no output live.
for i, c := range body.Chunks {
if c.Stream != "out" && c.Stream != "err" {
http.Error(w, fmt.Sprintf("chunk %d: invalid stream %q (want \"out\" or \"err\")", i, c.Stream), http.StatusBadRequest)
return
}
}
b.mu.Lock()
for _, c := range body.Chunks {
d, _ := base64.StdEncoding.DecodeString(c.Data)
b.logs[r.PathValue("id")] += string(d)
}
b.mu.Unlock()
})
return mux
}
func task(id, checkin string) polled {
in, _ := json.Marshal(map[string]string{"repo": "alpha", "checkin": checkin})
return polled{TaskID: id, RunID: "run-" + id, StepID: "s", Input: in}
}
// THE RUNNER RUNS WHAT THE MANIFEST AT THE CHECK-IN SAYS, STREAMS IT,
// AND REPORTS THE VERDICT.
//
// f24d78d9: a generic worker reads ci/runner.json at the check-in under
// test, so the command cannot drift from the code. A task whose command
// succeeds is completed with its output in the run's log; one that
// fails is failed with the reason; a fail_on_output task that prints is
// a failure; a type the check-in's manifest does not name is a failure
// that says which types it does name.
func TestTheRunnerRunsWhatTheManifestAtTheCheckinSays(t *testing.T) {
b := &fakeBridge{queue: map[string][]polled{
"t-test": {task("ok", "c1"), task("bad", "c2"), task("absent", "c3")},
"t-fmt": {task("fmt", "c1")},
}, resolved: map[string]map[string]any{}, logs: map[string]string{}}
srv := httptest.NewServer(b.handler())
defer srv.Close()
good := `{"tasks":{"t-test":["sh","-c","echo hello from c1"],"t-fmt":{"argv":["sh","-c","echo dirty.go"],"fail_on_output":true}}}`
bad := `{"tasks":{"t-test":["sh","-c","echo broke >&2; exit 3"]}}`
other := `{"tasks":{"t-lint":["true"]}}`
r, err := New(Config{
Bridge: srv.URL, Token: "WT", Tasks: []string{"t-test", "t-fmt"}, MaxTasks: 2,
WorkDir: t.TempDir(), Checkout: fakeCheckout{map[string]string{"c1": good, "c2": bad, "c3": other}},
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) }()
deadline := time.Now().Add(10 * time.Second)
for time.Now().Before(deadline) {
b.mu.Lock()
n := len(b.resolved)
b.mu.Unlock()
if n == 4 {
break
}
time.Sleep(20 * time.Millisecond)
}
cancel()
<-done
b.mu.Lock()
defer b.mu.Unlock()
if got := b.resolved["ok"]; got["action"] != "complete" {
t.Errorf("FAIL: a passing task resolved %v", got)
}
if !strings.Contains(b.logs["ok"], "hello from c1") || !strings.Contains(b.logs["ok"], "$ sh -c echo hello from c1") {
t.Errorf("FAIL: the passing task's log = %q", b.logs["ok"])
}
if got := b.resolved["bad"]; got["action"] != "fail" || !strings.Contains(b.logs["bad"], "broke") {
t.Errorf("FAIL: a failing task resolved %v with log %q", got, b.logs["bad"])
}
if got := b.resolved["fmt"]; got["action"] != "fail" || !strings.Contains(got["error"].(string), "must print nothing") {
t.Errorf("FAIL: a fail_on_output task that printed resolved %v", got)
}
if got := b.resolved["absent"]; got["action"] != "fail" || !strings.Contains(got["error"].(string), `names no task "t-test"`) || !strings.Contains(got["error"].(string), "t-lint") {
t.Errorf("FAIL: a type the manifest lacks resolved %v", got)
}
if b.connects == 0 || b.auth[0] != "Bearer WT" {
t.Errorf("FAIL: the runner did not register with its token: %d %v", b.connects, b.auth)
}
}
// A manifest names tasks as argv arrays or objects, and refuses an
// empty manifest or an entry with no command.
func TestManifestParsing(t *testing.T) {
m, err := ParseManifest([]byte(`{"tasks":{"a":["go","test","./..."],"b":{"argv":["gofmt","-l","."],"fail_on_output":true}}}`))
if err != nil || len(m.Tasks) != 2 || m.Tasks["a"].Argv[0] != "go" || !m.Tasks["b"].FailOnOutput {
t.Fatalf("FAIL: %+v %v", m, err)
}
for _, bad := range []string{`{}`, `{"tasks":{}}`, `{"tasks":{"a":[]}}`, `{"tasks":{"a":"go test"}}`, `not json`,
`{"tasks":{"a":{"argv":["true"],"dir":"../elsewhere"}}}`, `{"tasks":{"a":{"argv":["true"],"dir":"/etc"}}}`,
`{"tasks":{"a":{"argv":["true"],"env":{"A=B":"c"}}}}`} {
if _, err := ParseManifest([]byte(bad)); err == nil {
t.Errorf("FAIL: %s was accepted", bad)
}
}
}
// subdirCheckout writes the manifest and an app/ directory, the shape a
// task with "dir" expects.
type subdirCheckout struct{ manifest string }
func (c subdirCheckout) Checkout(_ context.Context, dir, _ string) error {
if err := os.MkdirAll(filepath.Join(dir, "ci"), 0o755); err != nil {
return err
}
if err := os.MkdirAll(filepath.Join(dir, "app"), 0o755); err != nil {
return err
}
return os.WriteFile(filepath.Join(dir, ManifestPath), []byte(c.manifest), 0o644)
}
func TestATaskRunsInItsDirWithItsEnvironment(t *testing.T) {
r, err := New(Config{
Bridge: "http://unused", Token: "t", Tasks: []string{"demo.env"}, WorkDir: t.TempDir(),
Checkout: subdirCheckout{manifest: `{"tasks":{"demo.env":{"argv":["sh","-c","basename \"$PWD\"; echo \"$GREETING\""],"env":{"GREETING":"from-the-manifest"},"dir":"app"}}}`},
})
if err != nil {
t.Fatal(err)
}
var logs, errs bytes.Buffer
if err := r.execute(context.Background(), "task", "demo.env", "c1", "", "", nil, &logs, &errs); err != nil {
t.Fatalf("%v: %s", err, errs.String())
}
out := logs.String()
if !strings.Contains(out, "\napp\n") {
t.Errorf("the task did not run in app/: %q", out)
}
if !strings.Contains(out, "from-the-manifest") {
t.Errorf("the task did not see its env: %q", out)
}
}