package ci
import (
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
)
func TestIsPipeline(t *testing.T) {
for name, want := range map[string]bool{
"ci/build.json": true,
"ci/test.json": true,
"ci/nested/thing.json": false, // a fixture, not a pipeline
"ci/README.md": false,
"ci/": false,
"ci/build.yml": true,
"ci/build.yaml": true,
"ci/nested/x.yml": false,
"ci/build.txt": false,
"cirrus/build.json": false,
"build.json": false,
"docs/ci/build.json": false,
} {
if got := IsPipeline(name); got != want {
t.Errorf("IsPipeline(%q) = %v, want %v", name, got, want)
}
}
}
// An unrecognised status must read as "still running". Reporting a run
// finished when it has not is the worse error: a verdict would be
// written against a build that is still going.
func TestUnknownStatusIsNotTerminal(t *testing.T) {
for _, s := range []string{"running", "queued", "", "paused", "something-new"} {
if (Run{Status: s}).Terminal() {
t.Errorf("status %q reported terminal", s)
}
}
for _, s := range []string{"completed", "failed", "cancelled", "error"} {
if !(Run{Status: s}).Terminal() {
t.Errorf("status %q not reported terminal", s)
}
}
if !(Run{Status: "completed"}).Passed() || (Run{Status: "failed"}).Passed() {
t.Fatal("Passed disagrees with the status")
}
}
// fakeRepo is a repository with one pipeline.
type fakeRepo struct {
files []string
blobs map[string][]byte
news []Checkin
}
func (f *fakeRepo) NewCheckins(after string, limit int) ([]Checkin, error) {
if after == "" && len(f.news) > 0 {
return f.news[len(f.news)-1:], nil
}
return f.news, nil
}
func (f *fakeRepo) Files(string) ([]string, error) { return f.files, nil }
func (f *fakeRepo) Read(p, _ string) ([]byte, error) { return f.blobs[p], nil }
type fakeStore struct {
mu sync.Mutex
mark map[string]string
runs []string
}
func (s *fakeStore) CIWatermark(repo string) (string, bool) {
s.mu.Lock()
defer s.mu.Unlock()
v, ok := s.mark[repo]
return v, ok
}
func (s *fakeStore) SetCIWatermark(repo, c string) error {
s.mu.Lock()
defer s.mu.Unlock()
s.mark[repo] = c
return nil
}
func (s *fakeStore) AddRun(runID, repo, checkin, workflow, pipeline, status string) error {
s.mu.Lock()
defer s.mu.Unlock()
s.runs = append(s.runs, runID+" "+checkin+" "+workflow)
return nil
}
// A repository seen for the first time must NOT launch its whole
// history — on an imported corpus that is tens of thousands of runs.
func TestFirstSightOnlySetsTheWatermark(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
t.Errorf("engine called on first sight: %s", r.URL.Path)
}))
defer srv.Close()
st := &fakeStore{mark: map[string]string{}}
repo := &fakeRepo{news: []Checkin{{Hash: "old"}, {Hash: "tip"}}}
if err := Trigger(New(srv.URL), st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if st.mark["alpha"] != "tip" {
t.Fatalf("watermark = %q, want tip", st.mark["alpha"])
}
if len(st.runs) != 0 {
t.Fatalf("launched %v on first sight", st.runs)
}
}
func TestLaunchesPipelineWithCheckinContext(t *testing.T) {
var gotInput map[string]any
var gotLabels map[string]string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/workflows":
w.WriteHeader(http.StatusCreated)
w.Write([]byte(`{"status":"registered"}`))
case "/runs":
if r.Method == http.MethodGet {
fmt.Fprint(w, `[]`) // the supersede sweep finds nothing
return
}
var body struct {
Workflow string `json:"workflow"`
Input map[string]any `json:"input"`
Labels map[string]string `json:"labels"`
}
json.NewDecoder(r.Body).Decode(&body)
gotInput = body.Input
gotLabels = body.Labels
w.WriteHeader(http.StatusCreated)
w.Write([]byte(`{"run_id":"r-1"}`))
}
}))
defer srv.Close()
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo := &fakeRepo{
files: []string{"README.md", "ci/build.json", "ci/notes.txt"},
blobs: map[string][]byte{"ci/build.json": []byte(`{"name":"build","version":"1.0","steps":[]}`)},
news: []Checkin{{Hash: "new1", Branch: "trunk", User: "dan"}},
}
if err := Trigger(New(srv.URL), st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if len(st.runs) != 1 || st.runs[0] != "r-1 new1 build" {
t.Fatalf("runs = %v", st.runs)
}
// The pipeline must be told what it is building and how to get it.
for _, k := range []string{"repo", "checkin", "branch", "cloneUrl", "pipeline"} {
if gotInput[k] == nil || gotInput[k] == "" {
t.Errorf("input missing %q: %+v", k, gotInput)
}
}
// The run must be findable — and cancellable — by what it was for.
for _, k := range []string{"repo", "checkin", "branch", "pipeline"} {
if gotLabels[k] == "" {
t.Errorf("labels missing %q: %+v", k, gotLabels)
}
}
if st.mark["alpha"] != "new1" {
t.Fatalf("watermark not advanced: %q", st.mark["alpha"])
}
}
// A repository with no ci/ directory is the common case and must be
// silent — no runs, no error, and the watermark still advances or every
// sync round would reconsider the same check-ins forever.
func TestNoPipelineIsQuietAndStillAdvances(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
t.Errorf("engine called for a repo with no pipelines: %s", r.URL.Path)
}))
defer srv.Close()
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo := &fakeRepo{files: []string{"README.md"}, news: []Checkin{{Hash: "new1"}}}
if err := Trigger(New(srv.URL), st, "alpha", repo, "u"); err != nil {
t.Fatal(err)
}
if st.mark["alpha"] != "new1" {
t.Fatalf("watermark stuck at %q", st.mark["alpha"])
}
}
// A disabled engine (no --dagnats) does nothing at all.
func TestDisabledEngineDoesNothing(t *testing.T) {
st := &fakeStore{mark: map[string]string{}}
if err := Trigger(New(""), st, "alpha", &fakeRepo{}, "u"); err != nil {
t.Fatal(err)
}
if len(st.mark) != 0 {
t.Fatalf("disabled engine touched the store: %v", st.mark)
}
}
// dagnats#624 as shipped: {chunks, next_cursor, eof} with base64 data;
// from=failure's strict 404 falls back to the whole step; and the
// old-binary catch-all (the RUN document) reads as "no logs yet".
func TestLogsParsesChunksAndOldBinaries(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case strings.Contains(r.URL.Path, "/logs") && r.URL.Query().Get("step") == "build":
if r.URL.Query().Get("from") == "failure" {
http.Error(w, `{"error":"attempt has no failure marker"}`, http.StatusNotFound)
return
}
fmt.Fprint(w, `{"chunks":[{"seq":1,"ts":"2026-08-29T01:00:00Z","stream":"out","data":"aGk="},{"seq":2,"stream":"marker","data":"ZmFpbGVk"}],"next_cursor":3,"eof":true}`)
case strings.Contains(r.URL.Path, "/logs"):
fmt.Fprint(w, `{"run_id":"x","status":"completed"}`) // pre-#624 catch-all
}
}))
defer srv.Close()
e := New(srv.URL)
chunks, err := e.Logs("x", "build", 0, true) // strict 404 -> plain retry
if err != nil || len(chunks) != 2 {
t.Fatalf("chunks: %v %v", chunks, err)
}
if chunks[0].Data != "hi" || chunks[1].Stream != "marker" || chunks[1].Data != "failed" {
t.Fatalf("base64 not decoded: %+v", chunks)
}
if _, err := e.Logs("x", "other", 0, false); !errors.Is(err, ErrNoLogs) {
t.Fatalf("old binary read as %v, want ErrNoLogs", err)
}
}
// A new check-in on a branch cancels the branch's still-running builds
// — and ONLY those. A run on another branch, a finished run, and a run
// from an engine that echoes no labels (pre-dagnats#629) are all left
// alone: label match is verified per run, never assumed from the query.
func TestSupersedeCancelsOnlyVerifiedRunningRuns(t *testing.T) {
var cancelled []string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case r.URL.Path == "/runs" && r.Method == http.MethodGet:
fmt.Fprint(w, `[
{"run_id":"r-stale","status":"running","labels":{"repo":"alpha","branch":"trunk","pipeline":"ci/build.json"}},
{"run_id":"r-other","status":"running","labels":{"repo":"alpha","branch":"side","pipeline":"ci/build.json"}},
{"run_id":"r-bare","status":"running"},
{"run_id":"r-done","status":"completed","labels":{"repo":"alpha","branch":"trunk","pipeline":"ci/build.json"}}]`)
case strings.HasSuffix(r.URL.Path, "/cancel"):
cancelled = append(cancelled, strings.TrimSuffix(strings.TrimPrefix(r.URL.Path, "/runs/"), "/cancel"))
}
}))
defer srv.Close()
n, err := New(srv.URL).Supersede("build", map[string]string{
"repo": "alpha", "branch": "trunk", "pipeline": "ci/build.json",
})
if err != nil || n != 1 {
t.Fatalf("superseded %d, %v", n, err)
}
if len(cancelled) != 1 || cancelled[0] != "r-stale" {
t.Fatalf("cancelled %v, want only r-stale", cancelled)
}
}
// A ci.yml pipeline goes through the engine's CI door (dagnats#633):
// compiled and registered in one call, never POST /workflows, and the
// run is named by the file stem.
func TestYmlPipelineCompilesAtTheEngine(t *testing.T) {
var gotName, gotSpec string
var gotRegister, jsonDoorUsed bool
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/v1/ci/compile":
var body struct {
Name string `json:"name"`
Spec string `json:"spec"`
Register bool `json:"register"`
}
json.NewDecoder(r.Body).Decode(&body)
gotName, gotSpec, gotRegister = body.Name, body.Spec, body.Register
fmt.Fprint(w, `{"workflow":{"name":"check"},"def_hash":"x","registered":true}`)
case "/workflows":
jsonDoorUsed = true
case "/runs":
if r.Method == http.MethodGet {
fmt.Fprint(w, `[]`)
return
}
w.WriteHeader(http.StatusCreated)
w.Write([]byte(`{"run_id":"r-y"}`))
}
}))
defer srv.Close()
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo := &fakeRepo{
files: []string{"ci/check.yml"},
blobs: map[string][]byte{"ci/check.yml": []byte("checks:\n test: { call: \"test\" }\n")},
news: []Checkin{{Hash: "new1", Branch: "trunk"}},
}
if err := Trigger(New(srv.URL), st, "alpha", repo, "u"); err != nil {
t.Fatal(err)
}
if gotName != "check" || !gotRegister || !strings.Contains(gotSpec, "checks:") {
t.Fatalf("compile request: name=%q register=%v spec=%q", gotName, gotRegister, gotSpec)
}
if jsonDoorUsed {
t.Fatal("a yml pipeline went through POST /workflows")
}
if len(st.runs) != 1 || st.runs[0] != "r-y new1 check" {
t.Fatalf("runs = %v", st.runs)
}
}
// The engine is the schema authority: its diagnostics ARE the error,
// and an engine predating #633 is named, not mistaken for a bad spec.
func TestYmlDiagnosticsComeBackWorded(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if strings.Contains(r.URL.Path, "compile") {
w.WriteHeader(http.StatusUnprocessableEntity)
fmt.Fprint(w, `{"diagnostics":[{"line":3,"message":"unknown needs: lint"},{"message":"spec has no checks and no deploy step"}]}`)
}
}))
defer srv.Close()
_, err := New(srv.URL).RegisterYAML("bad", []byte("x"))
if err == nil || !strings.Contains(err.Error(), "line 3: unknown needs: lint") || !strings.Contains(err.Error(), "no deploy step") {
t.Fatalf("diagnostics lost: %v", err)
}
old := httptest.NewServer(http.NotFoundHandler())
defer old.Close()
if _, err := New(old.URL).RegisterYAML("x", []byte("y")); err == nil || !strings.Contains(err.Error(), "dagnats#633") {
t.Fatalf("pre-#633 engine not named: %v", err)
}
}
// dagnats#627's admin-gated token routes: every call carries the admin
// bearer; the engine's two refusals — no credential configured there,
// wrong credential here — come back in words a settings page can show.
func TestWorkerTokenRoutesCarryTheAdminBearer(t *testing.T) {
var auths []string
var mintBody, revoked string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
auths = append(auths, r.Header.Get("Authorization"))
switch {
case r.Method == http.MethodPost:
b, _ := io.ReadAll(r.Body)
mintBody = string(b)
w.WriteHeader(http.StatusCreated)
fmt.Fprint(w, `{"id":"t1","token":"secret-once","label":"blacksmith","task_type_prefixes":["e2e-"],"created_at":"2026-08-29T01:00:00Z"}`)
case r.Method == http.MethodDelete:
revoked = r.URL.Path
w.WriteHeader(http.StatusNoContent)
default:
fmt.Fprint(w, `{"tokens":[{"id":"t1","label":"blacksmith","task_type_prefixes":["e2e-"],"created_at":"2026-08-29T01:00:00Z"}]}`)
}
}))
defer srv.Close()
e := New(srv.URL)
e.SetAdminToken("adm")
tok, err := e.MintWorkerToken("blacksmith", []string{"e2e-"})
if err != nil || tok.Token != "secret-once" {
t.Fatalf("mint: %+v %v", tok, err)
}
if !strings.Contains(mintBody, `"task_type_prefixes":["e2e-"]`) {
t.Fatalf("mint body: %s", mintBody)
}
ts, err := e.WorkerTokens()
if err != nil || len(ts) != 1 || ts[0].Token != "" {
t.Fatalf("list leaked or failed: %+v %v", ts, err)
}
if err := e.RevokeWorkerToken("t1"); err != nil || revoked != "/v1/tokens/t1" {
t.Fatalf("revoke: %v %q", err, revoked)
}
for _, a := range auths {
if a != "Bearer adm" {
t.Fatalf("a call went out without the admin bearer: %q", auths)
}
}
}
func TestWorkerTokenRefusalsAreWorded(t *testing.T) {
noCfg := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Error(w, "token management requires DAGNATS_BRIDGE_TOKEN", http.StatusServiceUnavailable)
}))
defer noCfg.Close()
if _, err := New(noCfg.URL).WorkerTokens(); err == nil || !strings.Contains(err.Error(), "no admin credential") {
t.Fatalf("503 not worded: %v", err)
}
wrong := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Error(w, "unauthorized", http.StatusUnauthorized)
}))
defer wrong.Close()
e := New(wrong.URL)
e.SetAdminToken("bad")
if _, err := e.WorkerTokens(); err == nil || !strings.Contains(err.Error(), "refused quarry's admin token") {
t.Fatalf("401 not worded: %v", err)
}
}