diff --git a/neste2e_test.go b/neste2e_test.go new file mode 100644 index 0000000..e1f1468 --- /dev/null +++ b/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 + } + } +}