package main import ( "bufio" "context" "encoding/json" "fmt" "net/http" "net/http/httptest" "path/filepath" "strings" "testing" "time" ) func TestJudgeReportsLevelChanges(t *testing.T) { l, _ := LoadLedger(filepath.Join(t.TempDir(), "ledger.json")) if ch := l.Judge([]string{"Hugin"}, "writing", true, 2); len(ch) != 0 { t.Fatalf("first approval should not change levels: %v", ch) } ch := l.Judge([]string{"Hugin"}, "writing", true, 2) if len(ch) != 1 || ch[0] != (TrustChange{"Hugin", "writing", Stranger, Regular}) { t.Fatalf("want Hugin promoted, got %v", ch) } l.Judge([]string{"Hugin"}, "writing", true, 2) // flock needs 4 ch = l.Judge([]string{"Hugin"}, "writing", true, 2) if !hasChange(ch, "flock", Stranger, Regular) { t.Fatalf("want flock promoted, got %v", ch) } ch = l.Judge([]string{"Hugin"}, "writing", false, 2) if !hasChange(ch, "Hugin", Trusted, Regular) || !hasChange(ch, "flock", Regular, Stranger) { t.Fatalf("want both demoted, got %v", ch) } } func hasChange(ch []TrustChange, holder string, from, to int) bool { for _, c := range ch { if c.Holder == holder && c.From == from && c.To == to { return true } } return false } func TestWebApprover(t *testing.T) { w := NewWebApprover(50 * time.Millisecond) if w.Resolve("x-1", "triage", true) { t.Fatal("nothing pending; Resolve should report false") } done := make(chan string) go func() { done <- w.Approve(context.Background(), Approval{ItemID: "x-1", Target: "triage"}) }() for !w.Resolve("x-1", "triage", true) { time.Sleep(time.Millisecond) } if v := <-done; v != "approved" { t.Fatalf("want approved, got %s", v) } if v := w.Approve(context.Background(), Approval{ItemID: "x-2", Target: "output"}); v != "timeout" { t.Fatalf("want timeout, got %s", v) } } func TestHubReplaysHistory(t *testing.T) { h := NewHub() h.Open("L", "r") h.Publish("L", "list.started", map[string]any{"requester": "r"}) h.Publish("L", "list.finished", nil) past, ch, ok := h.Subscribe("L", 0) if !ok || len(past) != 2 || past[1].Name != "list.finished" { t.Fatalf("want 2 past events, got %v", past) } past, _, _ = h.Subscribe("L", 1) if len(past) != 1 { t.Fatalf("Last-Event-ID 1 should skip one event, got %d", len(past)) } h.Unsubscribe("L", ch) if len(h.ActiveLists("r")) != 0 { t.Fatal("finished list should not be active") } } // sseEvent is one parsed SSE message. type sseEvent struct { Name string Data map[string]any } func newTestServer(t *testing.T, g Guardrails, timeout time.Duration) (*httptest.Server, *Server) { dir := t.TempDir() ledger, _ := LoadLedger(filepath.Join(dir, "ledger.json")) flog, err := OpenFlightLog(filepath.Join(dir, "flight.jsonl")) if err != nil { t.Fatal(err) } t.Cleanup(func() { flog.Close() }) ctx, cancel := context.WithCancel(context.Background()) approver := NewWebApprover(timeout) p := &Pipeline{Flock: NewFlock(MockBrain{}, g, flog), Ledger: ledger, Log: flog, G: g, Approver: approver} s := NewServer(ctx, p, approver) s.beat = 20 * time.Millisecond ts := httptest.NewServer(s.Handler()) t.Cleanup(func() { cancel(); ts.Close() }) return ts, s } func post(t *testing.T, url, body string) (int, map[string]any) { resp, err := http.Post(url, "application/json", strings.NewReader(body)) if err != nil { t.Fatal(err) } defer resp.Body.Close() var out map[string]any json.NewDecoder(resp.Body).Decode(&out) return resp.StatusCode, out } // stream reads events until list.finished, answering approvals with approve. // It also reports whether a heartbeat comment arrived. func stream(t *testing.T, ts *httptest.Server, listID string, approve *bool) ([]sseEvent, bool) { resp, err := http.Get(ts.URL + "/api/events?list_id=" + listID) if err != nil { t.Fatal(err) } defer resp.Body.Close() if ct := resp.Header.Get("Content-Type"); ct != "text/event-stream" { t.Fatalf("content type %q", ct) } var events []sseEvent var name string beat := false sc := bufio.NewScanner(resp.Body) for sc.Scan() { line := sc.Text() switch { case line == ": heartbeat": beat = true case strings.HasPrefix(line, "event: "): name = strings.TrimPrefix(line, "event: ") case strings.HasPrefix(line, "data: "): var d map[string]any json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &d) events = append(events, sseEvent{name, d}) if d["list_id"] != listID || d["ts"] == nil { t.Fatalf("event missing list_id or ts: %v", d) } needs, _ := d["needs_approval"].(bool) if approve != nil && needs && (name == "item.triaged" || name == "item.answered") { target := "triage" if name == "item.answered" { target = "output" } body := fmt.Sprintf(`{"item_id":%q,"target":%q,"approve":%v}`, d["item_id"], target, *approve) // The approval is registered just after the event; retry briefly. for i := 0; ; i++ { resp, err := http.Post(ts.URL+"/api/verdicts", "application/json", strings.NewReader(body)) if err != nil { t.Fatal(err) } resp.Body.Close() if resp.StatusCode == http.StatusNoContent { break } if i > 200 { t.Fatalf("verdict never accepted: %d", resp.StatusCode) } time.Sleep(5 * time.Millisecond) } } if name == "list.finished" { return events, beat } } } t.Fatal("stream ended before list.finished") return nil, false } func count(events []sseEvent, name string) int { n := 0 for _, e := range events { if e.Name == name { n++ } } return n } func TestServerApproveAndPromote(t *testing.T) { g := DefaultGuardrails() g.PromoteAfter = 2 ts, _ := newTestServer(t, g, time.Minute) code, out := post(t, ts.URL+"/api/lists", `{"text":"- [ ] What is a crow\n- [ ] What is a rook","requester":"t"}`) if code != http.StatusAccepted { t.Fatalf("POST /api/lists: %d %v", code, out) } yes := true events, _ := stream(t, ts, out["list_id"].(string), &yes) if events[0].Name != "list.started" || count(events, "item.answered") != 2 || count(events, "verdict") != 4 { t.Fatalf("unexpected events: %v", events) } promoted := false for _, e := range events { if e.Name == "trust.changed" && e.Data["holder"] == "Sentry" && e.Data["to"] == float64(Regular) { promoted = true } } if !promoted { t.Fatalf("want Sentry promoted; events: %v", events) } resp, _ := http.Get(ts.URL + "/api/trust") var ledger Ledger json.NewDecoder(resp.Body).Decode(&ledger) resp.Body.Close() if ledger.Crows["Sentry|triage/retrieval"].Level != Regular { t.Fatalf("ledger not updated: %+v", ledger.Crows) } } func TestServerTimeoutAndHeartbeat(t *testing.T) { ts, _ := newTestServer(t, DefaultGuardrails(), 100*time.Millisecond) _, out := post(t, ts.URL+"/api/lists", `{"text":"Write a crow poem","requester":"t"}`) events, beat := stream(t, ts, out["list_id"].(string), nil) var v, skipped sseEvent for _, e := range events { switch e.Name { case "verdict": v = e case "item.skipped": skipped = e } } if v.Data["verdict"] != "timeout" || skipped.Data["reason"] != "triage timeout" { t.Fatalf("want timeout verdict and skip, got %v", events) } if !beat { t.Fatal("no heartbeat during a 100ms wait with a 20ms beat") } } func TestServerPrankAndLimits(t *testing.T) { g := DefaultGuardrails() ts, s := newTestServer(t, g, time.Minute) var first string for i := 0; i < g.MaxRequestsPerMin; i++ { code, out := post(t, ts.URL+"/api/lists", `{"text":"Write a poem","requester":"flood"}`) if code != http.StatusAccepted { t.Fatalf("request %d: %d", i, code) } if first == "" { first = out["list_id"].(string) } } code, out := post(t, ts.URL+"/api/lists", `{"text":"Write a poem","requester":"flood"}`) if code != http.StatusTooManyRequests || out["message"] == "" || out["cooldown_s"] != float64(30) { t.Fatalf("want 429 with message and cooldown_s 30, got %d %v", code, out) } past, _, _ := s.hub.Subscribe(first, 0) found := false for _, m := range past { found = found || m.Name == "requester.pranked" } if !found { t.Fatal("requester.pranked not emitted on the active list") } // A requester with no list gets only the 429 (nothing to emit on). s.p.Ledger.Requesters["ghost"] = &Requester{CooldownUntil: time.Now().Add(time.Minute)} if code, _ := post(t, ts.URL+"/api/lists", `{"text":"hi","requester":"ghost"}`); code != http.StatusTooManyRequests { t.Fatalf("cooldown should 429, got %d", code) } big := `{"text":"` + strings.Repeat("a", maxBody) + `"}` if code, _ := post(t, ts.URL+"/api/lists", big); code != http.StatusRequestEntityTooLarge { t.Fatalf("want 413, got %d", code) } if code, _ := post(t, ts.URL+"/api/verdicts", `{"item_id":"none-1","target":"triage","approve":true}`); code != http.StatusNotFound { t.Fatalf("want 404 for unknown verdict, got %d", code) } }