neste2e_test.go

@@ -0,0 +1,129 @@
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+)
+
+// TestNestConversationEndToEndMock drives the whole loop with the offline mock:
+// an owner request auto-gets exactly one crow response, and an SSE reconnect
+// replays the same records the store holds.
+func TestNestConversationEndToEndMock(t *testing.T) {
+	ts, s := newTestNestServer(t)
+	if s.p.Flock.brain.Name() != "offline mock" {
+		t.Fatalf("B5 must run on the offline mock, got %q", s.p.Flock.brain.Name())
+	}
+	id := newNest(t, ts, "Lab", `"Hugin"`)
+
+	code, req := post(t, ts.URL+"/api/nests/"+id+"/messages",
+		`{"author":"owner","kind":"request","text":"count the eggs","to":"Hugin"}`)
+	if code != http.StatusCreated || req["kind"] != "request" {
+		t.Fatalf("request post: %d %v", code, req)
+	}
+
+	// The addressed crow responds (must-respond), asynchronously.
+	msgs := awaitMessages(t, ts, id, 2)
+	if len(msgs) != 2 {
+		t.Fatalf("want exactly 2 messages, got %d: %v", len(msgs), msgs)
+	}
+	if msgs[0]["kind"] != "request" || msgs[1]["author"] != "Hugin" {
+		t.Fatalf("bad conversation shape: %v", msgs)
+	}
+	respKind := msgs[1]["kind"].(string)
+	if respKind != "claim" && respKind != "deliver" && respKind != "decline" {
+		t.Fatalf("crow response must be responsive, got %v", msgs[1]["kind"])
+	}
+	if msgs[1]["refers_to"] != req["msg_id"] || msgs[1]["to"] != "owner" {
+		t.Fatalf("response must answer the request and its author: %v", msgs[1])
+	}
+
+	// The crow response is appended and emitted asynchronously; make sure the
+	// hub has caught up before replaying from it.
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		past, ch, _ := s.hub.Subscribe(id, 0)
+		n := len(past)
+		s.hub.Unsubscribe(id, ch)
+		if n >= 2 || time.Now().After(deadline) {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	// SSE reconnect with Last-Event-ID replays both records in order.
+	events := readNestSSE(t, ts, id, 0)
+	if len(events) != 2 {
+		t.Fatalf("want 2 replayed events, got %d: %v", len(events), events)
+	}
+	for i, ev := range events {
+		if ev.Name != "nest.message" || ev.Data["msg_id"] != msgs[i]["msg_id"] || ev.Data["text"] != msgs[i]["text"] {
+			t.Fatalf("replay mismatch at %d: event %v vs stored %v", i, ev.Data, msgs[i])
+		}
+	}
+
+	// Partial replay from Last-Event-ID: only the second message.
+	tail := readNestSSE(t, ts, id, 1)
+	if len(tail) != 1 || tail[0].Data["msg_id"] != msgs[1]["msg_id"] {
+		t.Fatalf("Last-Event-ID 1 should skip the request, got %v", tail)
+	}
+}
+
+// readNestSSE connects to the nest stream, reads the replayed history, and
+// closes once a heartbeat marks the end of what was already there.
+func readNestSSE(t *testing.T, ts *httptest.Server, nestID string, lastID int) []sseEvent {
+	t.Helper()
+	req, err := http.NewRequest("GET", ts.URL+"/api/events?nest_id="+nestID, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if lastID > 0 {
+		req.Header.Set("Last-Event-ID", strconv.Itoa(lastID))
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+
+	events := make(chan sseEvent, 32)
+	go func() {
+		var name string
+		sc := bufio.NewScanner(resp.Body)
+		for sc.Scan() {
+			line := sc.Text()
+			switch {
+			case line == ": heartbeat":
+				close(events)
+				return
+			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 <- sseEvent{name, d}
+			}
+		}
+		close(events)
+	}()
+
+	var out []sseEvent
+	deadline := time.After(2 * time.Second)
+	for {
+		select {
+		case ev, ok := <-events:
+			if !ok {
+				return out
+			}
+			out = append(out, ev)
+		case <-deadline:
+			t.Fatal("nest SSE stalled before heartbeat")
+			return out
+		}
+	}
+}