package proxy

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"net/url"
	"testing"

	gkev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/gin-gonic/gin"
	"github.com/stretchr/testify/require"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	k8slabels "k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/yaml"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/gkeheterogeneous"
	gkestatus "github.com/tidbcloud/infra-provider/pkg/gkeheterogeneous/status"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	nodepoolutil "github.com/tidbcloud/infra-provider/pkg/nodepool"
)

type recordingProxyClusterListClient struct {
	client.Client
	clusterListSelectors []string
}

func TestHTTPStatusForGRPCCode(t *testing.T) {
	tests := []struct {
		name string
		code codes.Code
		want int
	}{
		{name: "invalid argument", code: codes.InvalidArgument, want: http.StatusBadRequest},
		{name: "not found", code: codes.NotFound, want: http.StatusNotFound},
		{name: "failed precondition", code: codes.FailedPrecondition, want: http.StatusPreconditionFailed},
		{name: "permission denied", code: codes.PermissionDenied, want: http.StatusForbidden},
		{name: "default", code: codes.Internal, want: http.StatusInternalServerError},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			require.Equal(t, tt.want, httpStatusForGRPCCode(tt.code))
		})
	}
}

func (c *recordingProxyClusterListClient) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
	if _, ok := list.(*cloudv1.ClusterList); ok {
		listOptions := &client.ListOptions{}
		listOptions.ApplyOptions(opts)
		if listOptions.LabelSelector == nil {
			c.clusterListSelectors = append(c.clusterListSelectors, "")
		} else {
			c.clusterListSelectors = append(c.clusterListSelectors, listOptions.LabelSelector.String())
		}
	}
	return c.Client.List(ctx, list, opts...)
}

func TestClusterAgentProxyDerivesTiDBGatekeeperAssignKind(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var gotBody []byte
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var err error
		gotBody, err = io.ReadAll(r.Body)
		require.NoError(t, err)
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiDBDeploy(t, cluster, "tidb-priority", "extra-group")...)
	computeClassName := cluster.GKEComputeClassNameForTiDBGroup("extra-group")
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns",
		"application/json",
		bytes.NewBufferString(`{"componentType":"TIDB","computeClassName":"`+computeClassName+`","tidbGroupId":"extra-group"}`),
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusOK, resp.StatusCode)
	require.JSONEq(t, `{"componentType":"TIDB","computeClassName":"`+computeClassName+`","tidbGroupId":"extra-group","tidbGroupKind":"TIDB_GROUP_KIND_EXTRA"}`, string(gotBody))
}

func TestClusterAgentProxyRejectsMismatchedTiDBGatekeeperAssignKind(t *testing.T) {
	gin.SetMode(gin.TestMode)

	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		t.Fatal("backend should not be called")
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiDBDeploy(t, cluster, "tidb-priority", "extra-group")...)
	objects = append(objects, addReadyProxyTestTiKVDeploy(t, cluster, "tikv-priority")...)
	computeClassName := cluster.GKEComputeClassNameForTiDBGroup("extra-group")
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns",
		"application/json",
		bytes.NewBufferString(`{"componentType":"TIDB","computeClassName":"`+computeClassName+`","tidbGroupId":"extra-group","tidbGroupKind":"TIDB_GROUP_KIND_DEFAULT"}`),
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusBadRequest, resp.StatusCode)
}

func TestClusterAgentProxyAcceptsSnakeCaseGatekeeperAssignFields(t *testing.T) {
	gin.SetMode(gin.TestMode)

	gotBodies := make([][]byte, 0)
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(r.Body)
		require.NoError(t, err)
		gotBodies = append(gotBodies, body)
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiDBDeploy(t, cluster, "tidb-priority", "extra-group")...)
	objects = append(objects, addReadyProxyTestTiKVDeploy(t, cluster, "tikv-priority")...)
	tidbComputeClassName := cluster.GKEComputeClassNameForTiDBGroup("extra-group")
	tikvComputeClassName := cluster.GKEComputeClassNameForTiKV()
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	tests := []struct {
		name     string
		body     string
		wantBody string
	}{
		{
			name:     "component type",
			body:     `{"component_type":"TIDB","computeClassName":"` + tidbComputeClassName + `","tidbGroupId":"extra-group"}`,
			wantBody: `{"componentType":"TIDB","computeClassName":"` + tidbComputeClassName + `","tidbGroupId":"extra-group","tidbGroupKind":"TIDB_GROUP_KIND_EXTRA"}`,
		},
		{
			name:     "TiDB group id",
			body:     `{"componentType":"TIDB","computeClassName":"` + tidbComputeClassName + `","tidb_group_id":"extra-group"}`,
			wantBody: `{"componentType":"TIDB","computeClassName":"` + tidbComputeClassName + `","tidbGroupId":"extra-group","tidbGroupKind":"TIDB_GROUP_KIND_EXTRA"}`,
		},
		{
			name:     "TiDB group kind",
			body:     `{"componentType":"TIDB","computeClassName":"` + tidbComputeClassName + `","tidbGroupId":"extra-group","tidb_group_kind":"TIDB_GROUP_KIND_EXTRA"}`,
			wantBody: `{"componentType":"TIDB","computeClassName":"` + tidbComputeClassName + `","tidbGroupId":"extra-group","tidbGroupKind":"TIDB_GROUP_KIND_EXTRA"}`,
		},
		{
			name:     "TiKV snake case is passed through without TiDB kind rewrite",
			body:     `{"component_type":"TIKV","computeClassName":"` + tikvComputeClassName + `"}`,
			wantBody: `{"component_type":"TIKV","computeClassName":"` + tikvComputeClassName + `"}`,
		},
		{
			name:     "TiKV camel case group id is ignored by readiness preflight",
			body:     `{"componentType":"TIKV","computeClassName":"` + tikvComputeClassName + `","tidbGroupId":"ignored"}`,
			wantBody: `{"componentType":"TIKV","computeClassName":"` + tikvComputeClassName + `","tidbGroupId":"ignored"}`,
		},
		{
			name:     "TiKV snake case group id is ignored by readiness preflight",
			body:     `{"component_type":"TIKV","computeClassName":"` + tikvComputeClassName + `","tidb_group_id":"ignored"}`,
			wantBody: `{"component_type":"TIKV","computeClassName":"` + tikvComputeClassName + `","tidb_group_id":"ignored"}`,
		},
	}
	for i, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			resp, err := http.Post(
				regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns",
				"application/json",
				bytes.NewBufferString(tt.body),
			)
			require.NoError(t, err)
			defer func() { _ = resp.Body.Close() }()
			body, err := io.ReadAll(resp.Body)
			require.NoError(t, err)

			require.Equal(t, http.StatusOK, resp.StatusCode, string(body))
			require.JSONEq(t, tt.wantBody, string(gotBodies[i]))
		})
	}
	require.Len(t, gotBodies, len(tests))
}

func TestClusterAgentProxyGetGatekeeperAssignProxiesWithoutDeployIntent(t *testing.T) {
	gin.SetMode(gin.TestMode)

	type proxiedRequest struct {
		path     string
		rawQuery string
	}
	var gotRequests []proxiedRequest
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotRequests = append(gotRequests, proxiedRequest{path: r.URL.Path, rawQuery: r.URL.RawQuery})
		http.Error(w, "agent gatekeeper assign not found", http.StatusNotFound)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	tests := []struct {
		name         string
		path         string
		wantPath     string
		wantRawQuery string
	}{
		{
			name:     "tikv",
			path:     "/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns/TIKV",
			wantPath: "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/gatekeeperAssigns/TIKV",
		},
		{
			name:         "tidb",
			path:         "/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns/TIDB?tidb_group_id=extra-group",
			wantPath:     "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/gatekeeperAssigns/TIDB",
			wantRawQuery: "tidb_group_id=extra-group",
		},
	}
	for i, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			resp, err := http.Get(regionalServer.URL + tt.path)
			require.NoError(t, err)
			defer func() { _ = resp.Body.Close() }()
			body, err := io.ReadAll(resp.Body)
			require.NoError(t, err)

			require.Equal(t, http.StatusNotFound, resp.StatusCode, string(body))
			require.Contains(t, string(body), "agent gatekeeper assign not found")
			require.Len(t, gotRequests, i+1)
			require.Equal(t, tt.wantPath, gotRequests[i].path)
			require.Equal(t, tt.wantRawQuery, gotRequests[i].rawQuery)
		})
	}
}

func TestClusterAgentProxyGetGatekeeperAssignRequiresGatewayWithoutDeployIntent(t *testing.T) {
	gin.SetMode(gin.TestMode)

	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		t.Fatal("backend should not be called without a ready gateway")
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	cluster.Status.Gateway = nil
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns/TIKV")
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode, string(body))
	require.Contains(t, string(body), "cluster 123 gateway endpoint is not ready")
}

func TestClusterAgentProxyGetGatekeeperAssignProxiesOrphanTiDBIntent(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalled bool
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalled = true
		http.Error(w, "agent gatekeeper assign not found", http.StatusNotFound)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	updateProxyTestGKEComputeClassBinding(t, cluster, cluster.TiDBGKEComputeClassBindingsAnnoKey("orphan"), "", []string{"n2-standard-8", "n2-standard-64"})
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns/TIDB?tidb_group_id=orphan")
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusNotFound, resp.StatusCode, string(body))
	require.Contains(t, string(body), "agent gatekeeper assign not found")
	require.True(t, backendCalled)
}

func TestClusterAgentProxyGetGatekeeperAssignDoesNotShortCircuitMissingTiDBGroupID(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalled bool
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalled = true
		http.Error(w, "tidb_group_id is required", http.StatusBadRequest)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns/TIDB")
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusBadRequest, resp.StatusCode, string(body))
	require.Contains(t, string(body), "tidb_group_id is required")
	require.True(t, backendCalled)
}

func TestClusterAgentProxyGetGatekeeperAssignProxiesWithDeployIntent(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var gotPath string
	var gotRawQuery string
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotPath = r.URL.Path
		gotRawQuery = r.URL.RawQuery
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte(`{"name":"tikv-assign"}`))
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiKVDeploy(t, cluster, "tikv-priority")...)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns/TIKV?tidb_group_id=extra-group&tidbGroupId=extra-group")
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusOK, resp.StatusCode, string(body))
	require.JSONEq(t, `{"name":"tikv-assign"}`, string(body))
	require.Equal(t, "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/gatekeeperAssigns/TIKV", gotPath)
	require.Empty(t, gotRawQuery)
}

func TestClusterAgentProxyRejectsConflictingGatekeeperAssignAliases(t *testing.T) {
	gin.SetMode(gin.TestMode)

	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		t.Fatal("backend should not be called")
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	tests := []struct {
		name string
		body string
	}{
		{
			name: "component type",
			body: `{"componentType":"TIDB","component_type":"TIKV","computeClassName":"tidb-priority","tidbGroupId":"extra-group"}`,
		},
		{
			name: "TiDB group id",
			body: `{"componentType":"TIDB","computeClassName":"tidb-priority","tidbGroupId":"extra-group","tidb_group_id":"default"}`,
		},
		{
			name: "TiDB group kind",
			body: `{"componentType":"TIDB","computeClassName":"tidb-priority","tidbGroupId":"extra-group","tidbGroupKind":"TIDB_GROUP_KIND_EXTRA","tidb_group_kind":"TIDB_GROUP_KIND_DEFAULT"}`,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			resp, err := http.Post(
				regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns",
				"application/json",
				bytes.NewBufferString(tt.body),
			)
			require.NoError(t, err)
			defer func() { _ = resp.Body.Close() }()

			require.Equal(t, http.StatusBadRequest, resp.StatusCode)
		})
	}
}

func TestClusterAgentProxyRejectsGatekeeperAssignWhenDeployNotReady(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiDBDeploy(t, cluster, "tidb-priority", "extra-group")...)
	computeClassName := cluster.GKEComputeClassNameForTiDBGroup("extra-group")
	nodepoolutil.ClearGKEAppliedComputeClassForNodePool(objects[1].(*gkev1beta1.NodePool))
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns",
		"application/json",
		bytes.NewBufferString(`{"componentType":"TIDB","computeClassName":"`+computeClassName+`","tidbGroupId":"extra-group"}`),
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
	requireProxyGRPCErrorInfoReason(t, body, gkeHeterogeneousGatekeeperAssignNotReadyReason)
	require.Equal(t, 0, backendCalls)
}

func TestClusterAgentProxyAllowsGatekeeperAssignWhenBaseNodePoolRegionLabelMissing(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiDBDeploy(t, cluster, "tidb-priority", "extra-group")...)
	computeClassName := cluster.GKEComputeClassNameForTiDBGroup("extra-group")
	delete(objects[1].(*gkev1beta1.NodePool).Labels, cloudv1.RegionLabel)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns",
		"application/json",
		bytes.NewBufferString(`{"componentType":"TIDB","computeClassName":"`+computeClassName+`","tidbGroupId":"extra-group"}`),
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusOK, resp.StatusCode)
	require.Equal(t, 1, backendCalls)
}

func TestClusterAgentProxyRejectsGatekeeperAssignWhenNodePoolScopeMismatches(t *testing.T) {
	gin.SetMode(gin.TestMode)

	tests := []struct {
		name        string
		configure   func(*gkev1beta1.NodePool)
		wantMessage string
	}{
		{
			name: "tenant label mismatch",
			configure: func(nodePool *gkev1beta1.NodePool) {
				nodePool.Labels[cloudv1.TenantTagKey] = "other-tenant"
			},
			wantMessage: cloudv1.TenantTagKey,
		},
		{
			name: "scope hash missing",
			configure: func(nodePool *gkev1beta1.NodePool) {
				delete(nodePool.Labels, consts.GKEHeterogeneousNodePoolScopeLabelKey)
			},
			wantMessage: consts.GKEHeterogeneousNodePoolScopeLabelKey,
		},
		{
			name: "scope hash mismatch",
			configure: func(nodePool *gkev1beta1.NodePool) {
				nodePool.Labels[consts.GKEHeterogeneousNodePoolScopeLabelKey] = "other-scope"
			},
			wantMessage: consts.GKEHeterogeneousNodePoolScopeLabelKey,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var backendCalls int
			agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				backendCalls++
				w.WriteHeader(http.StatusOK)
			}))
			defer agentBackend.Close()

			cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
			objects := append([]client.Object{cluster}, addReadyProxyTestTiDBDeploy(t, cluster, "tidb-priority", "extra-group")...)
			computeClassName := cluster.GKEComputeClassNameForTiDBGroup("extra-group")
			tt.configure(objects[1].(*gkev1beta1.NodePool))
			kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
			regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
			defer regionalServer.Close()

			resp, err := http.Post(
				regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns",
				"application/json",
				bytes.NewBufferString(`{"componentType":"TIDB","computeClassName":"`+computeClassName+`","tidbGroupId":"extra-group"}`),
			)
			require.NoError(t, err)
			defer func() { _ = resp.Body.Close() }()
			body, err := io.ReadAll(resp.Body)
			require.NoError(t, err)

			require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
			got := requireProxyGRPCErrorInfoReason(t, body, gkeHeterogeneousGatekeeperAssignNotReadyReason)
			require.Contains(t, got.Message, tt.wantMessage)
			require.Equal(t, 0, backendCalls)
		})
	}
}

func TestClusterAgentProxyRejectsGatekeeperAssignWhenNodePoolClusterRefMismatches(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiDBDeploy(t, cluster, "tidb-priority", "extra-group")...)
	computeClassName := cluster.GKEComputeClassNameForTiDBGroup("extra-group")
	objects[1].(*gkev1beta1.NodePool).Spec.ForProvider.Cluster = "projects/other-project/locations/us-central1/clusters/other-gke"
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns",
		"application/json",
		bytes.NewBufferString(`{"componentType":"TIDB","computeClassName":"`+computeClassName+`","tidbGroupId":"extra-group"}`),
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusForbidden, resp.StatusCode)
	got := requireProxyGRPCStatus(t, body)
	require.Equal(t, int32(codes.PermissionDenied), got.Code)
	require.Contains(t, got.Message, "GKE cluster ref")
	require.Contains(t, got.Message, regionalGKEHeterogeneousNodePoolClusterRef(cluster))
	require.Equal(t, 0, backendCalls)
}

func TestClusterAgentProxyPatchesGatekeeperAssignOnlyForReadyDeployName(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		require.Equal(t, "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/gatekeeperAssigns/"+gkeheterogeneous.ResourceName("cluster-123", cloudv1.ComponentTiDB, "extra-group", "compute-class-assign"), r.URL.Path)
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiDBDeploy(t, cluster, "tidb-priority", "extra-group")...)
	computeClassName := cluster.GKEComputeClassNameForTiDBGroup("extra-group")
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	assignName := gkeheterogeneous.ResourceName(cluster.Name, cloudv1.ComponentTiDB, "extra-group", "compute-class-assign")
	req, err := http.NewRequest(http.MethodPatch, regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns/"+assignName, bytes.NewBufferString(`{"computeClassName":"`+computeClassName+`"}`))
	require.NoError(t, err)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusOK, resp.StatusCode)
	require.Equal(t, 1, backendCalls)
}

func TestClusterAgentProxyRejectsGatekeeperAssignPatchForUnmatchedName(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiDBDeploy(t, cluster, "tidb-priority", "extra-group")...)
	computeClassName := cluster.GKEComputeClassNameForTiDBGroup("extra-group")
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	req, err := http.NewRequest(http.MethodPatch, regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/gatekeeperAssigns/custom-assign", bytes.NewBufferString(`{"computeClassName":"`+computeClassName+`"}`))
	require.NoError(t, err)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
	requireProxyGRPCErrorInfoReason(t, body, gkeHeterogeneousGatekeeperAssignNotReadyReason)
	require.Equal(t, 0, backendCalls)
}

func TestClusterAgentProxyDeleteComputeClassRejectsActiveDeployIntent(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	anchor := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	anchor.Namespace = "default"
	anchor.Spec.NetworkName = "shared-network"
	anchor.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	peer := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	peer.Namespace = "default"
	peer.Name = "cluster-peer"
	peer.Labels[cloudv1.ClusterTagKey] = "peer"
	peer.Spec.NetworkName = "shared-network"
	peer.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	computeClassName := updateProxyTestGKEComputeClassBinding(t, peer, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-16", "n2-standard-64"})
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(anchor, peer).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	req, err := http.NewRequest(http.MethodDelete, regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/computeClasses/"+computeClassName, nil)
	require.NoError(t, err)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
	got := requireProxyGRPCErrorInfoReason(t, body, gkeHeterogeneousComputeClassInUseByDeployIntentReason)
	require.Contains(t, got.Message, "cluster=default/cluster-peer")
	require.Contains(t, got.Message, "component=TIKV")
	require.Equal(t, 0, backendCalls)
}

func TestClusterAgentProxyDeleteComputeClassAllowsAgentDeleteWithoutIntent(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		require.Equal(t, "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/computeClasses/tikv-priority", r.URL.Path)
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	cluster.Namespace = "default"
	cluster.Spec.NetworkName = "shared-network"
	cluster.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	req, err := http.NewRequest(http.MethodDelete, regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/computeClasses/tikv-priority", nil)
	require.NoError(t, err)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusOK, resp.StatusCode)
	require.Equal(t, 1, backendCalls)
}

func TestClusterAgentProxyDeleteComputeClassRejectsNodePoolMetadataRef(t *testing.T) {
	gin.SetMode(gin.TestMode)

	tests := []struct {
		name      string
		configure func(*gkev1beta1.NodePool)
	}{
		{
			name: "desired only",
			configure: func(nodePool *gkev1beta1.NodePool) {
				nodepoolutil.SetGKEDesiredComputeClassForNodePool(nodePool, "retained-priority")
			},
		},
		{
			name: "applied only",
			configure: func(nodePool *gkev1beta1.NodePool) {
				nodepoolutil.SetGKEAppliedComputeClassForNodePool(nodePool, "retained-priority")
			},
		},
		{
			name: "spec label only",
			configure: func(nodePool *gkev1beta1.NodePool) {
				nodepoolutil.SetGKEComputeClassSpecLabelForNodePool(nodePool, "retained-priority")
			},
		},
		{
			name: "spec taint only",
			configure: func(nodePool *gkev1beta1.NodePool) {
				nodepoolutil.SetGKEComputeClassSpecTaintForNodePool(nodePool, "retained-priority")
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var backendCalls int
			agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				backendCalls++
				w.WriteHeader(http.StatusOK)
			}))
			defer agentBackend.Close()

			cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
			cluster.Namespace = "default"
			cluster.Spec.NetworkName = "shared-network"
			nodePool := newReadyProxyTestNodePool(cluster, "retained-extra-pool", "n2-standard-64", "", true)
			nodepoolutil.ClearGKEComputeClassForNodePool(nodePool)
			tt.configure(nodePool)
			kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster, nodePool).Build()
			regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
			defer regionalServer.Close()

			req, err := http.NewRequest(http.MethodDelete, regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/computeClasses/retained-priority", nil)
			require.NoError(t, err)
			resp, err := http.DefaultClient.Do(req)
			require.NoError(t, err)
			defer func() { _ = resp.Body.Close() }()
			body, err := io.ReadAll(resp.Body)
			require.NoError(t, err)

			require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
			got := requireProxyGRPCErrorInfoReason(t, body, gkeHeterogeneousComputeClassInUseByNodePoolReason)
			require.Contains(t, got.Message, "nodepool=retained-extra-pool")
			require.Equal(t, 0, backendCalls)
		})
	}
}

func TestClusterAgentProxyDeleteComputeClassAllowsAgentDeleteForNodePoolWithoutMatchingMetadata(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		require.Equal(t, "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/computeClasses/retained-priority", r.URL.Path)
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	cluster.Namespace = "default"
	cluster.Spec.NetworkName = "shared-network"
	nodePool := newReadyProxyTestNodePool(cluster, "retained-extra-pool", "n2-standard-64", "", true)
	nodepoolutil.ClearGKEComputeClassForNodePool(nodePool)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster, nodePool).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	req, err := http.NewRequest(http.MethodDelete, regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/computeClasses/retained-priority", nil)
	require.NoError(t, err)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusOK, resp.StatusCode)
	require.Equal(t, 1, backendCalls)
}

func TestClusterAgentProxyDeleteComputeClassAllowsAgentDeleteForNodePoolOutsideScopeHash(t *testing.T) {
	gin.SetMode(gin.TestMode)

	tests := []struct {
		name      string
		configure func(*gkev1beta1.NodePool)
	}{
		{
			name: "scope hash missing",
			configure: func(nodePool *gkev1beta1.NodePool) {
				delete(nodePool.Labels, consts.GKEHeterogeneousNodePoolScopeLabelKey)
			},
		},
		{
			name: "scope hash mismatch",
			configure: func(nodePool *gkev1beta1.NodePool) {
				nodePool.Labels[consts.GKEHeterogeneousNodePoolScopeLabelKey] = "other-scope"
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var backendCalls int
			agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				backendCalls++
				require.Equal(t, "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/computeClasses/retained-priority", r.URL.Path)
				w.WriteHeader(http.StatusOK)
			}))
			defer agentBackend.Close()

			cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
			cluster.Namespace = "default"
			cluster.Spec.NetworkName = "shared-network"
			nodePool := newReadyProxyTestNodePool(cluster, "retained-extra-pool", "n2-standard-64", "retained-priority", true)
			tt.configure(nodePool)
			kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster, nodePool).Build()
			regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
			defer regionalServer.Close()

			req, err := http.NewRequest(http.MethodDelete, regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/computeClasses/retained-priority", nil)
			require.NoError(t, err)
			resp, err := http.DefaultClient.Do(req)
			require.NoError(t, err)
			defer func() { _ = resp.Body.Close() }()

			require.Equal(t, http.StatusOK, resp.StatusCode)
			require.Equal(t, 1, backendCalls)
		})
	}
}

func TestClusterAgentProxyDeleteComputeClassAllowsAgentDeleteForUnrelatedIntent(t *testing.T) {
	gin.SetMode(gin.TestMode)

	tests := []struct {
		name          string
		configurePeer func(*cloudv1.Cluster)
	}{
		{
			name: "peer outside Kubernetes cluster scope",
			configurePeer: func(peer *cloudv1.Cluster) {
				setProxyTestGKEAPPBoxLabels(peer, "project-2")
				updateProxyTestGKEComputeClassBinding(t, peer, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-16", "n2-standard-64"})
			},
		},
		{
			name: "peer references different compute class",
			configurePeer: func(peer *cloudv1.Cluster) {
				updateProxyTestGKEComputeClassBinding(t, peer, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "other-priority", []string{"n2-standard-16", "n2-standard-64"})
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var backendCalls int
			agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				backendCalls++
				require.Equal(t, "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/computeClasses/tikv-priority", r.URL.Path)
				w.WriteHeader(http.StatusOK)
			}))
			defer agentBackend.Close()

			anchor := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
			anchor.Namespace = "default"
			anchor.Spec.NetworkName = "shared-network"
			anchor.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
			peer := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
			peer.Namespace = "default"
			peer.Name = "cluster-peer"
			peer.Labels[cloudv1.ClusterTagKey] = "peer"
			peer.Spec.NetworkName = "shared-network"
			peer.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
			tt.configurePeer(peer)
			kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(anchor, peer).Build()
			regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
			defer regionalServer.Close()

			req, err := http.NewRequest(http.MethodDelete, regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/computeClasses/tikv-priority", nil)
			require.NoError(t, err)
			resp, err := http.DefaultClient.Do(req)
			require.NoError(t, err)
			defer func() { _ = resp.Body.Close() }()

			require.Equal(t, http.StatusOK, resp.StatusCode)
			require.Equal(t, 1, backendCalls)
		})
	}
}

func TestClusterAgentProxyDeleteComputeClassRejectsMalformedSharedIntent(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	anchor := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	anchor.Namespace = "default"
	anchor.Spec.NetworkName = "shared-network"
	anchor.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	peer := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	peer.Namespace = "default"
	peer.Name = "cluster-peer"
	peer.Labels[cloudv1.ClusterTagKey] = "peer"
	peer.Spec.NetworkName = "shared-network"
	peer.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	peer.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `{"computeClassName":`)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(anchor, peer).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	req, err := http.NewRequest(http.MethodDelete, regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/computeClasses/tikv-priority", nil)
	require.NoError(t, err)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
	got := requireProxyGRPCStatus(t, body)
	require.Equal(t, int32(codes.FailedPrecondition), got.Code)
	require.Contains(t, got.Message, "cannot inspect gke heterogeneous deploy intent for cluster default/cluster-peer")
	require.Equal(t, 0, backendCalls)
}

func TestClusterAgentProxyComputeClassIntentRefsUsesHeterogeneousNodePoolScopeBoundedPeerLookup(t *testing.T) {
	anchor := newGKEHeterogeneousProxyTestCluster("")
	anchor.Namespace = "default"
	anchor.Spec.NetworkName = "shared-network"
	anchor.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"

	peer := newGKEHeterogeneousProxyTestCluster("")
	peer.Namespace = "default"
	peer.Name = "cluster-peer"
	peer.Labels[cloudv1.ClusterTagKey] = "peer"
	peer.Spec.NetworkName = "shared-network"
	peer.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	computeClassName := updateProxyTestGKEComputeClassBinding(t, peer, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-16", "n2-standard-64"})

	differentNetworkMalformed := newGKEHeterogeneousProxyTestCluster("")
	differentNetworkMalformed.Namespace = "default"
	differentNetworkMalformed.Name = "cluster-different-network"
	differentNetworkMalformed.Labels[cloudv1.ClusterTagKey] = "different-network"
	differentNetworkMalformed.Spec.NetworkName = "isolated-network"
	differentNetworkMalformed.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	differentNetworkMalformed.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `not-json`)

	differentAPPBoxMalformed := newGKEHeterogeneousProxyTestCluster("")
	differentAPPBoxMalformed.Namespace = "default"
	differentAPPBoxMalformed.Name = "cluster-different-appbox"
	differentAPPBoxMalformed.Labels[cloudv1.ClusterTagKey] = "different-appbox"
	differentAPPBoxMalformed.Spec.NetworkName = "shared-network"
	differentAPPBoxMalformed.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	differentAPPBoxMalformed.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `not-json`)
	setProxyTestGKEAPPBoxLabels(differentAPPBoxMalformed, "project-2")

	baseClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(anchor, peer, differentNetworkMalformed, differentAPPBoxMalformed).Build()
	recordingClient := &recordingProxyClusterListClient{Client: baseClient}
	proxy := NewClusterAgentProxy(recordingClient)

	peers, err := proxy.listGKEClustersInSameKubernetesClusterScopeForSharedNodePool(context.Background(), anchor)
	require.NoError(t, err)
	refs, err := gkeheterogeneous.ComputeClassIntentRefs(peers, computeClassName)
	require.NoError(t, err)
	expectedSelector := k8slabels.SelectorFromSet(k8slabels.Set{
		cloudv1.TenantTagKey:  "tenant-1",
		cloudv1.ProjectTagKey: "project-1",
	}).String()
	require.Equal(t, []string{expectedSelector}, recordingClient.clusterListSelectors)
	require.Len(t, refs, 1)
	require.Equal(t, "cluster-peer", refs[0].ClusterName)
}

func TestClusterAgentProxyComputeClassIntentRefsRequiresKubernetesClusterScopeLabels(t *testing.T) {
	anchor := newGKEHeterogeneousProxyTestCluster("")
	anchor.Namespace = "default"
	anchor.Spec.NetworkName = ""
	anchor.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	delete(anchor.Labels, cloudv1.TenantTagKey)
	delete(anchor.Labels, cloudv1.ProjectTagKey)
	updateProxyTestGKEComputeClassBinding(t, anchor, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-16"})

	baseClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(anchor).Build()
	recordingClient := &recordingProxyClusterListClient{Client: baseClient}
	proxy := NewClusterAgentProxy(recordingClient)

	peers, err := proxy.listGKEClustersInSameKubernetesClusterScopeForSharedNodePool(context.Background(), anchor)
	require.Error(t, err)
	require.Equal(t, codes.FailedPrecondition, status.Code(err))
	require.Contains(t, err.Error(), "missing required GKE heterogeneous nodepool scope labels")
	require.Empty(t, recordingClient.clusterListSelectors)
	require.Nil(t, peers)
}

func TestClusterAgentProxyDeleteComputeClassRejectsCurrentClusterIntent(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	cluster.Namespace = "default"
	cluster.Spec.NetworkName = "shared-network"
	cluster.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	computeClassName := updateProxyTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-16", "n2-standard-64"})
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	regionalServer := newGKEHeterogeneousProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	req, err := http.NewRequest(http.MethodDelete, regionalServer.URL+"/api/dedicated/v1/clusters/123/gkeHeterogeneous/computeClasses/"+computeClassName, nil)
	require.NoError(t, err)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
	got := requireProxyGRPCErrorInfoReason(t, body, gkeHeterogeneousComputeClassInUseByDeployIntentReason)
	require.Contains(t, got.Message, "cluster=default/cluster-123")
	require.Contains(t, got.Message, "component=TIKV")
	require.Equal(t, 0, backendCalls)
}

func TestClusterAgentProxyEnableGKEPolicyControlPreparesIAMFromClusterThenProxies(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var gotBody []byte
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var err error
		gotBody, err = io.ReadAll(r.Body)
		require.NoError(t, err)
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{projectIdentities: fakeGKEPolicyControlProjectIdentities()}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControl:enable",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusOK, resp.StatusCode)
	require.JSONEq(t, `{"cluster_id":"cluster-123","project_id":"cluster-project","location":"us-west1","gke_cluster_name":"dedicated-dev-us-west1-abc","fleet_project_id":"cluster-project"}`, string(gotBody))
	require.Len(t, resourceManager.ensureCalls, 2)
	require.Equal(t, "cluster-project", resourceManager.ensureCalls[0].project)
	require.Equal(t, "tidbCloudGKEPolicyControl", resourceManager.ensureCalls[0].roleID)
	require.Contains(t, resourceManager.ensureCalls[0].role.IncludedPermissions, "gkehub.memberships.create")
	require.Len(t, resourceManager.addCalls, 2)
	require.Equal(t, "serviceAccount:agent@cluster-project.iam.gserviceaccount.com", resourceManager.addCalls[0].member)
	require.Equal(t, gkePolicyControlClusterAgentWorkloadIdentityMember("cluster-project", "123456789", cluster), resourceManager.addCalls[1].member)
	require.Equal(t, []string{"cluster-project"}, resourceManager.identityCalls)
}

func TestClusterAgentProxyEnableGKEPolicyControlRejectsUnexpectedActionPathBeforeIAM(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{projectIdentities: fakeGKEPolicyControlProjectIdentities()}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControlXYZ",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusNotFound, resp.StatusCode)
	require.Empty(t, resourceManager.identityCalls)
	require.Empty(t, resourceManager.ensureCalls)
	require.Empty(t, resourceManager.addCalls)
	require.Equal(t, 0, backendCalls)
}

func TestClusterAgentProxyEnableGKEPolicyControlPreparesAdminNodeServiceAccount(t *testing.T) {
	gin.SetMode(gin.TestMode)

	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	cluster.Spec.Kubernetes.Cloud.GCP = &cloudv1.CloudGCP{
		AdminNodeSa: "admin@cluster-project.iam.gserviceaccount.com",
	}
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{projectIdentities: fakeGKEPolicyControlProjectIdentities()}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControl:enable",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusOK, resp.StatusCode)
	require.Len(t, resourceManager.addCalls, 3)
	require.Equal(t, "serviceAccount:agent@cluster-project.iam.gserviceaccount.com", resourceManager.addCalls[0].member)
	require.Equal(t, "serviceAccount:admin@cluster-project.iam.gserviceaccount.com", resourceManager.addCalls[1].member)
	require.Equal(t, gkePolicyControlClusterAgentWorkloadIdentityMember("cluster-project", "123456789", cluster), resourceManager.addCalls[2].member)
}

func TestClusterAgentProxyEnableGKEPolicyControlUsesWorkloadIdentityWhenNodeIdentityIsMissing(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	cluster.Status.NodeIdentity = nil
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{projectIdentities: fakeGKEPolicyControlProjectIdentities()}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControl:enable",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusOK, resp.StatusCode)
	require.Equal(t, 1, backendCalls)
	require.Len(t, resourceManager.ensureCalls, 1)
	require.Len(t, resourceManager.addCalls, 1)
	require.Equal(t, gkePolicyControlClusterAgentWorkloadIdentityMember("cluster-project", "123456789", cluster), resourceManager.addCalls[0].member)
}

func TestClusterAgentProxyEnableGKEPolicyControlCanonicalizesClusterProjectNumber(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var gotBody []byte
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var err error
		gotBody, err = io.ReadAll(r.Body)
		require.NoError(t, err)
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	cluster.Spec.Kubernetes.Account = "123456789"
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{
		projectIdentities: map[string]gcp.ProjectIdentity{
			"123456789": {ProjectID: "cluster-project", ProjectNumber: "123456789"},
		},
	}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControl:enable",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusOK, resp.StatusCode)
	require.JSONEq(t, `{"cluster_id":"cluster-123","project_id":"cluster-project","location":"us-west1","gke_cluster_name":"dedicated-dev-us-west1-abc","fleet_project_id":"cluster-project"}`, string(gotBody))
	require.Len(t, resourceManager.ensureCalls, 2)
	require.Equal(t, "cluster-project", resourceManager.ensureCalls[0].project)
	require.Len(t, resourceManager.addCalls, 2)
	require.Equal(t, gkePolicyControlClusterAgentWorkloadIdentityMember("cluster-project", "123456789", cluster), resourceManager.addCalls[1].member)
	require.Equal(t, []string{"123456789"}, resourceManager.identityCalls)
}

func TestClusterAgentProxyEnableGKEPolicyControlCanonicalizesClusterProjectResourceName(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var gotBody []byte
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var err error
		gotBody, err = io.ReadAll(r.Body)
		require.NoError(t, err)
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	cluster.Spec.Kubernetes.Account = "projects/cluster-project"
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{projectIdentities: fakeGKEPolicyControlProjectIdentities()}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControl:enable",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusOK, resp.StatusCode)
	require.JSONEq(t, `{"cluster_id":"cluster-123","project_id":"cluster-project","location":"us-west1","gke_cluster_name":"dedicated-dev-us-west1-abc","fleet_project_id":"cluster-project"}`, string(gotBody))
	require.Len(t, resourceManager.ensureCalls, 2)
	require.Equal(t, "cluster-project", resourceManager.ensureCalls[0].project)
	require.Len(t, resourceManager.addCalls, 2)
	require.Equal(t, gkePolicyControlClusterAgentWorkloadIdentityMember("cluster-project", "123456789", cluster), resourceManager.addCalls[1].member)
	require.Equal(t, []string{"cluster-project"}, resourceManager.identityCalls)
}

func TestClusterAgentProxyGetGKEPolicyControlReturnsGatewayPreconditionWithoutDeployIntent(t *testing.T) {
	gin.SetMode(gin.TestMode)

	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		t.Fatal("backend should not be called")
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	cluster.Status.Gateway = nil
	cluster.Spec.Kubernetes.Account = ""
	cluster.Spec.Kubernetes.Name = ""
	delete(cluster.Labels, cloudv1.RegionLabel)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{identityErr: errors.New("lookup denied")}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkePolicyControl?project_id=ignored&membership_id=ignored")
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode, string(body))
	require.Contains(t, string(body), "gateway endpoint is not ready")
	require.Empty(t, resourceManager.identityCalls)
	require.Empty(t, resourceManager.ensureCalls)
	require.Empty(t, resourceManager.addCalls)
}

func TestClusterAgentProxyGetGKEPolicyControlProxiesWithoutDeployIntentWhenTargetReady(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var gotPath string
	var gotQuery string
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotPath = r.URL.Path
		gotQuery = r.URL.RawQuery
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write([]byte(`{"state":"GKE_POLICY_CONTROL_STATE_ENABLED"}`))
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkePolicyControl?project_id=ignored&membership_id=ignored")
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusOK, resp.StatusCode, string(body))
	require.JSONEq(t, `{"state":"GKE_POLICY_CONTROL_STATE_ENABLED"}`, string(body))
	require.Equal(t, "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkePolicyControl", gotPath)
	query, err := url.ParseQuery(gotQuery)
	require.NoError(t, err)
	require.Equal(t, "cluster-project", query.Get("project_id"))
	require.Equal(t, "us-west1", query.Get("location"))
	require.Equal(t, "dedicated-dev-us-west1-abc", query.Get("gke_cluster_name"))
	require.Equal(t, "cluster-project", query.Get("fleet_project_id"))
	require.Empty(t, query.Get("membership_id"))
}

func TestClusterAgentProxyGetGKEPolicyControlReturnsGatewayPreconditionForOrphanTiDBIntent(t *testing.T) {
	gin.SetMode(gin.TestMode)

	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		t.Fatal("backend should not be called")
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	cluster.Status.Gateway = nil
	updateProxyTestGKEComputeClassBinding(t, cluster, cluster.TiDBGKEComputeClassBindingsAnnoKey("orphan"), "", []string{"n2-standard-8", "n2-standard-64"})
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkePolicyControl")
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode, string(body))
	require.Contains(t, string(body), "gateway endpoint is not ready")
}

func TestClusterAgentProxyGetGKEPolicyControlDerivesTargetFromClusterWithoutPreparingIAM(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var gotQuery string
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotQuery = r.URL.RawQuery
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiKVDeploy(t, cluster, "tikv-priority")...)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{identityErr: errors.New("lookup denied")}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkePolicyControl?project_id=ignored&membership_id=ignored")
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusOK, resp.StatusCode)
	query, err := url.ParseQuery(gotQuery)
	require.NoError(t, err)
	require.Equal(t, "cluster-project", query.Get("project_id"))
	require.Equal(t, "us-west1", query.Get("location"))
	require.Equal(t, "dedicated-dev-us-west1-abc", query.Get("gke_cluster_name"))
	require.Equal(t, "cluster-project", query.Get("fleet_project_id"))
	require.Empty(t, query.Get("membership_id"))
	require.Empty(t, resourceManager.identityCalls)
	require.Empty(t, resourceManager.ensureCalls)
	require.Empty(t, resourceManager.addCalls)
}

func TestClusterAgentProxyMarksGKEHeterogeneousDeployReadyForSingleAndList(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var computeClassName string
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/computeClasses":
			require.Equal(t, computeClassName, r.URL.Query().Get("name"))
			writeProxyTestProtoJSON(t, w, &regionalpb.GetGKEHeterogeneousComputeClassResponse{
				ComputeClasses: []*regionalpb.GKEHeterogeneousComputeClass{{Name: computeClassName}},
			})
		case "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/gatekeeperAssigns/TIKV":
			require.Empty(t, r.URL.RawQuery)
			writeProxyTestProtoJSON(t, w, &regionalpb.GKEHeterogeneousGatekeeperAssign{
				ComponentType:    regionalpb.ComponentType_TIKV,
				ComputeClassName: computeClassName,
			})
		default:
			http.NotFound(w, r)
		}
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiKVDeploy(t, cluster, "tikv-priority")...)
	ref, err := regionalGKEHeterogeneousDeployRefForRequest(cluster, regionalpb.ComponentType_TIKV, "")
	require.NoError(t, err)
	computeClassName = ref.computeClassName
	objects = append(objects, newProxyTestGKEHeterogeneousStatusConfigMap(t, cluster, gkestatus.Snapshot{
		Enabled:           true,
		ValidIntent:       true,
		ClusterFullSynced: true,
		ProbeStatus:       gkestatus.ProbeStatusSuccess,
		Components: []gkestatus.ComponentStatus{
			{
				ComponentType:    string(cloudv1.ComponentTiKV),
				ComputeClassName: computeClassName,
				NodeUsages: []gkestatus.NodeUsage{
					{
						MachineType: "n2-standard-32",
						PodCount:    2,
					},
				},
			},
		},
	}))

	clusterBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/v1/clusters/123/gkeHeterogeneous/deploys":
			writeProxyTestProtoJSON(t, w, &regionalpb.ListGKEHeterogeneousDeploysResponse{
				Deploys: []*regionalpb.GKEHeterogeneousDeploy{
					newProxyTestGKEHeterogeneousReadyForGatekeeperDeploy(regionalpb.ComponentType_TIKV, "", computeClassName),
					{
						Enabled:          true,
						ComponentType:    regionalpb.ComponentType_TIKV,
						ComputeClassName: "other-compute-class",
						Status:           regionalpb.GKEHeterogeneousDeployStatus_GKE_HETEROGENEOUS_DEPLOY_STATUS_INTENT_ENABLED_NOT_READY,
					},
				},
			})
		case "/api/v1/clusters/123/gkeHeterogeneous/deploys/TIKV":
			writeProxyTestProtoJSON(t, w, newProxyTestGKEHeterogeneousReadyForGatekeeperDeploy(regionalpb.ComponentType_TIKV, "", computeClassName))
		default:
			http.NotFound(w, r)
		}
	}))
	defer clusterBackend.Close()

	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousDeployProxyTestServer(t, clusterBackend.URL, NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	singleResp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkeHeterogeneous/deploys/TIKV")
	require.NoError(t, err)
	defer func() { _ = singleResp.Body.Close() }()
	gotSingle := &regionalpb.GKEHeterogeneousDeploy{}
	readProxyTestProtoJSONResponse(t, singleResp, gotSingle)
	require.Equal(t, regionalpb.GKEHeterogeneousDeployStatus_GKE_HETEROGENEOUS_DEPLOY_STATUS_HETEROGENEOUS_READY, gotSingle.GetStatus())
	require.Len(t, gotSingle.GetPodCounts(), 1)
	require.Equal(t, "n2-standard-32", gotSingle.GetPodCounts()[0].GetMachineType())
	require.Equal(t, int32(2), gotSingle.GetPodCounts()[0].GetPodCount())

	listResp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkeHeterogeneous/deploys")
	require.NoError(t, err)
	defer func() { _ = listResp.Body.Close() }()
	gotList := &regionalpb.ListGKEHeterogeneousDeploysResponse{}
	readProxyTestProtoJSONResponse(t, listResp, gotList)
	require.Len(t, gotList.GetDeploys(), 2)
	require.Equal(t, regionalpb.GKEHeterogeneousDeployStatus_GKE_HETEROGENEOUS_DEPLOY_STATUS_HETEROGENEOUS_READY, gotList.GetDeploys()[0].GetStatus())
	require.Len(t, gotList.GetDeploys()[0].GetPodCounts(), 1)
	require.Equal(t, "n2-standard-32", gotList.GetDeploys()[0].GetPodCounts()[0].GetMachineType())
	require.Equal(t, regionalpb.GKEHeterogeneousDeployStatus_GKE_HETEROGENEOUS_DEPLOY_STATUS_INTENT_ENABLED_NOT_READY, gotList.GetDeploys()[1].GetStatus())
	require.Empty(t, gotList.GetDeploys()[1].GetPodCounts())
}

func TestClusterAgentProxyBackfillsGKEHeterogeneousDeployPodCountsFromUnsyncedSnapshot(t *testing.T) {
	gin.SetMode(gin.TestMode)

	cluster := newGKEHeterogeneousProxyTestCluster("")
	computeClassName := "tikv-priority"
	statusConfigMap := newProxyTestGKEHeterogeneousStatusConfigMap(t, cluster, gkestatus.Snapshot{
		Enabled:           true,
		ValidIntent:       true,
		ClusterFullSynced: false,
		ProbeStatus:       gkestatus.ProbeStatusSuccess,
		Components: []gkestatus.ComponentStatus{
			{
				ComponentType:    string(cloudv1.ComponentTiKV),
				ComputeClassName: computeClassName,
				NodeUsages: []gkestatus.NodeUsage{
					{
						MachineType: "n2-standard-32",
						PodCount:    2,
					},
				},
			},
		},
	})
	clusterBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		deploy := &regionalpb.GKEHeterogeneousDeploy{
			Enabled:          true,
			ComponentType:    regionalpb.ComponentType_TIKV,
			ComputeClassName: computeClassName,
			Status:           regionalpb.GKEHeterogeneousDeployStatus_GKE_HETEROGENEOUS_DEPLOY_STATUS_INTENT_ENABLED_NOT_READY,
		}
		switch r.URL.Path {
		case "/api/v1/clusters/123/gkeHeterogeneous/deploys":
			writeProxyTestProtoJSON(t, w, &regionalpb.ListGKEHeterogeneousDeploysResponse{
				Deploys: []*regionalpb.GKEHeterogeneousDeploy{deploy},
			})
		case "/api/v1/clusters/123/gkeHeterogeneous/deploys/TIKV":
			writeProxyTestProtoJSON(t, w, deploy)
		default:
			http.NotFound(w, r)
		}
	}))
	defer clusterBackend.Close()

	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster, statusConfigMap).Build()
	regionalServer := newGKEHeterogeneousDeployProxyTestServer(t, clusterBackend.URL, NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	singleResp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkeHeterogeneous/deploys/TIKV")
	require.NoError(t, err)
	defer func() { _ = singleResp.Body.Close() }()
	gotSingle := &regionalpb.GKEHeterogeneousDeploy{}
	readProxyTestProtoJSONResponse(t, singleResp, gotSingle)
	require.Equal(t, regionalpb.GKEHeterogeneousDeployStatus_GKE_HETEROGENEOUS_DEPLOY_STATUS_INTENT_ENABLED_NOT_READY, gotSingle.GetStatus())
	require.Len(t, gotSingle.GetPodCounts(), 1)
	require.Equal(t, "n2-standard-32", gotSingle.GetPodCounts()[0].GetMachineType())
	require.Equal(t, int32(2), gotSingle.GetPodCounts()[0].GetPodCount())

	listResp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkeHeterogeneous/deploys")
	require.NoError(t, err)
	defer func() { _ = listResp.Body.Close() }()
	gotList := &regionalpb.ListGKEHeterogeneousDeploysResponse{}
	readProxyTestProtoJSONResponse(t, listResp, gotList)
	require.Len(t, gotList.GetDeploys(), 1)
	require.Len(t, gotList.GetDeploys()[0].GetPodCounts(), 1)
	require.Equal(t, "n2-standard-32", gotList.GetDeploys()[0].GetPodCounts()[0].GetMachineType())
	require.Equal(t, int32(2), gotList.GetDeploys()[0].GetPodCounts()[0].GetPodCount())
}

func TestClusterAgentProxyKeepsGKEHeterogeneousDeployReadyForGatekeeperWhenClusterAgentResourceMissing(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var computeClassName string
	var gatekeeperChecked bool
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/computeClasses":
			require.Equal(t, computeClassName, r.URL.Query().Get("name"))
			writeProxyTestProtoJSON(t, w, &regionalpb.GetGKEHeterogeneousComputeClassResponse{
				ComputeClasses: []*regionalpb.GKEHeterogeneousComputeClass{{Name: computeClassName}},
			})
		case "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/gatekeeperAssigns/TIKV":
			gatekeeperChecked = true
			http.NotFound(w, r)
		default:
			http.NotFound(w, r)
		}
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiKVDeploy(t, cluster, "tikv-priority")...)
	ref, err := regionalGKEHeterogeneousDeployRefForRequest(cluster, regionalpb.ComponentType_TIKV, "")
	require.NoError(t, err)
	computeClassName = ref.computeClassName

	rawBody := []byte(`{"enabled":true,"componentType":"TIKV","computeClassName":"` + computeClassName + `","status":"GKE_HETEROGENEOUS_DEPLOY_STATUS_READY_FOR_GATEKEEPER","futureClusterServerField":"kept"}`)
	clusterBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, "/api/v1/clusters/123/gkeHeterogeneous/deploys/TIKV", r.URL.Path)
		w.Header().Set("Content-Type", "application/json")
		_, err := w.Write(rawBody)
		require.NoError(t, err)
	}))
	defer clusterBackend.Close()

	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousDeployProxyTestServer(t, clusterBackend.URL, NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkeHeterogeneous/deploys/TIKV")
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	require.Equal(t, http.StatusOK, resp.StatusCode)
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	require.Equal(t, rawBody, body)
	got := &regionalpb.GKEHeterogeneousDeploy{}
	require.NoError(t, (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(body, got), string(body))

	require.True(t, gatekeeperChecked)
	require.Equal(t, regionalpb.GKEHeterogeneousDeployStatus_GKE_HETEROGENEOUS_DEPLOY_STATUS_READY_FOR_GATEKEEPER, got.GetStatus())
}

func TestClusterAgentProxyClosesOriginalGKEHeterogeneousDeployResponseBody(t *testing.T) {
	rawBody := []byte(`{}`)
	originalBody := &trackingReadCloser{Reader: bytes.NewReader(rawBody)}
	resp := &http.Response{
		StatusCode: http.StatusOK,
		Header:     make(http.Header),
		Body:       originalBody,
	}

	proxy := NewClusterAgentProxy(nil)
	require.NoError(t, proxy.markGKEHeterogeneousDeployReady(context.Background(), "123", resp, false))
	defer func() { _ = resp.Body.Close() }()

	require.True(t, originalBody.closed)
	restoredBody, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	require.Equal(t, rawBody, restoredBody)
}

func TestClusterAgentProxyForwardsTiDBGroupWhenCheckingGKEHeterogeneousDeployReady(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var computeClassName string
	var gotTiDBGroupID string
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/computeClasses":
			require.Equal(t, computeClassName, r.URL.Query().Get("name"))
			writeProxyTestProtoJSON(t, w, &regionalpb.GetGKEHeterogeneousComputeClassResponse{
				ComputeClasses: []*regionalpb.GKEHeterogeneousComputeClass{{Name: computeClassName}},
			})
		case "/cluster-agent/tidbcluster-123/api/v1/clusters/cluster-123/gkeHeterogeneous/gatekeeperAssigns/TIDB":
			gotTiDBGroupID = r.URL.Query().Get("tidbGroupId")
			writeProxyTestProtoJSON(t, w, &regionalpb.GKEHeterogeneousGatekeeperAssign{
				ComponentType:    regionalpb.ComponentType_TIDB,
				TidbGroupId:      "extra-group",
				ComputeClassName: computeClassName,
			})
		default:
			http.NotFound(w, r)
		}
	}))
	defer agentBackend.Close()

	cluster := newGKEHeterogeneousProxyTestCluster(agentBackend.URL)
	objects := append([]client.Object{cluster}, addReadyProxyTestTiDBDeploy(t, cluster, "tidb-priority", "extra-group")...)
	ref, err := regionalGKEHeterogeneousDeployRefForRequest(cluster, regionalpb.ComponentType_TIDB, "extra-group")
	require.NoError(t, err)
	computeClassName = ref.computeClassName
	objects = append(objects, newProxyTestGKEHeterogeneousStatusConfigMap(t, cluster, gkestatus.Snapshot{
		Enabled:           true,
		ValidIntent:       true,
		ClusterFullSynced: true,
		ProbeStatus:       gkestatus.ProbeStatusSuccess,
		Components: []gkestatus.ComponentStatus{
			{
				ComponentType:    string(cloudv1.ComponentTiDB),
				TiDBGroupID:      "default-group",
				ComputeClassName: "default-compute-class",
				NodeUsages: []gkestatus.NodeUsage{
					{MachineType: "n2-standard-8", PodCount: 9},
				},
			},
			{
				ComponentType:    string(cloudv1.ComponentTiDB),
				TiDBGroupID:      "extra-group",
				ComputeClassName: computeClassName,
				NodeUsages: []gkestatus.NodeUsage{
					{
						MachineType: "n2-standard-64",
						PodCount:    1,
					},
				},
			},
		},
	}))

	clusterBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, "/api/v1/clusters/123/gkeHeterogeneous/deploys/TIDB", r.URL.Path)
		writeProxyTestProtoJSON(t, w, newProxyTestGKEHeterogeneousReadyForGatekeeperDeploy(regionalpb.ComponentType_TIDB, "extra-group", computeClassName))
	}))
	defer clusterBackend.Close()

	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(objects...).Build()
	regionalServer := newGKEHeterogeneousDeployProxyTestServer(t, clusterBackend.URL, NewClusterAgentProxy(kubeClient))
	defer regionalServer.Close()

	resp, err := http.Get(regionalServer.URL + "/api/dedicated/v1/clusters/123/gkeHeterogeneous/deploys/TIDB?tidb_group_id=extra-group")
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	got := &regionalpb.GKEHeterogeneousDeploy{}
	readProxyTestProtoJSONResponse(t, resp, got)

	require.Equal(t, "extra-group", gotTiDBGroupID)
	require.Equal(t, regionalpb.GKEHeterogeneousDeployStatus_GKE_HETEROGENEOUS_DEPLOY_STATUS_HETEROGENEOUS_READY, got.GetStatus())
	require.Len(t, got.GetPodCounts(), 1)
	require.Equal(t, "n2-standard-64", got.GetPodCounts()[0].GetMachineType())
	require.Equal(t, int32(1), got.GetPodCounts()[0].GetPodCount())
}

func TestClusterAgentProxyEnableGKEPolicyControlReturnsPreconditionFailedWhenIAMPreparationFails(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{ensureErr: errors.New("permission denied")}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControl:enable",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
	require.Contains(t, string(body), gkePolicyControlIAMPreconditionMessage)
	require.Equal(t, 0, backendCalls)
	require.Len(t, resourceManager.ensureCalls, 1)
	require.Empty(t, resourceManager.addCalls)
}

func TestClusterAgentProxyEnableGKEPolicyControlRejectsMissingClusterProject(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	cluster.Spec.Kubernetes.Account = ""
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{projectIdentities: fakeGKEPolicyControlProjectIdentities()}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControl:enable",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
	require.Equal(t, 0, backendCalls)
	require.Empty(t, resourceManager.ensureCalls)
	require.Empty(t, resourceManager.addCalls)
	require.Empty(t, resourceManager.identityCalls)
}

func TestClusterAgentProxyEnableGKEPolicyControlRejectsMissingClusterLocation(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	delete(cluster.Labels, cloudv1.RegionLabel)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{projectIdentities: fakeGKEPolicyControlProjectIdentities()}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControl:enable",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
	require.Equal(t, 0, backendCalls)
	require.Empty(t, resourceManager.ensureCalls)
	require.Empty(t, resourceManager.addCalls)
	require.Empty(t, resourceManager.identityCalls)
}

func TestClusterAgentProxyEnableGKEPolicyControlRejectsMissingGKEClusterName(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	cluster.Spec.Kubernetes.Name = ""
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{projectIdentities: fakeGKEPolicyControlProjectIdentities()}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControl:enable",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
	require.Equal(t, 0, backendCalls)
	require.Empty(t, resourceManager.ensureCalls)
	require.Empty(t, resourceManager.addCalls)
	require.Empty(t, resourceManager.identityCalls)
}

func TestClusterAgentProxyEnableGKEPolicyControlRejectsUnverifiedProjectIdentity(t *testing.T) {
	gin.SetMode(gin.TestMode)

	var backendCalls int
	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		backendCalls++
		w.WriteHeader(http.StatusOK)
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{identityErr: errors.New("lookup denied")}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControl:enable",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
	require.Contains(t, string(body), "cannot be verified")
	require.Contains(t, string(body), "lookup denied")
	require.Equal(t, 0, backendCalls)
	require.Empty(t, resourceManager.ensureCalls)
	require.Empty(t, resourceManager.addCalls)
	require.Equal(t, []string{"cluster-project"}, resourceManager.identityCalls)
}

func TestClusterAgentProxyEnableGKEPolicyControlRejectsNonGKECluster(t *testing.T) {
	gin.SetMode(gin.TestMode)

	agentBackend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		t.Fatal("backend should not be called")
	}))
	defer agentBackend.Close()

	cluster := newGKEPolicyControlProxyTestCluster(agentBackend.URL)
	cluster.Labels[cloudv1.ProviderTypeLabel] = cloudv1.ProviderTypeEKS
	kubeClient := fake.NewClientBuilder().WithScheme(newGKEHeterogeneousProxyTestScheme(t)).WithObjects(cluster).Build()
	resourceManager := &fakeGKEPolicyControlResourceManager{}
	regionalServer := newGKEPolicyControlProxyTestServer(NewClusterAgentProxy(kubeClient).WithGKEPolicyControlResourceManager(resourceManager))
	defer regionalServer.Close()

	resp, err := http.Post(
		regionalServer.URL+"/api/dedicated/v1/clusters/123/gkePolicyControl:enable",
		"application/json",
		nil,
	)
	require.NoError(t, err)
	defer func() { _ = resp.Body.Close() }()

	require.Equal(t, http.StatusPreconditionFailed, resp.StatusCode)
	require.Empty(t, resourceManager.ensureCalls)
	require.Empty(t, resourceManager.addCalls)
}

type proxyGRPCStatusResponse struct {
	Code    int32                      `json:"code"`
	Message string                     `json:"message"`
	Details []proxyGRPCErrorInfoDetail `json:"details"`
}

type proxyGRPCErrorInfoDetail struct {
	Type   string `json:"@type"`
	Reason string `json:"reason"`
	Domain string `json:"domain"`
}

func requireProxyGRPCStatus(t *testing.T, body []byte) proxyGRPCStatusResponse {
	t.Helper()
	var got proxyGRPCStatusResponse
	require.NoError(t, json.Unmarshal(body, &got), string(body))
	require.NotZero(t, got.Code, string(body))
	require.NotEmpty(t, got.Message, string(body))
	return got
}

func requireProxyGRPCErrorInfoReason(t *testing.T, body []byte, reason string) proxyGRPCStatusResponse {
	t.Helper()
	got := requireProxyGRPCStatus(t, body)
	require.Equal(t, int32(codes.FailedPrecondition), got.Code)
	require.NotEmpty(t, got.Details, string(body))
	require.Equal(t, "type.googleapis.com/google.rpc.ErrorInfo", got.Details[0].Type)
	require.Equal(t, reason, got.Details[0].Reason)
	require.Equal(t, "infra-provider", got.Details[0].Domain)
	return got
}

func newGKEHeterogeneousProxyTestScheme(t *testing.T) *runtime.Scheme {
	t.Helper()
	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	require.NoError(t, cloudv1.AddToScheme(scheme))
	require.NoError(t, gkev1beta1.SchemeBuilder.AddToScheme(scheme))
	return scheme
}

func newGKEHeterogeneousProxyTestCluster(endpoint string) *cloudv1.Cluster {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster-123",
			Labels: map[string]string{
				cloudv1.ClusterTagKey:     "123",
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default-group",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Kubernetes: cloudv1.Kubernetes{
				Name:    "gke-main",
				Account: "gcp-project",
			},
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{InstanceType: "n2-standard-8"},
				TiDB: cloudv1.TiDB{InstanceType: "n2-standard-8"},
			},
			ExtraGroups: []cloudv1.ExtraGroupSpec{
				{
					Name: "extra-group",
					TiDB: cloudv1.TiDB{InstanceType: "n2-standard-8"},
				},
			},
		},
		Status: cloudv1.ClusterStatus{
			Gateway: &cloudv1.GatewayStatus{Endpoint: endpoint},
		},
	}
	setProxyTestGKEAPPBoxLabels(cluster, "project-1")
	return cluster
}

func addReadyProxyTestTiDBDeploy(t *testing.T, cluster *cloudv1.Cluster, computeClassName, groupID string) []client.Object {
	t.Helper()
	instanceTypes := []string{"n2-standard-8", "n2-standard-64"}
	nodePoolNames := []string{
		gkeheterogeneous.NodePoolName(cluster, instanceTypes[0]),
		gkeheterogeneous.NodePoolName(cluster, instanceTypes[1]),
	}
	computeClassName = updateProxyTestGKEComputeClassBinding(t, cluster, cluster.TiDBGKEComputeClassBindingsAnnoKey(groupID), computeClassName, instanceTypes)
	return []client.Object{
		newReadyProxyTestNodePool(cluster, nodePoolNames[0], instanceTypes[0], computeClassName, true),
		newReadyProxyTestNodePool(cluster, nodePoolNames[1], instanceTypes[1], computeClassName, true),
	}
}

func addReadyProxyTestTiKVDeploy(t *testing.T, cluster *cloudv1.Cluster, computeClassName string) []client.Object {
	t.Helper()
	cluster.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	instanceTypes := []string{"n2-standard-16", "n2-standard-32"}
	nodePoolNames := []string{
		gkeheterogeneous.NodePoolName(cluster, instanceTypes[0]),
		gkeheterogeneous.NodePoolName(cluster, instanceTypes[1]),
	}
	computeClassName = updateProxyTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, computeClassName, instanceTypes)
	return []client.Object{
		newReadyProxyTestNodePool(cluster, nodePoolNames[0], instanceTypes[0], computeClassName, true),
		newReadyProxyTestNodePool(cluster, nodePoolNames[1], instanceTypes[1], computeClassName, true),
	}
}

func newReadyProxyTestNodePool(cluster *cloudv1.Cluster, name, instanceType, computeClassName string, apiManaged bool) *gkev1beta1.NodePool {
	labels := map[string]string{
		namer.ComponentSelectorKey: "shared",
		namer.ClusterSelectorKey:   "shared",
	}
	scopeLabels, _ := nodepoolutil.GKEHeterogeneousNodePoolScopeLabels(cluster)
	for key, value := range scopeLabels {
		labels[key] = value
	}
	if apiManaged {
		labels[consts.GKEHeterogeneousNodePoolManagedByLabelKey] = consts.GKEHeterogeneousNodePoolManagedByLabelValue
	}
	nodePool := &gkev1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: labels,
		},
		Spec: gkev1beta1.NodePoolSpec{
			ForProvider: gkev1beta1.NodePoolParameters{
				Cluster: regionalGKEHeterogeneousNodePoolClusterRef(cluster),
				Config: &gkev1beta1.NodeConfig{
					MachineType: ptr.To(instanceType),
				},
			},
		},
		Status: gkev1beta1.NodePoolStatus{
			ResourceStatus: xpv1.ResourceStatus{
				ConditionedStatus: xpv1.ConditionedStatus{
					Conditions: []xpv1.Condition{
						{Type: xpv1.TypeReady, Status: corev1.ConditionTrue},
						{Type: xpv1.TypeSynced, Status: corev1.ConditionTrue},
					},
				},
			},
		},
	}
	nodepoolutil.SetGKEComputeClassForNodePool(nodePool, computeClassName)
	nodepoolutil.SetGKEAppliedComputeClassForNodePool(nodePool, computeClassName)
	return nodePool
}

func setProxyTestGKEAPPBoxLabels(cluster *cloudv1.Cluster, projectID string) {
	if cluster.Labels == nil {
		cluster.Labels = map[string]string{}
	}
	cluster.Labels[cloudv1.RegionLabel] = "us-central1"
	cluster.Labels[cloudv1.TenantTagKey] = "tenant-1"
	cluster.Labels[cloudv1.ProjectTagKey] = projectID
}

func newGKEHeterogeneousProxyTestServer(proxy *ClusterAgentProxy) *httptest.Server {
	eng := gin.Default()
	eng.POST("/api/dedicated/v1/clusters/:clusterId/gkeHeterogeneous/gatekeeperAssigns", proxy.Proxy)
	eng.GET("/api/dedicated/v1/clusters/:clusterId/gkeHeterogeneous/gatekeeperAssigns/:componentType", proxy.Proxy)
	eng.PATCH("/api/dedicated/v1/clusters/:clusterId/gkeHeterogeneous/gatekeeperAssigns/:name", proxy.Proxy)
	eng.DELETE("/api/dedicated/v1/clusters/:clusterId/gkeHeterogeneous/computeClasses/:name", proxy.Proxy)
	return httptest.NewServer(eng)
}

type trackingReadCloser struct {
	*bytes.Reader
	closed bool
}

func (r *trackingReadCloser) Close() error {
	r.closed = true
	return nil
}

func newGKEHeterogeneousDeployProxyTestServer(t *testing.T, clusterBackendURL string, proxy *ClusterAgentProxy) *httptest.Server {
	t.Helper()
	rp, err := NewReverseProxy(clusterBackendURL)
	require.NoError(t, err)
	rp = rp.WithPathPerfixRewrite("/api/dedicated/v1", "/api/v1")
	eng := gin.Default()
	eng.GET("/api/dedicated/v1/clusters/:clusterId/gkeHeterogeneous/deploys", proxy.ProxyGKEHeterogeneousDeploy(rp, true))
	eng.GET("/api/dedicated/v1/clusters/:clusterId/gkeHeterogeneous/deploys/:componentType", proxy.ProxyGKEHeterogeneousDeploy(rp, false))
	return httptest.NewServer(eng)
}

func newProxyTestGKEHeterogeneousStatusConfigMap(t *testing.T, cluster *cloudv1.Cluster, snapshot gkestatus.Snapshot) *corev1.ConfigMap {
	t.Helper()
	data, err := yaml.Marshal(snapshot)
	require.NoError(t, err)
	return &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: cluster.Namespace,
			Name:      gkestatus.ConfigMapNamePrefix + cluster.Name,
		},
		Data: map[string]string{
			gkestatus.ConfigMapDataKey: string(data),
		},
	}
}

func newProxyTestGKEHeterogeneousReadyForGatekeeperDeploy(componentType regionalpb.ComponentType, tidbGroupID, computeClassName string) *regionalpb.GKEHeterogeneousDeploy {
	return &regionalpb.GKEHeterogeneousDeploy{
		Enabled:          true,
		ComponentType:    componentType,
		TidbGroupId:      tidbGroupID,
		ComputeClassName: computeClassName,
		Status:           regionalpb.GKEHeterogeneousDeployStatus_GKE_HETEROGENEOUS_DEPLOY_STATUS_READY_FOR_GATEKEEPER,
	}
}

func writeProxyTestProtoJSON(t *testing.T, w http.ResponseWriter, msg proto.Message) {
	t.Helper()
	body, err := (protojson.MarshalOptions{UseProtoNames: false}).Marshal(msg)
	require.NoError(t, err)
	w.Header().Set("Content-Type", "application/json")
	_, err = w.Write(body)
	require.NoError(t, err)
}

func readProxyTestProtoJSONResponse(t *testing.T, resp *http.Response, msg proto.Message) {
	t.Helper()
	require.Equal(t, http.StatusOK, resp.StatusCode)
	body, err := io.ReadAll(resp.Body)
	require.NoError(t, err)
	require.NoError(t, (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(body, msg), string(body))
}

func updateProxyTestGKEComputeClassBinding(t *testing.T, cluster *cloudv1.Cluster, key, _ string, instanceTypes []string) string {
	t.Helper()
	require.NotEmpty(t, instanceTypes)
	if len(instanceTypes) == 1 {
		instanceTypes = append(append([]string{}, instanceTypes...), "n2-standard-64")
	}
	annotation := cloudv1.GKEComputeClassBindingAnnotation{
		NodePools: make([]cloudv1.GKEComputeClassNodePoolBinding, 0, len(instanceTypes)),
	}
	for i, instanceType := range instanceTypes {
		role := cloudv1.GKEComputeClassNodePoolRoleExtra
		if i == 0 {
			role = cloudv1.GKEComputeClassNodePoolRoleBase
		}
		annotation.NodePools = append(annotation.NodePools, cloudv1.GKEComputeClassNodePoolBinding{
			Role:         role,
			InstanceType: instanceType,
			NodePoolName: gkeheterogeneous.NodePoolName(cluster, instanceType),
		})
	}
	computeClassName, err := cloudv1.GKEComputeClassNameForBindings(annotation.NodePools)
	require.NoError(t, err)
	annotation.ComputeClassName = computeClassName
	value, err := cloudv1.EncodeGKEComputeClassBindingAnnotation(annotation, key, instanceTypes[0], gkeheterogeneous.NodePoolName(cluster, instanceTypes[0]))
	require.NoError(t, err)
	cluster.UpdateAnnotation(key, value)
	return computeClassName
}

type gkePolicyControlEnsureCall struct {
	project string
	roleID  string
	role    gcp.ProjectCustomRole
}

type gkePolicyControlBindingCall struct {
	project string
	role    string
	member  string
}

type fakeGKEPolicyControlResourceManager struct {
	ensureErr   error
	addErr      error
	identityErr error

	projectIdentities map[string]gcp.ProjectIdentity

	ensureCalls   []gkePolicyControlEnsureCall
	addCalls      []gkePolicyControlBindingCall
	identityCalls []string
}

func fakeGKEPolicyControlProjectIdentities() map[string]gcp.ProjectIdentity {
	return map[string]gcp.ProjectIdentity{
		"cluster-project": {ProjectID: "cluster-project", ProjectNumber: "123456789"},
	}
}

func (f *fakeGKEPolicyControlResourceManager) GetProjectIdentity(_ context.Context, project string) (gcp.ProjectIdentity, error) {
	f.identityCalls = append(f.identityCalls, project)
	if f.identityErr != nil {
		return gcp.ProjectIdentity{}, f.identityErr
	}
	if identity, ok := f.projectIdentities[project]; ok {
		if identity.Input == "" {
			identity.Input = project
		}
		return identity, nil
	}
	return gcp.ProjectIdentity{
		Input:         project,
		ProjectID:     project,
		ProjectNumber: project,
	}, nil
}

func (f *fakeGKEPolicyControlResourceManager) EnsureProjectCustomRole(_ context.Context, project, roleID string, role gcp.ProjectCustomRole) (string, error) {
	f.ensureCalls = append(f.ensureCalls, gkePolicyControlEnsureCall{project: project, roleID: roleID, role: role})
	if f.ensureErr != nil {
		return "", f.ensureErr
	}
	return "projects/" + project + "/roles/" + roleID, nil
}

func (f *fakeGKEPolicyControlResourceManager) AddProjectPolicyBinding(_ context.Context, project, role, member string) error {
	f.addCalls = append(f.addCalls, gkePolicyControlBindingCall{project: project, role: role, member: member})
	return f.addErr
}

func (f *fakeGKEPolicyControlResourceManager) DelProjectPolicyBinding(_ context.Context, project, role, member string) error {
	return nil
}

func newGKEPolicyControlProxyTestCluster(endpoint string) *cloudv1.Cluster {
	cluster := newGKEHeterogeneousProxyTestCluster(endpoint)
	cluster.Namespace = "default"
	cluster.Labels[cloudv1.RegionLabel] = "us-west1"
	cluster.Spec.Kubernetes.Account = "cluster-project"
	cluster.Spec.Kubernetes.Name = "dedicated-dev-us-west1-abc"
	cluster.Spec.Kubernetes.Cloud.GCP = &cloudv1.CloudGCP{}
	cluster.Status.NodeIdentity = &cloudv1.NodeIdentity{
		GCP: &cloudv1.GCPIdentity{ServiceAccount: "agent@cluster-project.iam.gserviceaccount.com"},
	}
	return cluster
}

func newGKEPolicyControlProxyTestServer(proxy *ClusterAgentProxy) *httptest.Server {
	eng := gin.Default()
	eng.POST("/api/dedicated/v1/clusters/:clusterId/gkePolicyControl:enable", proxy.EnableGKEPolicyControl)
	eng.GET("/api/dedicated/v1/clusters/:clusterId/gkePolicyControl", proxy.GetGKEPolicyControl)
	return httptest.NewServer(eng)
}