pkg/util_stream_json_test.go
Rama main · réplica de lectura del repositorio original.
package pkg
import (
"encoding/json"
"errors"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"github.com/stretchr/testify/require"
"go.uber.org/zap"
yamlLib "gopkg.in/yaml.v3"
"go.keploy.io/server/v3/pkg/models"
"go.keploy.io/server/v3/pkg/provenswitch"
)
// A stream test case that reaches the replay as JSON (a cloud replay's, or
// any carried as JSON) is compared by what its recording kept: a stream that
// changed fails, under either switch. JSON carries its body, not its frames
// (StreamBody), and the replay compared the app's stream with no recorded
// frame: with the switch off it passed on the first frame that came, however
// different, and under the switch it failed, however alike.
// recordedThroughYAML is resp as a recording keeps it: written as a test
// case's YAML and read back, so a stream's body is kept as its frames.
func recordedThroughYAML(t *testing.T, resp models.HTTPResp) models.HTTPResp {
t.Helper()
out, err := yamlLib.Marshal(resp)
require.NoError(t, err)
var back models.HTTPResp
require.NoError(t, yamlLib.Unmarshal(out, &back))
return back
}
// throughJSON is tc as the replay gets it from JSON.
func throughJSON(t *testing.T, tc *models.TestCase) *models.TestCase {
t.Helper()
out, err := json.Marshal(tc)
require.NoError(t, err)
var back models.TestCase
require.NoError(t, json.Unmarshal(out, &back))
return &back
}
// streamingApp answers every request with body under contentType, frame by
// frame (each flushed), and ends the response.
func streamingApp(t *testing.T, contentType string, frames []string) *httptest.Server {
t.Helper()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", contentType)
flusher, ok := w.(http.Flusher)
require.True(t, ok)
w.WriteHeader(http.StatusOK)
for _, f := range frames {
_, _ = w.Write([]byte(f))
flusher.Flush()
}
}))
t.Cleanup(srv.Close)
return srv
}
// compareAgainstApp sends tc to an app streaming frames and compares its
// answer as the replay does.
func compareAgainstApp(t *testing.T, tc *models.TestCase, contentType string, frames []string) (bool, *StreamMismatchInfo, error) {
t.Helper()
srv := streamingApp(t, contentType, frames)
sent := *tc
sent.HTTPReq.URL = srv.URL + "/events"
resp, err := SimulateHTTPStreaming(t.Context(), &sent, "test-set-0", zap.NewNop(), SimulationConfig{APITimeout: 5})
require.NoError(t, err)
defer resp.Reader.Close()
require.NotEqual(t, HTTPStreamModeNone, resp.StreamConfig.Mode, "the answer is read as a stream")
ok, _, info, err := CompareHTTPStream(sent.HTTPResp, resp.Reader, resp.StreamConfig, map[string]struct{}{}, zap.NewNop())
return ok, info, err
}
type jsonStreamCase struct {
name string
contentType string
recorded []string // the frames the recording saw, as they went out
changed []string // the same stream, its content changed throughout
}
var jsonStreamCases = []jsonStreamCase{
{
name: "sse",
contentType: "text/event-stream",
recorded: []string{"id: 1\nevent: price\ndata: {\"sku\":\"a\",\"price\":10}\n\n", "id: 2\nevent: price\ndata: {\"sku\":\"b\",\"price\":20}\n\n"},
changed: []string{"id: 7\nevent: stock\ndata: {\"sku\":\"x\",\"price\":99}\n\n", "id: 8\nevent: stock\ndata: {\"sku\":\"y\",\"price\":98}\n\n"},
},
{
name: "ndjson",
contentType: "application/x-ndjson",
recorded: []string{"{\"sku\":\"a\",\"price\":10}\n", "{\"sku\":\"b\",\"price\":20}\n"},
changed: []string{"{\"sku\":\"x\",\"price\":99}\n", "{\"sku\":\"y\",\"price\":98}\n"},
},
// No text/plain row: a text body is a stream only on its recording's
// frames (models.HTTPStreamOf), which JSON does not carry, so carried as
// JSON it is compared whole (TestATextStreamCarriedAsJSONIsComparedWhole).
}
// A text/plain test case is a stream test only on the frames its recording
// kept (an older keploy's: a recording now keeps a text body whole), and JSON
// does not carry them. Carried as JSON it is no stream test, and is compared
// whole, as any response is, by its body: the wire of its lines, a blank one
// last included, so the unchanged answer is that body byte for byte.
func TestATextStreamCarriedAsJSONIsComparedWhole(t *testing.T) {
const wire = "started\n\ndone\n\n"
var frames []models.HTTPStreamChunk
for _, line := range []string{"started", "", "done", ""} {
frames = append(frames, models.HTTPStreamChunk{Data: []models.HTTPStreamDataField{{Key: "raw", Value: line}}})
}
switchStates(t, func(t *testing.T) {
recorded := streamTestCase("text/plain", "")
recorded.HTTPResp.StreamBody = frames
recorded.HTTPResp = recordedThroughYAML(t, recorded.HTTPResp)
require.Len(t, recorded.HTTPResp.StreamBody, len(frames), "its YAML keeps the frames an older keploy recorded")
require.True(t, IsHTTPStreamingTestCase(recorded), "framed, it is a stream test")
carried := throughJSON(t, recorded)
require.Empty(t, carried.HTTPResp.StreamBody)
require.False(t, IsHTTPStreamingTestCase(carried), "carried as JSON, a text body is no stream: it is compared whole")
require.Equal(t, wire, carried.HTTPResp.Body, "its body is the wire of its lines")
})
}
func switchStates(t *testing.T, run func(t *testing.T)) {
for _, on := range []bool{false, true} {
name := "switch-off"
if on {
name = "switch-on"
}
t.Run(name, func(t *testing.T) {
if on {
t.Setenv(provenswitch.Env, "1")
} else {
t.Setenv(provenswitch.Env, "")
}
run(t)
})
}
}
func streamTestCase(contentType, body string) *models.TestCase {
return &models.TestCase{
Name: "stream-1",
Kind: models.HTTP,
HTTPReq: models.HTTPReq{
Method: "GET",
URL: "http://localhost/events",
Header: map[string]string{"Accept": contentType},
},
HTTPResp: models.HTTPResp{
StatusCode: http.StatusOK,
Header: map[string]string{"Content-Type": contentType},
Body: body,
Timestamp: time.Date(2026, 10, 20, 9, 0, 0, 0, time.UTC),
},
}
}
// A recorded stream test case carried as JSON fails when its stream changed,
// and passes when it did not.
func TestAStreamTestCarriedAsJSONFailsWhenItsStreamChanged(t *testing.T) {
for _, c := range jsonStreamCases {
t.Run(c.name, func(t *testing.T) {
switchStates(t, func(t *testing.T) {
recorded := streamTestCase(c.contentType, strings.Join(c.recorded, ""))
recorded.HTTPResp = recordedThroughYAML(t, recorded.HTTPResp)
require.Len(t, recorded.HTTPResp.StreamBody, len(c.recorded), "the recording kept the stream's frames")
carried := throughJSON(t, recorded)
ok, info, err := compareAgainstApp(t, carried, c.contentType, c.changed)
require.NoError(t, err)
require.False(t, ok, "a stream whose content changed throughout must not pass")
require.NotNil(t, info)
require.Equal(t, 0, info.FrameIndex, "the first frame differs")
ok, info, err = compareAgainstApp(t, carried, c.contentType, c.recorded)
require.NoError(t, err)
require.True(t, ok, "the recorded stream passes: %+v", info)
})
})
}
}
// A cloud test case arrives as JSON with its recorded body and no frames (the
// DaemonSet agent records a body, and api-server keeps and serves it as
// JSON): it is compared by its body split into frames, as a recording of it
// would have kept them.
func TestACloudStreamTestCaseIsComparedByItsRecordedBody(t *testing.T) {
for _, c := range jsonStreamCases {
t.Run(c.name, func(t *testing.T) {
switchStates(t, func(t *testing.T) {
wire, err := json.Marshal(map[string]any{
"name": "stream-1",
"kind": "Http",
"http_req": map[string]any{
"method": "GET",
"url": "http://localhost/events",
"header": map[string]string{"Accept": c.contentType},
},
"http_resp": map[string]any{
"status_code": http.StatusOK,
"header": map[string]string{"Content-Type": c.contentType},
"body": strings.Join(c.recorded, ""),
},
})
require.NoError(t, err)
var tc models.TestCase
require.NoError(t, json.Unmarshal(wire, &tc))
require.Empty(t, tc.HTTPResp.StreamBody)
ok, info, err := compareAgainstApp(t, &tc, c.contentType, c.changed)
require.NoError(t, err)
require.False(t, ok, "a stream whose content changed throughout must not pass")
require.NotNil(t, info)
require.Equal(t, 0, info.FrameIndex)
ok, info, err = compareAgainstApp(t, &tc, c.contentType, c.recorded)
require.NoError(t, err)
require.True(t, ok, "the recorded stream passes: %+v", info)
// One frame changed, the last: still a failure, at that frame.
lastChanged := append(append([]string{}, c.recorded[:len(c.recorded)-1]...), c.changed[len(c.changed)-1])
ok, info, err = compareAgainstApp(t, &tc, c.contentType, lastChanged)
require.NoError(t, err)
require.False(t, ok)
require.NotNil(t, info)
require.Equal(t, len(c.recorded)-1, info.FrameIndex)
})
})
}
}
// The body JSON carries of a stream read from its recorded frames reads, as
// the app's stream is read, as those frames: an SSE event whose data spans
// lines, a blank one among them, or starts with a space, and a comment, are
// no other event read back from it. Carried as JSON, the recorded stream
// passes, and a changed one fails at the frame that changed.
func TestAStreamCarriedAsJSONIsComparedByTheFramesItsRecordingKept(t *testing.T) {
const contentType = "text/event-stream"
recordedFrames := []string{
"id: 1\nevent: log\ndata: line one\ndata: line two\n\n",
"data: indented\n\n",
"data: before\ndata:\ndata: after\n\n",
": keepalive\n\n",
"event: tick\ndata:\n\n",
}
switchStates(t, func(t *testing.T) {
recorded := streamTestCase(contentType, strings.Join(recordedFrames, ""))
recorded.HTTPResp = recordedThroughYAML(t, recorded.HTTPResp)
require.Len(t, recorded.HTTPResp.StreamBody, len(recordedFrames))
carried := throughJSON(t, recorded)
ok, info, err := compareAgainstApp(t, carried, contentType, recordedFrames)
require.NoError(t, err)
require.True(t, ok, "the recorded stream passes: %+v", info)
changed := append([]string{}, recordedFrames...)
changed[2] = "data: before\ndata:\ndata: AFTER\n\n"
ok, info, err = compareAgainstApp(t, carried, contentType, changed)
require.NoError(t, err)
require.False(t, ok)
require.NotNil(t, info)
require.Equal(t, 2, info.FrameIndex)
})
}
// A recorded body that cannot be read as the answer is read (a frame or line
// longer than a stream's read takes) fails the test with that reason: it is
// neither compared by the frames before it alone nor taken for a recording of
// nothing.
func TestARecordedStreamBodyThatCannotBeReadFails(t *testing.T) {
long := strings.Repeat("a", maxStreamTokenSize+1)
for _, c := range []struct {
name, contentType, body string
cfg HTTPStreamConfig
}{
{"sse", "text/event-stream", "data: " + long + "\n\n", HTTPStreamConfig{Mode: HTTPStreamModeSSE}},
{"ndjson", "application/x-ndjson", "{\"a\":1}\n" + long + "\n", HTTPStreamConfig{Mode: HTTPStreamModeNDJSON}},
{"plain-text", "text/plain", "first\n" + long + "\n", HTTPStreamConfig{Mode: HTTPStreamModePlainText}},
} {
t.Run(c.name, func(t *testing.T) {
tc := streamTestCase(c.contentType, c.body)
ok, _, info, err := CompareHTTPStream(tc.HTTPResp, strings.NewReader(c.body), c.cfg, map[string]struct{}{}, zap.NewNop())
require.False(t, ok)
require.Nil(t, info)
require.Error(t, err)
require.False(t, errors.Is(err, ErrStreamNotRecorded), "not a recording of nothing: %v", err)
require.Contains(t, err.Error(), "failed to read the recorded")
})
}
}
// A stream test whose recording saved nothing of its stream (a body over 1 MB,
// which a recording does not save: BodySkipped) has nothing to compare the
// app's stream with. It is refused, with the reason, under either switch, and
// never passes. A recording that saved its body is compared with it, however
// little it holds (TestAStreamRecordedEmptyIsComparedWithItsEmptyStream,
// TestAStreamRecordedWithKeepalivesAloneIsComparedWithWhatItHolds).
func TestAStreamTestWithNothingRecordedIsRefused(t *testing.T) {
modes := []struct {
name string
contentType string
cfg HTTPStreamConfig
answer string
}{
{"sse", "text/event-stream", HTTPStreamConfig{Mode: HTTPStreamModeSSE}, "data: {\"a\":1}\n\n"},
{"ndjson", "application/x-ndjson", HTTPStreamConfig{Mode: HTTPStreamModeNDJSON}, "{\"a\":1}\n"},
{"plain-text", "text/plain", HTTPStreamConfig{Mode: HTTPStreamModePlainText}, "line one\n"},
{"multipart", "multipart/x-mixed-replace; boundary=frame", HTTPStreamConfig{Mode: HTTPStreamModeMultipart, Boundary: "frame"}, "--frame\r\nContent-Type: text/plain\r\n\r\npart one\r\n--frame--\r\n"},
{"binary", "application/octet-stream", HTTPStreamConfig{Mode: HTTPStreamModeBinary}, "\x00\x01\x02"},
}
recordings := []struct {
name string
resp func(contentType string) models.HTTPResp
why string
}{
{"body not saved", func(ct string) models.HTTPResp {
return models.HTTPResp{StatusCode: 200, Header: map[string]string{"Content-Type": ct}, BodySkipped: true, BodySize: 2 << 20}
}, "its body, 2097152 bytes, was over 1 MB"},
{"body not saved, its size not either", func(ct string) models.HTTPResp {
return models.HTTPResp{StatusCode: 200, Header: map[string]string{"Content-Type": ct}, BodySkipped: true}
}, "(its body was over 1 MB"},
}
for _, m := range modes {
for _, rec := range recordings {
t.Run(m.name+"/"+rec.name, func(t *testing.T) {
switchStates(t, func(t *testing.T) {
resp := rec.resp(m.contentType)
// What a recording and JSON make of it.
resp = recordedThroughYAML(t, resp)
tc := throughJSON(t, &models.TestCase{Kind: models.HTTP, HTTPResp: resp})
answer := &countingReader{r: strings.NewReader(m.answer)}
ok, _, info, err := CompareHTTPStream(tc.HTTPResp, answer, m.cfg, map[string]struct{}{}, zap.NewNop())
require.False(t, ok, "nothing recorded must never pass")
require.Nil(t, info, "refused, not judged different")
require.True(t, errors.Is(err, ErrStreamNotRecorded), "refused: %v", err)
require.Contains(t, err.Error(), rec.why)
require.Zero(t, answer.n, "refused before its stream is read")
})
})
}
}
}
// A stream recorded empty (its body saved, and empty: the app ended the
// stream with nothing in it) is compared with that, as any recording is, under
// either switch, YAML or carried as JSON: the app's stream is read to its end,
// and an answer that ends with nothing passes, one that streams a frame fails,
// judged different, and one of a keepalive has nothing of it to compare, not
// judged. It is not refused: re-recording gives the same empty stream. With
// the switch off a stream that came passed, compared with no recorded frame.
func TestAStreamRecordedEmptyIsComparedWithItsEmptyStream(t *testing.T) {
modes := []struct {
name string
contentType string
cfg HTTPStreamConfig
frame string // an answer of a frame
keepalive string // an answer of a keepalive, where the format has one
differs string // the mismatch's reason
}{
{"sse", "text/event-stream", HTTPStreamConfig{Mode: HTTPStreamModeSSE}, "data: {\"a\":1}\n\n", ": ping\n\n", "the stream went on after its 0 recorded frames"},
{"ndjson", "application/x-ndjson", HTTPStreamConfig{Mode: HTTPStreamModeNDJSON}, "{\"a\":1}\n", "\n", "the stream went on after its 0 recorded frames"},
{"plain-text", "text/plain", HTTPStreamConfig{Mode: HTTPStreamModePlainText}, "\n", "", "the stream went on after its 0 recorded frames"},
{"multipart", "multipart/x-mixed-replace; boundary=frame", HTTPStreamConfig{Mode: HTTPStreamModeMultipart, Boundary: "frame"}, "--frame\r\nContent-Type: text/plain\r\n\r\npart one\r\n--frame--\r\n", "", "the stream went on after its 0 recorded frames"},
{"binary", "application/octet-stream", HTTPStreamConfig{Mode: HTTPStreamModeBinary}, "\x00", "", "size mismatch: expected 0 bytes, got 1 bytes"},
}
for _, m := range modes {
for _, carried := range []string{"yaml", "json"} {
t.Run(m.name+"/"+carried, func(t *testing.T) {
switchStates(t, func(t *testing.T) {
resp := recordedThroughYAML(t, models.HTTPResp{StatusCode: 200, Header: map[string]string{"Content-Type": m.contentType}})
if carried == "json" {
resp = throughJSON(t, &models.TestCase{Kind: models.HTTP, HTTPResp: resp}).HTTPResp
}
compare := func(answer string) (bool, *StreamMismatchInfo, error) {
ok, _, info, err := CompareHTTPStream(resp, strings.NewReader(answer), m.cfg, map[string]struct{}{}, zap.NewNop())
return ok, info, err
}
ok, info, err := compare("")
require.NoError(t, err)
require.True(t, ok, "an empty stream, as recorded, passes: %+v", info)
ok, info, err = compare(m.frame)
require.NoError(t, err, "judged, not refused")
require.False(t, ok, "a stream that came where none was recorded must not pass")
require.NotNil(t, info)
require.Equal(t, m.differs, info.Reason)
if m.keepalive != "" {
ok, info, err = compare(m.keepalive)
require.False(t, ok, "a keepalive is nothing of the answer to compare")
require.Nil(t, info)
require.True(t, errors.Is(err, ErrAnswerStalled), "not judged: %v", err)
require.False(t, errors.Is(err, ErrStreamNotRecorded), "compared, not refused: %v", err)
}
})
})
}
}
}
// A stream recorded with keepalives or blank lines alone, its body saved, is
// compared with what it holds, kept as YAML or carried as JSON, never
// refused: re-recording the app gives the same recording.
//
// An SSE body of comments (": ok", which a notification endpoint that had no
// event sends as it opens the stream) holds frames, its comments, and is
// compared by them in order, as any recording is: with the switch off the
// comparison stops at the last of them (a stream that goes on is not waited
// for), and under the switch it reads the stream to its end, an event past
// them failing it and a keepalive past them ending the read, not judged (the
// app held the stream open past its recording).
//
// A body that holds no frame (an SSE body of blank lines, an NDJSON body of
// blank lines, which a recording keeps none of, a multipart body of no part)
// has none to stop at, so the stream is read to its end under either switch:
// a stream that ends as recorded passes, one with an event (a line, a part)
// fails, judged different, and one of a keepalive has nothing of it to
// compare, not judged. With the switch off such a stream passed on its first
// frame, compared with nothing.
func TestAStreamRecordedWithKeepalivesAloneIsComparedWithWhatItHolds(t *testing.T) {
const (
pass = "passes"
notJudged = "not judged"
heldOpen = "held open"
part = "--frame\r\nContent-Type: text/plain\r\n\r\npart one\r\n--frame--\r\n"
)
sse := HTTPStreamConfig{Mode: HTTPStreamModeSSE}
ndjson := HTTPStreamConfig{Mode: HTTPStreamModeNDJSON}
multi := HTTPStreamConfig{Mode: HTTPStreamModeMultipart, Boundary: "frame"}
type answer struct {
body string
// want is pass, notJudged, heldOpen, or the reason it differs;
// wantOff, when set, is that with the switch off.
want, wantOff string
}
for _, c := range []struct {
name, contentType, body string
cfg HTTPStreamConfig
answers []answer
}{
{"sse of a keepalive", "text/event-stream", ": ok\n\n", sse, []answer{
{": ok\n\n", pass, ""},
{": ok\n\ndata: x\n\n", "the stream went on after its 1 recorded frames", pass},
{"data: x\n\n", "comment-position mismatch", ""},
{": ok\n\n: ok\n\n", heldOpen, pass},
{"", "expected 1 frames but only received 0", ""},
}},
{"sse of keepalives", "text/event-stream", ": ping\n\n: ping\n\n", sse, []answer{
{": ping\n\n: ping\n\n", pass, ""},
{": ping\n\n: ping\n\nevent: tick\ndata: 1\n\n", "the stream went on after its 2 recorded frames", pass},
{": ping\n\n: ping\n\n: ping\n\n", heldOpen, pass},
{": ping\n\n", "expected 2 frames but only received 1", ""},
}},
{"sse of blank lines", "text/event-stream", "\n\n\n\n", sse, []answer{
{"", pass, ""},
{"\n\n", pass, ""},
{"data: x\n\n", "the stream went on after its 0 recorded frames", ""},
{": ping\n\n", notJudged, ""},
}},
{"ndjson of blank lines", "application/x-ndjson", "\n\n", ndjson, []answer{
{"", pass, ""},
{"{\"a\":1}\n", "the stream went on after its 0 recorded frames", ""},
{"\n", notJudged, ""},
}},
{"multipart of no part", "multipart/x-mixed-replace; boundary=frame", "--frame--\r\n", multi, []answer{
{"", pass, ""},
{"--frame--\r\n", pass, ""},
{part, "the stream went on after its 0 recorded frames", ""},
}},
{"multipart of blank lines", "multipart/x-mixed-replace; boundary=frame", "\r\n\r\n", multi, []answer{
{"", pass, ""},
{part, "the stream went on after its 0 recorded frames", ""},
}},
} {
for _, carried := range []string{"yaml", "json"} {
t.Run(c.name+"/"+carried, func(t *testing.T) {
switchStates(t, func(t *testing.T) {
resp := recordedThroughYAML(t, models.HTTPResp{StatusCode: 200, Header: map[string]string{"Content-Type": c.contentType}, Body: c.body})
if carried == "json" {
resp = throughJSON(t, &models.TestCase{Kind: models.HTTP, HTTPResp: resp}).HTTPResp
}
for _, a := range c.answers {
want := a.want
if a.wantOff != "" && !provenswitch.On() {
want = a.wantOff
}
ok, _, info, err := CompareHTTPStream(resp, strings.NewReader(a.body), c.cfg, map[string]struct{}{}, zap.NewNop())
require.False(t, errors.Is(err, ErrStreamNotRecorded), "answer %q: compared, not refused: %v", a.body, err)
switch want {
case pass:
require.NoError(t, err, "answer %q", a.body)
require.True(t, ok, "answer %q, as recorded, passes: %+v", a.body, info)
case notJudged:
require.False(t, ok, "answer %q: keepalives alone are nothing of the answer to compare", a.body)
require.Nil(t, info, "answer %q", a.body)
require.True(t, errors.Is(err, ErrAnswerStalled), "answer %q: not judged: %v", a.body, err)
case heldOpen:
require.True(t, ok, "answer %q: what came matched", a.body)
require.Nil(t, info, "answer %q: what came matched", a.body)
require.True(t, errors.Is(err, ErrStreamHeldOpen), "answer %q: held open past its recording, not judged: %v", a.body, err)
default:
require.NoError(t, err, "answer %q: judged", a.body)
require.False(t, ok, "answer %q must not pass", a.body)
require.NotNil(t, info, "answer %q", a.body)
require.Equal(t, want, info.Reason, "answer %q", a.body)
}
}
})
})
}
}
}
// readPast is a stream that streams body, all of it in its first read, and is
// then held open: a read past body is one keploy waits on the app for, which
// returns errReadPast here.
type readPast struct {
body string
read bool
past int
}
var errReadPast = errors.New("read past the stream's recorded frames")
func (r *readPast) Read(p []byte) (int, error) {
if !r.read {
r.read = true
return copy(p, r.body), nil
}
r.past++
return 0, errReadPast
}
// With the switch off a stream's comparison stops at its last recorded frame
// (line, part, byte) and does not wait on a stream the app keeps open past
// it, whatever that frame is, an SSE comment among them: a recording of SSE
// comments alone holds frames, and is compared as any recording is. Read on,
// the stream the app held open ended only on the client's timeout, and the
// test failed on that. Under the switch every stream is read to its end.
func TestWithTheSwitchOffAStreamIsNotReadPastItsLastRecordedFrame(t *testing.T) {
for _, c := range []struct {
name, contentType, body string
cfg HTTPStreamConfig
answer string // what the app streams before it holds the stream open, when not body
}{
{"sse of a keepalive", "text/event-stream", ": ok\n\n", HTTPStreamConfig{Mode: HTTPStreamModeSSE}, ""},
{"sse of keepalives", "text/event-stream", ": ok\n\n: ping\n\n", HTTPStreamConfig{Mode: HTTPStreamModeSSE}, ""},
{"sse of an event", "text/event-stream", "data: a\n\n", HTTPStreamConfig{Mode: HTTPStreamModeSSE}, ""},
{"ndjson", "application/x-ndjson", "{\"a\":1}\n", HTTPStreamConfig{Mode: HTTPStreamModeNDJSON}, ""},
{"plain-text", "text/plain", "line one\n", HTTPStreamConfig{Mode: HTTPStreamModePlainText}, ""},
{"multipart", "multipart/x-mixed-replace; boundary=frame", "--frame\r\nContent-Type: text/plain\r\n\r\npart one\r\n--frame--\r\n", HTTPStreamConfig{Mode: HTTPStreamModeMultipart, Boundary: "frame"},
// The next part's boundary, and no part yet.
"--frame\r\nContent-Type: text/plain\r\n\r\npart one\r\n--frame\r\n"},
{"binary", "application/octet-stream", "\x00\x01\x02", HTTPStreamConfig{Mode: HTTPStreamModeBinary}, ""},
} {
for _, carried := range []string{"yaml", "json"} {
t.Run(c.name+"/"+carried, func(t *testing.T) {
switchStates(t, func(t *testing.T) {
resp := recordedThroughYAML(t, models.HTTPResp{StatusCode: 200, Header: map[string]string{"Content-Type": c.contentType}, Body: c.body})
if carried == "json" {
resp = throughJSON(t, &models.TestCase{Kind: models.HTTP, HTTPResp: resp}).HTTPResp
}
answer := c.answer
if answer == "" {
answer = c.body
}
stream := &readPast{body: answer}
ok, _, info, err := CompareHTTPStream(resp, stream, c.cfg, map[string]struct{}{}, zap.NewNop())
if provenswitch.On() {
require.NotZero(t, stream.past, "under the switch a stream is read to its end")
return
}
require.Zero(t, stream.past, "read past the last recorded frame of a stream the app holds open")
require.NoError(t, err)
require.True(t, ok, "every recorded frame came: %+v", info)
})
})
}
}
}
// A multipart stream recorded empty fails an answer of bytes that hold no
// part, none being multipart (no boundary), or a boundary line or a part cut
// off: those bytes are no empty stream. multipart.Reader reads the first two
// as the error "multipart: NextPart: EOF", which an answer of no byte is too.
func TestAMultipartStreamRecordedEmptyFailsAnAnswerOfNoPart(t *testing.T) {
cfg := HTTPStreamConfig{Mode: HTTPStreamModeMultipart, Boundary: "frame"}
for _, carried := range []string{"yaml", "json"} {
t.Run(carried, func(t *testing.T) {
switchStates(t, func(t *testing.T) {
resp := recordedThroughYAML(t, models.HTTPResp{StatusCode: 200, Header: map[string]string{"Content-Type": "multipart/x-mixed-replace; boundary=frame"}})
if carried == "json" {
resp = throughJSON(t, &models.TestCase{Kind: models.HTTP, HTTPResp: resp}).HTTPResp
}
for _, answer := range []string{
"hello",
"--frame",
"--frame\r\nContent-Type: text/plain\r\n\r\npart one",
} {
ok, _, info, err := CompareHTTPStream(resp, strings.NewReader(answer), cfg, map[string]struct{}{}, zap.NewNop())
require.False(t, ok, "answer %q holds no part, and must not pass", answer)
require.True(t, info != nil || err != nil, "answer %q: failed, with why", answer)
require.False(t, errors.Is(err, ErrStreamNotRecorded), "answer %q: compared, not refused: %v", answer, err)
}
})
})
}
}
// A response HTTP says has no body (the answer to a HEAD, a 1xx, 204 or 304)
// is no stream, whatever its Content-Type says: it is compared as any
// response is, by its status and headers, YAML or carried as JSON. Which
// response has none the answer says, not the recording. A recorded HEAD, 204,
// 304 or 1xx is a stream test as any of its Content-Type is, and is sent as
// one: an answer with no body (any answer to a HEAD) is compared as a
// response, and an answer of a stream to another method is read as a stream,
// bounded as a stream's read is (its first keepalive, or its first frame past
// the recording, ends it). Read whole, as
// a response's is, a stream the app never ends is never read to its end. An
// answer of no body to a recorded stream is not read as a stream either: it
// differs from the recording by its status.
func TestAResponseWithNoBodyIsNoStream(t *testing.T) {
contentTypes := []string{"text/event-stream", "application/x-ndjson", "multipart/x-mixed-replace; boundary=frame"}
answers := []struct {
name string
method string
status int
}{
{"HEAD", http.MethodHead, http.StatusOK},
{"204", http.MethodGet, http.StatusNoContent},
{"304", http.MethodGet, http.StatusNotModified},
{"103", http.MethodGet, http.StatusEarlyHints},
}
for _, ct := range contentTypes {
for _, a := range answers {
for _, carried := range []string{"yaml", "json"} {
t.Run(ct+"/"+a.name+"/"+carried, func(t *testing.T) {
tc := streamTestCase(ct, "")
tc.HTTPReq.Method = models.Method(a.method)
tc.HTTPResp.StatusCode = a.status
tc.HTTPResp = recordedThroughYAML(t, tc.HTTPResp)
if carried == "json" {
tc = throughJSON(t, tc)
}
require.True(t, IsHTTPStreamingTestCase(tc), "a recorded %s answered %d is sent as a stream test: the app's answer says whether it has a body", a.method, a.status)
answer := &http.Response{StatusCode: a.status, Header: http.Header{"Content-Type": {ct}}, Request: &http.Request{Method: a.method}}
require.Equal(t, HTTPStreamModeNone, DetectHTTPStreamConfig(tc, answer).Mode, "an answer with no body is no stream")
answer.Request = nil
require.Equal(t, HTTPStreamModeNone, DetectHTTPStreamConfig(tc, answer).Mode, "an answer naming no request answers the test's, of its recorded method")
if a.method != http.MethodHead {
streamed := &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": {ct}}, Request: &http.Request{Method: a.method}}
require.NotEqual(t, HTTPStreamModeNone, DetectHTTPStreamConfig(tc, streamed).Mode, "a recorded %d answered with a stream is read as one", a.status)
}
})
}
}
}
t.Run("a recorded stream answered with no body", func(t *testing.T) {
recorded := streamTestCase("text/event-stream", "data: a\n\n")
require.True(t, IsHTTPStreamingTestCase(recorded))
for _, status := range []int{http.StatusNoContent, http.StatusNotModified} {
answer := &http.Response{StatusCode: status, Header: http.Header{"Content-Type": {"text/event-stream"}}, Request: &http.Request{Method: http.MethodGet}}
require.Equal(t, HTTPStreamModeNone, DetectHTTPStreamConfig(recorded, answer).Mode, "a %d is compared as a response, by its status", status)
}
ok := &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": {"text/event-stream"}}, Request: &http.Request{Method: http.MethodGet}}
require.Equal(t, HTTPStreamModeSSE, DetectHTTPStreamConfig(recorded, ok).Mode, "a 200 of the stream is read as one")
})
}
// countingReader counts the bytes read from r.
type countingReader struct {
r io.Reader
n int
}
func (c *countingReader) Read(p []byte) (int, error) {
n, err := c.r.Read(p)
c.n += n
return n, err
}
// A stream recorded in one format and answered in another (frames kept as
// SSE events, an answer of NDJSON lines) is compared by its recorded body
// read as the answer is read: judged different, neither passed nor refused
// as a recording of nothing.
func TestAStreamAnsweredInAnotherFormatThanItsRecordingFails(t *testing.T) {
switchStates(t, func(t *testing.T) {
recorded := streamTestCase("text/event-stream", "id: 1\ndata: {\"sku\":\"a\"}\n\n")
recorded.HTTPResp = recordedThroughYAML(t, recorded.HTTPResp)
require.Len(t, recorded.HTTPResp.StreamBody, 1)
ok, info, err := compareAgainstApp(t, recorded, "application/x-ndjson", []string{"{\"sku\":\"a\"}\n"})
require.NoError(t, err)
require.False(t, ok)
require.NotNil(t, info, "judged different")
})
}
// A multipart stream is read by its own boundary, its recording by the one it
// was recorded with: a multipart writer picks a new boundary for each response
// (Go's does). Read by the answer's, the recorded body held no part, and the
// stream was compared with none: with the switch off a changed stream passed.
func TestAMultipartStreamIsComparedWithItsRecordingReadByItsRecordedBoundary(t *testing.T) {
part := func(boundary, body string) string {
return "--" + boundary + "\r\nContent-Type: text/plain\r\n\r\n" + body + "\r\n"
}
recordedBody := part("rec0rded", "frame one") + part("rec0rded", "frame two") + "--rec0rded--\r\n"
const answerType = "multipart/x-mixed-replace; boundary=answ3r"
switchStates(t, func(t *testing.T) {
recorded := streamTestCase("multipart/x-mixed-replace; boundary=rec0rded", recordedBody)
recorded.HTTPResp = recordedThroughYAML(t, recorded.HTTPResp)
carried := throughJSON(t, recorded)
require.Equal(t, recordedBody, carried.HTTPResp.Body, "a multipart body is kept whole")
ok, info, err := compareAgainstApp(t, carried, answerType, []string{part("answ3r", "frame one"), part("answ3r", "frame two"), "--answ3r--\r\n"})
require.NoError(t, err)
require.True(t, ok, "the recorded parts pass: %+v", info)
ok, info, err = compareAgainstApp(t, carried, answerType, []string{part("answ3r", "frame one"), part("answ3r", "frame 2"), "--answ3r--\r\n"})
require.NoError(t, err)
require.False(t, ok, "a changed part must not pass")
require.NotNil(t, info)
require.Equal(t, 1, info.FrameIndex)
})
}
// A recorded body JSON carries (a cloud test case's, as the app sent it) is
// read as the app's stream is read, so a stream answered as recorded passes,
// whatever its bytes: an SSE field line with no colon, a field named
// "comment" (no comment, which starts with ':'), a text stream's lines
// carriage returns run within.
func TestARecordedStreamBodyIsReadAsTheAnswerIsRead(t *testing.T) {
for _, c := range []struct {
name, contentType, body, changed string
cfg HTTPStreamConfig
}{
{"sse field with no colon", "text/event-stream", "retry\ndata: a\n\n", "retry\ndata: b\n\n", HTTPStreamConfig{Mode: HTTPStreamModeSSE}},
{"sse field named comment", "text/event-stream", "comment: hi\ndata: a\n\n", "comment: ho\ndata: a\n\n", HTTPStreamConfig{Mode: HTTPStreamModeSSE}},
{"text with carriage returns", "text/plain", "10%\r20%\r30%\ndone\n", "10%\r20%\r31%\ndone\n", HTTPStreamConfig{Mode: HTTPStreamModePlainText}},
} {
t.Run(c.name, func(t *testing.T) {
switchStates(t, func(t *testing.T) {
tc := throughJSON(t, streamTestCase(c.contentType, c.body))
ok, _, info, err := CompareHTTPStream(tc.HTTPResp, strings.NewReader(c.body), c.cfg, map[string]struct{}{}, zap.NewNop())
require.NoError(t, err)
require.True(t, ok, "the recorded stream passes: %+v", info)
ok, _, info, err = CompareHTTPStream(tc.HTTPResp, strings.NewReader(c.changed), c.cfg, map[string]struct{}{}, zap.NewNop())
require.NoError(t, err)
require.False(t, ok, "a changed stream must not pass")
require.NotNil(t, info)
})
})
}
}