package ci
import (
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/cronexpr"
)
// CRON LINES ARE READ FROM on: schedule, GITHUB'S 5 FIELDS, UTC, NO
// MORE OFTEN THAN EVERY 5 MINUTES, REFUSED WITH A POSITION OTHERWISE.
func TestSchedulesAreReadWithTheFiveMinuteFloor(t *testing.T) {
spec := []byte("on:\n schedule:\n - cron: \"20 3 * * *\"\n - cron: '*/15 * * * 1-5'\njobs:\n t:\n steps:\n - run: x\n")
got, err := schedulesOf(spec)
if err != nil || len(got) != 2 || got[0] != "20 3 * * *" || got[1] != "*/15 * * * 1-5" {
t.Errorf("FAIL: schedules = %v, %v; want both cron lines", got, err)
}
for _, c := range []struct{ spec, pos string }{
{"on:\n schedule:\n - cron: '* * * * *'\njobs: {}\n", "line 3, column 13"},
{"on:\n schedule:\n - cron: '0,2 * * * *'\njobs: {}\n", "line 3, column 13"},
{"on:\n schedule:\n - cron: 'nonsense'\njobs: {}\n", "line 3, column 13"},
{"on:\n schedule:\n - every: day\njobs: {}\n", "line 3, column 7"},
} {
if _, err := schedulesOf([]byte(c.spec)); err == nil || !strings.Contains(err.Error(), c.pos) {
t.Errorf("FAIL: %q gave %v; want a refusal at %s", c.spec, err, c.pos)
}
}
if got, err := schedulesOf([]byte("on: push\njobs: {}\n")); err != nil || len(got) != 0 {
t.Errorf("FAIL: a push workflow has schedules %v, %v", got, err)
}
// on: with a schedule list is a workflow quarry honours, not a malformed on:.
on := json.RawMessage(`{"schedule":[{"cron":"17 3 * * *"}],"push":{}}`)
if err := CheckOn(on); err != nil {
t.Errorf("FAIL: CheckOn refused a schedule: %v", err)
}
if ok, err := runsOn(on, event{kind: "schedule"}); err != nil || !ok {
t.Errorf("FAIL: runsOn(schedule) = %v, %v", ok, err)
}
if ok, _ := runsOn(json.RawMessage(`{"push":{}}`), event{kind: "schedule"}); ok {
t.Errorf("FAIL: a push-only workflow runs on a schedule")
}
t.Log("ok: cron lines read, the 5-minute floor and bad lines refused with a position, on: schedule honoured")
}
// DUE IS THE MINUTE, UTC: 03:17 is due, 03:16:59 and 03:18 are not.
func TestDueIsTheScheduledMinute(t *testing.T) {
expr, err := parseCron("17 3 * * *")
if err != nil {
t.Fatal(err)
}
crons := []*cronexpr.CronExpr{expr}
at := func(s string) time.Time { tm, _ := time.Parse(time.RFC3339, s); return tm }
for when, want := range map[string]bool{
"2026-10-03T03:17:00Z": true, "2026-10-03T03:17:59Z": true,
"2026-10-03T03:16:59Z": false, "2026-10-03T03:18:00Z": false,
"2026-10-03T05:17:00+02:00": true, // the same instant in another zone
} {
if got := due(crons, at(when)); got != want {
t.Errorf("FAIL: due at %s = %v; want %v", when, got, want)
}
}
if next := nextFire(crons, at("2026-10-03T03:17:30Z")); !next.Equal(at("2026-10-04T03:17:00Z")) {
t.Errorf("FAIL: next fire after 03:17:30 = %s; want the next day's 03:17", next)
}
t.Log("ok: due by the minute, UTC; next fire computed")
}
type scheduleRig struct {
st *fakeStore
claims map[string]bool // shared across "restarts"
running map[string]bool
mu *sync.Mutex
}
func (s scheduleRig) ClaimSchedule(repo, pipeline string, minute time.Time) (bool, error) {
s.mu.Lock()
defer s.mu.Unlock()
k := repo + " " + pipeline + " " + minute.UTC().Format(time.RFC3339)
if s.claims[k] {
return false, nil
}
s.claims[k] = true
return true, nil
}
func (s scheduleRig) ScheduleRunning(repo, pipeline string) bool { return s.running[pipeline] }
func (s scheduleRig) ScheduleFailed(repo, pipeline string, minute time.Time, why string) {}
func always() bool { return true }
// EACH SCHEDULED MINUTE FIRES ONCE, FROM THE DEFAULT BRANCH TIP'S
// DEFINITION, WITH NO STARTER; A STILL-RUNNING PREVIOUS RUN SKIPS.
func TestSchedulesFireOnceFromTheDefaultTip(t *testing.T) {
var mu sync.Mutex
var specs []string
var labels []map[string]string
n := 0
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
switch {
case r.URL.Path == "/v1/ci/compile":
var body struct{ Name, Spec string }
json.NewDecoder(r.Body).Decode(&body)
specs = append(specs, body.Spec)
fmt.Fprintf(w, `{"workflow":{"name":%q},"registered":true,"on":{"schedule":[{"cron":"0 3 * * *"}]}}`, body.Name)
case r.URL.Path == "/runs" && r.Method == http.MethodPost:
var body struct {
Labels map[string]string `json:"labels"`
}
json.NewDecoder(r.Body).Decode(&body)
labels = append(labels, body.Labels)
n++
w.WriteHeader(http.StatusCreated)
fmt.Fprintf(w, `{"run_id":"s%d"}`, n)
case r.URL.Path == "/runs":
fmt.Fprint(w, `[]`)
}
}))
defer srv.Close()
e := New(srv.URL)
nightly := func(tag string) []byte {
return []byte("on:\n schedule:\n - cron: '0 3 * * *'\njobs:\n t:\n steps:\n - run: echo " + tag + "\n")
}
repo := &fakeRepo{
files: []string{"ci/nightly.yml", "ci/test.yml"},
blobs: map[string][]byte{"ci/nightly.yml": nightly("trunk"), "ci/test.yml": []byte("on: push\njobs:\n t:\n steps:\n - run: x\n")},
tips: map[string]string{"trunk": "t9", "feat": "b9"},
// A branch changed the schedule; the default tip did not.
versioned: map[string]map[string][]byte{"b9": {"ci/nightly.yml": nightly("branch")}},
}
rig := scheduleRig{st: &fakeStore{mark: map[string]string{}}, claims: map[string]bool{}, running: map[string]bool{}, mu: &sync.Mutex{}}
type store struct {
*fakeStore
scheduleRig
}
st := store{rig.st, rig}
minute := time.Date(2026, 10, 3, 3, 0, 0, 0, time.UTC)
cache := NewScheduleCache()
if err := FireSchedules(e, st, cache, "alpha", repo, "u", []time.Time{minute}, always); err != nil {
t.Fatal(err)
}
if err := FireSchedules(e, st, cache, "alpha", repo, "u", []time.Time{minute}, always); err != nil { // a second tick in the same minute
t.Fatal(err)
}
// A restart: a new process, the same claims on record.
again := store{&fakeStore{mark: map[string]string{}}, rig}
cache = NewScheduleCache()
if err := FireSchedules(e, again, cache, "alpha", repo, "u", []time.Time{minute.Add(30 * time.Second)}, always); err != nil {
t.Fatal(err)
}
if n != 1 {
t.Fatalf("FAIL: %d runs started for one scheduled minute; want 1", n)
}
if len(specs) != 1 || !strings.Contains(specs[0], "echo trunk") {
t.Errorf("FAIL: the schedule registered %q; want the default tip's definition", specs)
}
got := rig.st.starts["s1"]
if got.Event != "schedule" || got.Starter != "" {
t.Errorf("FAIL: the scheduled run recorded %+v; want event schedule, no starter", got)
}
if len(rig.st.runs) != 1 || !strings.HasPrefix(rig.st.runs[0], "s1 t9 ") {
t.Errorf("FAIL: recorded %v; want one run of the default tip t9", rig.st.runs)
}
if labels[0]["subject"] != "schedule" || labels[0]["branch"] != "" {
t.Errorf("FAIL: labels %v; want subject schedule and no branch to supersede by", labels[0])
}
// Not due: nothing.
if err := FireSchedules(e, st, cache, "alpha", repo, "u", []time.Time{minute.Add(time.Minute)}, always); err != nil || n != 1 {
t.Errorf("FAIL: a minute that is not due started a run (%d), %v", n, err)
}
// The next day's minute, while the previous run still runs: skipped.
rig.running["ci/nightly.yml"] = true
if err := FireSchedules(e, st, cache, "alpha", repo, "u", []time.Time{minute.Add(24 * time.Hour)}, always); err != nil || n != 1 {
t.Errorf("FAIL: a schedule fired while its previous run still ran (%d), %v", n, err)
}
rig.running["ci/nightly.yml"] = false
if err := FireSchedules(e, st, cache, "alpha", repo, "u", []time.Time{minute.Add(48 * time.Hour)}, always); err != nil || n != 2 {
t.Errorf("FAIL: the schedule did not fire once its previous run ended (%d), %v", n, err)
}
t.Log("ok: once per minute across ticks and a restart, default tip only, no starter, skipped while the last still runs")
}
// A repository with no runner draining its group is not scheduled; an
// engine that reports no groups at all cannot say, and is not judged.
func TestARunnerServesARepositoryByItsGroup(t *testing.T) {
now := time.Now()
seen := now.Add(-10 * time.Second).UTC().Format(time.RFC3339)
old := now.Add(-time.Hour).UTC().Format(time.RFC3339)
ws := []Worker{{WorkerGroups: []string{GroupFor("alpha")}, LastSeen: seen}, {WorkerGroups: []string{GroupFor("beta")}, LastSeen: old}}
if !RunnerFor(ws, "alpha", now) || RunnerFor(ws, "beta", now) || RunnerFor(ws, "gamma", now) {
t.Errorf("FAIL: RunnerFor misjudged live, stale and absent runners")
}
if !RunnerFor([]Worker{{LastSeen: seen}}, "gamma", now) {
t.Errorf("FAIL: an engine reporting no groups was judged to have no runner")
}
t.Log("ok: a runner serves the repositories whose group it drains")
}