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)
}
})
}
}
// 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", ci.DefHash([]byte(`{"name":"test","version":"1.0","steps":[]}`)), "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")
// The inherited verdict is tagged on the merge as a run's would be,
// once, however many passes the poller makes.
repos := api.NewRepos(map[string]*api.Repo{"r": {Index: ix, Zit: cli}})
for range 2 {
pollCI(engine, db, repos, true, t.TempDir(), nil, nil)
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
}
if v, err := ix.VerdictOf(merged); err != nil || v != "ci-pass" {
t.Fatalf("FAIL: the merge's verdict tag is %q, %v; want ci-pass", v, err)
}
history, err := ix.TagHistory(merged)
if err != nil {
t.Fatal(err)
}
n := 0
for _, e := range history {
if e.Name == "ci-pass" || e.Name == "ci-fail" {
n++
}
}
if n != 1 {
t.Fatalf("FAIL: the merge carries %d verdict tags; want one", n)
}
t.Log("ok: the inherited pass is tagged on the merge, once")
}
// A LANDING CLOSING CANCELS ONLY THE MERGE RUNS ITS SWEEP STARTED, AND
// NEVER LEAVES ITS TIP WITHOUT A VERDICT.
//
// A merge run a push made is the push's (subject push-pair): it stays.
// One whose labels cannot be read is not known to be the landing's, and
// stays too. If, with the landing's runs cancelled, the tip has nothing
// that passed, failed or is still going, it is tested as itself.
func TestALandingClosingKeepsItsTipAnswered(t *testing.T) {
for name, tc := range map[string]struct {
pushStatus string // "" is no push-made run
unknown bool // a run whose labels the engine cannot show
wantTip bool
}{
"only the landing's run": {"", false, true},
"a push run still going": {"running", false, false},
"a push run that passed": {"completed", false, false},
"a push run cancelled": {"cancelled", false, true},
"a run of unknown kind": {"", true, false},
} {
t.Run(name, func(t *testing.T) {
dir, cli, ix, pipe, id := onceStore(t, true)
rg := newTestRegistrar(t, 25)
engine, started, cancelled := pipelineEngine(t, map[string]map[string]string{
"lp": {"subject": "pair", "branch": "feat"},
"pp": {"subject": "push-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}
if err := st.AddRun("lp", "r", ci.Pair(feat, root), "w", "ci/test.json", "", "running"); err != nil {
t.Fatal(err)
}
if tc.unknown {
if err := st.AddRun("unknown", "r", ci.Pair(feat, root), "w", "ci/race.json", "", "running"); err != nil {
t.Fatal(err)
}
}
if tc.pushStatus != "" {
if err := st.AddRun("pp", "r", ci.Pair(feat, root), "w", "ci/test.json", "", "running"); err != nil {
t.Fatal(err)
}
if err := rg.odb.SetCIStatus("pp", tc.pushStatus); err != nil {
t.Fatal(err)
}
}
rg.landingSweep("r", ix) // seeds the open set
started()
zitIn(t, dir, "landing", "reject", id, "--user", "dan")
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
rg.landingSweep("r", ix)
if got := cancelled(); !slices.Equal(got, []string{"lp"}) {
t.Fatalf("FAIL: closing cancelled %v; want the landing's own run lp only", got)
}
got := started()
if tc.wantTip && !slices.Equal(got, []string{feat + " ci/test.json"}) {
t.Fatalf("FAIL: the tip left with no verdict started %v; want its tip tested", got)
}
if !tc.wantTip && len(got) != 0 {
t.Fatalf("FAIL: a tip with a push verdict started %v", got)
}
})
}
}
// A PAIR TESTS WHAT WOULD LAND; TRUNK INHERITS ONLY THE SAME DEFINITION.
//
// Through the real store: trunk changes ci/test.json after feat forked.
// The proposed branch's push runs the merged tree's ci/test.json, the
// one trunk's merge will hold, and records it. A pair run that ran the
// branch's older definition (recorded before pairs read the merged
// tree) is no verdict for trunk's merge, though the trees are equal.
func TestAPairRunsTheMergedDefinition(t *testing.T) {
dir, cli, ix, pipe, _ := onceStore(t, true)
feat, _ := ix.BranchTip("feat")
zitIn(t, dir, "update", "trunk")
changed := `{"name":"test","version":"2.0","steps":[]}`
if err := os.WriteFile(filepath.Join(dir, "ci/test.json"), []byte(changed), 0o644); err != nil {
t.Fatal(err)
}
zitIn(t, dir, "commit", "-m", "trunk changes its tests", "--user", "u")
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
moved, _ := 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}
engine, started, _ := pipelineEngine(t, nil)
if err := ci.RunTip(engine, st, "r", ciRepo{ix: ix, zit: cli}, "http://x/r/r", ci.Checkin{Hash: feat, Branch: "feat"}); err != nil {
t.Fatal(err)
}
pair := ci.Pair(feat, moved)
if got := started(); !slices.Contains(got, pair+" ci/test.json") {
t.Fatalf("FAIL: the proposed push started %v; want its merge with trunk", got)
}
runs, _ := db.CIRunsForPair("r", pair)
for _, r := range runs {
if r.Pipeline == "ci/test.json" && r.DefHash != ci.DefHash([]byte(changed)) {
t.Fatalf("FAIL: the merge ran ci/test.json %s; want trunk's changed definition", r.DefHash)
}
}
t.Log("ok: the merge ran the definition trunk will hold")
// The same pair as recorded by an older forge: the branch's file.
stale := ci.Pair(feat, moved)
if err := db.CompleteCIRun(runs[0].RunID, "cancelled", ""); err != nil {
t.Fatal(err)
}
if err := st.AddRun("old", "r", stale, "w", "ci/test.json", ci.DefHash([]byte(`{"name":"test","version":"1.0","steps":[]}`)), "running"); err != nil {
t.Fatal(err)
}
if err := db.SetCIStatus("old", "completed"); err != nil {
t.Fatal(err)
}
for _, r := range runs[1:] {
if err := db.CompleteCIRun(r.RunID, "cancelled", ""); err != nil {
t.Fatal(err)
}
}
if err := st.SetCIWatermark("r", moved); err != nil {
t.Fatal(err)
}
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")
if err := ci.Trigger(engine, st, "r", ciRepo{ix: ix, zit: cli}, "http://x/r/r"); err != nil {
t.Fatal(err)
}
if got := started(); !slices.Contains(got, merged+" ci/test.json") {
t.Fatalf("FAIL: trunk's merge started %v; want ci/test.json run, the pair ran another definition", got)
}
t.Log("ok: a verdict on another definition is not inherited")
}
// A PUSH'S MERGE VERDICT IS ITS TIP'S BADGE (RFC 0009 §once).
//
// A proposed branch's push is tested as its merge, so the verdict tag
// that marks the tip in the timeline comes from the push's merge runs,
// as it came from the tip's own runs before. A merge the landing sweep
// ran is the landing's question and tags nothing, as before.
func TestAPushedMergeTagsItsTip(t *testing.T) {
for _, subject := range []string{"push-pair", "pair"} {
t.Run(subject, func(t *testing.T) {
_, cli, ix, pipe, _ := onceStore(t, true)
feat, _ := ix.BranchTip("feat")
root, _ := ix.BranchTip("trunk")
dagnats := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
json.NewEncoder(w).Encode(ci.Run{RunID: "m1", Status: "failed",
CompletedAt: "2026-10-01T12:00:00Z", Labels: map[string]string{"subject": subject}})
}))
t.Cleanup(dagnats.Close)
db, err := oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { db.Close() })
if err := (ciStore{db: db}).AddRun("m1", "r", ci.Pair(feat, root), "w", "ci/test.json", "", "running"); err != nil {
t.Fatal(err)
}
repos := api.NewRepos(map[string]*api.Repo{"r": {Index: ix, Zit: cli}})
for range 2 {
pollCI(ci.New(dagnats.URL), db, repos, true, t.TempDir(), nil, nil)
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
}
v, err := ix.VerdictOf(feat)
if err != nil {
t.Fatal(err)
}
want := "ci-fail"
if subject == "pair" {
want = ""
}
if v != want {
t.Fatalf("FAIL: a %s run's tip carries verdict %q; want %q", subject, v, want)
}
})
}
}
// ONE MERGE, ONE VERDICT TAG PER VERDICT.
//
// The carry-forward may already have tagged a landing's merge with the
// premerge verdict, and two pipelines may inherit the same one: the
// merge still carries ci-pass once, never a pile of equal tags.
func TestInheritedVerdictsDoNotRepeatATag(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}
for p, def := range map[string]string{
"ci/test.json": `{"name":"test","version":"1.0","steps":[]}`,
"ci/race.json": `{"name":"race","version":"1.0","quarry_when":"landing","steps":[]}`,
} {
if err := st.AddRun("p-"+p, "r", ci.Pair(feat, root), "w", p, ci.DefHash([]byte(def)), "running"); err != nil {
t.Fatal(err)
}
if err := db.SetCIStatus("p-"+p, "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 := started(); len(got) != 0 {
t.Fatalf("FAIL: trunk's merge started %v; want both inherited", got)
}
repo := &api.Repo{Index: ix, Zit: cli}
if err := repo.WriteTag("ci-pass", merged, "premerge"); err != nil { // the carry-forward's
t.Fatal(err)
}
repos := api.NewRepos(map[string]*api.Repo{"r": repo})
for range 2 {
pollCI(engine, db, repos, true, t.TempDir(), nil, nil)
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
}
history, err := ix.TagHistory(merged)
if err != nil {
t.Fatal(err)
}
n := 0
for _, e := range history {
if e.Name == "ci-pass" || e.Name == "ci-fail" {
n++
}
}
if n != 1 {
t.Fatalf("FAIL: the merge carries %d verdict tags; want the one ci-pass", n)
}
if owed, _ := db.InheritedOwingVerdict(); len(owed) != 0 {
t.Fatalf("FAIL: inherited verdicts still owed: %+v", owed)
}
t.Log("ok: one ci-pass, however many said so")
}