package grpctool_test

import (
	"bufio"
	"context"
	"errors"
	"io"
	"net/http"
	"net/url"
	"slices"
	"strings"
	"testing"
	"time"

	"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/grpctool/test"
	"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_kubernetes_api"
	"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_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/protobuf/proto"
	"google.golang.org/protobuf/types/known/anypb"
	"k8s.io/utils/ptr"
)

const (
	requestPath = "/test"
)

func TestHTTP2GRPC_HappyPath(t *testing.T) {
	mrClient, w, r, x := setupHTTP2GRPC(t, false)
	headerExtra := &test.Request{}
	send := mockSendHappy(t, mrClient, headerExtra, false)
	wh := http.Header{httpz.ContentTypeHeader: []string{"text/plain"}}
	recv := []any{
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Header_{
					Header: &grpctool.HttpResponse_Header{
						Response: &prototool.HttpResponse{
							StatusCode: http.StatusOK,
							Status:     http.StatusText(http.StatusOK),
							Header:     respHeaderKV(),
						},
					},
				},
			})),
		w.EXPECT().
			Header().
			Return(wh),
		w.EXPECT().
			WriteHeader(http.StatusOK).
			Do(func(status int) {
				// when WriteHeader is called, headers should have been set already
				expectedResponseHeaders := respHTTPHeader()
				expectedResponseHeaders[httpz.ContentTypeHeader] = []string{"text/plain"}
				assert.Equal(t, expectedResponseHeaders, wh)
			}),
		w.EXPECT().
			Flush(),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Data_{
					Data: &grpctool.HttpResponse_Data{
						Data: []byte(responseBodyData),
					},
				},
			})),
		w.EXPECT().
			Write([]byte(responseBodyData)),
		w.EXPECT().
			Flush(),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Trailer_{
					Trailer: &grpctool.HttpResponse_Trailer{},
				},
			})),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Return(io.EOF),
	}
	calls := send
	calls = append(calls, recv...)
	gomock.InOrder(calls...)
	abort := x.Pipe(w, r, headerExtra)
	assert.False(t, abort)
}

func TestHTTP2GRPC_DropConnection_WhenBody_WithoutContentType(t *testing.T) {
	tests := []struct {
		name string
		h    http.Header
	}{
		{
			name: "without content type header",
			h:    make(http.Header),
		},
		{
			name: "with empty content type header",
			h:    http.Header{httpz.ContentTypeHeader: []string{""}},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			mrClient, w, r, x := setupHTTP2GRPC(t, false)

			var expectedErr error
			x.HTTP2GRPC.HandleProcessingErrorFunc = func(msg string, err error) {
				// we are okay with only receiving the last error and we don't care about concurrency.
				expectedErr = err
			}

			headerExtra := &test.Request{}
			send := mockSendHappy(t, mrClient, headerExtra, false)
			wh := make(http.Header) // no content type header
			recv := []any{
				mrClient.EXPECT().
					RecvMsg(gomock.Any()).
					Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
						Message: &grpctool.HttpResponse_Header_{
							Header: &grpctool.HttpResponse_Header{
								Response: &prototool.HttpResponse{
									StatusCode: http.StatusOK,
									Status:     http.StatusText(http.StatusOK),
									Header:     respHeaderKV(),
								},
							},
						},
					})),
				w.EXPECT().
					Header().
					Return(wh),
				w.EXPECT().
					WriteHeader(http.StatusOK),
				w.EXPECT().
					Flush(),
				mrClient.EXPECT().
					RecvMsg(gomock.Any()).
					Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
						Message: &grpctool.HttpResponse_Data_{
							Data: &grpctool.HttpResponse_Data{
								Data: []byte(responseBodyData),
							},
						},
					})),
			}
			calls := send
			calls = append(calls, recv...)
			gomock.InOrder(calls...)
			abort := x.Pipe(w, r, headerExtra)
			require.True(t, abort)
			assert.EqualError(t, expectedErr, "response is missing Content-Type header, but attempted to send content. Dropping connection due to security concerns")
		})
	}
}

func TestHTTP2GRPC_HappyPath_GETNoBody(t *testing.T) {
	ctrl := gomock.NewController(t)
	mrClient := mock_kubernetes_api.NewMockKubernetesApi_MakeRequestClient(ctrl)
	w := mock_stdlib.NewMockResponseWriterFlusher(ctrl)
	r := &http.Request{
		Method: http.MethodGet,
		URL: &url.URL{
			Scheme: "http",
			Host:   "example.com",
			Path:   requestPath,
		},
		Body: nil, // No body for a GET request is allowed.
	}

	x := grpctool.InboundHTTPToOutboundGRPC{
		HTTP2GRPC: grpctool.HTTPToOutboundGRPC{
			Log: testlogger.New(t),
			HandleProcessingErrorFunc: func(msg string, err error) {
				t.Error(msg, err)
			},
			CheckHeader: func(statusCode int32, header http.Header) error {
				return nil
			},
		},
		NewClient: func(ctx context.Context) (grpctool.HTTPRequestClient, error) {
			return mrClient, nil
		},
		WriteErrorResponse: func(eResp *grpctool.ErrResp) {
			w.WriteHeader(int(eResp.StatusCode))
		},
		MergeHeaders:  mergeHeaders,
		TrackResponse: func(statusCode int) {},
	}
	send := mockSendHTTP2GRPCStream(t, mrClient, true,
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Header_{
				Header: &grpctool.HttpRequest_Header{
					Request: &prototool.HttpRequest{
						Method:  http.MethodGet,
						UrlPath: requestPath,
					},
					ContentLength: ptr.To(int64(0)),
				},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Trailer_{
				Trailer: &grpctool.HttpRequest_Trailer{},
			},
		},
	)
	wh := make(http.Header)
	recv := []any{
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Header_{
					Header: &grpctool.HttpResponse_Header{
						Response: &prototool.HttpResponse{
							StatusCode: http.StatusOK,
							Status:     http.StatusText(http.StatusOK),
						},
					},
				},
			})),
		w.EXPECT().
			Header().
			Return(wh),
		w.EXPECT().
			WriteHeader(http.StatusOK),
		w.EXPECT().
			Flush(),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Trailer_{
					Trailer: &grpctool.HttpResponse_Trailer{},
				},
			})),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Return(io.EOF),
	}
	calls := send
	calls = append(calls, recv...)
	gomock.InOrder(calls...)
	abort := x.Pipe(w, r, nil)
	assert.False(t, abort)
}

func TestHTTP2GRPC_Upgrade_HappyPath(t *testing.T) {
	ctrl := gomock.NewController(t)
	mrClient, w, r, x := setupHTTP2GRPC(t, true)
	conn := mock_stdlib.NewMockConn(ctrl)
	headerExtra := &test.Request{}
	wh := make(http.Header)
	setReadDeadlineCall := conn.EXPECT().
		SetReadDeadline(time.Time{})
	send := mockSendHappy(t, mrClient, headerExtra, true)
	recv := []any{
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Header_{
					Header: &grpctool.HttpResponse_Header{
						Response: &prototool.HttpResponse{
							StatusCode: http.StatusSwitchingProtocols,
							Status:     http.StatusText(http.StatusSwitchingProtocols),
							Header: append(
								respHeaderKV(),
								mock_prototool.NewHeaderKV(httpz.UpgradeHeader, "http/x"),
								mock_prototool.NewHeaderKV(httpz.ConnectionHeader, "upgrade"),
							),
						},
					},
				},
			})),
		w.EXPECT().
			Header().
			Return(wh),
		w.EXPECT().
			WriteHeader(http.StatusSwitchingProtocols).
			Do(func(status int) {
				// when WriteHeader is called, headers should have been set already
				expected := respHTTPHeader()
				expected[httpz.UpgradeHeader] = []string{"http/x"}
				expected[httpz.ConnectionHeader] = []string{"upgrade"}
				assert.Equal(t, expected, wh)
			}),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Trailer_{
					Trailer: &grpctool.HttpResponse_Trailer{},
				},
			})),
		w.EXPECT().
			Hijack().
			Return(conn, bufio.NewReadWriter(bufio.NewReader(conn), nil), nil),
		setReadDeadlineCall,
	}
	calls := send
	calls = append(calls, recv...)
	gomock.InOrder(calls...)
	connCloseCall := conn.EXPECT().Close()
	// pipeOutboundToInboundUpgraded
	gomock.InOrder(
		setReadDeadlineCall,
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_UpgradeData_{
					UpgradeData: &grpctool.HttpResponse_UpgradeData{
						Data: []byte(responseUpgradeBodyData),
					},
				},
			})),
		conn.EXPECT().
			SetWriteDeadline(gomock.Any()),
		conn.EXPECT().
			Write([]byte(responseUpgradeBodyData)),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Return(io.EOF),
		connCloseCall,
	)
	// pipeInboundToOutboundUpgraded
	gomock.InOrder(
		setReadDeadlineCall,
		conn.EXPECT().
			Read(gomock.Any()).
			DoAndReturn(func(b []byte) (int, error) {
				return copy(b, requestUpgradeBodyData), io.EOF
			}),
		mrClient.EXPECT().
			Send(matcher.ProtoEq(t, &grpctool.HttpRequest{
				Message: &grpctool.HttpRequest_UpgradeData_{
					UpgradeData: &grpctool.HttpRequest_UpgradeData{
						Data: []byte(requestUpgradeBodyData),
					},
				},
			})),
		mrClient.EXPECT().CloseSend(),
		connCloseCall,
	)
	abort := x.Pipe(w, r, headerExtra)
	assert.False(t, abort)
}

func TestHTTP2GRPC_Upgrade_DropConnection_WhenBody_WithoutContentType(t *testing.T) {
	mrClient, w, r, x := setupHTTP2GRPC(t, true)

	var expectedErr error
	x.HTTP2GRPC.HandleProcessingErrorFunc = func(msg string, err error) {
		// we are okay with only receiving the last error and we don't care about concurrency.
		expectedErr = err
	}

	headerExtra := &test.Request{}
	wh := make(http.Header)
	send := mockSendHappy(t, mrClient, headerExtra, true)
	recv := []any{
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Header_{
					Header: &grpctool.HttpResponse_Header{
						Response: &prototool.HttpResponse{
							StatusCode: http.StatusSwitchingProtocols,
							Status:     http.StatusText(http.StatusSwitchingProtocols),
							Header: append(
								respHeaderKV(),
								mock_prototool.NewHeaderKV(httpz.UpgradeHeader, "http/x"),
								mock_prototool.NewHeaderKV(httpz.ConnectionHeader, "upgrade"),
							),
						},
					},
				},
			})),
		w.EXPECT().
			Header().
			Return(wh),
		w.EXPECT().
			WriteHeader(http.StatusSwitchingProtocols).
			Do(func(status int) {
				// when WriteHeader is called, headers should have been set already
				expected := respHTTPHeader()
				expected[httpz.UpgradeHeader] = []string{"http/x"}
				expected[httpz.ConnectionHeader] = []string{"upgrade"}
				assert.Equal(t, expected, wh)
			}),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Data_{
					Data: &grpctool.HttpResponse_Data{
						Data: []byte(responseBodyData),
					},
				},
			})),
	}
	calls := send
	calls = append(calls, recv...)
	gomock.InOrder(calls...)
	abort := x.Pipe(w, r, headerExtra)
	require.True(t, abort)
	assert.EqualError(t, expectedErr, "response is missing Content-Type header, but attempted to send content. Dropping connection due to security concerns")
}

func TestHTTP2GRPC_Upgrade_InboundConnectionBreak(t *testing.T) {
	ctrl := gomock.NewController(t)
	mrClient := mock_kubernetes_api.NewMockKubernetesApi_MakeRequestClient(ctrl)
	w := mock_stdlib.NewMockResponseWriterFlusher(ctrl)
	r := &http.Request{
		Method: http.MethodGet,
		URL:    reqURL(),
		Header: http.Header{
			"A":                    []string{"a1", "a2"},
			httpz.ConnectionHeader: []string{"upgrade"},
			httpz.UpgradeHeader:    []string{"http/x"},
		},
		ContentLength: int64(len(requestBodyData)),
		Body:          io.NopCloser(strings.NewReader(requestBodyData)),
	}

	var mrClientCtx context.Context
	x := grpctool.InboundHTTPToOutboundGRPC{
		HTTP2GRPC: grpctool.HTTPToOutboundGRPC{
			Log: testlogger.New(t),
			HandleProcessingErrorFunc: func(msg string, err error) {
				t.Error(msg, err)
			},
			CheckHeader: func(statusCode int32, header http.Header) error {
				return nil
			},
		},
		NewClient: func(ctx context.Context) (grpctool.HTTPRequestClient, error) {
			mrClientCtx = ctx //nolint:fatcontext
			return mrClient, nil
		},
		WriteErrorResponse: func(eResp *grpctool.ErrResp) {
			w.WriteHeader(int(eResp.StatusCode))
		},
		MergeHeaders:  mergeHeaders,
		TrackResponse: func(statusCode int) {},
	}
	conn := mock_stdlib.NewMockConn(ctrl)
	headerExtra := &test.Request{}
	wh := make(http.Header)
	setReadDeadlineCall := conn.EXPECT().
		SetReadDeadline(time.Time{})
	send := mockSendHappy(t, mrClient, headerExtra, true)
	recv := []any{
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Header_{
					Header: &grpctool.HttpResponse_Header{
						Response: &prototool.HttpResponse{
							StatusCode: http.StatusSwitchingProtocols,
							Status:     http.StatusText(http.StatusSwitchingProtocols),
							Header: []*prototool.HeaderKV{
								mock_prototool.NewHeaderKV(httpz.UpgradeHeader, "http/x"),
								mock_prototool.NewHeaderKV(httpz.ConnectionHeader, "upgrade"),
							},
						},
					},
				},
			})),
		w.EXPECT().
			Header().
			Return(wh),
		w.EXPECT().
			WriteHeader(http.StatusSwitchingProtocols),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Trailer_{
					Trailer: &grpctool.HttpResponse_Trailer{},
				},
			})),
		w.EXPECT().
			Hijack().
			Return(conn, bufio.NewReadWriter(bufio.NewReader(conn), nil), nil),
		setReadDeadlineCall,
	}
	calls := send
	calls = append(calls, recv...)
	gomock.InOrder(calls...)
	connCloseCall := conn.EXPECT().Close()
	// pipeOutboundToInboundUpgraded
	gomock.InOrder(
		setReadDeadlineCall,
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(func(msg any) error {
				<-mrClientCtx.Done() // block!
				return mrClientCtx.Err()
			}),
		connCloseCall,
	)
	// pipeInboundToOutboundUpgraded
	gomock.InOrder(
		setReadDeadlineCall,
		conn.EXPECT().
			Read(gomock.Any()).
			DoAndReturn(func(b []byte) (int, error) {
				return 0, errors.New("broken read")
			}),
		connCloseCall,
	)
	abort := x.Pipe(w, r, headerExtra)
	assert.True(t, abort)
}

func TestHTTP2GRPC_ServerRefusesToUpgrade(t *testing.T) {
	mrClient, w, r, x := setupHTTP2GRPC(t, true)
	headerExtra := &test.Request{}
	wh := make(http.Header)
	extra, err := anypb.New(headerExtra)
	require.NoError(t, err)
	contentLength := int64(len(requestBodyData))
	send := mockSendHTTP2GRPCStream(t, mrClient, false,
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Header_{
				Header: &grpctool.HttpRequest_Header{
					Request: &prototool.HttpRequest{
						Method: http.MethodGet,
						Header: []*prototool.HeaderKV{
							mock_prototool.NewHeaderKV("A", "a1", "a2"),
							mock_prototool.NewHeaderKV(httpz.UpgradeHeader, "http/x"),
							mock_prototool.NewHeaderKV(httpz.ConnectionHeader, "upgrade"),
						},
						UrlPath: requestPath,
						Query: []*prototool.QueryKV{
							mock_prototool.NewQueryKV(binKey(), binVal()),
						},
					},
					Extra:         extra,
					ContentLength: &contentLength,
				},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Data_{
				Data: &grpctool.HttpRequest_Data{
					Data: []byte(requestBodyData),
				},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Trailer_{
				Trailer: &grpctool.HttpRequest_Trailer{},
			},
		},
	)
	recv := []any{
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Header_{
					Header: &grpctool.HttpResponse_Header{
						Response: &prototool.HttpResponse{
							StatusCode: http.StatusOK,
							Status:     http.StatusText(http.StatusOK),
							Header:     respHeaderKV(),
						},
					},
				},
			})),
		mrClient.EXPECT().
			CloseSend(),
		w.EXPECT().
			Header().
			Return(wh),
		w.EXPECT().
			WriteHeader(http.StatusOK).
			Do(func(status int) {
				// when WriteHeader is called, headers should have been set already
				assert.Equal(t, respHTTPHeader(), wh)
			}),
		w.EXPECT().
			Flush(),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Trailer_{
					Trailer: &grpctool.HttpResponse_Trailer{},
				},
			})),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Return(io.EOF),
	}
	calls := send
	calls = append(calls, recv...)
	gomock.InOrder(calls...)
	abort := x.Pipe(w, r, headerExtra)
	assert.False(t, abort)
}

func TestHTTP2GRPC_HeaderRecvError(t *testing.T) {
	mrClient, w, r, x := setupHTTP2GRPC(t, false)
	headerExtra := &test.Request{}
	send := mockSendHappy(t, mrClient, headerExtra, false)
	recv := []any{
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Return(errors.New("no headers for you")),
		w.EXPECT().
			WriteHeader(http.StatusBadGateway),
	}
	calls := send
	calls = append(calls, recv...)
	gomock.InOrder(calls...)

	abort := x.Pipe(w, r, headerExtra)
	assert.False(t, abort)
}

func TestHTTP2GRPC_ErrorAfterHeaderWritten(t *testing.T) {
	mrClient, w, r, x := setupHTTP2GRPC(t, false)
	headerExtra := &test.Request{}
	send := mockSendHappy(t, mrClient, headerExtra, false)
	wh := http.Header{httpz.ContentTypeHeader: []string{"text/plain"}}
	recv := []any{
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Header_{
					Header: &grpctool.HttpResponse_Header{
						Response: &prototool.HttpResponse{
							StatusCode: http.StatusOK,
							Status:     http.StatusText(http.StatusOK),
							Header:     respHeaderKV(),
						},
					},
				},
			})),
		w.EXPECT().
			Header().
			Return(wh),
		w.EXPECT().
			WriteHeader(http.StatusOK).
			Do(func(status int) {
				// when WriteHeader is called, headers should have been set already
				expectedResponseHeaders := respHTTPHeader()
				expectedResponseHeaders[httpz.ContentTypeHeader] = []string{"text/plain"}
				assert.Equal(t, expectedResponseHeaders, wh)
			}),
		w.EXPECT().
			Flush(),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Return(errors.New("no body for you")),
	}
	calls := send
	calls = append(calls, recv...)
	gomock.InOrder(calls...)

	abort := x.Pipe(w, r, headerExtra)
	assert.True(t, abort)
}

func TestHTTP2GRPC_ErrorAfterBodyWritten(t *testing.T) {
	mrClient, w, r, x := setupHTTP2GRPC(t, false)
	headerExtra := &test.Request{}
	send := mockSendHappy(t, mrClient, headerExtra, false)
	wh := http.Header{httpz.ContentTypeHeader: []string{"text/plain"}}
	recv := []any{
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Header_{
					Header: &grpctool.HttpResponse_Header{
						Response: &prototool.HttpResponse{
							StatusCode: http.StatusOK,
							Status:     http.StatusText(http.StatusOK),
							Header:     respHeaderKV(),
						},
					},
				},
			})),
		w.EXPECT().
			Header().
			Return(wh),
		w.EXPECT().
			WriteHeader(http.StatusOK).
			Do(func(status int) {
				// when WriteHeader is called, headers should have been set already
				expectedResponseHeaders := respHTTPHeader()
				expectedResponseHeaders[httpz.ContentTypeHeader] = []string{"text/plain"}
				assert.Equal(t, expectedResponseHeaders, wh)
			}),
		w.EXPECT().
			Flush(),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Do(testhelpers.RecvMsg(&grpctool.HttpResponse{
				Message: &grpctool.HttpResponse_Data_{
					Data: &grpctool.HttpResponse_Data{
						Data: []byte(responseBodyData),
					},
				},
			})),
		w.EXPECT().
			Write([]byte(responseBodyData)),
		w.EXPECT().
			Flush(),
		mrClient.EXPECT().
			RecvMsg(gomock.Any()).
			Return(errors.New("no body for you")),
	}
	calls := send
	calls = append(calls, recv...)
	gomock.InOrder(calls...)

	abort := x.Pipe(w, r, headerExtra)
	assert.True(t, abort)
}

func setupHTTP2GRPC(t *testing.T, isUpgrade bool) (*mock_kubernetes_api.MockKubernetesApi_MakeRequestClient, *mock_stdlib.MockResponseWriterFlusher, *http.Request, grpctool.InboundHTTPToOutboundGRPC) {
	ctrl := gomock.NewController(t)
	mrClient := mock_kubernetes_api.NewMockKubernetesApi_MakeRequestClient(ctrl)
	w := mock_stdlib.NewMockResponseWriterFlusher(ctrl)
	r := &http.Request{
		Method: http.MethodGet,
		URL:    reqURL(),
		Header: http.Header{
			"A": []string{"a1", "a2"},
		},
		ContentLength: int64(len(requestBodyData)),
		Body:          io.NopCloser(strings.NewReader(requestBodyData)),
	}
	if isUpgrade {
		r.Header[httpz.ConnectionHeader] = []string{"upgrade"}
		r.Header[httpz.UpgradeHeader] = []string{"http/x"}
	}

	x := grpctool.InboundHTTPToOutboundGRPC{
		HTTP2GRPC: grpctool.HTTPToOutboundGRPC{
			Log: testlogger.New(t),
			HandleProcessingErrorFunc: func(msg string, err error) {
				t.Error(msg, err)
			},
			CheckHeader: func(statusCode int32, header http.Header) error {
				return nil
			},
		},
		NewClient: func(ctx context.Context) (grpctool.HTTPRequestClient, error) {
			return mrClient, nil
		},
		WriteErrorResponse: func(eResp *grpctool.ErrResp) {
			w.WriteHeader(int(eResp.StatusCode))
		},
		MergeHeaders:  mergeHeaders,
		TrackResponse: func(statusCode int) {},
	}
	return mrClient, w, r, x
}

func mergeHeaders(outboundResponse, inboundResponse http.Header) {
	for k, v := range outboundResponse {
		inboundResponse[k] = append(inboundResponse[k], v...)
	}
}

func mockSendHappy(t *testing.T, mrClient *mock_kubernetes_api.MockKubernetesApi_MakeRequestClient, headerExtra proto.Message, isUpgrade bool) []any {
	extra, err := anypb.New(headerExtra)
	require.NoError(t, err)
	header := []*prototool.HeaderKV{
		mock_prototool.NewHeaderKV("A", "a1", "a2"),
	}
	if isUpgrade {
		header = append(header,
			mock_prototool.NewHeaderKV(httpz.UpgradeHeader, "http/x"),
			mock_prototool.NewHeaderKV(httpz.ConnectionHeader, "upgrade"),
		)
	}
	contentLength := int64(len(requestBodyData))
	return mockSendHTTP2GRPCStream(t, mrClient, !isUpgrade,
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Header_{
				Header: &grpctool.HttpRequest_Header{
					Request: &prototool.HttpRequest{
						Method:  http.MethodGet,
						Header:  header,
						UrlPath: requestPath,
						Query: []*prototool.QueryKV{
							mock_prototool.NewQueryKV(binKey(), binVal()),
						},
					},
					Extra:         extra,
					ContentLength: &contentLength,
				},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Data_{
				Data: &grpctool.HttpRequest_Data{
					Data: []byte(requestBodyData),
				},
			},
		},
		&grpctool.HttpRequest{
			Message: &grpctool.HttpRequest_Trailer_{
				Trailer: &grpctool.HttpRequest_Trailer{},
			},
		},
	)
}

func mockSendHTTP2GRPCStream(t *testing.T, client *mock_kubernetes_api.MockKubernetesApi_MakeRequestClient, closeSend bool, msgs ...*grpctool.HttpRequest) []any {
	res := make([]any, 0, len(msgs)+1)
	for _, msg := range msgs {
		call := client.EXPECT().
			Send(mock_prototool.ProtoEq(t, msg))
		res = append(res, call)
	}
	if closeSend {
		res = append(res, client.EXPECT().CloseSend())
	}
	return res
}

func binKey() string {
	key := make([]byte, 256)
	for i := range 256 {
		key[i] = byte(i)
	}
	return string(key)
}

func binVal() string {
	val := slices.Clone([]byte(binKey()))
	slices.Reverse(val)
	return string(val)
}

func reqURL() *url.URL {
	return &url.URL{
		Scheme: "http",
		Host:   "example.com",
		Path:   requestPath,
		RawQuery: url.Values{
			binKey(): []string{binVal()},
		}.Encode(),
	}
}
