package main
import (
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"os"
"os/exec"
"path/filepath"
"slices"
"strings"
"sync"
"testing"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/ci"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/ingest"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/zitcli"
)
// A RUN'S VERDICT IS REUSED; ANYTHING THAT IS NO VERDICT IS NOT.
//
// The pipelines a merge already has an answer for, or will have, are
// the ones a landing opening over it does not start again: passed,
// failed, still going. An error, a cancelled run and a deferred one
// are no answer, and run again. An inherited verdict reads as its
// check-in's own run, naming the run it came from, and a run's note
// is its check's summary.
func TestTheStoreReusesVerdictsOnly(t *testing.T) {
db, err := oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { db.Close() })
st := ciStore{db: db}
pair := ci.Pair("b", "t")
for i, status := range []string{"completed", "failed", "running", "error", "cancelled", "deferred"} {
if err := st.AddRun(fmt.Sprintf("r%d", i), "alpha", pair, "w", "ci/"+status+".json", "running"); err != nil {
t.Fatal(err)
}
if err := db.SetCIStatus(fmt.Sprintf("r%d", i), status); err != nil {
t.Fatal(err)
}
}
got, err := st.RunPipelines("alpha", pair)
if err != nil {
t.Fatal(err)
}
slices.Sort(got)
if want := []string{"ci/completed.json", "ci/failed.json", "ci/running.json"}; !slices.Equal(got, want) {
t.Fatalf("FAIL: reusable pipelines %v; want %v", got, want)
}
recs, err := st.RunsOf("alpha", pair)
if err != nil || len(recs) != 6 || recs[0].RunID != "r5" {
t.Fatalf("FAIL: RunsOf = %+v, %v; want all six, newest first", recs, err)
}
if err := st.Inherit("alpha", "m1", "w", "ci/completed.json", "completed", "r0"); err != nil {
t.Fatal(err)
}
if err := st.Note("r1", "tested as the branch tip"); err != nil {
t.Fatal(err)
}
runs, err := ciRuns{db: db}.CIRunsFor("alpha", "m1")
if err != nil || len(runs) != 1 || runs[0].RunID != "r0" || runs[0].InheritedFrom != "r0" || runs[0].Status != "completed" {
t.Fatalf("FAIL: m1's checks %+v, %v; want r0's verdict, inherited", runs, err)
}
runs, _ = ciRuns{db: db}.CIRunsForPair("alpha", pair)
for _, r := range runs {
if r.RunID == "r1" && r.Summary != "tested as the branch tip" {
t.Fatalf("FAIL: the noted run's summary is %q", r.Summary)
}
}
t.Log("ok: verdicts reused, non-answers re-run; inherited and noted runs read back")
}
// pipelineEngine records "checkin pipeline" for every run it starts,
// lists none running, and answers a run's labels from labels.
func pipelineEngine(t *testing.T, labels map[string]map[string]string) (*ci.Engine, func() []string, func() []string) {
t.Helper()
var mu sync.Mutex
var started, cancelled []string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
switch {
case r.URL.Path == "/workflows":
w.WriteHeader(http.StatusCreated)
w.Write([]byte(`{"status":"registered"}`))
case r.URL.Path == "/runs" && r.Method == http.MethodPost:
var body struct {
Input map[string]any `json:"input"`
}
json.NewDecoder(r.Body).Decode(&body)
started = append(started, fmt.Sprint(body.Input["checkin"], " ", body.Input["pipeline"]))
w.WriteHeader(http.StatusCreated)
fmt.Fprintf(w, `{"run_id":"r-%d"}`, len(started))
case r.URL.Path == "/runs":
w.Write([]byte(`[]`))
case strings.HasSuffix(r.URL.Path, "/cancel"):
cancelled = append(cancelled, strings.TrimSuffix(strings.TrimPrefix(r.URL.Path, "/runs/"), "/cancel"))
case strings.HasPrefix(r.URL.Path, "/runs/"):
id := strings.TrimPrefix(r.URL.Path, "/runs/")
l, ok := labels[id]
if !ok {
http.NotFound(w, r)
return
}
json.NewEncoder(w).Encode(ci.Run{RunID: id, Status: "running", Labels: l})
default:
http.NotFound(w, r)
}
}))
t.Cleanup(srv.Close)
take := func(l *[]string) func() []string {
return func() []string {
mu.Lock()
defer mu.Unlock()
out := append([]string(nil), (*l)...)
*l = nil
slices.Sort(out)
return out
}
}
return ci.New(srv.URL), take(&started), take(&cancelled)
}
// onceRepo is a store with a tested pipeline and a landing-only one,
// trunk at root and feat one check-in ahead of it, proposed or not.
func onceStore(t *testing.T, propose bool) (dir string, cli *zitcli.Client, ix *index.Index, pipe *ingest.Pipeline, landing string) {
t.Helper()
if _, err := exec.LookPath("zit"); err != nil {
t.Skip("zit not on PATH")
}
dir = t.TempDir()
write := func(name, body string) {
t.Helper()
os.MkdirAll(filepath.Dir(filepath.Join(dir, name)), 0o755)
if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0o644); err != nil {
t.Fatal(err)
}
}
zitIn(t, dir, "init", ".")
write("ci/test.json", `{"name":"test","version":"1.0","steps":[]}`)
write("ci/race.json", `{"name":"race","version":"1.0","quarry_when":"landing","steps":[]}`)
write("f.txt", "root\n")
zitIn(t, dir, "add", "ci/test.json", "ci/race.json", "f.txt")
zitIn(t, dir, "commit", "-m", "root", "--user", "u")
write("g.txt", "feat\n")
zitIn(t, dir, "add", "g.txt")
zitIn(t, dir, "commit", "-m", "feat", "--branch", "feat", "--user", "u")
cli = zitcli.New(dir)
if propose {
id, err := cli.LandingOpen(lease.Unguarded("test"), "dan", "feat", "trunk", "land feat", "")
if err != nil {
t.Fatal(err)
}
landing = id
}
ev, err := events.Start(filepath.Join(t.TempDir(), "nats"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
ix, err = index.Open(filepath.Join(t.TempDir(), "index.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { ix.Close() })
pipe = ingest.NewPipeline(cli, ev, ix, "r", "n1")
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
return dir, cli, ix, pipe, landing
}
// A LANDING OPENING OVER A MERGE ITS PUSH TESTED REUSES THE VERDICTS.
//
// What the pushed merge passed or failed is not run again; the
// landing-only pipeline, and one whose run errored, run once. A target
// that has moved since is a new merge, tested whole. A landing that is
// not opening starts nothing over a merge already tested.
func TestAnOpeningLandingReusesThePushedMerge(t *testing.T) {
for _, status := range []string{"completed", "failed", "error", "cancelled", "deferred"} {
t.Run(status, func(t *testing.T) {
dir, cli, ix, pipe, _ := onceStore(t, true)
rg := newTestRegistrar(t, 25)
engine, started, _ := pipelineEngine(t, nil)
rg.engine = engine
rg.deps = &api.Deps{Repos: api.NewRepos(map[string]*api.Repo{"r": {Index: ix, Zit: cli}})}
feat, _ := ix.BranchTip("feat")
root, _ := ix.BranchTip("trunk")
pair := ci.Pair(feat, root)
st := ciStore{db: rg.odb}
if err := st.AddRun("pushed", "r", pair, "w", "ci/test.json", "running"); err != nil {
t.Fatal(err)
}
if err := rg.odb.SetCIStatus("pushed", status); err != nil {
t.Fatal(err)
}
open, err := ix.Landings("open", 500)
if err != nil || len(open) != 1 {
t.Fatalf("open landings %v, %v", open, err)
}
rg.pairSweep("r", ix, open, nil)
if got := started(); len(got) != 0 {
t.Fatalf("FAIL: a landing not opening started %v over a tested merge", got)
}
rg.pairSweep("r", ix, open, map[string]bool{open[0].ID: true})
want := []string{pair + " ci/race.json"}
if status != "completed" && status != "failed" {
want = append(want, pair+" ci/test.json")
}
slices.Sort(want)
if got := started(); !slices.Equal(got, want) {
t.Fatalf("FAIL: opening over a %s run started %v; want %v", status, got, want)
}
zitIn(t, dir, "update", "trunk")
if err := os.WriteFile(filepath.Join(dir, "f.txt"), []byte("moved\n"), 0o644); err != nil {
t.Fatal(err)
}
zitIn(t, dir, "commit", "-m", "trunk moves", "--user", "u")
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
moved, _ := ix.BranchTip("trunk")
rg.pairSweep("r", ix, open, map[string]bool{open[0].ID: true})
want = []string{ci.Pair(feat, moved) + " ci/race.json", ci.Pair(feat, moved) + " ci/test.json"}
if got := started(); !slices.Equal(got, want) {
t.Fatalf("FAIL: opening against a moved target started %v; want %v", got, want)
}
})
}
}
// A PUSH'S MERGE WITH NO LANDING IS NOT A LANDING'S TO CANCEL.
//
// A landing closing stops its merge runs; a run of the same tip's merge
// that a push made while nothing was proposed is keyed apart
// (branch-pair), and only the branch's next push supersedes it.
func TestALandingClosingSparesABranchPair(t *testing.T) {
_, cli, ix, _, _ := onceStore(t, false)
rg := newTestRegistrar(t, 25)
engine, _, cancelled := pipelineEngine(t, map[string]map[string]string{
"lp": {"subject": "pair", "branch": "feat"},
"bp": {"subject": "branch-pair", "branch": "feat"},
})
rg.engine = engine
rg.deps = &api.Deps{Repos: api.NewRepos(map[string]*api.Repo{"r": {Index: ix, Zit: cli}})}
feat, _ := ix.BranchTip("feat")
root, _ := ix.BranchTip("trunk")
st := ciStore{db: rg.odb}
for _, id := range []string{"lp", "bp"} {
if err := st.AddRun(id, "r", ci.Pair(feat, root), "w", "ci/"+id+".json", "running"); err != nil {
t.Fatal(err)
}
}
rg.cancelClosedPairs("r", ix, "feat")
if got := cancelled(); !slices.Equal(got, []string{"lp"}) {
t.Fatalf("FAIL: a landing closing cancelled %v; want its own merge run lp only", got)
}
t.Log("ok: the landing's merge run cancelled, the push's branch-pair spared")
}
// TRUNK'S MERGE OF A TESTED PAIR INHERITS ITS VERDICT, END TO END.
//
// Through the real store: the merge zit writes from the two tips is
// the pair's tree, so the pipeline the pair passed is not run again
// and reads as the merge's own check, naming the pair's run; the
// landing-only pipeline the pair never ran, runs.
func TestTrunksMergeInheritsItsPair(t *testing.T) {
dir, cli, ix, pipe, _ := onceStore(t, false)
feat, _ := ix.BranchTip("feat")
root, _ := ix.BranchTip("trunk")
db, err := oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { db.Close() })
st := ciStore{db: db}
if err := st.AddRun("p-test", "r", ci.Pair(feat, root), "w", "ci/test.json", "running"); err != nil {
t.Fatal(err)
}
if err := db.SetCIStatus("p-test", "completed"); err != nil {
t.Fatal(err)
}
if err := st.SetCIWatermark("r", feat); err != nil {
t.Fatal(err)
}
zitIn(t, dir, "update", "trunk")
zitIn(t, dir, "merge", "feat")
zitIn(t, dir, "commit", "-m", "land feat", "--user", "u")
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
merged, _ := ix.BranchTip("trunk")
engine, started, _ := pipelineEngine(t, nil)
if err := ci.Trigger(engine, st, "r", ciRepo{ix: ix, zit: cli}, "http://x/r/r"); err != nil {
t.Fatal(err)
}
if got, want := started(), []string{merged + " ci/race.json"}; !slices.Equal(got, want) {
t.Fatalf("FAIL: trunk's merge started %v; want %v", got, want)
}
runs, err := ciRuns{db: db}.CIRunsFor("r", merged)
if err != nil {
t.Fatal(err)
}
var inherited []string
for _, r := range runs {
if r.InheritedFrom != "" {
inherited = append(inherited, r.Pipeline+" "+r.RunID+" "+r.Status)
}
}
if want := []string{"ci/test.json p-test completed"}; !slices.Equal(inherited, want) {
t.Fatalf("FAIL: the merge's inherited checks %v; want %v", inherited, want)
}
t.Log("ok: the pair's pass is the merge's; what the pair did not run, runs")
}