// provision_test.go proves RFC 0011 §7/Phase 0 end to end: quarry, not
// an operator, keeps notify's subscribers, topic membership and
// workflows in step with its own accounts and grants, and a chat post's
// mentions reach only people notify can actually address. The fake
// notify server here is built against notify's documented routes
// (docs/running.md, pre-0011-phase-0 revision) — regenerate it from
// notify's OpenAPI if those routes change (RFC 0012 §Risks).
package harness
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"net/url"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/chat"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/ingest"
"craftdesign.group/quarry/internal/notify"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/web"
"craftdesign.group/quarry/internal/zitcli"
)
// fakeNotifyReq is one call the fake notify server saw.
type fakeNotifyReq struct {
method, path string
body map[string]any
}
// fakeNotifyServer is a recording stand-in for notify's HTTP surface:
// every request is remembered, and workflow presence is stateful (a
// slug can be pre-seeded as "already there", the way an operator's
// earlier edit would be) so EnsureWorkflow's GET-then-POST path is
// exercised for real.
type fakeNotifyServer struct {
mu sync.Mutex
reqs []fakeNotifyReq
workflows map[string]bool
}
func (f *fakeNotifyServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
var body map[string]any
if r.Body != nil {
b, _ := io.ReadAll(r.Body)
if len(b) > 0 {
json.Unmarshal(b, &body)
}
}
f.mu.Lock()
f.reqs = append(f.reqs, fakeNotifyReq{method: r.Method, path: r.URL.Path, body: body})
defer f.mu.Unlock()
switch {
case r.Method == "GET" && strings.HasPrefix(r.URL.Path, "/v1/workflows/"):
slug := strings.TrimPrefix(r.URL.Path, "/v1/workflows/")
if f.workflows[slug] {
w.WriteHeader(http.StatusOK)
return
}
w.WriteHeader(http.StatusNotFound)
case r.Method == "POST" && r.URL.Path == "/v1/workflows":
name, _ := body["name"].(string)
f.workflows[name] = true
w.WriteHeader(http.StatusCreated)
default:
// subscribers, topics, topic members, and events/trigger all
// just need to be recorded and answered success.
w.WriteHeader(http.StatusCreated)
}
}
func (f *fakeNotifyServer) snapshot() []fakeNotifyReq {
f.mu.Lock()
defer f.mu.Unlock()
return append([]fakeNotifyReq(nil), f.reqs...)
}
func (f *fakeNotifyServer) find(method, path string) []fakeNotifyReq {
var out []fakeNotifyReq
for _, r := range f.snapshot() {
if r.method == method && r.path == path {
out = append(out, r)
}
}
return out
}
// waitForCount polls until n's request count for (method, path) reaches
// want or the deadline passes — the same shape notifyCapture.waitFor
// uses, because provisioning here is exactly as best-effort-async as a
// notify trigger.
func waitForCount(t *testing.T, srv *fakeNotifyRecorder, method, path string, want int) {
t.Helper()
for i := 0; i < 100; i++ {
if len(srv.f.find(method, path)) >= want {
return
}
time.Sleep(20 * time.Millisecond)
}
t.Fatalf("never saw %d %s %s (saw %+v)", want, method, path, srv.f.snapshot())
}
// fakeNotifyRecorder pairs the httptest server with its handler so
// tests can both hit it over HTTP (via notify.Client) and inspect what
// it recorded.
type fakeNotifyRecorder struct {
srv *httptest.Server
f *fakeNotifyServer
}
func startFakeNotify(t *testing.T, preseed ...string) *fakeNotifyRecorder {
t.Helper()
f := &fakeNotifyServer{workflows: map[string]bool{}}
for _, w := range preseed {
f.workflows[w] = true
}
srv := httptest.NewServer(f)
t.Cleanup(srv.Close)
return &fakeNotifyRecorder{srv: srv, f: f}
}
// harnessAccounts adapts forge OLTP to api.AccountStore, mirroring
// cmd/quarry's own accountStore (unexported there, so restated here —
// this IS the seam a fresh harness test needs, per the brief).
type harnessAccounts struct{ db *oltp.DB }
func (a harnessAccounts) Accounts() ([]api.Account, error) {
rows, err := a.db.Accounts()
if err != nil {
return nil, err
}
out := make([]api.Account, 0, len(rows))
for _, r := range rows {
out = append(out, api.Account{Name: r.Name, Admin: r.Admin, Created: r.Created})
}
return out, nil
}
func (a harnessAccounts) AddAccount(name, password string, admin bool) error {
return a.db.AddAccount(name, password, admin)
}
func (a harnessAccounts) SetPassword(name, password string) error {
return a.db.SetPassword(name, password)
}
func (a harnessAccounts) DeleteAccount(name string) error { return a.db.DeleteAccount(name) }
func (a harnessAccounts) Grant(account, repo, role string) error {
r, err := oltp.ParseRole(role)
if err != nil {
return err
}
return a.db.Grant(account, repo, r)
}
func (a harnessAccounts) Revoke(account, repo string) error { return a.db.Revoke(account, repo) }
func (a harnessAccounts) GrantsFor(account string) (map[string]string, error) {
grants, err := a.db.GrantsFor(account)
if err != nil {
return nil, err
}
out := make(map[string]string, len(grants))
for repo, role := range grants {
out[repo] = role.String()
}
return out, nil
}
// harnessProvision adapts api.ProvisionPlane to the notify client,
// mirroring cmd/quarry's provisionAdapter.
type harnessProvision struct{ client *notify.Client }
func (p *harnessProvision) EnsureSubscriber(ctx context.Context, account string) error {
return p.client.EnsureSubscriber(ctx, account)
}
func (p *harnessProvision) SetTopicMember(ctx context.Context, repo, account string, member bool) error {
key := "repo-" + repo
if err := p.client.EnsureTopic(ctx, key, repo+" activity"); err != nil {
return err
}
if member {
return p.client.AddTopicMember(ctx, key, account)
}
return p.client.RemoveTopicMember(ctx, key, account)
}
// harnessNotify adapts api.NotifyPlane to the notify client, mirroring
// cmd/quarry's notifyAdapter but without the repo: topic addressing
// (Phase 0's mention triggers always address a person, never a topic).
type harnessNotify struct{ client *notify.Client }
func (n *harnessNotify) Trigger(ctx context.Context, e api.NotifyEvent) error {
return n.client.Trigger(ctx, e.Workflow, notify.Recipient{SubscriberID: e.To}, e.Payload, e.TxnID)
}
// buildProvisionServer stands up a real forge (one repo, real chat and
// accounts planes, real notify.Client wired to the fake server) — the
// harness proof this phase's spec asks for, built independently of
// ingestAndServeAuthMod because wiring Accounts/CapsFor there would
// retroactively filter mentions in every OTHER harness test that
// mentions a handle with no real account row (TestForumSubscriptions'
// @carol, notably).
func buildProvisionServer(t *testing.T, notifyURL string) (*httptest.Server, *oltp.DB) {
t.Helper()
dataDir := t.TempDir()
ev, err := events.Start(filepath.Join(dataDir, "nats"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
odb, err := oltp.Open(filepath.Join(dataDir, "oltp.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { odb.Close() })
room, err := chat.New(ev.Conn(), ev.JS())
if err != nil {
t.Fatal(err)
}
nclient := notify.New(notifyURL, "k")
deps := api.Deps{
Version: "0.0.0-harness",
Repos: api.NewRepos(nil),
Chat: room,
Notify: &harnessNotify{client: nclient},
Provision: &harnessProvision{client: nclient},
Accounts: harnessAccounts{db: odb},
Forum: odb,
}
deps.CapsFor = func(account, repo string) api.CapSet { return odb.EffectiveCaps(account, repo) }
repoPath := buildStore(t, "provseed")
ix, err := index.Open(filepath.Join(dataDir, "alpha.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { ix.Close() })
cli := zitcli.New(repoPath)
pipe := ingest.NewPipeline(cli, ev, ix, "alpha", "harness")
deps.Repos.Put("alpha", &api.Repo{Index: ix, Zit: cli, Sync: pipe.Sync, Code: "alpha"})
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
// dan bootstraps as the first admin exactly as docs/running.md's
// `quarry account add --admin` does — an existing admin account is
// the precondition for account.add/grant/revoke to work at all.
if err := odb.AddAccount("dan", "pw", true); err != nil {
t.Fatal(err)
}
verify := func(repo, user, pass string, need ops.Auth) api.Decision {
if user == "" || pass != "pw" {
return api.Denied
}
return api.Allowed
}
r := ops.NewRegistry()
if err := api.RegisterAll(r, deps); err != nil {
t.Fatal(err)
}
apiRouter, err := api.RouterWithCaps(r, "0.0.0-harness", deps.CapsFor, verify)
if err != nil {
t.Fatal(err)
}
mux := http.NewServeMux()
mux.Handle("/api/", apiRouter)
mux.Handle("/", web.HandlerWith(r, nil, nil, nil, web.Verify(verify)))
srv := httptest.NewServer(mux)
t.Cleanup(srv.Close)
return srv, odb
}
func TestProvisionEndToEnd(t *testing.T) {
fake := startFakeNotify(t, "quarry-mention") // pre-seed one workflow as "already there"
srv, _ := buildProvisionServer(t, fake.srv.URL)
client := srv.Client()
call := func(user, method, path string, form url.Values) (int, []byte) {
t.Helper()
var body io.Reader
if form != nil {
body = strings.NewReader(form.Encode())
}
req, _ := http.NewRequest(method, srv.URL+path, body)
if user != "" {
req.SetBasicAuth(user, "pw")
}
if form != nil {
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
}
resp, err := client.Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
b, _ := io.ReadAll(resp.Body)
return resp.StatusCode, b
}
// --- startup: EnsureWorkflow for each of notify.Workflows(), as
// cmd/quarry's serve does right after the notify client exists.
nclient := notify.New(fake.srv.URL, "k")
for _, wf := range notify.Workflows() {
if err := nclient.EnsureWorkflow(context.Background(), wf); err != nil {
t.Fatalf("EnsureWorkflow(%s): %v", wf.Name, err)
}
}
// The pre-seeded workflow was never POSTed...
for _, r := range fake.f.find("POST", "/v1/workflows") {
if r.body["name"] == "quarry-mention" {
t.Fatalf("a pre-existing workflow must never be recreated: %+v", r)
}
}
// ...and every absent one was.
created := map[string]bool{}
for _, r := range fake.f.find("POST", "/v1/workflows") {
created[fmt.Sprint(r.body["name"])] = true
}
for _, wf := range notify.Workflows() {
if wf.Name == "quarry-mention" {
continue
}
if !created[wf.Name] {
t.Fatalf("workflow %q was never created", wf.Name)
}
}
// --- account.add creates the subscriber.
if code, body := call("dan", "POST", "/api/v1/accounts",
url.Values{"name": {"alice"}, "password": {"pw12345678"}}); code != http.StatusOK {
t.Fatalf("account.add: %d: %s", code, body)
}
waitForCount(t, fake, "POST", "/v1/subscribers", 1)
found := false
for _, r := range fake.f.find("POST", "/v1/subscribers") {
if r.body["subscriberId"] == "alice" {
found = true
}
}
if !found {
t.Fatalf("no subscriber request for alice: %+v", fake.f.find("POST", "/v1/subscribers"))
}
// --- account.grant adds alice to repo-alpha's topic.
if code, body := call("dan", "POST", "/api/v1/accounts/grant",
url.Values{"name": {"alice"}, "repo": {"alpha"}, "role": {"reader"}}); code != http.StatusOK {
t.Fatalf("account.grant: %d: %s", code, body)
}
waitForCount(t, fake, "POST", "/v1/topics/repo-alpha/subscribers", 1)
found = false
for _, r := range fake.f.find("POST", "/v1/topics/repo-alpha/subscribers") {
if r.body["subscriberId"] == "alice" {
found = true
}
}
if !found {
t.Fatal("alice was never added to repo-alpha's topic")
}
// --- account.revoke removes alice from the topic.
if code, body := call("dan", "POST", "/api/v1/accounts/revoke",
url.Values{"name": {"alice"}, "repo": {"alpha"}}); code != http.StatusOK {
t.Fatalf("account.revoke: %d: %s", code, body)
}
waitForCount(t, fake, "DELETE", "/v1/topics/repo-alpha/subscribers/alice", 1)
// alice needs to be re-granted for the mention tests below (revoke
// above proved removal; the mention path needs her capable again).
if code, body := call("dan", "POST", "/api/v1/accounts/grant",
url.Values{"name": {"alice"}, "repo": {"alpha"}, "role": {"reader"}}); code != http.StatusOK {
t.Fatalf("account.grant (re-grant): %d: %s", code, body)
}
// --- anonymous chat.post is refused.
if code, _ := call("", "POST", "/api/v1/r/alpha/chat", url.Values{"text": {"hello"}}); code != http.StatusUnauthorized && code != http.StatusForbidden {
t.Fatalf("anonymous chat.post: status %d, want 401 or 403", code)
}
// --- @unknown triggers nothing.
before := len(fake.f.find("POST", "/v1/events/trigger"))
if code, body := call("dan", "POST", "/api/v1/r/alpha/chat",
url.Values{"text": {"hi @nobody"}}); code != http.StatusOK {
t.Fatalf("chat.post: %d: %s", code, body)
}
time.Sleep(200 * time.Millisecond) // fireNotify is async; give it a beat
if got := len(fake.f.find("POST", "/v1/events/trigger")); got != before {
t.Fatalf("@nobody fired %d triggers, want 0 new", got-before)
}
// --- @alice (granted, capable) triggers exactly once with an
// address-based TxnID.
if code, body := call("dan", "POST", "/api/v1/r/alpha/chat",
url.Values{"text": {"hey @alice"}}); code != http.StatusOK {
t.Fatalf("chat.post: %d: %s", code, body)
}
waitForCount(t, fake, "POST", "/v1/events/trigger", before+1)
var toAlice []fakeNotifyReq
for _, r := range fake.f.find("POST", "/v1/events/trigger")[before:] {
to, _ := r.body["to"].(map[string]any)
if to != nil && to["subscriberId"] == "alice" {
toAlice = append(toAlice, r)
}
}
if len(toAlice) != 1 {
t.Fatalf("expected exactly one trigger to alice, got %d: %+v", len(toAlice), toAlice)
}
txn, _ := toAlice[0].body["transactionId"].(string)
if !strings.HasPrefix(txn, "mention:chat:alpha:") || !strings.HasSuffix(txn, ":alice") {
t.Fatalf("TxnID %q is not address-based (mention:chat:<repo>:<at>:<who>)", txn)
}
}