http2: factor out frame read/write test functions

Client and server tests both write frames to a test connection
and read frames back. Frame reads are usually paired with
test expectations.

Unify the API used for frame reads/writes in tests.

Introduce a testConnFramer type that implements a common set
of read/write methods, and embed it in both client and server
test types.

Change-Id: I6927c43459ba24f150a21c058a92797754f82bf1
Reviewed-on: https://go-review.googlesource.com/c/net/+/586249
Reviewed-by: Jonathan Amsterdam <jba@google.com>
LUCI-TryBot-Result: Go LUCI <golang-scoped@luci-project-accounts.iam.gserviceaccount.com>

http2/clientconn_test.go

@@ -13,10 +13,8 @@ import (
 	"fmt"
 	"io"
 	"net/http"
-	"os"
 	"reflect"
 	"runtime"
-	"slices"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -62,6 +60,7 @@ func TestTestClientConn(t *testing.T) {
 		streamID:  rt.streamID(),
 		endStream: true,
 		size:      10,
+		multiple:  true,
 	})
 
 	// tc.writeHeaders sends a HEADERS frame back to the client.
@@ -97,6 +96,7 @@ type testClientConn struct {
 	fr    *Framer
 	cc    *ClientConn
 	group *synctestGroup
+	testConnFramer
 
 	encbuf bytes.Buffer
 	enc    *hpack.Encoder
@@ -115,6 +115,7 @@ func newTestClientConnFromClientConn(t *testing.T, cc *ClientConn) *testClientCo
 	}
 	cli, srv := synctestNetPipe(tc.group)
 	srv.SetReadDeadline(tc.group.Now())
+	srv.autoWait = true
 	tc.netconn = srv
 	tc.enc = hpack.NewEncoder(&tc.encbuf)
 
@@ -123,8 +124,12 @@ func newTestClientConnFromClientConn(t *testing.T, cc *ClientConn) *testClientCo
 	// cli is the ClientConn's side, srv is the side controlled by the test.
 	cc.tconn = cli
 	tc.fr = NewFramer(srv, srv)
+	tc.testConnFramer = testConnFramer{
+		t:   t,
+		fr:  tc.fr,
+		dec: hpack.NewDecoder(initialHeaderTableSize, nil),
+	}
 
-	tc.fr.ReadMetaHeaders = hpack.NewDecoder(initialHeaderTableSize, nil)
 	tc.fr.SetMaxReadFrameSize(10 << 20)
 	t.Cleanup(func() {
 		tc.closeWrite()
@@ -174,169 +179,15 @@ func (tc *testClientConn) hasFrame() bool {
 	return len(tc.netconn.Peek()) > 0
 }
 
+// isClosed reports whether the peer has closed the connection.
 func (tc *testClientConn) isClosed() bool {
 	return tc.netconn.IsClosedByPeer()
 }
 
-// readFrame reads the next frame from the conn.
-func (tc *testClientConn) readFrame() Frame {
-	tc.t.Helper()
-	tc.sync()
-	fr, err := tc.fr.ReadFrame()
-	if err == io.EOF || err == os.ErrDeadlineExceeded {
-		return nil
-	}
-	if err != nil {
-		tc.t.Fatalf("ReadFrame: %v", err)
-	}
-	return fr
-}
-
-// testClientConnReadFrame reads a frame of a specific type from the conn.
-func testClientConnReadFrame[T any](tc *testClientConn) T {
-	tc.t.Helper()
-	var v T
-	fr := tc.readFrame()
-	if fr == nil {
-		tc.t.Fatalf("got no frame, want frame %T", v)
-	}
-	v, ok := fr.(T)
-	if !ok {
-		tc.t.Fatalf("got frame %T, want %T", fr, v)
-	}
-	return v
-}
-
-// wantFrameType reads the next frame from the conn.
-// It produces an error if the frame type is not the expected value.
-func (tc *testClientConn) wantFrameType(want FrameType) {
-	tc.t.Helper()
-	fr := tc.readFrame()
-	if fr == nil {
-		tc.t.Fatalf("got no frame, want frame %v", want)
-	}
-	if got := fr.Header().Type; got != want {
-		tc.t.Fatalf("got frame %v, want %v", got, want)
-	}
-}
-
-// wantUnorderedFrames reads frames from the conn until every condition in want has been satisfied.
-//
-// want is a list of func(*SomeFrame) bool.
-// wantUnorderedFrames will call each func with frames of the appropriate type
-// until the func returns true.
-// It calls t.Fatal if an unexpected frame is received (no func has that frame type,
-// or all funcs with that type have returned true), or if the conn runs out of frames
-// with unsatisfied funcs.
-//
-// Example:
-//
-//	// Read a SETTINGS frame, and any number of DATA frames for a stream.
-//	// The SETTINGS frame may appear anywhere in the sequence.
-//	// The last DATA frame must indicate the end of the stream.
-//	tc.wantUnorderedFrames(
-//		func(f *SettingsFrame) bool {
-//			return true
-//		},
-//		func(f *DataFrame) bool {
-//			return f.StreamEnded()
-//		},
-//	)
-func (tc *testClientConn) wantUnorderedFrames(want ...any) {
-	tc.t.Helper()
-	want = slices.Clone(want)
-	seen := 0
-frame:
-	for seen < len(want) && !tc.t.Failed() {
-		fr := tc.readFrame()
-		if fr == nil {
-			break
-		}
-		for i, f := range want {
-			if f == nil {
-				continue
-			}
-			typ := reflect.TypeOf(f)
-			if typ.Kind() != reflect.Func ||
-				typ.NumIn() != 1 ||
-				typ.NumOut() != 1 ||
-				typ.Out(0) != reflect.TypeOf(true) {
-				tc.t.Fatalf("expected func(*SomeFrame) bool, got %T", f)
-			}
-			if typ.In(0) == reflect.TypeOf(fr) {
-				out := reflect.ValueOf(f).Call([]reflect.Value{reflect.ValueOf(fr)})
-				if out[0].Bool() {
-					want[i] = nil
-					seen++
-				}
-				continue frame
-			}
-		}
-		tc.t.Errorf("got unexpected frame type %T", fr)
-	}
-	if seen < len(want) {
-		for _, f := range want {
-			if f == nil {
-				continue
-			}
-			tc.t.Errorf("did not see expected frame: %v", reflect.TypeOf(f).In(0))
-		}
-		tc.t.Fatalf("did not see %v expected frame types", len(want)-seen)
-	}
-}
-
-type wantHeader struct {
-	streamID  uint32
-	endStream bool
-	header    http.Header
-}
-
-// wantHeaders reads a HEADERS frame and potential CONTINUATION frames,
-// and asserts that they contain the expected headers.
-func (tc *testClientConn) wantHeaders(want wantHeader) {
-	tc.t.Helper()
-	got := testClientConnReadFrame[*MetaHeadersFrame](tc)
-	if got, want := got.StreamID, want.streamID; got != want {
-		tc.t.Fatalf("got stream ID %v, want %v", got, want)
-	}
-	if got, want := got.StreamEnded(), want.endStream; got != want {
-		tc.t.Fatalf("got stream ended %v, want %v", got, want)
-	}
-	gotHeader := make(http.Header)
-	for _, f := range got.Fields {
-		gotHeader[f.Name] = append(gotHeader[f.Name], f.Value)
-	}
-	for k, v := range want.header {
-		if !reflect.DeepEqual(v, gotHeader[k]) {
-			tc.t.Fatalf("got header %q = %q; want %q", k, v, gotHeader[k])
-		}
-	}
-}
-
-type wantData struct {
-	streamID  uint32
-	endStream bool
-	size      int
-}
-
-// wantData reads zero or more DATA frames, and asserts that they match the expectation.
-func (tc *testClientConn) wantData(want wantData) {
-	tc.t.Helper()
-	gotSize := 0
-	gotEndStream := false
-	for tc.hasFrame() && !gotEndStream {
-		data := testClientConnReadFrame[*DataFrame](tc)
-		gotSize += len(data.Data())
-		if data.StreamEnded() {
-			gotEndStream = true
-		}
-	}
-	if gotSize != want.size {
-		tc.t.Fatalf("got %v bytes of DATA frames, want %v", gotSize, want.size)
-	}
-	if gotEndStream != want.endStream {
-		tc.t.Fatalf("after %v bytes of DATA frames, got END_STREAM=%v; want %v", gotSize, gotEndStream, want.endStream)
-	}
+// closeWrite causes the net.Conn used by the ClientConn to return a error
+// from Read calls.
+func (tc *testClientConn) closeWrite() {
+	tc.netconn.Close()
 }
 
 // testRequestBody is a Request.Body for use in tests.
@@ -468,38 +319,6 @@ func (tc *testClientConn) greet(settings ...Setting) {
 	tc.wantFrameType(FrameSettings) // acknowledgement
 }
 
-func (tc *testClientConn) writeSettings(settings ...Setting) {
-	tc.t.Helper()
-	if err := tc.fr.WriteSettings(settings...); err != nil {
-		tc.t.Fatal(err)
-	}
-	tc.sync()
-}
-
-func (tc *testClientConn) writeSettingsAck() {
-	tc.t.Helper()
-	if err := tc.fr.WriteSettingsAck(); err != nil {
-		tc.t.Fatal(err)
-	}
-	tc.sync()
-}
-
-func (tc *testClientConn) writeData(streamID uint32, endStream bool, data []byte) {
-	tc.t.Helper()
-	if err := tc.fr.WriteData(streamID, endStream, data); err != nil {
-		tc.t.Fatal(err)
-	}
-	tc.sync()
-}
-
-func (tc *testClientConn) writeDataPadded(streamID uint32, endStream bool, data, pad []byte) {
-	tc.t.Helper()
-	if err := tc.fr.WriteDataPadded(streamID, endStream, data, pad); err != nil {
-		tc.t.Fatal(err)
-	}
-	tc.sync()
-}
-
 // makeHeaderBlockFragment encodes headers in a form suitable for inclusion
 // in a HEADERS or CONTINUATION frame.
 //
@@ -515,87 +334,6 @@ func (tc *testClientConn) makeHeaderBlockFragment(s ...string) []byte {
 	return tc.encbuf.Bytes()
 }
 
-func (tc *testClientConn) writeHeaders(p HeadersFrameParam) {
-	tc.t.Helper()
-	if err := tc.fr.WriteHeaders(p); err != nil {
-		tc.t.Fatal(err)
-	}
-	tc.sync()
-}
-
-// writeHeadersMode writes header frames, as modified by mode:
-//
-//   - noHeader: Don't write the header.
-//   - oneHeader: Write a single HEADERS frame.
-//   - splitHeader: Write a HEADERS frame and CONTINUATION frame.
-func (tc *testClientConn) writeHeadersMode(mode headerType, p HeadersFrameParam) {
-	tc.t.Helper()
-	switch mode {
-	case noHeader:
-	case oneHeader:
-		tc.writeHeaders(p)
-	case splitHeader:
-		if len(p.BlockFragment) < 2 {
-			panic("too small")
-		}
-		contData := p.BlockFragment[1:]
-		contEnd := p.EndHeaders
-		p.BlockFragment = p.BlockFragment[:1]
-		p.EndHeaders = false
-		tc.writeHeaders(p)
-		tc.writeContinuation(p.StreamID, contEnd, contData)
-	default:
-		panic("bogus mode")
-	}
-}
-
-func (tc *testClientConn) writeContinuation(streamID uint32, endHeaders bool, headerBlockFragment []byte) {
-	tc.t.Helper()
-	if err := tc.fr.WriteContinuation(streamID, endHeaders, headerBlockFragment); err != nil {
-		tc.t.Fatal(err)
-	}
-	tc.sync()
-}
-
-func (tc *testClientConn) writeRSTStream(streamID uint32, code ErrCode) {
-	tc.t.Helper()
-	if err := tc.fr.WriteRSTStream(streamID, code); err != nil {
-		tc.t.Fatal(err)
-	}
-	tc.sync()
-}
-
-func (tc *testClientConn) writePing(ack bool, data [8]byte) {
-	tc.t.Helper()
-	if err := tc.fr.WritePing(ack, data); err != nil {
-		tc.t.Fatal(err)
-	}
-	tc.sync()
-}
-
-func (tc *testClientConn) writeGoAway(maxStreamID uint32, code ErrCode, debugData []byte) {
-	tc.t.Helper()
-	if err := tc.fr.WriteGoAway(maxStreamID, code, debugData); err != nil {
-		tc.t.Fatal(err)
-	}
-	tc.sync()
-}
-
-func (tc *testClientConn) writeWindowUpdate(streamID, incr uint32) {
-	tc.t.Helper()
-	if err := tc.fr.WriteWindowUpdate(streamID, incr); err != nil {
-		tc.t.Fatal(err)
-	}
-	tc.sync()
-}
-
-// closeWrite causes the net.Conn used by the ClientConn to return a error
-// from Read calls.
-func (tc *testClientConn) closeWrite() {
-	tc.netconn.Close()
-	tc.sync()
-}
-
 // inflowWindow returns the amount of inbound flow control available for a stream,
 // or for the connection if streamID is 0.
 func (tc *testClientConn) inflowWindow(streamID uint32) int32 {

http2/connframes_test.go

@@ -0,0 +1,414 @@
+// Copyright 2024 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package http2
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net/http"
+	"os"
+	"reflect"
+	"slices"
+	"testing"
+
+	"golang.org/x/net/http2/hpack"
+)
+
+type testConnFramer struct {
+	t   testing.TB
+	fr  *Framer
+	dec *hpack.Decoder
+}
+
+// readFrame reads the next frame.
+// It returns nil if the conn is closed or no frames are available.
+func (tf *testConnFramer) readFrame() Frame {
+	tf.t.Helper()
+	fr, err := tf.fr.ReadFrame()
+	if err == io.EOF || err == os.ErrDeadlineExceeded {
+		return nil
+	}
+	if err != nil {
+		tf.t.Fatalf("ReadFrame: %v", err)
+	}
+	return fr
+}
+
+type readFramer interface {
+	readFrame() Frame
+}
+
+// readFrame reads a frame of a specific type.
+func readFrame[T any](t testing.TB, framer readFramer) T {
+	t.Helper()
+	var v T
+	fr := framer.readFrame()
+	if fr == nil {
+		t.Fatalf("got no frame, want frame %T", v)
+	}
+	v, ok := fr.(T)
+	if !ok {
+		t.Fatalf("got frame %T, want %T", fr, v)
+	}
+	return v
+}
+
+// wantFrameType reads the next frame.
+// It produces an error if the frame type is not the expected value.
+func (tf *testConnFramer) wantFrameType(want FrameType) {
+	tf.t.Helper()
+	fr := tf.readFrame()
+	if fr == nil {
+		tf.t.Fatalf("got no frame, want frame %v", want)
+	}
+	if got := fr.Header().Type; got != want {
+		tf.t.Fatalf("got frame %v, want %v", got, want)
+	}
+}
+
+// wantUnorderedFrames reads frames until every condition in want has been satisfied.
+//
+// want is a list of func(*SomeFrame) bool.
+// wantUnorderedFrames will call each func with frames of the appropriate type
+// until the func returns true.
+// It calls t.Fatal if an unexpected frame is received (no func has that frame type,
+// or all funcs with that type have returned true), or if the framer runs out of frames
+// with unsatisfied funcs.
+//
+// Example:
+//
+//	// Read a SETTINGS frame, and any number of DATA frames for a stream.
+//	// The SETTINGS frame may appear anywhere in the sequence.
+//	// The last DATA frame must indicate the end of the stream.
+//	tf.wantUnorderedFrames(
+//		func(f *SettingsFrame) bool {
+//			return true
+//		},
+//		func(f *DataFrame) bool {
+//			return f.StreamEnded()
+//		},
+//	)
+func (tf *testConnFramer) wantUnorderedFrames(want ...any) {
+	tf.t.Helper()
+	want = slices.Clone(want)
+	seen := 0
+frame:
+	for seen < len(want) && !tf.t.Failed() {
+		fr := tf.readFrame()
+		if fr == nil {
+			break
+		}
+		for i, f := range want {
+			if f == nil {
+				continue
+			}
+			typ := reflect.TypeOf(f)
+			if typ.Kind() != reflect.Func ||
+				typ.NumIn() != 1 ||
+				typ.NumOut() != 1 ||
+				typ.Out(0) != reflect.TypeOf(true) {
+				tf.t.Fatalf("expected func(*SomeFrame) bool, got %T", f)
+			}
+			if typ.In(0) == reflect.TypeOf(fr) {
+				out := reflect.ValueOf(f).Call([]reflect.Value{reflect.ValueOf(fr)})
+				if out[0].Bool() {
+					want[i] = nil
+					seen++
+				}
+				continue frame
+			}
+		}
+		tf.t.Errorf("got unexpected frame type %T", fr)
+	}
+	if seen < len(want) {
+		for _, f := range want {
+			if f == nil {
+				continue
+			}
+			tf.t.Errorf("did not see expected frame: %v", reflect.TypeOf(f).In(0))
+		}
+		tf.t.Fatalf("did not see %v expected frame types", len(want)-seen)
+	}
+}
+
+type wantHeader struct {
+	streamID  uint32
+	endStream bool
+	header    http.Header
+}
+
+// wantHeaders reads a HEADERS frame and potential CONTINUATION frames,
+// and asserts that they contain the expected headers.
+func (tf *testConnFramer) wantHeaders(want wantHeader) {
+	tf.t.Helper()
+
+	hf := readFrame[*HeadersFrame](tf.t, tf)
+	if got, want := hf.StreamID, want.streamID; got != want {
+		tf.t.Fatalf("got stream ID %v, want %v", got, want)
+	}
+	if got, want := hf.StreamEnded(), want.endStream; got != want {
+		tf.t.Fatalf("got stream ended %v, want %v", got, want)
+	}
+
+	gotHeader := make(http.Header)
+	tf.dec.SetEmitFunc(func(hf hpack.HeaderField) {
+		gotHeader[hf.Name] = append(gotHeader[hf.Name], hf.Value)
+	})
+	defer tf.dec.SetEmitFunc(nil)
+	if _, err := tf.dec.Write(hf.HeaderBlockFragment()); err != nil {
+		tf.t.Fatalf("decoding HEADERS frame: %v", err)
+	}
+	headersEnded := hf.HeadersEnded()
+	for !headersEnded {
+		cf := readFrame[*ContinuationFrame](tf.t, tf)
+		if cf == nil {
+			tf.t.Fatalf("got end of frames, want CONTINUATION")
+		}
+		if _, err := tf.dec.Write(cf.HeaderBlockFragment()); err != nil {
+			tf.t.Fatalf("decoding CONTINUATION frame: %v", err)
+		}
+		headersEnded = cf.HeadersEnded()
+	}
+	if err := tf.dec.Close(); err != nil {
+		tf.t.Fatalf("hpack decoding error: %v", err)
+	}
+
+	for k, v := range want.header {
+		if !reflect.DeepEqual(v, gotHeader[k]) {
+			tf.t.Fatalf("got header %q = %q; want %q", k, v, gotHeader[k])
+		}
+	}
+}
+
+// decodeHeader supports some older server tests.
+// TODO: rewrite those tests to use newer, more convenient test APIs.
+func (tf *testConnFramer) decodeHeader(headerBlock []byte) (pairs [][2]string) {
+	tf.dec.SetEmitFunc(func(hf hpack.HeaderField) {
+		if hf.Name == "date" {
+			return
+		}
+		pairs = append(pairs, [2]string{hf.Name, hf.Value})
+	})
+	defer tf.dec.SetEmitFunc(nil)
+	if _, err := tf.dec.Write(headerBlock); err != nil {
+		tf.t.Fatalf("hpack decoding error: %v", err)
+	}
+	if err := tf.dec.Close(); err != nil {
+		tf.t.Fatalf("hpack decoding error: %v", err)
+	}
+	return pairs
+}
+
+type wantData struct {
+	streamID  uint32
+	endStream bool
+	size      int
+	data      []byte
+	multiple  bool // data may be spread across multiple DATA frames
+}
+
+// wantData reads zero or more DATA frames, and asserts that they match the expectation.
+func (tf *testConnFramer) wantData(want wantData) {
+	tf.t.Helper()
+	gotSize := 0
+	gotEndStream := false
+	if want.data != nil {
+		want.size = len(want.data)
+	}
+	var gotData []byte
+	for {
+		fr := tf.readFrame()
+		if fr == nil {
+			break
+		}
+		data, ok := fr.(*DataFrame)
+		if !ok {
+			tf.t.Fatalf("got frame %T, want DataFrame", fr)
+		}
+		if want.data != nil {
+			gotData = append(gotData, data.Data()...)
+		}
+		gotSize += len(data.Data())
+		if data.StreamEnded() {
+			gotEndStream = true
+			break
+		}
+		if !want.endStream && gotSize >= want.size {
+			break
+		}
+		if !want.multiple {
+			break
+		}
+	}
+	if gotSize != want.size {
+		tf.t.Fatalf("got %v bytes of DATA frames, want %v", gotSize, want.size)
+	}
+	if gotEndStream != want.endStream {
+		tf.t.Fatalf("after %v bytes of DATA frames, got END_STREAM=%v; want %v", gotSize, gotEndStream, want.endStream)
+	}
+	if want.data != nil && !bytes.Equal(gotData, want.data) {
+		tf.t.Fatalf("got data %q, want %q", gotData, want.data)
+	}
+}
+
+func (tf *testConnFramer) wantRSTStream(streamID uint32, code ErrCode) {
+	tf.t.Helper()
+	fr := readFrame[*RSTStreamFrame](tf.t, tf)
+	if fr.StreamID != streamID || fr.ErrCode != code {
+		tf.t.Fatalf("got %v, want RST_STREAM StreamID=%v, code=%v", summarizeFrame(fr), streamID, code)
+	}
+}
+
+func (tf *testConnFramer) wantSettingsAck() {
+	tf.t.Helper()
+	fr := readFrame[*SettingsFrame](tf.t, tf)
+	if !fr.Header().Flags.Has(FlagSettingsAck) {
+		tf.t.Fatal("Settings Frame didn't have ACK set")
+	}
+}
+
+func (tf *testConnFramer) wantGoAway(maxStreamID uint32, code ErrCode) {
+	tf.t.Helper()
+	fr := readFrame[*GoAwayFrame](tf.t, tf)
+	if fr.LastStreamID != maxStreamID || fr.ErrCode != code {
+		tf.t.Fatalf("got %v, want GOAWAY LastStreamID=%v, code=%v", summarizeFrame(fr), maxStreamID, code)
+	}
+}
+
+func (tf *testConnFramer) wantWindowUpdate(streamID, incr uint32) {
+	tf.t.Helper()
+	wu := readFrame[*WindowUpdateFrame](tf.t, tf)
+	if wu.FrameHeader.StreamID != streamID {
+		tf.t.Fatalf("WindowUpdate StreamID = %d; want %d", wu.FrameHeader.StreamID, streamID)
+	}
+	if wu.Increment != incr {
+		tf.t.Fatalf("WindowUpdate increment = %d; want %d", wu.Increment, incr)
+	}
+}
+
+func (tf *testConnFramer) wantClosed() {
+	tf.t.Helper()
+	fr, err := tf.fr.ReadFrame()
+	if err == nil {
+		tf.t.Fatalf("got unexpected frame (want closed connection): %v", fr)
+	}
+	if err == context.DeadlineExceeded {
+		tf.t.Fatalf("connection is not closed; want it to be")
+	}
+}
+
+func (tf *testConnFramer) wantIdle() {
+	tf.t.Helper()
+	fr, err := tf.fr.ReadFrame()
+	if err == nil {
+		tf.t.Fatalf("got unexpected frame (want idle connection): %v", fr)
+	}
+	if err != context.DeadlineExceeded {
+		tf.t.Fatalf("got unexpected frame error (want idle connection): %v", err)
+	}
+}
+
+func (tf *testConnFramer) writeSettings(settings ...Setting) {
+	tf.t.Helper()
+	if err := tf.fr.WriteSettings(settings...); err != nil {
+		tf.t.Fatal(err)
+	}
+}
+
+func (tf *testConnFramer) writeSettingsAck() {
+	tf.t.Helper()
+	if err := tf.fr.WriteSettingsAck(); err != nil {
+		tf.t.Fatal(err)
+	}
+}
+
+func (tf *testConnFramer) writeData(streamID uint32, endStream bool, data []byte) {
+	tf.t.Helper()
+	if err := tf.fr.WriteData(streamID, endStream, data); err != nil {
+		tf.t.Fatal(err)
+	}
+}
+
+func (tf *testConnFramer) writeDataPadded(streamID uint32, endStream bool, data, pad []byte) {
+	tf.t.Helper()
+	if err := tf.fr.WriteDataPadded(streamID, endStream, data, pad); err != nil {
+		tf.t.Fatal(err)
+	}
+}
+
+func (tf *testConnFramer) writeHeaders(p HeadersFrameParam) {
+	tf.t.Helper()
+	if err := tf.fr.WriteHeaders(p); err != nil {
+		tf.t.Fatal(err)
+	}
+}
+
+// writeHeadersMode writes header frames, as modified by mode:
+//
+//   - noHeader: Don't write the header.
+//   - oneHeader: Write a single HEADERS frame.
+//   - splitHeader: Write a HEADERS frame and CONTINUATION frame.
+func (tf *testConnFramer) writeHeadersMode(mode headerType, p HeadersFrameParam) {
+	tf.t.Helper()
+	switch mode {
+	case noHeader:
+	case oneHeader:
+		tf.writeHeaders(p)
+	case splitHeader:
+		if len(p.BlockFragment) < 2 {
+			panic("too small")
+		}
+		contData := p.BlockFragment[1:]
+		contEnd := p.EndHeaders
+		p.BlockFragment = p.BlockFragment[:1]
+		p.EndHeaders = false
+		tf.writeHeaders(p)
+		tf.writeContinuation(p.StreamID, contEnd, contData)
+	default:
+		panic("bogus mode")
+	}
+}
+
+func (tf *testConnFramer) writeContinuation(streamID uint32, endHeaders bool, headerBlockFragment []byte) {
+	tf.t.Helper()
+	if err := tf.fr.WriteContinuation(streamID, endHeaders, headerBlockFragment); err != nil {
+		tf.t.Fatal(err)
+	}
+}
+
+func (tf *testConnFramer) writePriority(id uint32, p PriorityParam) {
+	if err := tf.fr.WritePriority(id, p); err != nil {
+		tf.t.Fatal(err)
+	}
+}
+
+func (tf *testConnFramer) writeRSTStream(streamID uint32, code ErrCode) {
+	tf.t.Helper()
+	if err := tf.fr.WriteRSTStream(streamID, code); err != nil {
+		tf.t.Fatal(err)
+	}
+}
+
+func (tf *testConnFramer) writePing(ack bool, data [8]byte) {
+	tf.t.Helper()
+	if err := tf.fr.WritePing(ack, data); err != nil {
+		tf.t.Fatal(err)
+	}
+}
+
+func (tf *testConnFramer) writeGoAway(maxStreamID uint32, code ErrCode, debugData []byte) {
+	tf.t.Helper()
+	if err := tf.fr.WriteGoAway(maxStreamID, code, debugData); err != nil {
+		tf.t.Fatal(err)
+	}
+}
+
+func (tf *testConnFramer) writeWindowUpdate(streamID, incr uint32) {
+	tf.t.Helper()
+	if err := tf.fr.WriteWindowUpdate(streamID, incr); err != nil {
+		tf.t.Fatal(err)
+	}
+}

http2/server_push_test.go

@@ -218,12 +218,12 @@ func TestServer_Push_Success(t *testing.T) {
 
 	consumed := map[uint32]int{}
 	for k := 0; len(expected) > 0; k++ {
-		f, err := st.readFrame()
-		if err != nil {
+		f := st.readFrame()
+		if f == nil {
 			for id, left := range expected {
 				t.Errorf("stream %d: missing %d frames", id, len(left))
 			}
-			t.Fatalf("readFrame %d: %v", k, err)
+			break
 		}
 		id := f.Header().StreamID
 		label := fmt.Sprintf("stream %d, frame %d", id, consumed[id])
@@ -339,10 +339,10 @@ func testServer_Push_RejectSingleRequest(t *testing.T, doPush func(http.Pusher,
 		t.Error(err)
 	}
 	// Should not get a PUSH_PROMISE frame.
-	hf := st.wantHeaders()
-	if !hf.StreamEnded() {
-		t.Error("stream should end after headers")
-	}
+	st.wantHeaders(wantHeader{
+		streamID:  1,
+		endStream: true,
+	})
 }
 
 func TestServer_Push_RejectIfDisabled(t *testing.T) {
@@ -459,7 +459,7 @@ func TestServer_Push_StateTransitions(t *testing.T) {
 	}
 	getSlash(st)
 	// After the PUSH_PROMISE is sent, the stream should be stateHalfClosedRemote.
-	st.wantPushPromise()
+	_ = readFrame[*PushPromiseFrame](t, st)
 	if got, want := st.streamState(2), stateHalfClosedRemote; got != want {
 		t.Fatalf("streamState(2)=%v, want %v", got, want)
 	}
@@ -468,10 +468,10 @@ func TestServer_Push_StateTransitions(t *testing.T) {
 	// the stream before we check st.streamState(2) -- should that happen, we'll
 	// see stateClosed and fail the above check.
 	close(gotPromise)
-	st.wantHeaders()
-	if df := st.wantData(); !df.StreamEnded() {
-		t.Fatal("expected END_STREAM flag on DATA")
-	}
+	st.wantHeaders(wantHeader{
+		streamID:  2,
+		endStream: false,
+	})
 	if got, want := st.streamState(2), stateClosed; got != want {
 		t.Fatalf("streamState(2)=%v, want %v", got, want)
 	}
@@ -554,9 +554,9 @@ func TestServer_Push_Underflow(t *testing.T) {
 	numPushPromises := 0
 	numHeaders := 0
 	for numHeaders < numRequests*2 || numPushPromises < numRequests {
-		f, err := st.readFrame()
-		if err != nil {
-			st.t.Fatal(err)
+		f := st.readFrame()
+		if f == nil {
+			st.t.Fatal("conn is idle, want frame")
 		}
 		switch f := f.(type) {
 		case *HeadersFrame:

http2/transport_test.go

@@ -151,7 +151,7 @@ func TestIdleConnTimeout(t *testing.T) {
 				}
 
 				// Respond to the client's request.
-				hf := testClientConnReadFrame[*MetaHeadersFrame](tc)
+				hf := readFrame[*HeadersFrame](t, tc)
 				tc.writeHeaders(HeadersFrameParam{
 					StreamID:   hf.StreamID,
 					EndHeaders: true,
@@ -865,6 +865,7 @@ func testTransportReqBodyAfterResponse(t *testing.T, status int) {
 			streamID:  rt.streamID(),
 			endStream: true,
 			size:      bodySize / 2,
+			multiple:  true,
 		})
 	} else {
 		// After a 403 response, client gives up and resets the stream.
@@ -1870,6 +1871,7 @@ func testTransportResponseHeaderTimeout(t *testing.T, body bool) {
 		tc.wantData(wantData{
 			endStream: true,
 			size:      bodySize,
+			multiple:  true,
 		})
 	}
 
@@ -2604,7 +2606,7 @@ func TestTransportAdjustsFlowControl(t *testing.T) {
 
 	gotBytes := int64(0)
 	for {
-		f := testClientConnReadFrame[*DataFrame](tc)
+		f := readFrame[*DataFrame](t, tc)
 		gotBytes += int64(len(f.Data()))
 		// After we've got half the client's initial flow control window's worth
 		// of request body data, give it just enough flow control to finish.
@@ -2700,7 +2702,7 @@ func TestTransportReturnsErrorOnBadResponseHeaders(t *testing.T) {
 		t.Fatalf("RoundTrip error = %#v; want %#v", err, want)
 	}
 
-	fr := testClientConnReadFrame[*RSTStreamFrame](tc)
+	fr := readFrame[*RSTStreamFrame](t, tc)
 	if fr.StreamID != 1 || fr.ErrCode != ErrCodeProtocol {
 		t.Errorf("Frame = %v; want RST_STREAM for stream 1 with ErrCodeProtocol", summarizeFrame(fr))
 	}
@@ -3071,7 +3073,7 @@ func TestTransportPingWhenReadingMultiplePings(t *testing.T) {
 
 		// ...ping now.
 		tc.advance(1 * time.Millisecond)
-		f := testClientConnReadFrame[*PingFrame](tc)
+		f := readFrame[*PingFrame](t, tc)
 		tc.writePing(true, f.Data)
 	}
 
@@ -3375,7 +3377,7 @@ func TestTransportMaxFrameReadSize(t *testing.T) {
 				tr.MaxReadFrameSize = test.maxReadFrameSize
 			})
 
-			fr := testClientConnReadFrame[*SettingsFrame](tc)
+			fr := readFrame[*SettingsFrame](t, tc)
 			got, ok := fr.Value(SettingMaxFrameSize)
 			if !ok {
 				t.Errorf("Transport.MaxReadFrameSize = %v; server got no setting, want %v", test.maxReadFrameSize, test.want)
@@ -3518,7 +3520,7 @@ func TestTransportMaxDecoderHeaderTableSize(t *testing.T) {
 		tr.MaxDecoderHeaderTableSize = reqSize
 	})
 
-	fr := testClientConnReadFrame[*SettingsFrame](tc)
+	fr := readFrame[*SettingsFrame](t, tc)
 	if v, ok := fr.Value(SettingHeaderTableSize); !ok {
 		t.Fatalf("missing SETTINGS_HEADER_TABLE_SIZE setting")
 	} else if v != reqSize {
@@ -4135,7 +4137,7 @@ func TestTransportBodyEagerEndStream(t *testing.T) {
 	tc.roundTrip(req)
 
 	tc.wantFrameType(FrameHeaders)
-	f := testClientConnReadFrame[*DataFrame](tc)
+	f := readFrame[*DataFrame](t, tc)
 	if !f.StreamEnded() {
 		t.Fatalf("data frame without END_STREAM %v", f)
 	}