From 12600617baf538c08b7d6db470250521e64fb4b2 Mon Sep 17 00:00:00 2001 From: Damian Debkowski Date: Tue, 7 Nov 2023 09:17:36 -0800 Subject: [PATCH] fix(plugin): remove unit test race condition (#4002) --- .../loopback/testing_grpc_stream_test.go | 555 +++++++++++------- 1 file changed, 351 insertions(+), 204 deletions(-) diff --git a/internal/plugin/loopback/testing_grpc_stream_test.go b/internal/plugin/loopback/testing_grpc_stream_test.go index 578b1bea1..0e97db64d 100644 --- a/internal/plugin/loopback/testing_grpc_stream_test.go +++ b/internal/plugin/loopback/testing_grpc_stream_test.go @@ -6,8 +6,10 @@ package loopback import ( "fmt" "io" + "strings" "sync" "testing" + "time" plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin" tr "github.com/stretchr/testify/require" @@ -16,195 +18,339 @@ import ( func Test_GetObjectStream_Client(t *testing.T) { require := tr.New(t) - stream := newGetObjectStream() - require.NotNil(stream) - require.NotNil(stream.client) + t.Run("header", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.client) - // Validate Header is not nil and does not return an error - header, err := stream.client.Header() - require.NotNil(header) - require.NoError(err) - - // Validate Trailer is not nil - trailer := stream.client.Trailer() - require.NotNil(trailer) - - // Validate Context is not nil - ctx := stream.client.Context() - require.NotNil(ctx) - - // validate SendMsg does not write to the channel messages - err = stream.client.SendMsg(nil) - require.NoError(err) - require.Empty(stream.messages) - - // Validate RecvMsg does not read from the channel messages - streamSize := len(stream.messages) - var msg *getObjectStreamResponse - err = stream.client.RecvMsg(msg) - require.NoError(err) - require.Nil(msg) - require.Equal(streamSize, len(stream.messages)) - - // Validate Recv reads a message from the channel messages - var wg sync.WaitGroup - wg.Add(1) - go func() { - stream.messages <- &getObjectStreamResponse{ - msg: &plgpb.GetObjectResponse{}, - } - }() - go func() { - defer wg.Done() - resp, err := stream.client.Recv() + // Validate Header is not nil and does not return an error + header, err := stream.client.Header() + require.NotNil(header) require.NoError(err) - require.NotNil(resp) - }() - wg.Wait() + }) - // Validates Recv reads a error from the channel messages - wg.Add(1) - go func() { - stream.messages <- &getObjectStreamResponse{ - err: fmt.Errorf("mock error"), - } - }() - go func() { - defer wg.Done() - resp, err := stream.client.Recv() + t.Run("trailer", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.client) + + // Validate Trailer is not nil + trailer := stream.client.Trailer() + require.NotNil(trailer) + }) + + t.Run("context", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.client) + + // Validate Context is not nil + ctx := stream.client.Context() + require.NotNil(ctx) + }) + + t.Run("sendMsg is ignored", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.client) + + // validate SendMsg does not write to the channel messages + err := stream.client.SendMsg(nil) + require.NoError(err) + require.Empty(stream.messages) + }) + + t.Run("recvMsg does not read from the channel", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.client) + + // Validate RecvMsg does not read from the channel messages + streamSize := len(stream.messages) + var msg *getObjectStreamResponse + err := stream.client.RecvMsg(msg) + require.NoError(err) + require.Nil(msg) + require.Equal(streamSize, len(stream.messages)) + }) + + t.Run("recv reads from the channel messages", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.client) + + // Validate Recv reads a message from the channel messages + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + stream.messages <- &getObjectStreamResponse{ + msg: &plgpb.GetObjectResponse{}, + } + }() + wg.Add(1) + go func() { + defer wg.Done() + resp, err := stream.client.Recv() + require.NoError(err) + require.NotNil(resp) + }() + wg.Wait() + require.Empty(stream.messages) + }) + + t.Run("recv reads an error from the channel messages", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.client) + + // Validates Recv reads a error from the channel messages + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + stream.messages <- &getObjectStreamResponse{ + err: fmt.Errorf("mock error"), + } + }() + wg.Add(1) + go func() { + defer wg.Done() + resp, err := stream.client.Recv() + require.Error(err) + require.Nil(resp) + }() + wg.Wait() + require.Empty(stream.messages) + }) + + t.Run("closeSend closes the channel messages", func(t *testing.T) { + stream := newGetObjectStream() + require.NotNil(stream) + require.NotNil(stream.client) + require.False(stream.streamClosed) + + // Validate CloseSend does not return an error + // Validate streamClosed is set to true + // Validate the channel messages is closed + err := stream.client.CloseSend() + require.NoError(err) + require.True(stream.streamClosed) + _, ok := <-stream.messages + require.False(ok) + require.Empty(stream.messages) + }) + + t.Run("closeSend returns a stream is closed error", func(t *testing.T) { + stream := newGetObjectStream() + require.NotNil(stream) + require.NotNil(stream.client) + + stream.Close() + require.True(stream.streamClosed) + + // Validate CloseSend does return an error when the channel is closed + // Validate streamClosed is set to true + // Validate the channel messages is closed + err := stream.client.CloseSend() require.Error(err) + require.Contains(err.Error(), "stream is closed") + require.True(stream.streamClosed) + _, ok := <-stream.messages + require.False(ok) + require.Empty(stream.messages) + }) + + t.Run("recv returns EOF error on closed channel", func(t *testing.T) { + stream := newGetObjectStream() + require.NotNil(stream) + require.NotNil(stream.client) + + stream.Close() + require.True(stream.streamClosed) + + // Validate Recv returns EOF error when the channel is closed + resp, err := stream.client.Recv() + require.Equal(io.EOF, err) require.Nil(resp) - }() - wg.Wait() - - // Validate CloseSend does not return an error - // Validate streamClosed is set to true - // Validate the channel messages is closed - var ok bool - err = stream.client.CloseSend() - require.NoError(err) - require.True(stream.streamClosed) - _, ok = <-stream.messages - require.False(ok) - - // Validate CloseSend does return an error when the channel is closed - // Validate streamClosed is set to true - // Validate the channel messages is closed - err = stream.client.CloseSend() - require.Error(err) - require.Contains(err.Error(), "stream is closed") - require.True(stream.streamClosed) - _, ok = <-stream.messages - require.False(ok) - - // Validate Recv returns EOF error when the channel is closed - resp, err := stream.client.Recv() - require.Equal(io.EOF, err) - require.Nil(resp) + require.Empty(stream.messages) + }) } func Test_GetObjectStream_Server(t *testing.T) { require := tr.New(t) - stream := newGetObjectStream() - require.NotNil(stream) - require.NotNil(stream.server) + t.Run("set header", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.server) - // Validate SetHeader does not return an error - err := stream.server.SetHeader(nil) - require.NoError(err) + // Validate SetHeader does not return an error + err := stream.server.SetHeader(nil) + require.NoError(err) + }) - // Validate SendHeader does not return an error - err = stream.server.SendHeader(nil) - require.NoError(err) + t.Run("send header", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.server) - // Validate Context returns a context - ctx := stream.server.Context() - require.NotNil(ctx) + // Validate SendHeader does not return an error + err := stream.server.SendHeader(nil) + require.NoError(err) + }) - // Validate RecvMsg does not read from the channel messages - streamSize := len(stream.messages) - var msg *getObjectStreamResponse - err = stream.server.RecvMsg(msg) - require.NoError(err) - require.Nil(msg) - require.Equal(streamSize, len(stream.messages)) + t.Run("context", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.server) - // Validate Send writes to the channel messages - var wg sync.WaitGroup - go func() { + // Validate Context returns a context + ctx := stream.server.Context() + require.NotNil(ctx) + }) + + t.Run("stream is closed error", func(t *testing.T) { + stream := newGetObjectStream() + require.NotNil(stream) + require.NotNil(stream.server) + stream.Close() + + // Validate Send returns an error + var err error + require.True(stream.streamClosed) err = stream.server.Send(&plgpb.GetObjectResponse{}) - require.NoError(err) - }() - wg.Add(1) - go func() { - defer wg.Done() - resp, ok := <-stream.messages - require.True(ok) - require.NotNil(resp) - require.NotNil(resp.msg) - require.Nil(resp.err) - }() - wg.Wait() + require.Error(err) + require.Contains(err.Error(), "stream is closed") - // Validate Send returns an error when the channel is closed - stream.Close() - require.True(stream.streamClosed) - err = stream.server.Send(&plgpb.GetObjectResponse{}) - require.Error(err) - require.Contains(err.Error(), "stream is closed") - - // Validate SendMsg returns an error when the channel is closed - err = stream.server.SendMsg(&plgpb.GetObjectResponse{}) - require.Error(err) - require.Contains(err.Error(), "stream is closed") - err = stream.server.SendMsg(fmt.Errorf("mock error")) - require.Error(err) - require.Contains(err.Error(), "stream is closed") - - // Reset stream - stream = newGetObjectStream() - - // Validate SendMsg returns an error for unknown message type - err = stream.server.SendMsg(map[string]string{}) - require.False(stream.streamClosed) - require.Error(err) - require.Contains(err.Error(), "invalid argument") - - // Validate SendMsg writes to the channel messages for GetObjectResponse type - go func() { + // Validate SendMsg returns an error when the channel is closed err = stream.server.SendMsg(&plgpb.GetObjectResponse{}) - require.NoError(err) - }() - wg.Add(1) - go func() { - defer wg.Done() - resp, ok := <-stream.messages - require.True(ok) - require.NotNil(resp) - require.NotNil(resp.msg) - require.Nil(resp.err) - }() - wg.Wait() - - // Validate SendMsg closes the channel messages for error type - go func() { + require.Error(err) + require.Contains(err.Error(), "stream is closed") err = stream.server.SendMsg(fmt.Errorf("mock error")) + require.Error(err) + require.Contains(err.Error(), "stream is closed") + }) + + t.Run("recieve empty message", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.server) + + // Validate RecvMsg does not read from the channel messages + streamSize := len(stream.messages) + var msg *getObjectStreamResponse + err := stream.server.RecvMsg(msg) require.NoError(err) - }() - wg.Add(1) - go func() { - defer wg.Done() - resp, ok := <-stream.messages - require.True(ok) - require.NotNil(resp) - require.NotNil(resp.err) - require.Nil(resp.msg) - }() - wg.Wait() - require.True(stream.streamClosed) - _, ok := <-stream.messages - require.False(ok) + require.Nil(msg) + require.Equal(streamSize, len(stream.messages)) + }) + + t.Run("send writes to channel messages", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.server) + + // Validate Send writes to the channel messages + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + err := stream.server.Send(&plgpb.GetObjectResponse{}) + require.NoError(err) + require.False(stream.streamClosed) + }() + wg.Add(1) + go func() { + defer wg.Done() + resp, ok := <-stream.messages + require.True(ok) + require.NotNil(resp) + require.NotNil(resp.msg) + require.Nil(resp.err) + }() + wg.Wait() + }) + + t.Run("sendMsg writes GetObjectResponse to channel messages", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.server) + + // Validate SendMsg writes to the channel messages for GetObjectResponse type + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + err := stream.server.SendMsg(&plgpb.GetObjectResponse{}) + require.NoError(err) + require.False(stream.streamClosed) + }() + wg.Add(1) + go func() { + defer wg.Done() + resp, ok := <-stream.messages + require.True(ok) + require.NotNil(resp) + require.NotNil(resp.msg) + require.Nil(resp.err) + }() + wg.Wait() + }) + + t.Run("sendMsg writes error to close the channel", func(t *testing.T) { + stream := newGetObjectStream() + require.NotNil(stream) + require.NotNil(stream.server) + + // Validate SendMsg closes the channel messages for error type + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + err := stream.server.SendMsg(fmt.Errorf("mock error")) + require.NoError(err) + }() + wg.Add(1) + go func() { + defer wg.Done() + resp, ok := <-stream.messages + require.True(ok) + require.NotNil(resp) + require.NotNil(resp.err) + require.Nil(resp.msg) + }() + wg.Wait() + require.True(stream.streamClosed) + _, ok := <-stream.messages + require.False(ok) + }) + + t.Run("sendMsg returns an error for unknown types", func(t *testing.T) { + stream := newGetObjectStream() + defer stream.Close() + require.NotNil(stream) + require.NotNil(stream.server) + + // Validate SendMsg returns an error for unknown message type + err := stream.server.SendMsg(map[string]string{}) + require.Error(err) + require.Contains(err.Error(), "invalid argument") + require.False(stream.streamClosed) + }) } func TestNewGetObjectStream(t *testing.T) { @@ -230,45 +376,46 @@ func TestNewGetObjectStream(t *testing.T) { func Test_GetObjectStream(t *testing.T) { require := tr.New(t) - stream := newGetObjectStream() + t.Run("server receives error when client closes stream", func(t *testing.T) { + stream := newGetObjectStream() - // Validate the server receives an error when the client closes the stream - var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() - for { + require.NoError(stream.client.CloseSend()) + + // Validate the server receives an error when the client closes the stream + require.Eventually(func() bool { err := stream.server.Send(&plgpb.GetObjectResponse{}) if err == nil { - continue + return false } - require.Error(err) - require.Contains(err.Error(), "stream is closed") - break - } - }() - stream.client.CloseSend() - wg.Wait() - require.True(stream.streamClosed) - require.True(stream.IsStreamClosed()) - _, ok := <-stream.messages - require.False(ok) + return strings.Contains(err.Error(), "stream is closed") + }, time.Second*10, time.Millisecond*500) - // Reset Stream - stream = newGetObjectStream() + require.True(stream.streamClosed) + require.True(stream.IsStreamClosed()) + _, ok := <-stream.messages + require.False(ok) + }) - // Validate the client recieves an error when the server sends an error - // Validate the channel messages is closed - go func() { - err := stream.server.SendMsg(fmt.Errorf("mock error")) - require.NoError(err) - }() - resp, err := stream.client.Recv() - require.Nil(resp) - require.Error(err) - require.Contains(err.Error(), "mock error") - require.True(stream.streamClosed) - require.True(stream.IsStreamClosed()) - _, ok = <-stream.messages - require.False(ok) + t.Run("client receives error when server sends error", func(t *testing.T) { + stream := newGetObjectStream() + + // Validate the client recieves an error when the server sends an error + // Validate the channel messages is closed + go func() { + err := stream.server.SendMsg(fmt.Errorf("mock error")) + require.NoError(err) + }() + + require.Eventually(func() bool { + resp, err := stream.client.Recv() + if err == nil { + return false + } + return resp == nil && strings.Contains(err.Error(), "mock error") + }, time.Second*10, time.Millisecond*50) + + require.Eventually(func() bool { + return stream.streamClosed && stream.IsStreamClosed() + }, time.Second*10, time.Millisecond*50) + }) }