so-many-crows/server_test.go

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() })
	nests, err := LoadNests(filepath.Join(dir, "nests.json"))
	if err != nil {
		t.Fatal(err)
	}
	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, nests)
	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)
	}
}