package grpctool_test

import (
	"bufio"
	"bytes"
	"context"
	"io"
	"net/http"
	"strings"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/grpctool"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/httpz"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/prototool"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/testing/matcher"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/testing/mock_prototool"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/testing/mock_rpc"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/testing/mock_stdlib"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/testing/testhelpers"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/testing/testlogger"
	"go.uber.org/mock/gomock"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/anypb"
)

const (
	requestBodyData  = "request_jkasdbfkadsbfkadbfkjasbfkasbdf"
	responseBodyData = "response_nlnflkwqnflkasdnflnasdlfnasldnflnl"

	requestUpgradeBodyData  = "upgrade_request_asdfjkasbfkasdf"
	responseUpgradeBodyData = "upgrade_response_asdfasdfadsf"
)

func TestGRPC2HTTP_HappyPath(t *testing.T) {
	ctrl := gomock.NewController(t)
	server := mock_rpc.NewMockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse](ctrl)
	server.EXPECT().
		Context().
		Return(context.Background()).
		MinTimes(1)
	sh := sendHeader()
	gomock.InOrder(mockRecvStream(server, true,
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Header_{
				Header: sh,
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Data_{
				Data: &grpctool.HttpRequest_Data{
					Data: []byte(requestBodyData[:1]),
				},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Data_{
				Data: &grpctool.HttpRequest_Data{
					Data: []byte(requestBodyData[1:]),
				},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Trailer_{
				Trailer: &grpctool.HttpRequest_Trailer{},
			},
		},
	)...)
	gomock.InOrder(mockSendStream(t, server,
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Header_{
				Header: &grpctool.HttpResponse_Header{
					Response: &prototool.HttpResponse{
						StatusCode: http.StatusOK,
						Status:     "OK!",
						Header:     respHeaderKV(),
					},
				},
			},
		},
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Data_{
				Data: &grpctool.HttpResponse_Data{
					Data: []byte(responseBodyData),
				},
			},
		},
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Trailer_{
				Trailer: &grpctool.HttpResponse_Trailer{},
			},
		},
	)...)
	grpc2http := makeGRPC2HTTP(t, func(ctx context.Context, header *grpctool.HttpRequest_Header, body io.Reader) (grpctool.DoResponse, error) {
		matcher.AssertProtoEqual(t, header, sh)
		data, err := io.ReadAll(body)
		if !assert.NoError(t, err) {
			return grpctool.DoResponse{}, err
		}
		assert.Equal(t, requestBodyData, string(data))
		return grpctool.DoResponse{
			Resp: &http.Response{
				Status:     "OK!",
				StatusCode: http.StatusOK,
				Header:     respHTTPHeader(),
				Body:       io.NopCloser(strings.NewReader(responseBodyData)),
			},
		}, nil
	})
	err := grpc2http.Pipe(server)
	require.NoError(t, err)
}

func TestGRPC2HTTP_HappyPathNoBody(t *testing.T) {
	ctrl := gomock.NewController(t)
	server := mock_rpc.NewMockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse](ctrl)
	server.EXPECT().
		Context().
		Return(context.Background()).
		MinTimes(1)
	sh := sendHeader()
	contentLength := int64(0)
	sh.ContentLength = &contentLength
	gomock.InOrder(mockRecvStream(server, true,
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Header_{
				Header: sh,
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Trailer_{
				Trailer: &grpctool.HttpRequest_Trailer{},
			},
		},
	)...)
	gomock.InOrder(mockSendStream(t, server,
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Header_{
				Header: &grpctool.HttpResponse_Header{
					Response: &prototool.HttpResponse{
						StatusCode: http.StatusOK,
						Status:     "OK!",
						Header:     respHeaderKV(),
					},
				},
			},
		},
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Trailer_{
				Trailer: &grpctool.HttpResponse_Trailer{},
			},
		},
	)...)
	grpc2http := makeGRPC2HTTP(t, func(ctx context.Context, header *grpctool.HttpRequest_Header, body io.Reader) (grpctool.DoResponse, error) {
		assert.IsType(t, http.NoBody, body)
		matcher.AssertProtoEqual(t, header, sh, mock_prototool.Equate())
		data, err := io.ReadAll(body)
		if !assert.NoError(t, err) {
			return grpctool.DoResponse{}, err
		}
		assert.Empty(t, data)
		return grpctool.DoResponse{
			Resp: &http.Response{
				Status:     "OK!",
				StatusCode: http.StatusOK,
				Header:     respHTTPHeader(),
				Body:       io.NopCloser(bytes.NewReader(nil)),
			},
		}, nil
	})
	err := grpc2http.Pipe(server)
	require.NoError(t, err)
}

func TestGRPC2HTTP_UpgradeHappyPathWithBody(t *testing.T) {
	ctrl := gomock.NewController(t)
	server := mock_rpc.NewMockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse](ctrl)
	server.EXPECT().
		Context().
		Return(context.Background()).
		MinTimes(1)
	conn := mock_stdlib.NewMockConn(ctrl)
	gomock.InOrder(
		conn.EXPECT().Write([]byte(requestUpgradeBodyData[:1])),
		conn.EXPECT().Write([]byte(requestUpgradeBodyData[1:])),
		conn.EXPECT().Close(),
	)
	sh := sendUpgradeHeader()
	gomock.InOrder(mockRecvStream(server, true,
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Header_{
				Header: sh,
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Data_{
				Data: &grpctool.HttpRequest_Data{
					Data: []byte(requestBodyData[:1]),
				},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Data_{
				Data: &grpctool.HttpRequest_Data{
					Data: []byte(requestBodyData[1:]),
				},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Trailer_{
				Trailer: &grpctool.HttpRequest_Trailer{},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_UpgradeData_{
				UpgradeData: &grpctool.HttpRequest_UpgradeData{
					Data: []byte(requestUpgradeBodyData[:1]),
				},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_UpgradeData_{
				UpgradeData: &grpctool.HttpRequest_UpgradeData{
					Data: []byte(requestUpgradeBodyData[1:]),
				},
			},
		},
	)...)
	gomock.InOrder(mockSendStream(t, server,
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Header_{
				Header: &grpctool.HttpResponse_Header{
					Response: &prototool.HttpResponse{
						StatusCode: http.StatusSwitchingProtocols,
						Status:     "OK!",
						Header:     respHeaderKV(),
					},
				},
			},
		},
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Data_{
				Data: &grpctool.HttpResponse_Data{
					Data: []byte(responseBodyData),
				},
			},
		},
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Trailer_{
				Trailer: &grpctool.HttpResponse_Trailer{},
			},
		},
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_UpgradeData_{
				UpgradeData: &grpctool.HttpResponse_UpgradeData{
					Data: []byte(responseUpgradeBodyData),
				},
			},
		},
	)...)
	grpc2http := makeGRPC2HTTP(t, func(ctx context.Context, header *grpctool.HttpRequest_Header, body io.Reader) (grpctool.DoResponse, error) {
		matcher.AssertProtoEqual(t, header, sh)
		data, err := io.ReadAll(body)
		if !assert.NoError(t, err) {
			return grpctool.DoResponse{}, err
		}
		assert.Equal(t, requestBodyData, string(data))
		return grpctool.DoResponse{
			Resp: &http.Response{
				Status:     "OK!",
				StatusCode: http.StatusSwitchingProtocols,
				Header:     respHTTPHeader(),
				Body:       io.NopCloser(strings.NewReader(responseBodyData)),
			},
			UpgradeConn: conn,
			ConnReader:  bufio.NewReader(strings.NewReader(responseUpgradeBodyData)),
		}, nil
	})
	err := grpc2http.Pipe(server)
	require.NoError(t, err)
}

func TestGRPC2HTTP_UpgradeHappyPathNoBody(t *testing.T) {
	ctrl := gomock.NewController(t)
	server := mock_rpc.NewMockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse](ctrl)
	server.EXPECT().
		Context().
		Return(context.Background()).
		MinTimes(1)
	conn := mock_stdlib.NewMockConn(ctrl)
	conn.EXPECT().Close()
	sh := sendUpgradeHeader()
	contentLength := int64(0)
	sh.ContentLength = &contentLength
	gomock.InOrder(mockRecvStream(server, true,
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Header_{
				Header: sh,
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Trailer_{
				Trailer: &grpctool.HttpRequest_Trailer{},
			},
		},
	)...)
	gomock.InOrder(mockSendStream(t, server,
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Header_{
				Header: &grpctool.HttpResponse_Header{
					Response: &prototool.HttpResponse{
						StatusCode: http.StatusSwitchingProtocols,
						Status:     "OK!",
						Header:     respHeaderKV(),
					},
				},
			},
		},
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Trailer_{
				Trailer: &grpctool.HttpResponse_Trailer{},
			},
		},
	)...)
	grpc2http := makeGRPC2HTTP(t, func(ctx context.Context, header *grpctool.HttpRequest_Header, body io.Reader) (grpctool.DoResponse, error) {
		assert.IsType(t, http.NoBody, body)
		matcher.AssertProtoEqual(t, header, sh)
		data, err := io.ReadAll(body)
		if !assert.NoError(t, err) {
			return grpctool.DoResponse{}, err
		}
		assert.Empty(t, data)
		return grpctool.DoResponse{
			Resp: &http.Response{
				Status:     "OK!",
				StatusCode: http.StatusSwitchingProtocols,
				Header:     respHTTPHeader(),
				Body:       io.NopCloser(bytes.NewReader(nil)),
			},
			UpgradeConn: conn,
			ConnReader:  bufio.NewReader(bytes.NewReader(nil)),
		}, nil
	})
	err := grpc2http.Pipe(server)
	require.NoError(t, err)
}

func TestGRPC2HTTP_ServerRefusesToUpgrade(t *testing.T) {
	ctrl := gomock.NewController(t)
	server := mock_rpc.NewMockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse](ctrl)
	server.EXPECT().
		Context().
		Return(context.Background()).
		MinTimes(1)
	gomock.InOrder(mockRecvStream(server, true,
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Header_{
				Header: sendUpgradeHeader(),
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Trailer_{
				Trailer: &grpctool.HttpRequest_Trailer{},
			},
		},
	)...)
	gomock.InOrder(mockSendStream(t, server,
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Header_{
				Header: &grpctool.HttpResponse_Header{
					Response: &prototool.HttpResponse{
						StatusCode: http.StatusOK,
						Status:     "OK!",
					},
				},
			},
		},
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Trailer_{
				Trailer: &grpctool.HttpResponse_Trailer{},
			},
		},
	)...)
	grpc2http := makeGRPC2HTTP(t, func(ctx context.Context, header *grpctool.HttpRequest_Header, body io.Reader) (grpctool.DoResponse, error) {
		return grpctool.DoResponse{
			Resp: &http.Response{
				Status:     "OK!",
				StatusCode: http.StatusOK,
				Body:       io.NopCloser(bytes.NewReader(nil)),
			},
		}, nil
	})
	err := grpc2http.Pipe(server)
	require.NoError(t, err)
}

func TestGRPC2HTTP_UpgradeMessageForNonUpgradeRequest(t *testing.T) {
	ctrl := gomock.NewController(t)
	server := mock_rpc.NewMockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse](ctrl)
	server.EXPECT().
		Context().
		Return(context.Background()).
		MinTimes(1)
	gomock.InOrder(mockRecvStream(server, false,
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Header_{
				Header: sendHeader(),
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Trailer_{
				Trailer: &grpctool.HttpRequest_Trailer{},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_UpgradeData_{
				UpgradeData: &grpctool.HttpRequest_UpgradeData{
					Data: []byte(requestUpgradeBodyData),
				},
			},
		},
	)...)
	gomock.InOrder(mockSendStream(t, server,
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Header_{
				Header: &grpctool.HttpResponse_Header{
					Response: &prototool.HttpResponse{
						StatusCode: http.StatusOK,
						Status:     "OK!",
					},
				},
			},
		},
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Trailer_{
				Trailer: &grpctool.HttpResponse_Trailer{},
			},
		},
	)...)
	grpc2http := makeGRPC2HTTP(t, func(ctx context.Context, header *grpctool.HttpRequest_Header, body io.Reader) (grpctool.DoResponse, error) {
		return grpctool.DoResponse{
			Resp: &http.Response{
				Status:     "OK!",
				StatusCode: http.StatusOK,
				Body:       io.NopCloser(bytes.NewReader(nil)),
			},
		}, nil
	})
	err := grpc2http.Pipe(server)
	require.EqualError(t, err, "rpc error: code = Internal desc = unexpected HttpRequest_UpgradeData message for non-upgrade request")
}

func TestGRPC2HTTP_FailureWhenDataWasSentForRequestNotExpectingData(t *testing.T) {
	ctrl := gomock.NewController(t)
	server := mock_rpc.NewMockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse](ctrl)
	server.EXPECT().
		Context().
		Return(context.Background()).
		MinTimes(1)
	sh := sendHeader()
	contentLength := int64(0)
	sh.ContentLength = &contentLength
	gomock.InOrder(mockRecvStream(server, false,
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Header_{
				Header: sh,
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Data_{
				Data: &grpctool.HttpRequest_Data{
					Data: []byte(requestBodyData),
				},
			},
		},
	)...)
	gomock.InOrder(mockSendStream(t, server,
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Header_{
				Header: &grpctool.HttpResponse_Header{
					Response: &prototool.HttpResponse{
						StatusCode: http.StatusOK,
						Status:     "OK!",
					},
				},
			},
		},
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Trailer_{
				Trailer: &grpctool.HttpResponse_Trailer{},
			},
		},
	)...)
	grpc2http := makeGRPC2HTTP(t, func(ctx context.Context, header *grpctool.HttpRequest_Header, body io.Reader) (grpctool.DoResponse, error) {
		return grpctool.DoResponse{
			Resp: &http.Response{
				Status:     "OK!",
				StatusCode: http.StatusOK,
				Body:       io.NopCloser(bytes.NewReader(nil)),
			},
		}, nil
	})
	err := grpc2http.Pipe(server)
	require.EqualError(t, err, "rpc error: code = Internal desc = unexpected HttpRequest_Data message received")
}

func TestGRPC2HTTP_UpgradeMessageWhenServerRefusesToUpgrade(t *testing.T) {
	ctrl := gomock.NewController(t)
	server := mock_rpc.NewMockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse](ctrl)
	server.EXPECT().
		Context().
		Return(context.Background()).
		MinTimes(1)
	gomock.InOrder(mockRecvStream(server, false,
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Header_{
				Header: sendUpgradeHeader(),
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Trailer_{
				Trailer: &grpctool.HttpRequest_Trailer{},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_UpgradeData_{
				UpgradeData: &grpctool.HttpRequest_UpgradeData{
					Data: []byte(requestUpgradeBodyData),
				},
			},
		},
	)...)
	gomock.InOrder(mockSendStream(t, server,
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Header_{
				Header: &grpctool.HttpResponse_Header{
					Response: &prototool.HttpResponse{
						StatusCode: http.StatusOK,
						Status:     "OK!",
					},
				},
			},
		},
		&grpctool.HttpResponse{
			Message: &grpctool.HttpResponse_Trailer_{
				Trailer: &grpctool.HttpResponse_Trailer{},
			},
		},
	)...)
	grpc2http := makeGRPC2HTTP(t, func(ctx context.Context, header *grpctool.HttpRequest_Header, body io.Reader) (grpctool.DoResponse, error) {
		return grpctool.DoResponse{
			Resp: &http.Response{
				Status:     "OK!",
				StatusCode: http.StatusOK,
				Body:       io.NopCloser(bytes.NewReader(nil)),
			},
		}, nil
	})
	err := grpc2http.Pipe(server)
	require.EqualError(t, err, "rpc error: code = Internal desc = unexpected HttpRequest_UpgradeData message for HTTP status code 200")
}

// This test ensures PipeOutboundToInbound goroutine is unblocked on error in PipeInboundToOutbound.
func TestGRPC2HTTP_ErrorReceivingHeader(t *testing.T) {
	ctrl := gomock.NewController(t)
	server := mock_rpc.NewMockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse](ctrl)
	server.EXPECT().
		Context().
		Return(context.Background()).
		MinTimes(1)
	server.EXPECT().
		RecvMsg(gomock.Any()).
		Return(status.Error(codes.DataLoss, "recv failed"))
	grpc2http := grpctool.InboundGRPCToOutboundHTTP{
		Log: testlogger.New(t),
		HandleProcessingError: func(msg string, err error) {
			t.Error(msg, err)
		},
		HandleIOError: func(msg string, err error) error {
			t.Error(msg, err)
			return nil
		},
		HTTPDo: func(ctx context.Context, header *grpctool.HttpRequest_Header, body io.Reader) (grpctool.DoResponse, error) {
			t.FailNow()
			return grpctool.DoResponse{}, nil
		},
	}
	err := grpc2http.Pipe(server)
	require.EqualError(t, err, "rpc error: code = DataLoss desc = recv failed")
}

// This test ensures PipeInboundToOutbound goroutine is unblocked on error in PipeOutboundToInbound.
func TestGRPC2HTTP_ErrorReceivingHTTPResponse(t *testing.T) {
	ctrl := gomock.NewController(t)
	server := mock_rpc.NewMockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse](ctrl)
	server.EXPECT().
		Context().
		Return(context.Background()).
		MinTimes(1)
	block := make(chan struct{})
	gomock.InOrder(
		server.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpRequest{
				Message: &grpctool.HttpRequest_Header_{
					Header: sendUpgradeHeader(),
				},
			})),
		server.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpRequest{
				Message: &grpctool.HttpRequest_Trailer_{
					Trailer: &grpctool.HttpRequest_Trailer{},
				},
			})),
		server.EXPECT().
			RecvMsg(gomock.Any()).
			Do(func(msg any) error {
				testhelpers.SetValue(msg, &grpctool.HttpRequest{
					Message: &grpctool.HttpRequest_UpgradeData_{
						UpgradeData: &grpctool.HttpRequest_UpgradeData{
							Data: []byte(requestUpgradeBodyData),
						},
					},
				})
				close(block)
				return nil
			}),
	)
	grpc2http := makeGRPC2HTTP(t, func(ctx context.Context, header *grpctool.HttpRequest_Header, body io.Reader) (grpctool.DoResponse, error) {
		return grpctool.DoResponse{}, status.Error(codes.DataLoss, "DO failed")
	})
	err := grpc2http.Pipe(server)
	require.EqualError(t, err, "rpc error: code = DataLoss desc = DO failed")
	<-block // wait for the last recv to avoid a race
}

// This test ensures PipeOutboundToInbound goroutine is unblocked on error in PipeInboundToOutbound.
func TestGRPC2HTTP_ErrorReceivingGRPCRequest(t *testing.T) {
	ctrl := gomock.NewController(t)
	server := mock_rpc.NewMockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse](ctrl)
	server.EXPECT().
		Context().
		Return(context.Background()).
		MinTimes(1)
	gomock.InOrder(
		server.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpRequest{
				Message: &grpctool.HttpRequest_Header_{
					Header: sendUpgradeHeader(),
				},
			})),
		server.EXPECT().
			RecvMsg(gomock.Any()).
			Return(status.Error(codes.DataLoss, "RecvMsg failed")),
	)
	grpc2http := makeGRPC2HTTP(t, func(ctx context.Context, header *grpctool.HttpRequest_Header, body io.Reader) (grpctool.DoResponse, error) {
		_, err := body.Read(make([]byte, 1))
		return grpctool.DoResponse{}, err
	})
	err := grpc2http.Pipe(server)
	assert.EqualError(t, err, "rpc error: code = DataLoss desc = RecvMsg failed")
}

func makeGRPC2HTTP(t *testing.T, f grpctool.HTTPDo) *grpctool.InboundGRPCToOutboundHTTP {
	return &grpctool.InboundGRPCToOutboundHTTP{
		Log: testlogger.New(t),
		HandleProcessingError: func(msg string, err error) {
			t.Error(msg, err)
		},
		HandleIOError: func(msg string, err error) error {
			t.Error(msg, err)
			return nil
		},
		HTTPDo: f,
	}
}

func sendHeader() *grpctool.HttpRequest_Header {
	return &grpctool.HttpRequest_Header{
		Request: &prototool.HttpRequest{
			Method: "BOOM",
			Header: []*prototool.HeaderKV{
				mock_prototool.NewHeaderKV("Req-Normal", "value"),
				mock_prototool.NewHeaderKV("Req-Empty", ""),
				mock_prototool.NewHeaderKV("Req-Binary", "\xFF"),
			},
			UrlPath: "/asd/asd/asd",
			Query: []*prototool.QueryKV{
				mock_prototool.NewQueryKV("adasd", "a"),
			},
		},
		Extra: &anypb.Any{
			TypeUrl: "sadfasdfasdfads",
			Value:   []byte{1, 2, 3},
		},
	}
}

func sendUpgradeHeader() *grpctool.HttpRequest_Header {
	sh := sendHeader()
	sh.Request.Header = append(sh.Request.Header,
		mock_prototool.NewHeaderKV(httpz.UpgradeHeader, "a"),
		mock_prototool.NewHeaderKV(httpz.ConnectionHeader, "upgrade"),
	)
	return sh
}

func respHeaderKV() []*prototool.HeaderKV {
	return []*prototool.HeaderKV{
		mock_prototool.NewHeaderKV("Resp-Normal", "a1", "a2"),
		mock_prototool.NewHeaderKV("Resp-Empty", ""),
		mock_prototool.NewHeaderKV("Resp-Binary", "\xFF"),
	}
}

func respHTTPHeader() http.Header {
	return http.Header{
		"Resp-Normal": []string{"a1", "a2"},
		"Resp-Empty":  []string{""},
		"Resp-Binary": []string{"\xFF"},
	}
}

func mockRecvStream(server *mock_rpc.MockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse], eof bool, msgs ...proto.Message) []any {
	res := make([]any, 0, len(msgs)+1)
	for _, msg := range msgs {
		call := server.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(msg))
		res = append(res, call)
	}
	if eof {
		call := server.EXPECT().
			RecvMsg(gomock.Any()).
			Return(io.EOF)
		res = append(res, call)
	}
	return res
}

func mockSendStream(t *testing.T, server *mock_rpc.MockInboundGRPCToOutboundHTTPStream[grpctool.HttpResponse], msgs ...*grpctool.HttpResponse) []any {
	res := make([]any, 0, len(msgs))
	for _, msg := range msgs {
		call := server.EXPECT().
			Send(mock_prototool.ProtoEq(t, msg))
		res = append(res, call)
	}
	return res
}
