package ci
import (
"encoding/json"
"errors"
"fmt"
"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,
"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
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":
var body struct {
Workflow string `json:"workflow"`
Input map[string]any `json:"input"`
}
json.NewDecoder(r.Body).Decode(&body)
gotInput = body.Input
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)
}
}
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's contract, and the old-binary fallback: an engine that
// answers the RUN document for /logs reads as "no logs yet", never as
// a parse error.
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":
fmt.Fprint(w, `[{"seq":1,"ts":"2026-08-29T01:00:00Z","stream":"out","data":"hi"},{"seq":2,"stream":"marker","data":"failed"}]`)
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)
if err != nil || len(chunks) != 2 || chunks[1].Stream != "marker" {
t.Fatalf("chunks: %v %v", chunks, err)
}
if _, err := e.Logs("x", "other", 0, false); !errors.Is(err, ErrNoLogs) {
t.Fatalf("old binary read as %v, want ErrNoLogs", err)
}
}