package api
import (
"context"
"errors"
"path/filepath"
"testing"
"craftdesign.group/quarry/internal/artifact"
"craftdesign.group/quarry/internal/attach"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/zitcli"
)
// testToken stands in for the lease where a test writes to a scratch
// store directly, with no coordinator.
var testToken = lease.Unguarded("test: scratch store, no coordinator")
// notHolder is the guard of a node that does not hold the store's
// lease: it counts that it was asked and refuses, never running the
// write. allow lets the first n writes through, for an op whose earlier
// writes must succeed to reach the one under test.
type notHolder struct {
allow int
calls int
}
func (g *notHolder) guard(fn func(lease.Token) error) error {
g.calls++
if g.calls <= g.allow {
return fn(lease.Unguarded("test: a guard standing in for the holder"))
}
return lease.ErrNotHolder
}
// refused asserts that an op was stopped by the lease, not by anything
// else, and that it asked.
func refused(t *testing.T, op string, g *notHolder, err error) {
t.Helper()
switch {
case g.calls == 0:
t.Errorf("FAIL %s: wrote to the store without asking the lease (err %v)", op, err)
case !errors.Is(err, lease.ErrNotHolder):
t.Errorf("FAIL %s: a node without the lease answered %v, want ErrNotHolder", op, err)
default:
t.Logf("ok %s refused without the lease", op)
}
}
// Ticket 8e6ecbb15c: these store writes ran outside the lease guard. On
// one node zit's file lock orders them; on several, a node that does not
// hold the lease wrote anyway. Each must now ask, and stop when refused.
func TestStoreWritesWithoutTheLeaseAreRefused(t *testing.T) {
ix, z, event := attachStore(t)
ctx := context.Background()
t.Run("attachment.remove", func(t *testing.T) {
art, blob, err := z.AttachAdd(testToken, ctx, event, "kept.txt", "text/plain", "lou", "", []byte("kept\n"))
if err != nil {
t.Fatal(err)
}
if err := ix.FoldNow([]index.Parsed{{Hash: art, Art: &artifact.Artifact{
Type: artifact.Attachment, Date: "2026-01-02T00:00:00", User: "lou", Mimetype: "text/plain",
Attachment: &artifact.Attach{Name: "kept.txt", Target: event, Source: blob}}}}); err != nil {
t.Fatal(err)
}
g := ¬Holder{}
r := ops.NewRegistry()
if err := RegisterAll(r, Deps{Version: "v", Repos: NewRepos(map[string]*Repo{
"test": {Index: ix, Zit: z, Guard: g.guard}})}); err != nil {
t.Fatal(err)
}
_, err = r.Invoke(as("lou", "bB"), "attachment.remove", attachmentRemoveIn{Repo: "test", Target: event, Name: "kept.txt"})
refused(t, "attachment.remove", g, err)
if rows, _ := z.AttachList(ctx, event); len(rows) != 1 {
t.Errorf("FAIL attachment.remove: the store changed under a refusal: %+v", rows)
}
})
t.Run("attachment.commit", func(t *testing.T) {
srv, objects := newObjectS3(t)
buckets := newMemBuckets()
buckets.rows["test"] = oltp.BucketBinding{Endpoint: srv.URL, Bucket: "files", Region: "auto", AccessKey: "AK", Secret: "SK", LimitBytes: 1 << 20}
store := newMemAttachments(buckets)
g := ¬Holder{}
r := ops.NewRegistry()
if err := RegisterAll(r, Deps{Version: "v", Repos: NewRepos(map[string]*Repo{
"test": {Index: ix, Zit: z, Guard: g.guard}}), Buckets: buckets, Attachments: store}); err != nil {
t.Fatal(err)
}
lou := as("lou", "b")
out, err := r.Invoke(lou, "attachment.begin", attachmentBeginIn{Repo: "test", Target: event[:16], Name: "clip.mp4", Size: 500, Type: "video/mp4", SHA256: sha})
if err != nil {
t.Fatal(err)
}
objects.objects[attach.Key(sha)] = 500
before, _ := z.AttachList(ctx, event)
_, err = r.Invoke(lou, "attachment.commit", attachmentCommitIn{Repo: "test", Ticket: out.(attachmentBeginOut).Ticket})
refused(t, "attachment.commit", g, err)
if after, _ := z.AttachList(ctx, event); len(after) != len(before) {
t.Errorf("FAIL attachment.commit: an A card was written under a refusal: %+v", after)
}
})
t.Run("ticket.promote attachments", func(t *testing.T) {
db, err := oltp.Open(filepath.Join(t.TempDir(), "oltp.db"))
if err != nil {
t.Fatal(err)
}
defer db.Close()
local, err := db.LocalTicketNew("test", "lou", "", map[string]string{"title": "held", "status": "Open"})
if err != nil {
t.Fatal(err)
}
if err := db.LocalAttachmentAdd(oltp.LocalAttachment{Repo: "test", Ticket: local, Name: "held.txt",
Size: 5, Type: "text/plain", SHA256: sha, Who: "lou"}); err != nil {
t.Fatal(err)
}
// The ticket itself may be written; the held file after it may not.
g := ¬Holder{allow: 1}
r := ops.NewRegistry()
if err := RegisterAll(r, Deps{Version: "v", Repos: NewRepos(map[string]*Repo{
"test": {Index: ix, Zit: z, Guard: g.guard}}), Local: db}); err != nil {
t.Fatal(err)
}
pr, err := r.Invoke(as("adm", "a"), "ticket.promote", ticketPromoteIn{Repo: "test", ID: local[:12]})
if err != nil {
t.Fatalf("promote: %v", err)
}
p := pr.(ticketPromoteOut)
switch {
case g.calls < 2:
t.Errorf("FAIL ticket.promote: the held file was written without asking the lease (%d asks)", g.calls)
case p.Attached != 0 || len(p.AttachErrors) != 1:
t.Errorf("FAIL ticket.promote: a refused attachment was not reported: %+v", p)
default:
t.Log("ok ticket.promote asks the lease for each held file")
}
if rows, _ := z.AttachList(ctx, p.ID); len(rows) != 0 {
t.Errorf("FAIL ticket.promote: an A card was written under a refusal: %+v", rows)
}
})
t.Run("settings and config", func(t *testing.T) {
dir, _, _, _ := landedStore(t)
g := ¬Holder{}
r := ops.NewRegistry()
if err := RegisterAll(r, Deps{Version: "v", Repos: NewRepos(map[string]*Repo{
"alpha": {Zit: zitcli.New(dir), Guard: g.guard}})}); err != nil {
t.Fatal(err)
}
_, err := r.Invoke(ctx, "settings.set", settingsSetIn{Repo: "alpha", Name: "binary-glob", Value: "*.bin"})
refused(t, "settings.set", g, err)
g.calls = 0
_, err = r.Invoke(ctx, "config.import", configImportIn{Repo: "alpha", File: "# empty\n"})
refused(t, "config.import", g, err)
})
t.Run("ci.pipeline.save", func(t *testing.T) {
g := ¬Holder{}
saved := 0
r := ops.NewRegistry()
if err := RegisterAll(r, Deps{Version: "v",
Repos: NewRepos(map[string]*Repo{"test": {Index: ix, Guard: g.guard}}),
SavePipeline: func(_ lease.Token, repo, branch, user, message, path string, body []byte) (string, error) {
saved++
return "", nil
}}); err != nil {
t.Fatal(err)
}
_, err := r.Invoke(ops.WithUser(ctx, "dan"), "ci.pipeline.save", ciPipelineSaveIn{
Repo: "test", Path: "ci/quarry.yml", Content: "checks:\n - name: t\n run: go test ./...\n",
})
refused(t, "ci.pipeline.save", g, err)
if saved != 0 {
t.Errorf("FAIL ci.pipeline.save: committed %d times under a refusal", saved)
}
})
}