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
landings map[string]bool
defaultBranch string // "" is trunk
}
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 }
func (f *fakeRepo) HasOpenLanding(branch string) bool { return f.landings[branch] }
func (f *fakeRepo) DefaultBranch() string {
if f.defaultBranch == "" {
return "trunk"
}
return f.defaultBranch
}
type fakeStore struct {
mu sync.Mutex
mark map[string]string
runs []string
waiting []Checkin
ran map[string][]string // checkin -> pipelines with a run that has (or will have) a verdict
}
func (s *fakeStore) Defer(_ string, c Checkin) error {
s.mu.Lock()
defer s.mu.Unlock()
for _, w := range s.waiting {
if w.Hash == c.Hash {
return nil
}
}
s.waiting = append(s.waiting, c)
return nil
}
func (s *fakeStore) Deferred(string) ([]Checkin, error) {
s.mu.Lock()
defer s.mu.Unlock()
return append([]Checkin(nil), s.waiting...), nil
}
func (s *fakeStore) Undefer(_, checkin string) error {
s.mu.Lock()
defer s.mu.Unlock()
kept := s.waiting[:0]
for _, w := range s.waiting {
if w.Hash != checkin {
kept = append(kept, w)
}
}
s.waiting = kept
return nil
}
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)
if s.ran == nil {
s.ran = map[string][]string{}
}
s.ran[checkin] = append(s.ran[checkin], pipeline)
return nil
}
func (s *fakeStore) RunPipelines(_, checkin string) ([]string, error) {
s.mu.Lock()
defer s.mu.Unlock()
return append([]string(nil), s.ran[checkin]...), 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)
}
// The workflow is named for the repository as well as the file, so
// two repositories with the same pipeline name stay apart on a
// shared engine. Asserted through ScopedName rather than as a
// literal: pinning the spelling would pin the scheme.
if len(st.runs) != 1 || st.runs[0] != "r-1 new1 "+ScopedName("alpha", "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, 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, 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 != ScopedName("alpha", "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 "+ScopedName("alpha", "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-"}, nil)
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)
}
}
// 2026-08-29's outage, pinned: a store that answers "no files" is a
// parser fault, not a repository without CI — the watermark must hold
// so nothing is lost, and the engine must not even be asked.
func TestEmptyListingHoldsTheWatermark(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
t.Errorf("engine consulted on a faulty listing: %s", r.URL.Path)
}))
defer srv.Close()
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo := &fakeRepo{files: nil, news: []Checkin{{Hash: "new1", Branch: "trunk"}}}
if err := Trigger(New(srv.URL), st, "alpha", repo, "u"); err != nil {
t.Fatal(err)
}
if st.mark["alpha"] != "old" {
t.Fatalf("watermark advanced past a faulty listing: %q", st.mark["alpha"])
}
}
// An unreachable engine is a bad moment, not a bad check-in: hold the
// watermark and retry the same check-in next round.
func TestEngineDownHoldsTheWatermark(t *testing.T) {
dead := httptest.NewServer(http.NotFoundHandler())
deadURL := dead.URL
dead.Close() // the address now refuses connections
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo := &fakeRepo{
files: []string{"ci/build.json"},
blobs: map[string][]byte{"ci/build.json": []byte(`{"name":"build","version":"1.0","steps":[]}`)},
news: []Checkin{{Hash: "new1", Branch: "trunk"}},
}
if err := Trigger(New(deadURL), st, "alpha", repo, "u"); err != nil {
t.Fatal(err)
}
if st.mark["alpha"] != "old" {
t.Fatalf("watermark advanced past an unreachable engine: %q", st.mark["alpha"])
}
if len(st.runs) != 0 {
t.Fatalf("runs recorded with no engine: %v", st.runs)
}
}
// A pipeline the engine refused is a content fault: advancing is right,
// or a bad definition would wedge the watermark forever.
func TestRefusedPipelineStillAdvances(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/workflows" {
http.Error(w, "no", http.StatusBadRequest)
}
}))
defer srv.Close()
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo := &fakeRepo{
files: []string{"ci/build.json"},
blobs: map[string][]byte{"ci/build.json": []byte(`{"name":"build","version":"1.0","steps":[]}`)},
news: []Checkin{{Hash: "new1", Branch: "trunk"}},
}
if err := Trigger(New(srv.URL), st, "alpha", repo, "u"); err != nil {
t.Fatal(err)
}
if st.mark["alpha"] != "new1" {
t.Fatalf("watermark wedged on a refused pipeline: %q", st.mark["alpha"])
}
}
// A repository's worker group must be ONE subject token. The subject is
// task.{task}.{group}, so a dotted group is the ambiguity that let a
// token holding the dotted task type "a.b" drain group "b" of type "a"
// (dagnats#695). Repository names may contain dots, so this is reachable
// from an ordinary repository name, not a contrived one.
func TestGroupForIsAlwaysOneSubjectToken(t *testing.T) {
for _, name := range []string{"quarry", "orchard", "my.repo", "a.b", "ab", "x y", "we*rd", "deep.a.b.c"} {
if got := GroupFor(name); strings.ContainsAny(got, ".*> /") {
t.Errorf("FAIL GroupFor(%q) = %q, which is not one subject token", name, got)
} else {
t.Logf("ok %q -> %q", name, got)
}
}
}
// dagnats encodes a group as task.{task}.={group}.{runID} (#704): "=" is
// the sentinel that separates a task type from a group, chosen because
// it is legal in subject tokens, consumer names and KV keys while being
// unreachable from user input. A repository name is user input, and a
// repository is named by whoever creates it — so the fold must keep "="
// out, or a repository could spell a group boundary of its own.
func TestGroupForCannotSpellTheGroupSentinel(t *testing.T) {
for _, name := range []string{"=", "a=b", "==", "evil=group", "a.=b"} {
got := GroupFor(name)
if strings.Contains(got, "=") {
t.Errorf("FAIL GroupFor(%q) = %q, which carries the group sentinel", name, got)
} else {
t.Logf("ok %q -> %q", name, got)
}
}
}
// Two names that fold to the same safe characters must not land in one
// queue: "a.b" and "a-b" are different repositories and their runners
// must not see each other's work.
func TestFoldedGroupsDoNotCollide(t *testing.T) {
if a, b := GroupFor("a.b"), GroupFor("a-b"); a == b {
t.Errorf("FAIL GroupFor(a.b) == GroupFor(a-b) == %q", a)
} else {
t.Logf("ok %q != %q", a, b)
}
if got := GroupFor("orchard"); got != "orchard" {
t.Errorf("FAIL GroupFor(orchard) = %q, want it unchanged", got)
} else {
t.Log("ok an ordinary name is its own group")
}
}
// Absent means "unscoped by group" and must stay absent: a
// present-but-empty list is refused at mint, so sending one would turn
// every fleet-wide mint into an error.
func TestMintOmitsWorkerGroupsWhenThereAreNone(t *testing.T) {
var body string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
b, _ := io.ReadAll(r.Body)
body = string(b)
fmt.Fprint(w, `{"id":"t1","token":"once","label":"l","task_type_prefixes":[]}`)
}))
defer srv.Close()
e := New(srv.URL)
e.SetAdminToken("adm")
if _, err := e.MintWorkerToken("l", nil, nil); err != nil {
t.Fatal(err)
}
if strings.Contains(body, "worker_groups") {
t.Errorf("FAIL empty groups were sent: %s", body)
} else {
t.Log("ok no groups, no worker_groups key")
}
if _, err := e.MintWorkerToken("l", nil, []string{"orchard"}); err != nil {
t.Fatal(err)
}
if !strings.Contains(body, `"worker_groups":["orchard"]`) {
t.Errorf("FAIL group not sent: %s", body)
} else {
t.Log("ok a named group is sent")
}
}
// The runner's manifest sits beside the pipelines and is JSON, but it is
// not a workflow: counting it as one failed every trigger (f24d78d9).
func TestTheRunnerManifestIsNotAPipeline(t *testing.T) {
if IsPipeline("ci/runner.json") {
t.Error("FAIL: ci/runner.json was taken for a pipeline")
}
for _, p := range []string{"ci/quarry.json", "ci/build.yml", "ci/x.yaml"} {
if !IsPipeline(p) {
t.Errorf("FAIL: %s is a pipeline", p)
}
}
}
// #283: how many attempts each step may take, read from the workflow
// definition the RUN started under and resolved the way dagnats
// resolves it — the step's retry, then the workflow default, then the
// legacy retries count — as max_attempts retries plus the first try.
// Read once and kept briefly: the run page asks on every render.
func TestStepLimitsResolveEachStepsPolicy(t *testing.T) {
var gets int
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != "GET" || r.URL.Path != "/workflows" {
http.NotFound(w, r)
return
}
gets++
fmt.Fprint(w, `[
{"name":"quarry","def_hash":"q1","steps":[{"id":"test","retry":{"max_attempts":4}},{"id":"fmt","retries":2},{"id":"once"}]},
{"name":"other","def_hash":"o1","default_retry":{"max_attempts":1},"steps":[{"id":"a"},{"id":"b","retry":{"max_attempts":0}}]}]`)
}))
defer srv.Close()
e := New(srv.URL)
got, err := e.StepLimits("quarry", "q1")
if err != nil {
t.Fatal(err)
}
if got["test"] != 5 || got["fmt"] != 3 || got["once"] != 1 {
t.Fatalf("quarry limits = %v, want test 5, fmt 3, once 1", got)
}
other, err := e.StepLimits("other", "o1")
if err != nil {
t.Fatal(err)
}
if other["a"] != 2 || other["b"] != 1 {
t.Fatalf("other limits = %v, want a 2, b 1", other)
}
if gets != 1 {
t.Fatalf("workflows read %d times, want once", gets)
}
}
// #283 review: dagnats pins a run to the definition it started under
// (def_hash); the listing holds only the latest one per name. Limits
// are trusted only for a run whose hash is the listed one. A miss is
// re-read once — the run may have started under a definition
// registered since the last read — and then answers nothing, so the
// page says "attempt N" alone and only the engine settles the step.
func TestStepLimitsTrustOnlyTheRunsOwnDefinition(t *testing.T) {
var gets int
hash := "new"
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gets++
fmt.Fprintf(w, `[{"name":"quarry","def_hash":%q,"steps":[{"id":"test","retry":{"max_attempts":0}}]}]`, hash)
}))
defer srv.Close()
e := New(srv.URL)
if got, err := e.StepLimits("quarry", "old"); err != nil || len(got) != 0 {
t.Fatalf("a run on an older definition got limits %v, %v", got, err)
}
if got, _ := e.StepLimits("quarry", "old"); len(got) != 0 {
t.Fatalf("second ask got %v", got)
}
if got, _ := e.StepLimits("quarry", ""); len(got) != 0 {
t.Fatalf("a run with no def_hash got limits %v", got)
}
if got, _ := e.StepLimits("gone", "x"); len(got) != 0 {
t.Fatalf("an unknown workflow got limits %v", got)
}
readsBefore := gets
hash = "newer" // registered since the last read
if got, _ := e.StepLimits("quarry", "newer"); got["test"] != 1 {
t.Fatalf("a run on a definition registered since the read = %v, want test 1", got)
}
if gets != readsBefore+1 {
t.Fatalf("a new hash cost %d reads, want one", gets-readsBefore)
}
}