pkg/models/http_answer_marks_test.go

Rama main · réplica de lectura del repositorio original.

package models

import (
	"encoding/json"
	"strings"
	"testing"
	"time"

	"go.mongodb.org/mongo-driver/v2/bson"
	yamlLib "gopkg.in/yaml.v3"
)

var marksT0 = time.Date(2026, 11, 5, 10, 0, 0, 123456789, time.UTC)

func sec(n int) time.Time { return marksT0.Add(time.Duration(n) * time.Second) }

func roundTripYAML(t *testing.T, resp HTTPResp) (HTTPResp, string) {
	t.Helper()
	out, err := yamlLib.Marshal(resp)
	if err != nil {
		t.Fatal(err)
	}
	var back HTTPResp
	if err := yamlLib.Unmarshal(out, &back); err != nil {
		t.Fatal(err)
	}
	return back, string(out)
}

// pieces is the body cut at its marks' offsets. A recording with no marks,
// or one whose offsets do not run in order inside its body, fails the row.
func pieces(t *testing.T, resp HTTPResp) []string {
	t.Helper()
	if resp.Marks == nil {
		t.Fatalf("no marks on %q", resp.Body)
	}
	var out []string
	start := 0
	for _, w := range resp.Marks.Writes {
		if w.Offset < start || w.Offset > len(resp.Body) {
			t.Fatalf("mark at %d does not fit after %d in a body of %d: %q", w.Offset, start, len(resp.Body), resp.Body)
		}
		out = append(out, resp.Body[start:w.Offset])
		start = w.Offset
	}
	return out
}

// An SSE frame takes the time of the write that completed it, with CRLF line
// ends, a frame cut across two writes, and a comment frame; the read-back
// body's pieces are its frames, each with the blank line after it.
func TestFramesTakeTheTimeOfTheWriteThatCompletedThem(t *testing.T) {
	body := "data: a\r\n\r\n: ping\r\n\r\nevent: x\r\ndata: b\r\n\r\n"
	// Write 1 ends in frame 3's first line, write 2 completes it.
	resp := HTTPResp{
		StatusCode: 200,
		Header:     map[string]string{"Content-Type": "text/event-stream"},
		Body:       body,
		Timestamp:  sec(9),
		Marks: &HTTPAnswerMarks{Head: sec(1), Writes: []HTTPAnswerWrite{
			{Offset: strings.Index(body, "data: b"), TS: sec(1)},
			{Offset: len(body), TS: sec(2)},
		}},
	}
	back, out := roundTripYAML(t, resp)
	if len(back.StreamBody) != 3 {
		t.Fatalf("frames = %d, want 3:\n%s", len(back.StreamBody), out)
	}
	for i, want := range []time.Time{sec(1), sec(1), sec(2)} {
		if !back.StreamBody[i].TS.Equal(want) {
			t.Errorf("frame %d ts = %v, want %v", i, back.StreamBody[i].TS, want)
		}
	}
	got := pieces(t, back)
	want := []string{"data:a\n\n: ping\n\n", "event:x\ndata:b\n\n"}
	if strings.Join(got, "|") != strings.Join(want, "|") {
		t.Errorf("pieces = %q, want %q", got, want)
	}
	if !back.Marks.Head.Equal(sec(1)) {
		t.Errorf("head = %v, want %v", back.Marks.Head, sec(1))
	}
}

// NDJSON keepalive (blank) lines are no frames: a line takes the time of the
// write that ended it.
func TestNDJSONLinesTakeTheirWritesTime(t *testing.T) {
	body := "{\"n\":1}\n\n{\"n\":2}\n"
	resp := HTTPResp{
		StatusCode: 200,
		Header:     map[string]string{"Content-Type": "application/x-ndjson"},
		Body:       body,
		Marks: &HTTPAnswerMarks{Head: sec(0), Writes: []HTTPAnswerWrite{
			{Offset: 8, TS: sec(1)},
			{Offset: 9, TS: sec(2)},
			{Offset: len(body), TS: sec(3)},
		}},
	}
	back, out := roundTripYAML(t, resp)
	if len(back.StreamBody) != 2 || !back.StreamBody[0].TS.Equal(sec(1)) || !back.StreamBody[1].TS.Equal(sec(3)) {
		t.Fatalf("frames = %+v:\n%s", back.StreamBody, out)
	}
	if got := pieces(t, back); strings.Join(got, "|") != "{\"n\":1}\n|{\"n\":2}" {
		t.Errorf("pieces = %q", got)
	}
}

// A body kept whole keeps its marks as they are: they index it.
func TestWholeBodyKeepsItsMarks(t *testing.T) {
	marks := &HTTPAnswerMarks{Head: sec(1), Writes: []HTTPAnswerWrite{{Offset: 5, TS: sec(2)}, {Offset: 11, TS: sec(3)}}}
	resp := HTTPResp{StatusCode: 200, Header: map[string]string{"Content-Type": "application/json"}, Body: `{"a":"done`, Marks: marks}
	back, out := roundTripYAML(t, resp)
	if back.Body != resp.Body || back.Marks == nil || len(back.Marks.Writes) != 2 || back.Marks.Writes[0].Offset != 5 ||
		!back.Marks.Writes[0].TS.Equal(sec(2)) || !back.Marks.Writes[1].TS.Equal(sec(3)) || !back.Marks.Head.Equal(sec(1)) {
		t.Fatalf("read back %+v, %+v:\n%s", back.Body, back.Marks, out)
	}
	if !strings.Contains(out, "- {offset: 5, ts: ") {
		t.Errorf("a write is not on one line:\n%s", out)
	}
}

// An SSE body kept as the bytes the upstream sent, never split into frames,
// keeps its marks as they are, for they index those bytes: a stream that
// ended inside its last event, and a fault's body sent before the fault.
func TestAnSSEBodyKeptAsItsBytesKeepsItsMarks(t *testing.T) {
	sse := map[string]string{"Content-Type": "text/event-stream"}
	marks := func() *HTTPAnswerMarks {
		return &HTTPAnswerMarks{Head: sec(1), Writes: []HTTPAnswerWrite{{Offset: 9, TS: sec(2)}, {Offset: 16, TS: sec(3)}}}
	}
	check := func(name string, back HTTPResp, out string) {
		t.Helper()
		if back.Body != "data: a\n\ndata: b" || len(back.StreamBody) != 0 {
			t.Fatalf("%s: body %q, frames %+v, want the bytes as sent:\n%s", name, back.Body, back.StreamBody, out)
		}
		if got := pieces(t, back); strings.Join(got, "|") != "data: a\n\n|data: b" || !back.Marks.Head.Equal(sec(1)) ||
			!back.Marks.Writes[0].TS.Equal(sec(2)) || !back.Marks.Writes[1].TS.Equal(sec(3)) {
			t.Errorf("%s: pieces %q, marks %+v", name, got, back.Marks)
		}
	}

	cut, out := roundTripYAML(t, HTTPResp{StatusCode: 200, Header: sse, Body: "data: a\n\ndata: b", Marks: marks()})
	check("a stream that ended inside its last event", cut, out)

	raw, err := yamlLib.Marshal(HTTPSchema{
		Response: HTTPResp{StatusCode: 200, Header: sse, Body: "data: a\n\ndata: b", Marks: marks()},
		Fault:    &Fault{Kind: FaultReset, AfterHead: true, BodyBytes: 16},
	})
	if err != nil {
		t.Fatal(err)
	}
	var fault HTTPSchema
	if err := yamlLib.Unmarshal(raw, &fault); err != nil {
		t.Fatal(err)
	}
	check("a fault's body", fault.Response, string(raw))
}

// A stream whose frames the marks cannot place (a line too long for the
// frames' parse, which keeps the body as one frame) keeps no marks: replay
// writes it at once.
func TestUnplacedFramesKeepNoMarks(t *testing.T) {
	body := "{\"n\":1}\n" + strings.Repeat("x", maxModelStreamTokenSize+1) + "\n"
	resp := HTTPResp{
		StatusCode: 200,
		Header:     map[string]string{"Content-Type": "application/x-ndjson"},
		Body:       body,
		Marks:      &HTTPAnswerMarks{Head: sec(0), Writes: []HTTPAnswerWrite{{Offset: 8, TS: sec(1)}, {Offset: len(body), TS: sec(2)}}},
	}
	back, _ := roundTripYAML(t, resp)
	if back.Marks != nil {
		t.Errorf("marks = %+v, want none", back.Marks)
	}
}

// BSON keeps the marks' times to the nanosecond, JSON keeps them, and a
// clone's marks are its own.
func TestMarksSurviveBSONJSONAndClone(t *testing.T) {
	resp := HTTPResp{StatusCode: 200, Body: "abcd", Marks: &HTTPAnswerMarks{Head: sec(1), Writes: []HTTPAnswerWrite{{Offset: 2, TS: sec(2)}, {Offset: 4, TS: sec(3)}}}}
	raw, err := bson.Marshal(resp)
	if err != nil {
		t.Fatal(err)
	}
	var fromBSON HTTPResp
	if err := bson.Unmarshal(raw, &fromBSON); err != nil {
		t.Fatal(err)
	}
	js, err := json.Marshal(resp)
	if err != nil {
		t.Fatal(err)
	}
	var fromJSON HTTPResp
	if err := json.Unmarshal(js, &fromJSON); err != nil {
		t.Fatal(err)
	}
	for name, got := range map[string]*HTTPAnswerMarks{"bson": fromBSON.Marks, "json": fromJSON.Marks} {
		if got == nil || !got.Head.Equal(sec(1)) || len(got.Writes) != 2 || got.Writes[1].Offset != 4 || !got.Writes[1].TS.Equal(sec(3)) {
			t.Errorf("%s: marks = %+v", name, got)
		}
	}
	var none HTTPResp
	raw, err = bson.Marshal(HTTPResp{StatusCode: 200})
	if err != nil {
		t.Fatal(err)
	}
	if err := bson.Unmarshal(raw, &none); err != nil || none.Marks != nil {
		t.Errorf("no marks read back as %+v (%v)", none.Marks, err)
	}
	c := resp.clone()
	if c.Marks == nil || !c.Marks.Head.Equal(sec(1)) || len(c.Marks.Writes) != 2 {
		t.Fatalf("a clone's marks = %+v, want the response's", c.Marks)
	}
	c.Marks.Writes[0].Offset = 3
	if resp.Marks.Writes[0].Offset != 2 {
		t.Error("a clone shares its marks")
	}
}

// A write that ends inside the CRLF blank line that ends an SSE frame did not
// complete it: the frame takes the time of the write that ends the blank line.
func TestAFrameEndsWithItsWholeCRLFBlankLine(t *testing.T) {
	body := "data: a\r\n\r\ndata: b\r\n\r\n"
	resp := HTTPResp{
		StatusCode: 200,
		Header:     map[string]string{"Content-Type": "text/event-stream"},
		Body:       body,
		Marks: &HTTPAnswerMarks{Head: sec(0), Writes: []HTTPAnswerWrite{
			{Offset: len("data: a\r\n\r"), TS: sec(1)},
			{Offset: len("data: a\r\n\r\n"), TS: sec(2)},
			{Offset: len(body), TS: sec(3)},
		}},
	}
	back, out := roundTripYAML(t, resp)
	if len(back.StreamBody) != 2 || !back.StreamBody[0].TS.Equal(sec(2)) || !back.StreamBody[1].TS.Equal(sec(3)) {
		t.Fatalf("frames = %+v, want frame 1 at write 2 and frame 2 at write 3:\n%s", back.StreamBody, out)
	}
	if got := pieces(t, back); strings.Join(got, "|") != "data:a\n\n|data:b\n\n" {
		t.Errorf("pieces = %q", got)
	}
	if !back.Marks.Writes[0].TS.Equal(sec(2)) {
		t.Errorf("frame 1's piece at %v, want %v", back.Marks.Writes[0].TS, sec(2))
	}
}

// The terminal write of an NDJSON stream, later than its last line, keeps its
// own mark at the body's end: replay ends the answer at its place, not at
// the last line's.
func TestTheTerminalWriteKeepsItsOwnMark(t *testing.T) {
	body := "{\"n\":1}\n{\"n\":2}\n"
	resp := HTTPResp{
		StatusCode: 200,
		Header:     map[string]string{"Content-Type": "application/x-ndjson"},
		Body:       body,
		Marks: &HTTPAnswerMarks{Head: sec(0), Writes: []HTTPAnswerWrite{
			{Offset: 8, TS: sec(1)},
			{Offset: len(body), TS: sec(2)},
			{Offset: len(body), TS: sec(5)},
		}},
	}
	back, out := roundTripYAML(t, resp)
	if len(back.StreamBody) != 2 || !back.StreamBody[1].TS.Equal(sec(2)) {
		t.Fatalf("frames = %+v:\n%s", back.StreamBody, out)
	}
	ws := back.Marks.Writes
	if len(ws) != 3 || ws[1].Offset != len(back.Body) || !ws[1].TS.Equal(sec(2)) ||
		ws[2].Offset != len(back.Body) || !ws[2].TS.Equal(sec(5)) {
		t.Errorf("writes = %+v, want the last line at 2s and the end at 5s, both at %d", ws, len(back.Body))
	}
}

// An NDJSON body of blank lines only has no line to place: it reads back as
// the body it was, and its marks index that body.
func TestBlankNDJSONKeepsMarksOnItsBody(t *testing.T) {
	body := "\n\n\n"
	resp := HTTPResp{
		StatusCode: 200,
		Header:     map[string]string{"Content-Type": "application/x-ndjson"},
		Body:       body,
		Marks:      &HTTPAnswerMarks{Head: sec(0), Writes: []HTTPAnswerWrite{{Offset: 1, TS: sec(1)}, {Offset: 3, TS: sec(2)}}},
	}
	back, out := roundTripYAML(t, resp)
	if back.Body != body || len(back.StreamBody) != 1 || back.Marks == nil {
		t.Fatalf("read back %q, %d frames, marks %+v:\n%s", back.Body, len(back.StreamBody), back.Marks, out)
	}
	if ws := back.Marks.Writes; len(ws) != 1 || ws[0].Offset != len(body) || !ws[0].TS.Equal(sec(2)) {
		t.Errorf("writes = %+v, want the whole body at 2s", ws)
	}
}

// A response a producer gave only as frames (StreamBody, no Body) keeps the
// marks it came with: they already index the body its frames read back as.
func TestFramesOnlyResponseKeepsItsMarks(t *testing.T) {
	frames := SplitStreamBody(HTTPStreamSSE, "data: a\n\ndata: b\n\n")
	frames[0].TS, frames[1].TS = sec(1), sec(2)
	marks := &HTTPAnswerMarks{Head: sec(0), Writes: []HTTPAnswerWrite{{Offset: 8, TS: sec(1)}, {Offset: 14, TS: sec(2)}}}
	resp := HTTPResp{StatusCode: 200, Header: map[string]string{"Content-Type": "text/event-stream"}, StreamBody: frames, Marks: marks}
	back, out := roundTripYAML(t, resp)
	if back.Marks == nil || len(back.Marks.Writes) != 2 || back.Marks.Writes[0].Offset != 8 || !back.Marks.Writes[1].TS.Equal(sec(2)) {
		t.Fatalf("marks = %+v, want them as given:\n%s", back.Marks, out)
	}
}

// recovered runs f and reports a panic in it as the row's failure.
func recovered(t *testing.T, what string, f func()) {
	t.Helper()
	defer func() {
		if r := recover(); r != nil {
			t.Errorf("%s: panicked: %v", what, r)
		}
	}()
	f()
}

// The end of no marks, or of marks with no write, is no time; else it is the
// last write's.
func TestAnswerMarksEndOfNoWriteIsZero(t *testing.T) {
	for name, m := range map[string]*HTTPAnswerMarks{"no marks": nil, "no write": {Head: sec(1)}} {
		recovered(t, name, func() {
			if got := m.End(); !got.IsZero() {
				t.Errorf("%s: End = %v, want zero", name, got)
			}
		})
	}
	m := &HTTPAnswerMarks{Head: sec(1), Writes: []HTTPAnswerWrite{{Offset: 4, TS: sec(2)}, {Offset: 8, TS: sec(3)}}}
	if got := m.End(); !got.Equal(sec(3)) {
		t.Errorf("End = %v, want %v", got, sec(3))
	}
}

// The BSON form of no marks is none, and of marks each of them.
func TestAnswerMarksBSONFormKeepsEachMark(t *testing.T) {
	recovered(t, "no marks", func() {
		if b := (*HTTPAnswerMarks)(nil).toBSON(); b != nil {
			t.Errorf("no marks: %+v, want none", b)
		}
	})
	m := &HTTPAnswerMarks{Head: sec(1), Writes: []HTTPAnswerWrite{{Offset: 4, TS: sec(2)}, {Offset: 8, TS: sec(3)}}}
	recovered(t, "marks", func() {
		b := m.toBSON()
		if b == nil || b.Head != FormatMockTimestamp(sec(1)) || len(b.Writes) != 2 || b.Writes[1].Offset != 8 {
			t.Errorf("marks: %+v, want the head and both writes", b)
		}
	})
}

// Each kept frame's piece of the read-back body ends after the blank line
// that follows it, and the last one at the body's end.
func TestLegacyFramePiecesEndAfterEachFrame(t *testing.T) {
	chunks := parseSSEBodyToChunks("data: a\n\ndata: b\n\ndata: c\n\n", time.Time{})
	body := streamChunksToLegacyBody(chunks, streamBodyKindSSE)
	got, ok := legacyFramePieces(chunks, streamBodyKindSSE)
	want := []int{strings.Index(body, "data:b"), strings.Index(body, "data:c"), len(body)}
	if !ok || len(got) != len(want) || got[0] != want[0] || got[1] != want[1] || got[2] != want[2] {
		t.Fatalf("pieces of %q = %v (%v), want %v", body, got, ok, want)
	}
}

// Chunks none of which is a frame (an SSE frame with no field written, a line
// with no raw field) have no pieces, though the body they read back as is
// empty or one line, and so ends where their no frames do: the answer keeps
// no marks.
func TestLegacyFramePiecesOfNoFrameAreNone(t *testing.T) {
	for _, tc := range []struct {
		name   string
		chunks []HTTPStreamChunk
		kind   streamBodyKind
	}{
		{"an SSE frame with no field", []HTTPStreamChunk{{Data: []HTTPStreamDataField{{Key: " ", Value: "a"}}}}, streamBodyKindSSE},
		{"a line with no raw field", []HTTPStreamChunk{{Data: []HTTPStreamDataField{{Key: "line", Value: "a"}}}}, streamBodyKindRaw},
		{"a line with no field, which reads back as no body", []HTTPStreamChunk{{}}, streamBodyKindRaw},
	} {
		if got, ok := legacyFramePieces(tc.chunks, tc.kind); ok || got != nil {
			t.Errorf("%s: pieces %v (%v), want none", tc.name, got, ok)
		}
	}
}

// Frames whose pieces stop short of the body they read back as have no
// pieces: a line with no raw field after a frame, which the body reads back
// by its first field, and a last raw line that is blank, which the body ends
// with one newline more. The answer keeps no marks, so none of them points
// into the wrong bytes.
func TestLegacyFramePiecesThatStopShortOfTheBodyAreNone(t *testing.T) {
	raw := func(v string) HTTPStreamChunk {
		return HTTPStreamChunk{Data: []HTTPStreamDataField{{Key: "raw", Value: v}}}
	}
	for _, tc := range []struct {
		name   string
		chunks []HTTPStreamChunk
		body   string
	}{
		{"a line with no raw field after a frame", []HTTPStreamChunk{raw("a"), {Data: []HTTPStreamDataField{{Key: "line", Value: "b"}}}}, "a\nb"},
		{"a blank last line", []HTTPStreamChunk{raw("a"), raw("")}, "a\n\n"},
	} {
		if body := streamChunksToLegacyBody(tc.chunks, streamBodyKindRaw); body != tc.body {
			t.Fatalf("%s: body %q, want %q", tc.name, body, tc.body)
		}
		if got, ok := legacyFramePieces(tc.chunks, streamBodyKindRaw); ok || got != nil {
			t.Errorf("%s: pieces %v (%v), want none", tc.name, got, ok)
		}
	}
}

// A chunk before the first frame has no bytes of the body: its piece ends at
// the body's start, not after a separator.
func TestLegacyFramePiecesBeforeTheFirstFrameEndAtTheStart(t *testing.T) {
	chunks := append([]HTTPStreamChunk{{Data: []HTTPStreamDataField{{Key: " ", Value: "x"}}}},
		parseSSEBodyToChunks("data: a\n\ndata: b\n\n", time.Time{})...)
	body := streamChunksToLegacyBody(chunks, streamBodyKindSSE)
	got, ok := legacyFramePieces(chunks, streamBodyKindSSE)
	want := []int{0, strings.Index(body, "data:b"), len(body)}
	if !ok || len(got) != len(want) || got[0] != want[0] || got[1] != want[1] || got[2] != want[2] {
		t.Fatalf("pieces of %q = %v (%v), want %v", body, got, ok, want)
	}
}

// A frame ends after the blank line that ends it, read with CR, LF or CRLF
// line ends; a blank line with no frame before it ends none.
func TestSSEFrameEndsAreTheFramesTheParserKeeps(t *testing.T) {
	for _, tc := range []struct {
		body string
		want []int
	}{
		{"data: a\r\rdata: b\r\r", []int{9, 18}},
		{"data: a\n\n\ndata: b\n\n", []int{9, 19}},
		{"data: a\r\n\r\ndata: b", []int{11, 18}},
	} {
		got := sseFrameEnds(tc.body)
		if len(got) != len(tc.want) || (len(got) > 0 && (got[0] != tc.want[0] || got[len(got)-1] != tc.want[len(tc.want)-1])) {
			t.Errorf("sseFrameEnds(%q) = %v, want %v", tc.body, got, tc.want)
		}
	}
}

// A stream is written as its frames with its marks moved onto them, or with
// none when it has none.
func TestStreamsMarshalWithAndWithoutMarks(t *testing.T) {
	body := "data: a\n\ndata: b\n\n"
	for name, marks := range map[string]*HTTPAnswerMarks{
		"no marks": nil,
		"marks":    {Head: sec(1), Writes: []HTTPAnswerWrite{{Offset: 9, TS: sec(1)}, {Offset: len(body), TS: sec(2)}}},
	} {
		recovered(t, name, func() {
			resp := HTTPResp{StatusCode: 200, Header: map[string]string{"Content-Type": "text/event-stream"}, Body: body, Timestamp: sec(9), Marks: marks}
			back, out := roundTripYAML(t, resp)
			if len(back.StreamBody) != 2 || (back.Marks != nil) != (marks != nil) {
				t.Errorf("%s: frames %d, marks %+v:\n%s", name, len(back.StreamBody), back.Marks, out)
			}
		})
	}
}