channelreadwriter test done.

pkg/testsupport/channelreadwritecloser.go

@@ -36,7 +36,7 @@ func (rw *ChannelReadWriter) Read(p []byte) (n int, err error) {
 
 	select {
 	case <-rw.ctx.Done():
-		return 0, io.EOF
+		return 0, io.ErrClosedPipe
 	case data, ok := <-rw.receiver:
 		if !ok {
 			return 0, io.EOF
@@ -54,7 +54,7 @@ func (rw *ChannelReadWriter) Write(p []byte) (n int, err error) {
 	select {
 	case <-rw.ctx.Done():
 		rw.Close()
-		return 0, io.EOF
+		return 0, io.ErrClosedPipe
 	case rw.sender <- p:
 	}
 	return len(p), nil

pkg/testsupport/channelreadwriter_test.go

@@ -3,6 +3,7 @@ package testsupport
 import (
 	"context"
 	"github.com/stretchr/testify/suite"
+	"io"
 	"log"
 	"net/http"
 	"strings"
@@ -26,8 +27,10 @@ func TestChannelReadWriterSuite(t *testing.T) {
 }
 
 func (s *ChannelReadWriterTestSuite) createChannel() {
-	toChannel := make(chan []byte)
+	// buffered channels provide more similar behavior to TCP connections
-	fromChannel := make(chan []byte)
+	// then unbuffered channels.
+	toChannel := make(chan []byte, 10)
+	fromChannel := make(chan []byte, 10)
 	s.toChannel = toChannel
 	s.fromChannel = fromChannel
 	ctx, cancelFunc := context.WithCancel(context.Background())
@@ -56,13 +59,8 @@ func (s *ChannelReadWriterTestSuite) TearDownTest() {
 	s.cancelFunc()
 }
 
-func (s *ChannelReadWriterTestSuite) Test_SuccessfulCommunication() {
+func (s *ChannelReadWriterTestSuite) Test_SuccessfulChannelToReadWriter() {
-	tests := []struct {
+	tests := []SuccessfulTest{
-		name       string
-		data       []string
-		chunkSizes []int
-		chunks     []string
-	}{
 		{
 			name:       "buffer_large_enough",
 			data:       []string{"hello"},
@@ -98,17 +96,72 @@ func (s *ChannelReadWriterTestSuite) Test_SuccessfulCommunication() {
 	}
 
 	for _, test := range tests {
-		s.Run(test.name, s.runSuccessfulCommunicationTest(test))
+		s.Run(test.name, s.runSuccessfulChannelToReadWrite(test))
 	}
 
 }
 
-func (s *ChannelReadWriterTestSuite) runSuccessfulCommunicationTest(test struct {
+func (s *ChannelReadWriterTestSuite) Test_SuccessfulReadWriterToChannel() {
+	tests := []SuccessfulTest{
+		{
+			name:       "buffer_large_enough",
+			data:       []string{"hello"},
+			chunkSizes: []int{1},
+		},
+		{
+			name:       "two_reads_required",
+			data:       []string{"hel", "lo"},
+			chunkSizes: []int{3, 2},
+		},
+		{
+			name:       "many_reads_required",
+			data:       []string{"h", "e", "l", "l", "o"},
+			chunkSizes: []int{1, 1, 1, 1, 1},
+		},
+	}
+
+	for _, test := range tests {
+		s.Run(test.name, s.runSuccessfulReadWriteToChannel(test))
+	}
+}
+
+func (s *ChannelReadWriterTestSuite) runSuccessfulReadWriteToChannel(test SuccessfulTest) func() {
+	return func() {
+		runAndWait(
+			&s.Suite,
+			func() any {
+				for _, d := range test.data {
+					n, err := s.conn.Write([]byte(d))
+					s.Nil(err)
+					s.Equal(len(d), n)
+				}
+				return nil
+			},
+			func() any {
+				for _, chunk := range test.data {
+					select {
+					case <-s.ctx.Done():
+						s.Fail("context canceled")
+					case d, ok := <-s.fromChannel:
+						s.True(ok)
+						s.Equal([]byte(chunk), d)
+					}
+				}
+				return nil
+			},
+		)
+	}
+
+}
+
+type SuccessfulTest struct {
 	name       string
 	data       []string
 	chunkSizes []int
 	chunks     []string
-}) func() {
+}
+
+func (s *ChannelReadWriterTestSuite) runSuccessfulChannelToReadWrite(test SuccessfulTest) func() {
 	return func() {
 		runAndWait(
 			&s.Suite,
@@ -139,14 +192,56 @@ func (s *ChannelReadWriterTestSuite) runSuccessfulCommunicationTest(test struct
 	}
 }
 
-func (s *ChannelReadWriterTestSuite) Test_Close() {
+func (s *ChannelReadWriterTestSuite) Test_ChannelCloseBeforeRead() {
-	s.FailNow("todo")
+	data := "hello"
+	// buffered channel
+	s.toChannel <- []byte(data)
+	close(s.toChannel)
+	buf := make([]byte, len(data))
+	n, err := s.conn.Read(buf)
+	s.Nil(err)
+	s.Equal(len(data), n)
+	s.Equal([]byte(data), buf[:n])
+}
+
+func (s *ChannelReadWriterTestSuite) Test_ChannelCloseAfterWrite() {
+	data := "hello"
+	n, err := s.conn.Write([]byte(data))
+	s.Nil(err)
+	s.Equal(len(data), n)
+	err = s.conn.Close()
+	s.Nil(err)
+	select {
+	case <-s.ctx.Done():
+		s.Fail("channel closed")
+	case d, ok := <-s.fromChannel:
+		s.True(ok)
+		s.Equal([]byte(data), d)
+	}
+	n, err = s.conn.Write([]byte(data))
+	s.NotNil(err)
+	s.Equal(0, n)
 }
 
 func (s *ChannelReadWriterTestSuite) Test_CloseTwice() {
-	s.FailNow("todo")
+	err := s.conn.Close()
+	s.Nil(err)
+	s.True(s.conn.closed)
+	err = s.conn.Close()
+	s.Nil(err)
 }
 
-func (s *ChannelReadWriterTestSuite) Test_ContextCanceled() {
+func (s *ChannelReadWriterTestSuite) Test_ContextCanceledRead() {
-	s.FailNow("todo")
+	s.cancelFunc()
+	buf := make([]byte, 100)
+	n, err := s.conn.Read(buf)
+	s.Equal(io.ErrClosedPipe, err)
+	s.Equal(0, n)
+}
+
+func (s *ChannelReadWriterTestSuite) Test_ContextCanceledWrite() {
+	s.cancelFunc()
+	n, err := s.conn.Write([]byte("hello"))
+	s.Equal(io.ErrClosedPipe, err)
+	s.Equal(0, n)
 }

pkg/testsupport/utils.go

@@ -42,7 +42,7 @@ func startPprof(port string) *http.Server {
 		if err := srv.ListenAndServe(); err != http.ErrServerClosed {
 			log.Fatalf("Could not start pprof listener: %v", err)
 		}
-		log.Println("Test pprof server started")
+		log.Println("Test pprof server started: " + port)
 	}()
 	return &srv
 }