package ci
import (
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"slices"
"strings"
"sync"
"testing"
)
// startedEngine accepts every registration and start, and records which
// pipeline each started run was for, at which check-in.
func startedEngine(t *testing.T) (*httptest.Server, func() []string) {
srv, started, _ := cancellingEngine(t)
return srv, started
}
// cancellingEngine is startedEngine that also lists the runs it started
// as running, with their labels, and records which ones were cancelled
// (by check-in and pipeline), so a supersede can be seen.
func cancellingEngine(t *testing.T) (*httptest.Server, func() []string, func() []string) {
t.Helper()
var mu sync.Mutex
var started, cancelled []string
var runs []Run
what := map[string]string{} // run id -> "checkin pipeline"
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case r.URL.Path == "/workflows":
w.WriteHeader(http.StatusCreated)
w.Write([]byte(`{"status":"registered"}`))
case r.URL.Path == "/runs" && r.Method == http.MethodGet:
mu.Lock()
json.NewEncoder(w).Encode(runs)
mu.Unlock()
case r.URL.Path == "/runs":
var body struct {
Input map[string]any `json:"input"`
Labels map[string]string `json:"labels"`
}
json.NewDecoder(r.Body).Decode(&body)
mu.Lock()
key := fmt.Sprint(body.Input["checkin"], " ", body.Input["pipeline"])
started = append(started, key)
id := fmt.Sprintf("r-%d", len(what)+1)
what[id] = key
runs = append(runs, Run{RunID: id, Status: "running", Labels: body.Labels})
mu.Unlock()
w.WriteHeader(http.StatusCreated)
fmt.Fprintf(w, `{"run_id":%q}`, id)
case strings.HasPrefix(r.URL.Path, "/runs/") && strings.HasSuffix(r.URL.Path, "/cancel"):
id := strings.TrimSuffix(strings.TrimPrefix(r.URL.Path, "/runs/"), "/cancel")
mu.Lock()
cancelled = append(cancelled, what[id])
for i := range runs {
if runs[i].RunID == id {
runs[i].Status = "cancelled"
}
}
mu.Unlock()
}
}))
t.Cleanup(srv.Close)
take := func(l *[]string) []string {
mu.Lock()
defer mu.Unlock()
out := append([]string(nil), (*l)...)
*l = (*l)[:0]
slices.Sort(out)
return out
}
return srv, func() []string { return take(&started) }, func() []string { return take(&cancelled) }
}
// A PIPELINE MARKED "quarry_when": "landing" RUNS ON THE DEFAULT BRANCH
// AND IN LANDINGS' MERGES, NEVER ON A BRANCH TIP.
//
// A proposed branch's checks are its merge with the target (RunPairAt),
// as a pull_request workflow runs on a pull request's test merge, so a
// push to it runs its push pipelines only. A re-run somebody asks for
// runs everything.
func TestALandingPipelineRunsOnTheDefaultBranchOnly(t *testing.T) {
srv, started := startedEngine(t)
e := New(srv.URL)
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo := landingRepo()
round := func(news ...Checkin) []string {
repo.news = news
if err := Trigger(e, st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
return started()
}
if got := round(Checkin{Hash: "b1", Branch: "feat"}); !slices.Equal(got, []string{"b1 ci/quarry.json"}) {
t.Errorf("FAIL: a bare branch push started %v; want only ci/quarry.json", got)
}
repo.landings["feat"] = true
if got := round(Checkin{Hash: "b2", Branch: "feat"}); !slices.Equal(got, []string{"b2 ci/quarry.json"}) {
t.Errorf("FAIL: a push to a proposed branch started %v; want only ci/quarry.json, its merge runs the rest", got)
}
if got := round(Checkin{Hash: "t1", Branch: "trunk"}); !slices.Equal(got, []string{"t1 ci/quarry.json", "t1 ci/race.json"}) {
t.Errorf("FAIL: a trunk check-in started %v; want both pipelines", got)
}
repo.landings["feat"] = false // the landing closed
if got := round(Checkin{Hash: "b3", Branch: "feat"}); !slices.Equal(got, []string{"b3 ci/quarry.json"}) {
t.Errorf("FAIL: a push after the landing closed started %v; want only ci/quarry.json", got)
}
repo.defaultBranch = "main"
if got := round(Checkin{Hash: "m1", Branch: "main"}, Checkin{Hash: "x1", Branch: "trunk"}); !slices.Equal(got, []string{"m1 ci/quarry.json", "m1 ci/race.json", "x1 ci/quarry.json"}) {
t.Errorf("FAIL: with main as the default branch, started %v; want race on main only", got)
}
repo.defaultBranch = ""
repo.news = nil
if err := RunAt(e, st, "alpha", repo, "http://x/r/alpha", "b9", ""); err != nil {
t.Fatal(err)
}
if got := started(); !slices.Equal(got, []string{"b9 ci/quarry.json", "b9 ci/race.json"}) {
t.Errorf("FAIL: an asked-for re-run started %v; want every pipeline", got)
}
t.Log("ok: the landing pipeline runs on trunk and on asked-for re-runs, never on a branch tip")
}
// SEVERAL CHECK-INS IN ONE ROUND ON A PROPOSED BRANCH EACH RUN, AND EACH
// SUPERSEDES THE ONE BEFORE: only the tip's runs are left going.
func TestSeveralCheckinsOnAProposedBranchLeaveOnlyTheTipRunning(t *testing.T) {
srv, started, cancelled := cancellingEngine(t)
e := New(srv.URL)
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo := landingRepo()
repo.landings["feat"] = true
repo.news = []Checkin{{Hash: "b4", Branch: "feat"}, {Hash: "b5", Branch: "feat"}}
if err := Trigger(e, st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if got := started(); !slices.Equal(got, []string{"b4 ci/quarry.json", "b5 ci/quarry.json"}) {
t.Errorf("FAIL: started %v; want the push pipeline at each check-in", got)
}
if got := cancelled(); !slices.Equal(got, []string{"b4 ci/quarry.json"}) {
t.Errorf("FAIL: cancelled %v; want the b4 runs superseded by b5", got)
}
t.Log("ok: each check-in runs, and the older runs on the branch are superseded")
}
// A ci.yml carrying quarry_when is refused in words: the key is read
// from JSON pipelines only, and silently running it everywhere would
// be the one outcome its author did not ask for. The refusal is that
// pipeline's alone: the check-in's other pipelines still start, and a
// wait the check-in was in is still ended.
func TestAYAMLPipelineCarryingQuarryWhenIsRefused(t *testing.T) {
srv, started := startedEngine(t)
e := New(srv.URL)
repo := &fakeRepo{files: []string{"ci/checks.yml", "ci/quarry.json"}, blobs: map[string][]byte{
"ci/checks.yml": []byte("quarry_when: landing\nsteps: []\n"),
"ci/quarry.json": []byte(`{"name":"quarry","version":"1.0","steps":[]}`),
}, landings: map[string]bool{}}
err := RunAt(e, &fakeStore{mark: map[string]string{}}, "alpha", repo, "http://x/r/alpha", "b1", "")
if err == nil || !strings.Contains(err.Error(), "quarry_when") {
t.Errorf("FAIL: a ci.yml with quarry_when answered %v; want a refusal naming the key", err)
}
if got := started(); !slices.Equal(got, []string{"b1 ci/quarry.json"}) {
t.Errorf("FAIL: beside the refused ci.yml, started %v; want ci/quarry.json", got)
}
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo.news = []Checkin{{Hash: "t1", Branch: "trunk"}}
if err := Trigger(e, st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if got := started(); !slices.Equal(got, []string{"t1 ci/quarry.json"}) || st.mark["alpha"] != "t1" {
t.Errorf("FAIL: a trunk check-in beside the refused ci.yml started %v, watermark %q; want ci/quarry.json and t1", got, st.mark["alpha"])
}
st.waiting = []Checkin{{Hash: "d1", Branch: "feat"}} // deferred earlier, now drained
repo.news = nil
if err := Trigger(e, st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if got := started(); !slices.Equal(got, []string{"d1 ci/quarry.json"}) || len(st.waiting) != 0 {
t.Errorf("FAIL: a drained check-in beside the refused ci.yml started %v, still waiting %v; want ci/quarry.json and no wait", got, st.waiting)
}
t.Log("ok: quarry_when in a ci.yml is refused alone; its siblings start")
}
func landingRepo() *fakeRepo {
return &fakeRepo{
files: []string{"ci/quarry.json", "ci/race.json"},
blobs: map[string][]byte{
"ci/quarry.json": []byte(`{"name":"quarry","version":"1.0","steps":[]}`),
"ci/race.json": []byte(`{"name":"race","version":"1.0","quarry_when":"landing","steps":[]}`),
},
landings: map[string]bool{},
}
}