package privatelinkservice

import (
	"context"
	"fmt"
	"testing"

	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/network/armnetwork/v6"
	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	route53v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/route53/v1alpha1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	regionalapiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	mockaws "github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/test/mockazure"
	"github.com/tidbcloud/infra-provider/regional-server/test/resourcemanager/mockapproute"
	"github.com/tidbcloud/infra-provider/regional-server/test/resourcemanager/mockprivatelinkservice"
	"github.com/tidbcloud/infra-provider/regional-server/test/resourcemanager/mockvpc"
	"go.uber.org/mock/gomock"
	"google.golang.org/protobuf/encoding/protojson"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
)

func TestCreatePrivateLinkService(t *testing.T) {
	ctl := gomock.NewController(t)
	approuter := mockapproute.NewMockAPPRouteReader(ctl)
	plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
	plssvc := NewPrivateLinkServiceService(nil, plsrw, approuter, nil, config.RootConfig{
		ControllerConfig: config.ControllerConfig{
			Provider: consts.ProviderTypeEKS,
			RegionConfig: config.RegionConfig{
				Name: "us-west-1",
			},
		},
	})
	t.Run("list approute failed", func(t *testing.T) {
		approuter.EXPECT().List(context.Background(), &v1alpha1.APPRouteList{}, client.MatchingLabels{
			v1alpha1.ClusterIDKey: "test-cluster",
		}).Return(assert.AnError)
		_, err := plssvc.CreatePrivateLinkService(context.Background(), &regionalapiv1.CreatePrivateLinkServiceRequest{
			PrivateLinkService: &regionalapiv1.PrivateLinkService{
				ClusterId:   "test-cluster",
				TidbGroupId: "test-lb",
			},
		})
		assert.ErrorContains(t, err, "failed to list cluster")
	})
	t.Run("list approute empty", func(t *testing.T) {
		approuter.EXPECT().List(context.Background(), &v1alpha1.APPRouteList{}, client.MatchingLabels{
			v1alpha1.ClusterIDKey: "test-cluster",
		}).Return(nil)
		_, err := plssvc.CreatePrivateLinkService(context.Background(), &regionalapiv1.CreatePrivateLinkServiceRequest{
			PrivateLinkService: &regionalapiv1.PrivateLinkService{
				ClusterId:   "test-cluster",
				TidbGroupId: "test-lb",
			},
		})
		assert.ErrorContains(t, err, "failed to find cluster")
	})
	t.Run("create pls failed", func(t *testing.T) {
		approuter.EXPECT().List(context.Background(), &v1alpha1.APPRouteList{}, client.MatchingLabels{
			v1alpha1.ClusterIDKey: "test-cluster",
		}).DoAndReturn(func(ctx context.Context, list *v1alpha1.APPRouteList, opts ...client.ListOption) error {
			list.Items = []v1alpha1.APPRoute{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name: "test-app-route",
						Labels: map[string]string{
							v1alpha1.ClusterIDKey: "test-cluster",
						},
					},
				},
			}
			return nil
		})
		plsrw.EXPECT().Create(context.Background(), gomock.Any()).Return(assert.AnError)
		_, err := plssvc.CreatePrivateLinkService(context.Background(), &regionalapiv1.CreatePrivateLinkServiceRequest{
			PrivateLinkService: &regionalapiv1.PrivateLinkService{
				ClusterId:   "test-cluster",
				TidbGroupId: "test-lb",
			},
		})
		assert.ErrorContains(t, err, "failed to create privatelinkservice")
	})
	t.Run("create pls success", func(t *testing.T) {
		req := &regionalapiv1.CreatePrivateLinkServiceRequest{
			PrivateLinkService: &regionalapiv1.PrivateLinkService{
				Id:          "test-pls",
				ClusterId:   "test-cluster",
				TidbGroupId: "test-lb",
				Connections: []*regionalapiv1.PrivateLinkService_Connection{
					{
						Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
						Message: "pending",
						Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
							Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
								VpcEndpointId: "https://test.com",
							},
						},
					},
				},
			},
		}
		approute := &v1alpha1.APPRoute{
			ObjectMeta: metav1.ObjectMeta{
				Name: "test-app-route",
				Labels: map[string]string{
					v1alpha1.ClusterIDKey: "test-cluster",
					v1alpha1.VPCIDKey:     "test-vpc",
					v1alpha1.VPCNameKey:   "test-vpc-name",
				},
				Namespace: "test-namespace",
			},
		}
		approuter.EXPECT().List(context.Background(), &v1alpha1.APPRouteList{}, client.MatchingLabels{
			v1alpha1.ClusterIDKey: "test-cluster",
		}).DoAndReturn(func(ctx context.Context, list *v1alpha1.APPRouteList, opts ...client.ListOption) error {
			list.Items = []v1alpha1.APPRoute{*approute}
			return nil
		})
		plsrw.EXPECT().Create(context.Background(), &v1alpha1.PrivateLinkService{
			ObjectMeta: metav1.ObjectMeta{
				Name:      req.PrivateLinkService.Id,
				Namespace: approute.Namespace,
				Labels: map[string]string{
					v1alpha1.VPCIDKey:          approute.GetLabels()[v1alpha1.VPCIDKey],
					v1alpha1.VPCNameKey:        approute.GetLabels()[v1alpha1.VPCNameKey],
					v1alpha1.ClusterIDKey:      req.PrivateLinkService.ClusterId,
					v1alpha1.LoadBalancerIDKey: req.PrivateLinkService.TidbGroupId,
					v1alpha1.TagTenantKey:      approute.GetLabels()[v1alpha1.TagTenantKey],
					v1alpha1.TagProjectKey:     approute.GetLabels()[v1alpha1.TagProjectKey],
					v1alpha1.TagClusterKey:     approute.GetLabels()[v1alpha1.TagClusterKey],
				},
			},
			Spec: v1alpha1.PrivateLinkServiceSpec{
				Provider: consts.ProviderTypeEKS,
				Region:   "us-west-1",
				VPCRef: v1alpha1.Reference{
					Name:      approute.GetLabels()[v1alpha1.VPCNameKey],
					Namespace: approute.Namespace,
				},
				LoadBalancers: []v1alpha1.LoadBalancer{
					{ID: req.PrivateLinkService.TidbGroupId, ClusterID: req.PrivateLinkService.ClusterId},
				},
				Connections: []v1alpha1.Connection{
					{Endpoint: "https://test.com"},
				},
			},
		}).Return(nil)
		_, err := plssvc.CreatePrivateLinkService(context.Background(), req)
		assert.NoError(t, err)
	})
}

func TestGetPrivateLinkService(t *testing.T) {
	ctl := gomock.NewController(t)
	plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
	plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{})
	namer := utils.NewNamerWithClusterID("", plssvc.config.RegionConfig.Name, plssvc.config.Env)
	t.Run("get privatelinkservice failed", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).Return(assert.AnError)
		_, err := plssvc.GetPrivateLinkService(context.Background(), &regionalapiv1.GetPrivateLinkServiceRequest{
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "failed to get privatelinkservice")
	})

	t.Run("fill privatelinkservice failed", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).Return(nil)
		_, err := plssvc.GetPrivateLinkService(context.Background(), &regionalapiv1.GetPrivateLinkServiceRequest{
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "failed to fill privatelinkservice")
	})

	t.Run("get privatelinkservice success", func(t *testing.T) {
		plssvc.config = config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{
					Name: "us-west-1",
				},
				Env: "test-env",
			},
		}
		namer = utils.NewNamerWithClusterID("", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).DoAndReturn(func(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
			*pls = v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "test-pls",
					Namespace: namer.GetNamespace(),
				},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					Provider: consts.ProviderTypeEKS,
					Region:   "us-west-1",
				},
			}
			return nil
		})
		_, err := plssvc.GetPrivateLinkService(context.Background(), &regionalapiv1.GetPrivateLinkServiceRequest{
			PrivateLinkServiceId: "test-pls",
		})
		assert.NoError(t, err)
	})
	t.Run("get azure privatelinkservice success", func(t *testing.T) {
		ctl := gomock.NewController(t)
		vpcr := mockvpc.NewMockVPCReader(ctl)
		plssvc.vpcr = vpcr
		plssvc.config = config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeAKS,
				RegionConfig: config.RegionConfig{
					Name: "eastus",
				},
				Env: "test-env",
			},
		}
		plscli := mockazure.NewMockPrivateLinkServiceClient(ctl)
		plssvc.plsClients.Store("test-subscription", plscli)
		namer = utils.NewNamerWithClusterID("", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).DoAndReturn(func(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
			*pls = v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "test-pls",
					Namespace: namer.GetNamespace(),
				},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					Provider: consts.ProviderTypeAKS,
					Region:   "us-west-1",
					VPCRef: v1alpha1.Reference{
						Name:      "test-vpc",
						Namespace: "test-namespace",
					},
				},
				Status: v1alpha1.PrivateLinkServiceStatus{
					AzurePrivateLinkService: v1alpha1.AzurePrivateLinkServiceStatus{
						Name: "test-pls-azure",
					},
					Connections: []v1alpha1.ConnectionStatus{
						{
							Phase:    v1alpha1.ConnectionPhasePending,
							Message:  "pending",
							Endpoint: "/endppint1",
						},
					},
				},
			}
			return nil
		})
		vpcr.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-vpc", Namespace: "test-namespace"}, gomock.Any()).DoAndReturn(func(ctx context.Context, key client.ObjectKey, vpc *v1alpha1.VPC) error {
			*vpc = v1alpha1.VPC{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "test-vpc",
					Namespace: "test-namespace",
				},
				Spec: v1alpha1.VPCSpec{
					Provider: consts.ProviderTypeAKS,
					Region:   "eastus",
					AccountRef: &v1alpha1.AccountReference{
						AccountID: "test-subscription",
					},
				},
			}
			return nil
		})
		resourceGroup := controllers.GetResourceGroupName(&v1alpha1.VPC{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-vpc",
				Namespace: "test-namespace",
			},
		})
		plscli.EXPECT().ListPrivateLinkEndpointConnections(context.Background(), resourceGroup, "test-pls-azure").DoAndReturn(func(ctx context.Context, resourceGroup, plsName string) ([]*armnetwork.PrivateEndpointConnection, error) {
			return []*armnetwork.PrivateEndpointConnection{
				{
					Name: ptr.To("test-conn1"),
					Properties: &armnetwork.PrivateEndpointConnectionProperties{
						PrivateEndpoint: &armnetwork.PrivateEndpoint{
							ID: ptr.To("/endppint1"),
						},
						PrivateLinkServiceConnectionState: &armnetwork.PrivateLinkServiceConnectionState{
							Status: ptr.To("Pending"),
						},
					},
				},
				{
					Name: ptr.To("test-conn2"),
					Properties: &armnetwork.PrivateEndpointConnectionProperties{
						PrivateEndpoint: &armnetwork.PrivateEndpoint{
							ID: ptr.To("/endppint2"),
						},
						PrivateLinkServiceConnectionState: &armnetwork.PrivateLinkServiceConnectionState{
							Status: ptr.To("Rejected"),
						},
					},
				},
				{
					Name: ptr.To("test-conn3"),
					Properties: &armnetwork.PrivateEndpointConnectionProperties{
						PrivateEndpoint: &armnetwork.PrivateEndpoint{
							ID: ptr.To("/endppint3"),
						},
						PrivateLinkServiceConnectionState: &armnetwork.PrivateLinkServiceConnectionState{
							Status: ptr.To("Pending"),
						},
					},
				},
			}, nil
		})
		output, err := plssvc.GetPrivateLinkService(context.Background(), &regionalapiv1.GetPrivateLinkServiceRequest{
			PrivateLinkServiceId: "test-pls",
		})
		assert.NoError(t, err)
		assert.Equal(t, []*regionalapiv1.PrivateLinkService_Connection{
			{
				Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
				Message: "pending",
				Info: &regionalapiv1.PrivateLinkService_Connection_Azure_{
					Azure: &regionalapiv1.PrivateLinkService_Connection_Azure{
						PrivateEndpointResourceId: "/endppint1",
					},
				},
			},
			{
				Status: regionalapiv1.PrivateLinkService_Connection_PENDING,
				Info: &regionalapiv1.PrivateLinkService_Connection_Azure_{
					Azure: &regionalapiv1.PrivateLinkService_Connection_Azure{
						PrivateEndpointResourceId: "/endppint3",
					},
				},
			},
		}, output.Connections)
	})
}

func TestListPrivateLinkService(t *testing.T) {
	ctl := gomock.NewController(t)
	plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
	plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{})

	t.Run("list privatelinkservice failed", func(t *testing.T) {
		plsrw.EXPECT().List(context.Background(), &v1alpha1.PrivateLinkServiceList{}, client.MatchingLabels{
			v1alpha1.ClusterIDKey: "test-cluster",
		}).Return(assert.AnError)
		_, err := plssvc.ListPrivateLinkService(context.Background(), &regionalapiv1.ListPrivateLinkServicesRequest{
			ClusterId: "test-cluster",
		})
		assert.ErrorContains(t, err, "failed to list privatelinkservice")
	})

	t.Run("fill privatelinkservice failed", func(t *testing.T) {
		plsrw.EXPECT().List(context.Background(), &v1alpha1.PrivateLinkServiceList{}, client.MatchingLabels{
			v1alpha1.ClusterIDKey: "test-cluster",
		}).DoAndReturn(func(ctx context.Context, list *v1alpha1.PrivateLinkServiceList, opts ...client.ListOption) error {
			list.Items = []v1alpha1.PrivateLinkService{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name: "test-pls",
					},
				},
			}
			return nil
		})
		_, err := plssvc.ListPrivateLinkService(context.Background(), &regionalapiv1.ListPrivateLinkServicesRequest{
			ClusterId: "test-cluster",
		})
		assert.ErrorContains(t, err, "failed to fill privatelinkservice")
	})

	t.Run("list privatelinkservice success", func(t *testing.T) {
		plssvc.config = config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{
					Name: "us-west-1",
				},
			},
		}
		plsrw.EXPECT().List(context.Background(), &v1alpha1.PrivateLinkServiceList{}, client.MatchingLabels{
			v1alpha1.ClusterIDKey: "test-cluster",
		}).DoAndReturn(func(ctx context.Context, list *v1alpha1.PrivateLinkServiceList, opts ...client.ListOption) error {
			list.Items = []v1alpha1.PrivateLinkService{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name: "test-pls",
					},
					Spec: v1alpha1.PrivateLinkServiceSpec{
						Provider: consts.ProviderTypeEKS,
						Region:   "us-west-1",
					},
				},
			}
			return nil
		})
		_, err := plssvc.ListPrivateLinkService(context.Background(), &regionalapiv1.ListPrivateLinkServicesRequest{
			ClusterId: "test-cluster",
		})
		assert.NoError(t, err)
	})
}

func TestUpdatePrivateLinkService(t *testing.T) {
	ctl := gomock.NewController(t)
	plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
	plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{})
	namer := utils.NewNamerWithClusterID("", plssvc.config.RegionConfig.Name, plssvc.config.Env)

	t.Run("get privatelinkservice failed", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).Return(assert.AnError)
		_, err := plssvc.UpdatePrivateLinkService(context.Background(), &regionalapiv1.UpdatePrivateLinkServiceRequest{
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "failed to get privatelinkservice")
	})

	t.Run("update privatelinkservice failed", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).Return(nil)
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(assert.AnError)
		_, err := plssvc.UpdatePrivateLinkService(context.Background(), &regionalapiv1.UpdatePrivateLinkServiceRequest{
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "failed to update privatelinkservice")
	})

	t.Run("update privatelinkservice success", func(t *testing.T) {
		plssvc.config = config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{
					Name: "us-west-1",
				},
			},
		}
		namer = utils.NewNamerWithClusterID("", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		req := &regionalapiv1.UpdatePrivateLinkServiceRequest{
			PrivateLinkServiceId: "test-pls",
			PrivateLinkConnections: []*regionalapiv1.PrivateLinkService_Connection{
				{
					Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
					Message: "pending",
					Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
						Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
							VpcEndpointId: "https://test.com",
						},
					},
				},
			},
		}
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).DoAndReturn(func(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
			*pls = v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "test-pls",
					Namespace: namer.GetNamespace(),
				},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					Provider: consts.ProviderTypeEKS,
					Region:   "us-west-1",
				},
			}
			return nil
		})
		plsrw.EXPECT().Update(context.Background(), &v1alpha1.PrivateLinkService{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-pls",
				Namespace: namer.GetNamespace(),
			},
			Spec: v1alpha1.PrivateLinkServiceSpec{
				Provider: consts.ProviderTypeEKS,
				Region:   "us-west-1",
				Connections: []v1alpha1.Connection{
					{Endpoint: "https://test.com"},
				},
			},
		}).Return(nil)
		_, err := plssvc.UpdatePrivateLinkService(context.Background(), req)
		assert.NoError(t, err)
	})
	t.Run("validate privatelinkservice connections", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		vpcr := mockvpc.NewMockVPCReader(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, vpcr,
			config.RootConfig{
				ControllerConfig: config.ControllerConfig{
					Provider: consts.ProviderTypeAKS,
					RegionConfig: config.RegionConfig{
						Name: "useast2",
					},
				},
			})
		namer = utils.NewNamerWithClusterID("", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		req := &regionalapiv1.UpdatePrivateLinkServiceRequest{
			PrivateLinkServiceId: "test-pls",
			ValidateOnly:         true,
			PrivateLinkConnections: []*regionalapiv1.PrivateLinkService_Connection{
				{
					Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
					Message: "pending",
					Info: &regionalapiv1.PrivateLinkService_Connection_Azure_{
						Azure: &regionalapiv1.PrivateLinkService_Connection_Azure{
							PrivateEndpointResourceId: "azure/endpoint-2",
						},
					},
				},
			},
		}
		plscli := mockazure.NewMockPrivateLinkServiceClient(ctl)
		plssvc.plsClients.Store("test-subscription", plscli)

		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).DoAndReturn(func(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
			*pls = v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "test-pls",
					Namespace: namer.GetNamespace(),
				},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					Provider: consts.ProviderTypeAKS,
					Region:   "useast2",
					VPCRef: v1alpha1.Reference{
						Name:      "test-vpc",
						Namespace: "test-namespace",
					},
				},
				Status: v1alpha1.PrivateLinkServiceStatus{
					AzurePrivateLinkService: v1alpha1.AzurePrivateLinkServiceStatus{
						Name: "test-pls-azure",
					},
				},
			}
			return nil
		}).Times(2)
		vpcr.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-vpc", Namespace: "test-namespace"}, gomock.Any()).DoAndReturn(func(ctx context.Context, key client.ObjectKey, vpc *v1alpha1.VPC) error {
			*vpc = v1alpha1.VPC{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "test-vpc",
					Namespace: "test-namespace",
				},
				Spec: v1alpha1.VPCSpec{
					Provider: consts.ProviderTypeAKS,
					Region:   "eastus",
					AccountRef: &v1alpha1.AccountReference{
						AccountID: "test-subscription",
					},
				},
			}
			return nil
		}).Times(2)
		resourceGroup := controllers.GetResourceGroupName(&v1alpha1.VPC{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-vpc",
				Namespace: "test-namespace",
			},
		})
		plscli.EXPECT().ListPrivateLinkEndpointConnections(context.Background(), resourceGroup, "test-pls-azure").DoAndReturn(func(ctx context.Context, resourceGroup, plsName string) ([]*armnetwork.PrivateEndpointConnection, error) {
			return []*armnetwork.PrivateEndpointConnection{
				{
					Name: ptr.To("test-conn1"),
					Properties: &armnetwork.PrivateEndpointConnectionProperties{
						PrivateEndpoint: &armnetwork.PrivateEndpoint{
							ID: ptr.To("azure/endppint1"),
						},
						PrivateLinkServiceConnectionState: &armnetwork.PrivateLinkServiceConnectionState{
							Status: ptr.To("Pending"),
						},
					},
				},
			}, nil
		})
		_, err := plssvc.UpdatePrivateLinkService(context.Background(), req)
		assert.ErrorContains(t, err, "connection azure/endpoint-2 not found")

		plscli.EXPECT().ListPrivateLinkEndpointConnections(context.Background(), resourceGroup, "test-pls-azure").DoAndReturn(func(ctx context.Context, resourceGroup, plsName string) ([]*armnetwork.PrivateEndpointConnection, error) {
			return []*armnetwork.PrivateEndpointConnection{
				{
					Name: ptr.To("test-conn2"),
					Properties: &armnetwork.PrivateEndpointConnectionProperties{
						PrivateEndpoint: &armnetwork.PrivateEndpoint{
							ID: ptr.To("azure/endpoint-2"),
						},
						PrivateLinkServiceConnectionState: &armnetwork.PrivateLinkServiceConnectionState{
							Status: ptr.To("Pending"),
						},
					},
				},
			}, nil
		})
		_, err = plssvc.UpdatePrivateLinkService(context.Background(), req)
		assert.NoError(t, err)
	})
}

func TestRejectVpcEndpointConnections(t *testing.T) {
	t.Run("provider not EKS", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeAKS,
			},
		})
		err := plssvc.RejectVpcEndpointConnections(context.Background(), &regionalapiv1.RejectVpcEndpointConnectionsRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
			VpcEndpointIds:       []string{"vpce-1"},
		})
		assert.ErrorContains(t, err, "only supported for AWS")
	})

	t.Run("aws client not initialized", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		err := plssvc.RejectVpcEndpointConnections(context.Background(), &regionalapiv1.RejectVpcEndpointConnectionsRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
			VpcEndpointIds:       []string{"vpce-1"},
		})
		assert.ErrorContains(t, err, "AWS client is not initialized")
	})

	t.Run("get privatelinkservice k8s error", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		mockAwsCli := mockaws.NewMockClient(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{
					Name: "us-west-1",
				},
			},
		})
		plssvc.WithAWSClient(mockAwsCli)
		namer := utils.NewNamerWithClusterID("test-cluster", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).Return(assert.AnError)
		err := plssvc.RejectVpcEndpointConnections(context.Background(), &regionalapiv1.RejectVpcEndpointConnectionsRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
			VpcEndpointIds:       []string{"vpce-1"},
		})
		assert.ErrorContains(t, err, "failed to get privatelinkservice")
	})

	t.Run("privatelinkservice not found", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		mockAwsCli := mockaws.NewMockClient(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{
					Name: "us-west-1",
				},
			},
		})
		plssvc.WithAWSClient(mockAwsCli)
		namer := utils.NewNamerWithClusterID("test-cluster", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).Return(
			apierrors.NewNotFound(schema.GroupResource{}, "test-pls"),
		)
		err := plssvc.RejectVpcEndpointConnections(context.Background(), &regionalapiv1.RejectVpcEndpointConnectionsRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
			VpcEndpointIds:       []string{"vpce-1"},
		})
		assert.ErrorContains(t, err, "not found")
	})

	t.Run("cluster id mismatch", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		mockAwsCli := mockaws.NewMockClient(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{
					Name: "us-west-1",
				},
			},
		})
		plssvc.WithAWSClient(mockAwsCli)
		namer := utils.NewNamerWithClusterID("test-cluster", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).DoAndReturn(
			func(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
				*pls = v1alpha1.PrivateLinkService{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "test-pls",
						Namespace: namer.GetNamespace(),
						Labels: map[string]string{
							v1alpha1.ClusterIDKey: "different-cluster",
						},
					},
				}
				return nil
			},
		)
		err := plssvc.RejectVpcEndpointConnections(context.Background(), &regionalapiv1.RejectVpcEndpointConnectionsRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
			VpcEndpointIds:       []string{"vpce-1"},
		})
		assert.ErrorContains(t, err, "not found")
	})

	t.Run("endpoint service not ready", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		mockAwsCli := mockaws.NewMockClient(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{
					Name: "us-west-1",
				},
			},
		})
		plssvc.WithAWSClient(mockAwsCli)
		namer := utils.NewNamerWithClusterID("test-cluster", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).DoAndReturn(
			func(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
				*pls = v1alpha1.PrivateLinkService{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "test-pls",
						Namespace: namer.GetNamespace(),
						Labels: map[string]string{
							v1alpha1.ClusterIDKey: "test-cluster",
						},
					},
					Status: v1alpha1.PrivateLinkServiceStatus{
						EndpointService: v1alpha1.EndpointServiceStatus{
							ServiceID: "",
						},
					},
				}
				return nil
			},
		)
		err := plssvc.RejectVpcEndpointConnections(context.Background(), &regionalapiv1.RejectVpcEndpointConnectionsRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
			VpcEndpointIds:       []string{"vpce-1"},
		})
		assert.ErrorContains(t, err, "VPC endpoint service is not ready")
	})

	t.Run("get vpc failed", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		vpcr := mockvpc.NewMockVPCReader(ctl)
		mockAwsCli := mockaws.NewMockClient(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, vpcr, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
				RegionConfig: config.RegionConfig{
					Name: "us-west-1",
				},
			},
		})
		plssvc.WithAWSClient(mockAwsCli)
		namer := utils.NewNamerWithClusterID("test-cluster", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).DoAndReturn(
			func(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
				*pls = v1alpha1.PrivateLinkService{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "test-pls",
						Namespace: namer.GetNamespace(),
						Labels: map[string]string{
							v1alpha1.ClusterIDKey: "test-cluster",
						},
					},
					Spec: v1alpha1.PrivateLinkServiceSpec{
						Provider: consts.ProviderTypeEKS,
						Region:   "us-west-1",
						VPCRef: v1alpha1.Reference{
							Name:      "test-vpc",
							Namespace: "test-ns",
						},
					},
					Status: v1alpha1.PrivateLinkServiceStatus{
						EndpointService: v1alpha1.EndpointServiceStatus{
							ServiceID: "vpce-svc-123",
						},
					},
				}
				return nil
			},
		)
		vpcr.EXPECT().Get(context.Background(), client.ObjectKey{Namespace: "test-ns", Name: "test-vpc"}, gomock.Any()).Return(assert.AnError)
		err := plssvc.RejectVpcEndpointConnections(context.Background(), &regionalapiv1.RejectVpcEndpointConnectionsRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
			VpcEndpointIds:       []string{"vpce-1"},
		})
		assert.ErrorContains(t, err, "failed to get vpc")
	})
}

func TestDeletePrivateLinkService(t *testing.T) {
	ctl := gomock.NewController(t)
	plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
	plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{})

	namer := utils.NewNamerWithClusterID("", plssvc.config.RegionConfig.Name, plssvc.config.Env)
	t.Run("delete privatelinkservice failed", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).Return(nil)
		plsrw.EXPECT().Delete(context.Background(), &v1alpha1.PrivateLinkService{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-pls",
				Namespace: namer.GetNamespace(),
			}}).Return(assert.AnError)
		err := plssvc.DeletePrivateLinkService(context.Background(), &regionalapiv1.DeletePrivateLinkServiceRequest{
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "failed to delete privatelinkservice")
	})
	t.Run("privatelinkservice not found", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).Return(apierrors.NewNotFound(schema.GroupResource{}, "test-pls"))
		err := plssvc.DeletePrivateLinkService(context.Background(), &regionalapiv1.DeletePrivateLinkServiceRequest{
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "not found")
	})

	t.Run("delete privatelinkservice success", func(t *testing.T) {
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).Return(nil)
		plsrw.EXPECT().Delete(context.Background(), &v1alpha1.PrivateLinkService{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-pls",
				Namespace: namer.GetNamespace(),
			},
		}).Return(nil)
		err := plssvc.DeletePrivateLinkService(context.Background(), &regionalapiv1.DeletePrivateLinkServiceRequest{
			PrivateLinkServiceId: "test-pls",
		})
		assert.NoError(t, err)
	})
}

func Test_genConnectionPB(t *testing.T) {
	cases := []struct {
		name        string
		provider    string
		conn        *v1alpha1.ConnectionStatus
		want        *regionalapiv1.PrivateLinkService_Connection
		expectedErr error
	}{
		{
			name:     "eks pending connection",
			provider: consts.ProviderTypeEKS,
			conn: &v1alpha1.ConnectionStatus{
				Phase:    v1alpha1.ConnectionPhasePending,
				Message:  "pending",
				Endpoint: "https://test.com",
			},
			want: &regionalapiv1.PrivateLinkService_Connection{
				Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
				Message: "pending",
				Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
					Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
						VpcEndpointId: "https://test.com",
					},
				},
			},
		},
		{
			name:     "eks active connection",
			provider: consts.ProviderTypeEKS,
			conn: &v1alpha1.ConnectionStatus{
				Phase:         v1alpha1.ConnectionPhaseActive,
				Message:       "active",
				Endpoint:      "https://test.com",
				IPAddressType: v1alpha1.EndpointIPAddressTypeIPv6,
			},
			want: &regionalapiv1.PrivateLinkService_Connection{
				Status:        regionalapiv1.PrivateLinkService_Connection_ACTIVE,
				Message:       "active",
				IpAddressType: ptr.To(regionalapiv1.PrivateLinkService_Connection_IP_ADDRESS_TYPE_IPV6),
				Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
					Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
						VpcEndpointId: "https://test.com",
					},
				},
			},
		},
		{
			name:     "eks failed connection",
			provider: consts.ProviderTypeEKS,
			conn: &v1alpha1.ConnectionStatus{
				Phase:    v1alpha1.ConnectionPhaseFailed,
				Message:  "failed",
				Endpoint: "https://test.com",
			},
			want: &regionalapiv1.PrivateLinkService_Connection{
				Status:  regionalapiv1.PrivateLinkService_Connection_FAILED,
				Message: "failed",
				Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
					Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
						VpcEndpointId: "https://test.com",
					},
				},
			},
		},
		{
			name:     "eks deleting connection",
			provider: consts.ProviderTypeEKS,
			conn: &v1alpha1.ConnectionStatus{
				Phase:    v1alpha1.ConnectionPhaseDeleting,
				Message:  "deleting",
				Endpoint: "https://test.com",
			},
			want: &regionalapiv1.PrivateLinkService_Connection{
				Status:  regionalapiv1.PrivateLinkService_Connection_DELETING,
				Message: "deleting",
				Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
					Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
						VpcEndpointId: "https://test.com",
					},
				},
			},
		},
		{
			name:     "eks unknown connection",
			provider: consts.ProviderTypeEKS,
			conn: &v1alpha1.ConnectionStatus{
				Phase:    "",
				Message:  "unknown",
				Endpoint: "https://test.com",
			},
			want: &regionalapiv1.PrivateLinkService_Connection{
				Status:  regionalapiv1.PrivateLinkService_Connection_STATE_UNSPECIFIED,
				Message: "unknown",
				Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
					Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
						VpcEndpointId: "https://test.com",
					},
				},
			},
		},
		{
			name:     "gke pending connection",
			provider: consts.ProviderTypeGKE,
			conn: &v1alpha1.ConnectionStatus{
				Phase:    v1alpha1.ConnectionPhasePending,
				Message:  "pending",
				Endpoint: "https://test.com",
			},
			want: &regionalapiv1.PrivateLinkService_Connection{
				Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
				Message: "pending",
				Info: &regionalapiv1.PrivateLinkService_Connection_Gcp_{
					Gcp: &regionalapiv1.PrivateLinkService_Connection_Gcp{
						PrivateServiceConnectEndpoint: "https://test.com",
					},
				},
			},
		},
		{
			name: "invalid provider",
			conn: &v1alpha1.ConnectionStatus{
				Phase:    v1alpha1.ConnectionPhasePending,
				Message:  "pending",
				Endpoint: "https://test.com",
			},
			expectedErr: bizerr.New(bizerr.InternalServerError, "provider config invalid"),
		},
	}

	for _, tt := range cases {
		t.Run(tt.name, func(t *testing.T) {
			got, err := genConnectionPB(tt.provider, v1alpha1.Connection{Endpoint: tt.conn.Endpoint}, tt.conn)
			assert.Equal(t, tt.expectedErr, err)
			assert.Equal(t, tt.want, got)
		})
	}
}

func TestConnectionIPAddressTypeJSONPresenceIsEKSSpecific(t *testing.T) {
	marshal := protojson.MarshalOptions{EmitUnpopulated: true}

	for _, provider := range []string{consts.ProviderTypeGKE, consts.ProviderTypeAKS} {
		t.Run(provider, func(t *testing.T) {
			connection, err := genConnectionPB(
				provider,
				v1alpha1.Connection{Endpoint: "private-endpoint"},
				&v1alpha1.ConnectionStatus{Phase: v1alpha1.ConnectionPhaseActive},
			)
			require.NoError(t, err)
			encoded, err := marshal.Marshal(connection)
			require.NoError(t, err)
			assert.NotContains(t, string(encoded), "ipAddressType")
		})
	}

	eks, err := genConnectionPB(
		consts.ProviderTypeEKS,
		v1alpha1.Connection{Endpoint: "vpce-1"},
		&v1alpha1.ConnectionStatus{
			Phase: v1alpha1.ConnectionPhaseActive, IPAddressType: v1alpha1.EndpointIPAddressTypeDualstack,
		},
	)
	require.NoError(t, err)
	eksJSON, err := marshal.Marshal(eks)
	require.NoError(t, err)
	assert.Contains(t, string(eksJSON), `"ipAddressType":"IP_ADDRESS_TYPE_DUALSTACK"`)
}

func Test_setConnectionProviderInfo(t *testing.T) {
	cases := []struct {
		name        string
		provider    string
		endpoint    string
		dnsName     string
		input       *regionalapiv1.PrivateLinkService_Connection
		want        *regionalapiv1.PrivateLinkService_Connection
		expectedErr error
	}{
		{
			name:     "eks connection",
			provider: consts.ProviderTypeEKS,
			endpoint: "https://test.com",
			input: &regionalapiv1.PrivateLinkService_Connection{
				Status: regionalapiv1.PrivateLinkService_Connection_PENDING,
			},
			want: &regionalapiv1.PrivateLinkService_Connection{
				Status: regionalapiv1.PrivateLinkService_Connection_PENDING,
				Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
					Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
						VpcEndpointId: "https://test.com",
					},
				},
			},
		},
		{
			name:     "gke connection",
			provider: consts.ProviderTypeGKE,
			endpoint: "https://test.com",
			input: &regionalapiv1.PrivateLinkService_Connection{
				Status: regionalapiv1.PrivateLinkService_Connection_PENDING,
			},
			want: &regionalapiv1.PrivateLinkService_Connection{
				Status: regionalapiv1.PrivateLinkService_Connection_PENDING,
				Info: &regionalapiv1.PrivateLinkService_Connection_Gcp_{
					Gcp: &regionalapiv1.PrivateLinkService_Connection_Gcp{
						PrivateServiceConnectEndpoint: "https://test.com",
					},
				},
			},
		},
		{
			name:     "invalid provider",
			provider: "invalid",
			endpoint: "https://test.com",
			input: &regionalapiv1.PrivateLinkService_Connection{
				Status: regionalapiv1.PrivateLinkService_Connection_PENDING,
			},
			expectedErr: bizerr.New(bizerr.InternalServerError, "provider config invalid"),
		},
		{
			name:     "gke connection with dns name",
			provider: consts.ProviderTypeGKE,
			endpoint: "https://test.com/endpoint",
			dnsName:  "endpoint.tidbcloud.com",
			input: &regionalapiv1.PrivateLinkService_Connection{
				Status: regionalapiv1.PrivateLinkService_Connection_ACTIVE,
			},
			want: &regionalapiv1.PrivateLinkService_Connection{
				Status: regionalapiv1.PrivateLinkService_Connection_ACTIVE,
				Info: &regionalapiv1.PrivateLinkService_Connection_Gcp_{
					Gcp: &regionalapiv1.PrivateLinkService_Connection_Gcp{
						PrivateServiceConnectEndpoint: "https://test.com/endpoint",
						Host:                          "endpoint.tidbcloud.com",
					},
				},
			},
		},
	}

	for _, tt := range cases {
		t.Run(tt.name, func(t *testing.T) {
			output, err := setConnectionProviderInfo(tt.input, tt.provider, v1alpha1.Connection{Endpoint: tt.endpoint}, &v1alpha1.ConnectionStatus{
				DNSName: tt.dnsName,
			})
			assert.Equal(t, tt.expectedErr, err)
			assert.Equal(t, tt.want, output)
		})
	}
}

func Test_genPrivateLinkServicePB(t *testing.T) {
	cases := []struct {
		name        string
		pls         *v1alpha1.PrivateLinkService
		want        *regionalapiv1.PrivateLinkService
		expectedErr error
	}{
		{
			name: "eks pending pls",
			pls: &v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{
					Name: "test",
					Labels: map[string]string{
						v1alpha1.ClusterIDKey:      "test-cluster",
						v1alpha1.LoadBalancerIDKey: "test-lb",
					},
				},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					Provider: consts.ProviderTypeEKS,
					Region:   "us-west-1",
					Connections: []v1alpha1.Connection{
						{Endpoint: "https://test.com"},
					},
				},
				Status: v1alpha1.PrivateLinkServiceStatus{
					Phase: v1alpha1.PrivateLinkServicePhasePending,
					EndpointService: v1alpha1.EndpointServiceStatus{
						ServiceName:       "test-service",
						AvailabilityZones: []string{"us-west-1a", "us-west-1b"},
						PrivateDNSName:    "test.com",
						Port:              4001,
					},
					Connections: []v1alpha1.ConnectionStatus{
						{
							Phase:    v1alpha1.ConnectionPhasePending,
							Message:  "pending",
							Endpoint: "https://test.com",
						},
					},
				},
			},
			want: &regionalapiv1.PrivateLinkService{
				Id:             "test",
				ClusterId:      "test-cluster",
				TidbGroupId:    "test-lb",
				Host:           "test.com",
				ServiceDnsName: "test.com",
				Info: &regionalapiv1.PrivateLinkService_Aws_{
					Aws: &regionalapiv1.PrivateLinkService_Aws{
						ServiceName: "test-service",
						AzIds:       []string{"us-west-1a", "us-west-1b"},
					},
				},
				Connections: []*regionalapiv1.PrivateLinkService_Connection{
					{
						Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
						Message: "pending",
						Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
							Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
								VpcEndpointId: "https://test.com",
							},
						},
					},
				},
				Status: regionalapiv1.PrivateLinkService_CREATING,
			},
		},
		{
			name: "gke pending pls",
			pls: &v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{
					Name: "test",
					Labels: map[string]string{
						v1alpha1.ClusterIDKey:      "test-cluster",
						v1alpha1.LoadBalancerIDKey: "test-lb",
					},
				},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					Provider: consts.ProviderTypeGKE,
					Region:   "us-west-1",
					Connections: []v1alpha1.Connection{
						{Endpoint: "https://test.com"},
					},
				},
				Status: v1alpha1.PrivateLinkServiceStatus{
					Phase: v1alpha1.PrivateLinkServicePhasePending,
					ServiceAttachment: v1alpha1.ServiceAttachmentStatus{
						Name:              "test-attachment",
						ServiceAttachment: "test-attachment",
						ConnectDomain:     "test.com",
						Port:              4001,
					},
					Connections: []v1alpha1.ConnectionStatus{
						{
							Phase:    v1alpha1.ConnectionPhasePending,
							Message:  "pending",
							Endpoint: "https://test.com",
						},
					},
				},
			},
			want: &regionalapiv1.PrivateLinkService{
				Id:             "test",
				ClusterId:      "test-cluster",
				TidbGroupId:    "test-lb",
				Host:           "test.com",
				ServiceDnsName: "test.com",
				Info: &regionalapiv1.PrivateLinkService_Gcp_{
					Gcp: &regionalapiv1.PrivateLinkService_Gcp{
						ServiceAttachment: "test-attachment",
					},
				},
				Connections: []*regionalapiv1.PrivateLinkService_Connection{
					{
						Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
						Message: "pending",
						Info: &regionalapiv1.PrivateLinkService_Connection_Gcp_{
							Gcp: &regionalapiv1.PrivateLinkService_Connection_Gcp{
								PrivateServiceConnectEndpoint: "https://test.com",
							},
						},
					},
				},
				Status: regionalapiv1.PrivateLinkService_CREATING,
			},
		},
		{
			name: "connection status empty",
			pls: &v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{
					Name: "test",
					Labels: map[string]string{
						v1alpha1.ClusterIDKey:      "test-cluster",
						v1alpha1.LoadBalancerIDKey: "test-lb",
					},
				},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					Provider: consts.ProviderTypeEKS,
					Region:   "us-west-1",
					Connections: []v1alpha1.Connection{
						{Endpoint: "https://test.com"},
					},
				},
				Status: v1alpha1.PrivateLinkServiceStatus{
					Phase: v1alpha1.PrivateLinkServicePhasePending,
					EndpointService: v1alpha1.EndpointServiceStatus{
						ServiceName:       "test-service",
						AvailabilityZones: []string{"us-west-1a", "us-west-1b"},
						PrivateDNSName:    "test.com",
						Port:              4001,
					},
				},
			},
			want: &regionalapiv1.PrivateLinkService{
				Id:             "test",
				ClusterId:      "test-cluster",
				TidbGroupId:    "test-lb",
				Host:           "test.com",
				ServiceDnsName: "test.com",
				Info: &regionalapiv1.PrivateLinkService_Aws_{
					Aws: &regionalapiv1.PrivateLinkService_Aws{
						ServiceName: "test-service",
						AzIds:       []string{"us-west-1a", "us-west-1b"},
					},
				},
				Connections: []*regionalapiv1.PrivateLinkService_Connection{
					{
						Status: regionalapiv1.PrivateLinkService_Connection_STATE_UNSPECIFIED,
						Info: &regionalapiv1.PrivateLinkService_Connection_Aws_{
							Aws: &regionalapiv1.PrivateLinkService_Connection_Aws{
								VpcEndpointId: "https://test.com",
							},
						},
					},
				},
				Status: regionalapiv1.PrivateLinkService_CREATING,
			},
		},
		{
			name: "exist pending connections which are not in allowed list",
			pls: &v1alpha1.PrivateLinkService{
				ObjectMeta: metav1.ObjectMeta{
					Name: "test",
					Labels: map[string]string{
						v1alpha1.ClusterIDKey:      "test-cluster",
						v1alpha1.LoadBalancerIDKey: "test-lb",
					},
				},
				Spec: v1alpha1.PrivateLinkServiceSpec{
					Provider: consts.ProviderTypeAKS,
					Region:   "eastus",
					Connections: []v1alpha1.Connection{
						{Endpoint: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/privateEndpoints/conn1"},
						{Endpoint: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/privateEndpoints/conn2", IPs: []string{"1.2.3.4"}},
					},
				},
				Status: v1alpha1.PrivateLinkServiceStatus{
					Phase: v1alpha1.PrivateLinkServicePhasePending,
					AzurePrivateLinkService: v1alpha1.AzurePrivateLinkServiceStatus{
						Name:           "pls1",
						ID:             "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/PrivateLinkServices/pls1",
						ServiceDNSName: "test.com",
					},
					Connections: []v1alpha1.ConnectionStatus{
						{
							Phase:    v1alpha1.ConnectionPhasePending,
							Message:  "pending",
							Endpoint: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/privateEndpoints/conn1",
						},
						{
							Phase:    v1alpha1.ConnectionPhaseActive,
							Message:  "active",
							Endpoint: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/privateEndpoints/conn2",
							DNSName:  "endpoint.test.com",
						},
						{
							Phase:    v1alpha1.ConnectionPhasePending,
							Message:  "pending and not in allowed list",
							Endpoint: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/privateEndpoints/conn3",
						},
						{
							Phase:    v1alpha1.ConnectionPhasePending,
							Message:  "pending and not in allowed list",
							Endpoint: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/privateEndpoints/conn4",
						},
						{
							Phase:    v1alpha1.ConnectionPhaseFailed,
							Message:  "falied",
							Endpoint: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/privateEndpoints/conn5",
						},
					},
				},
			},
			want: &regionalapiv1.PrivateLinkService{
				Id:             "test",
				ClusterId:      "test-cluster",
				TidbGroupId:    "test-lb",
				Host:           "test.com",
				ServiceDnsName: "test.com",
				Status:         regionalapiv1.PrivateLinkService_CREATING,
				Info: &regionalapiv1.PrivateLinkService_Azure_{
					Azure: &regionalapiv1.PrivateLinkService_Azure{
						ServiceResourceId: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/PrivateLinkServices/pls1",
					},
				},
				Connections: []*regionalapiv1.PrivateLinkService_Connection{
					{
						Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
						Message: "pending",
						Info: &regionalapiv1.PrivateLinkService_Connection_Azure_{
							Azure: &regionalapiv1.PrivateLinkService_Connection_Azure{
								PrivateEndpointResourceId: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/privateEndpoints/conn1",
							},
						},
					},
					{
						Status:  regionalapiv1.PrivateLinkService_Connection_ACTIVE,
						Message: "active",
						Info: &regionalapiv1.PrivateLinkService_Connection_Azure_{
							Azure: &regionalapiv1.PrivateLinkService_Connection_Azure{
								PrivateEndpointResourceId: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/privateEndpoints/conn2",
								Host:                      "endpoint.test.com",
								PrivateIpAddress:          "1.2.3.4",
							},
						},
					},
					{
						Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
						Message: "pending and not in allowed list",
						Info: &regionalapiv1.PrivateLinkService_Connection_Azure_{
							Azure: &regionalapiv1.PrivateLinkService_Connection_Azure{
								PrivateEndpointResourceId: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/privateEndpoints/conn3",
							},
						},
					},
					{
						Status:  regionalapiv1.PrivateLinkService_Connection_PENDING,
						Message: "pending and not in allowed list",
						Info: &regionalapiv1.PrivateLinkService_Connection_Azure_{
							Azure: &regionalapiv1.PrivateLinkService_Connection_Azure{
								PrivateEndpointResourceId: "/subscriptions/xxx/resourceGroups/xxx/providers/Microsoft.Network/privateEndpoints/conn4",
							},
						},
					},
				},
			},
		},
	}

	for _, tt := range cases {
		t.Run(tt.name, func(t *testing.T) {
			got, err := genPrivateLinkServicePB(tt.pls)
			assert.Equal(t, tt.expectedErr, err)
			assert.Equal(t, tt.want, got)
		})
	}
}

func Test_coverToPrivateLinkServicePBStatus(t *testing.T) {
	cases := []struct {
		name  string
		phase v1alpha1.PrivateLinkServicePhase
		want  regionalapiv1.PrivateLinkService_Status
	}{
		{
			name:  "empty",
			phase: "",
			want:  regionalapiv1.PrivateLinkService_CREATING,
		},
		{
			name:  "pending",
			phase: v1alpha1.PrivateLinkServicePhasePending,
			want:  regionalapiv1.PrivateLinkService_CREATING,
		},
		{
			name:  "readyforcreate",
			phase: v1alpha1.PrivateLinkServicePhaseReadyForCreate,
			want:  regionalapiv1.PrivateLinkService_CREATING,
		},
		{
			name:  "creating",
			phase: v1alpha1.PrivateLinkServicePhaseCreating,
			want:  regionalapiv1.PrivateLinkService_CREATING,
		},
		{
			name:  "failed",
			phase: v1alpha1.PrivateLinkServicePhaseFailed,
			want:  regionalapiv1.PrivateLinkService_CREATING,
		},
		{
			name:  "active",
			phase: v1alpha1.PrivateLinkServicePhaseActive,
			want:  regionalapiv1.PrivateLinkService_ACTIVE,
		},
		{
			name:  "deleting",
			phase: v1alpha1.PrivateLinkServicePhaseDeleting,
			want:  regionalapiv1.PrivateLinkService_DELETING,
		},
		{
			name:  "deleted",
			phase: v1alpha1.PrivateLinkServicePhaseDeleted,
			want:  regionalapiv1.PrivateLinkService_DELETING,
		},
		{
			name:  "readyfordelete",
			phase: v1alpha1.PrivateLinkServicePhaseReadyForDelete,
			want:  regionalapiv1.PrivateLinkService_DELETING,
		},
	}

	for _, tt := range cases {
		t.Run(tt.name, func(t *testing.T) {
			got := coverToPrivateLinkServicePBStatus(tt.phase)
			assert.Equal(t, tt.want, got)
		})
	}
}

func TestPausePrivateLinkService(t *testing.T) {
	t.Run("unsupported provider", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeGKE,
			},
		})
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "only supported for AWS")
	})

	t.Run("privatelinkservice not found", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		namer := utils.NewNamerWithClusterID("test-cluster", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			Return(apierrors.NewNotFound(schema.GroupResource{}, "test-pls"))
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "not found")
	})

	t.Run("cluster id not match", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		namer := utils.NewNamerWithClusterID("test-cluster", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "other-cluster"}
				return nil
			})
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "not found")
	})

	t.Run("pause privatelinkservice update failed", func(t *testing.T) {
		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(nil, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		namer := utils.NewNamerWithClusterID("test-cluster", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				return nil
			})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(assert.AnError)
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "failed to pause privatelinkservice")
	})

	t.Run("crossplane vpce not found - returns success", func(t *testing.T) {
		s := runtime.NewScheme()
		ec2v1alpha1.SchemeBuilder.AddToScheme(s)
		fakeCli := fake.NewClientBuilder().WithScheme(s).Build()

		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(fakeCli, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		namer := utils.NewNamerWithClusterID("test-cluster", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				obj.Name = "test-pls"
				obj.Namespace = namer.GetNamespace()
				return nil
			})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(nil)
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.NoError(t, err)
	})

	t.Run("crossplane vpce get error", func(t *testing.T) {
		s := runtime.NewScheme()
		ec2v1alpha1.SchemeBuilder.AddToScheme(s)
		fakeCli := fake.NewClientBuilder().WithScheme(s).
			WithInterceptorFuncs(interceptor.Funcs{
				Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
					if _, ok := obj.(*ec2v1alpha1.VPCEndpointServiceConfiguration); ok {
						return assert.AnError
					}
					return c.Get(ctx, key, obj, opts...)
				},
			}).Build()

		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(fakeCli, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		namer := utils.NewNamerWithClusterID("test-cluster", plssvc.config.RegionConfig.Name, plssvc.config.Env)
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				obj.Name = "test-pls"
				obj.Namespace = namer.GetNamespace()
				return nil
			})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(nil)
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "failed to get crossplane VPCEndpointServiceConfiguration")
	})

	t.Run("crossplane vpce already paused - idempotent", func(t *testing.T) {
		s := runtime.NewScheme()
		ec2v1alpha1.SchemeBuilder.AddToScheme(s)
		namer := utils.NewNamerWithClusterID("test-cluster", "", "")
		vpceSvcName := fmt.Sprintf("%s-%s", namer.GetNamespace(), "test-pls")
		existingVes := &ec2v1alpha1.VPCEndpointServiceConfiguration{
			ObjectMeta: metav1.ObjectMeta{
				Name: vpceSvcName,
				Annotations: map[string]string{
					meta.AnnotationKeyReconciliationPaused: "true",
				},
			},
		}
		fakeCli := fake.NewClientBuilder().WithScheme(s).WithObjects(existingVes).Build()

		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(fakeCli, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				obj.Name = "test-pls"
				obj.Namespace = namer.GetNamespace()
				// already paused
				obj.Annotations = map[string]string{
					v1alpha1.AnnotationKeyReconciliationPaused: "true",
				}
				return nil
			})
		// no plsrw.Update expected - PLS already paused
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.NoError(t, err)
	})

	t.Run("crossplane vpce update error", func(t *testing.T) {
		s := runtime.NewScheme()
		ec2v1alpha1.SchemeBuilder.AddToScheme(s)
		namer := utils.NewNamerWithClusterID("test-cluster", "", "")
		vpceSvcName := fmt.Sprintf("%s-%s", namer.GetNamespace(), "test-pls")
		existingVes := &ec2v1alpha1.VPCEndpointServiceConfiguration{
			ObjectMeta: metav1.ObjectMeta{
				Name: vpceSvcName,
			},
		}
		fakeCli := fake.NewClientBuilder().WithScheme(s).WithObjects(existingVes).
			WithInterceptorFuncs(interceptor.Funcs{
				Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
					if _, ok := obj.(*ec2v1alpha1.VPCEndpointServiceConfiguration); ok {
						return assert.AnError
					}
					return c.Patch(ctx, obj, patch, opts...)
				},
			}).Build()

		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(fakeCli, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				obj.Name = "test-pls"
				obj.Namespace = namer.GetNamespace()
				return nil
			})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(nil)
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "failed to pause crossplane VPCEndpointServiceConfiguration")
	})

	t.Run("full success - both pls and crossplane paused", func(t *testing.T) {
		s := runtime.NewScheme()
		ec2v1alpha1.SchemeBuilder.AddToScheme(s)
		namer := utils.NewNamerWithClusterID("test-cluster", "", "")
		vpceSvcName := fmt.Sprintf("%s-%s", namer.GetNamespace(), "test-pls")
		existingVes := &ec2v1alpha1.VPCEndpointServiceConfiguration{
			ObjectMeta: metav1.ObjectMeta{
				Name: vpceSvcName,
			},
		}
		fakeCli := fake.NewClientBuilder().WithScheme(s).WithObjects(existingVes).Build()

		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(fakeCli, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				obj.Name = "test-pls"
				obj.Namespace = namer.GetNamespace()
				return nil
			})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(nil)
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.NoError(t, err)

		// Verify the crossplane object was annotated
		updatedVes := &ec2v1alpha1.VPCEndpointServiceConfiguration{}
		err = fakeCli.Get(context.Background(), client.ObjectKey{Name: vpceSvcName}, updatedVes)
		assert.NoError(t, err)
		assert.Equal(t, "true", updatedVes.GetAnnotations()[meta.AnnotationKeyReconciliationPaused])
	})

	t.Run("crossplane resourcerecordset not found - returns success", func(t *testing.T) {
		s := runtime.NewScheme()
		ec2v1alpha1.SchemeBuilder.AddToScheme(s)
		route53v1alpha1.SchemeBuilder.AddToScheme(s)
		namer := utils.NewNamerWithClusterID("test-cluster", "", "")
		vpceSvcName := fmt.Sprintf("%s-%s", namer.GetNamespace(), "test-pls")
		existingVes := &ec2v1alpha1.VPCEndpointServiceConfiguration{
			ObjectMeta: metav1.ObjectMeta{
				Name: vpceSvcName,
			},
		}
		fakeCli := fake.NewClientBuilder().WithScheme(s).WithObjects(existingVes).Build()

		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(fakeCli, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				obj.Name = "test-pls"
				obj.Namespace = namer.GetNamespace()
				obj.Status.EndpointService.PrivateDNSName = "test.example.com"
				return nil
			})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(nil)
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.NoError(t, err)
	})

	t.Run("crossplane resourcerecordset get error", func(t *testing.T) {
		s := runtime.NewScheme()
		ec2v1alpha1.SchemeBuilder.AddToScheme(s)
		route53v1alpha1.SchemeBuilder.AddToScheme(s)
		namer := utils.NewNamerWithClusterID("test-cluster", "", "")
		vpceSvcName := fmt.Sprintf("%s-%s", namer.GetNamespace(), "test-pls")
		existingVes := &ec2v1alpha1.VPCEndpointServiceConfiguration{
			ObjectMeta: metav1.ObjectMeta{
				Name: vpceSvcName,
			},
		}
		fakeCli := fake.NewClientBuilder().WithScheme(s).WithObjects(existingVes).
			WithInterceptorFuncs(interceptor.Funcs{
				Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
					if _, ok := obj.(*route53v1alpha1.ResourceRecordSet); ok {
						return assert.AnError
					}
					return c.Get(ctx, key, obj, opts...)
				},
			}).Build()

		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(fakeCli, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				obj.Name = "test-pls"
				obj.Namespace = namer.GetNamespace()
				obj.Status.EndpointService.PrivateDNSName = "test.example.com"
				return nil
			})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(nil)
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "failed to get crossplane ResourceRecordSet")
	})

	t.Run("crossplane resourcerecordset already paused - idempotent", func(t *testing.T) {
		s := runtime.NewScheme()
		ec2v1alpha1.SchemeBuilder.AddToScheme(s)
		route53v1alpha1.SchemeBuilder.AddToScheme(s)
		namer := utils.NewNamerWithClusterID("test-cluster", "", "")
		vpceSvcName := fmt.Sprintf("%s-%s", namer.GetNamespace(), "test-pls")
		recordName := "test.example.com"
		existingVes := &ec2v1alpha1.VPCEndpointServiceConfiguration{
			ObjectMeta: metav1.ObjectMeta{
				Name: vpceSvcName,
				Annotations: map[string]string{
					meta.AnnotationKeyReconciliationPaused: "true",
				},
			},
		}
		existingRRS := &route53v1alpha1.ResourceRecordSet{
			ObjectMeta: metav1.ObjectMeta{
				Name: recordName,
				Annotations: map[string]string{
					meta.AnnotationKeyReconciliationPaused: "true",
				},
			},
		}
		fakeCli := fake.NewClientBuilder().WithScheme(s).WithObjects(existingVes, existingRRS).
			WithInterceptorFuncs(interceptor.Funcs{
				Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
					// no patches expected because both are already paused
					return fmt.Errorf("unexpected Patch call for %T", obj)
				},
			}).Build()

		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(fakeCli, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				obj.Name = "test-pls"
				obj.Namespace = namer.GetNamespace()
				obj.Status.EndpointService.PrivateDNSName = recordName
				// pls already paused - no update expected
				obj.Annotations = map[string]string{
					v1alpha1.AnnotationKeyReconciliationPaused: "true",
				}
				return nil
			})
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.NoError(t, err)
	})

	t.Run("crossplane resourcerecordset patch error", func(t *testing.T) {
		s := runtime.NewScheme()
		ec2v1alpha1.SchemeBuilder.AddToScheme(s)
		route53v1alpha1.SchemeBuilder.AddToScheme(s)
		namer := utils.NewNamerWithClusterID("test-cluster", "", "")
		vpceSvcName := fmt.Sprintf("%s-%s", namer.GetNamespace(), "test-pls")
		recordName := "test.example.com"
		existingVes := &ec2v1alpha1.VPCEndpointServiceConfiguration{
			ObjectMeta: metav1.ObjectMeta{
				Name: vpceSvcName,
			},
		}
		existingRRS := &route53v1alpha1.ResourceRecordSet{
			ObjectMeta: metav1.ObjectMeta{
				Name: recordName,
			},
		}
		fakeCli := fake.NewClientBuilder().WithScheme(s).WithObjects(existingVes, existingRRS).
			WithInterceptorFuncs(interceptor.Funcs{
				Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
					if _, ok := obj.(*route53v1alpha1.ResourceRecordSet); ok {
						return assert.AnError
					}
					return c.Patch(ctx, obj, patch, opts...)
				},
			}).Build()

		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(fakeCli, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				obj.Name = "test-pls"
				obj.Namespace = namer.GetNamespace()
				obj.Status.EndpointService.PrivateDNSName = recordName
				return nil
			})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(nil)
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.ErrorContains(t, err, "failed to pause crossplane ResourceRecordSet")
	})

	t.Run("full success - pls, vpce and resourcerecordset paused", func(t *testing.T) {
		s := runtime.NewScheme()
		ec2v1alpha1.SchemeBuilder.AddToScheme(s)
		route53v1alpha1.SchemeBuilder.AddToScheme(s)
		namer := utils.NewNamerWithClusterID("test-cluster", "", "")
		vpceSvcName := fmt.Sprintf("%s-%s", namer.GetNamespace(), "test-pls")
		recordName := "test.example.com"
		existingVes := &ec2v1alpha1.VPCEndpointServiceConfiguration{
			ObjectMeta: metav1.ObjectMeta{
				Name: vpceSvcName,
			},
		}
		existingRRS := &route53v1alpha1.ResourceRecordSet{
			ObjectMeta: metav1.ObjectMeta{
				Name: recordName,
			},
		}
		fakeCli := fake.NewClientBuilder().WithScheme(s).WithObjects(existingVes, existingRRS).Build()

		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(fakeCli, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				obj.Name = "test-pls"
				obj.Namespace = namer.GetNamespace()
				obj.Status.EndpointService.PrivateDNSName = recordName
				return nil
			})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(nil)
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.NoError(t, err)

		// Verify the crossplane VPCEndpointServiceConfiguration is annotated
		updatedVes := &ec2v1alpha1.VPCEndpointServiceConfiguration{}
		err = fakeCli.Get(context.Background(), client.ObjectKey{Name: vpceSvcName}, updatedVes)
		assert.NoError(t, err)
		assert.Equal(t, "true", updatedVes.GetAnnotations()[meta.AnnotationKeyReconciliationPaused])

		// Verify the crossplane ResourceRecordSet is annotated
		updatedRRS := &route53v1alpha1.ResourceRecordSet{}
		err = fakeCli.Get(context.Background(), client.ObjectKey{Name: recordName}, updatedRRS)
		assert.NoError(t, err)
		assert.Equal(t, "true", updatedRRS.GetAnnotations()[meta.AnnotationKeyReconciliationPaused])
	})

	t.Run("resourcerecordset name from annotation override", func(t *testing.T) {
		s := runtime.NewScheme()
		ec2v1alpha1.SchemeBuilder.AddToScheme(s)
		route53v1alpha1.SchemeBuilder.AddToScheme(s)
		namer := utils.NewNamerWithClusterID("test-cluster", "", "")
		vpceSvcName := fmt.Sprintf("%s-%s", namer.GetNamespace(), "test-pls")
		overrideRecordName := "override.example.com"
		existingVes := &ec2v1alpha1.VPCEndpointServiceConfiguration{
			ObjectMeta: metav1.ObjectMeta{
				Name: vpceSvcName,
			},
		}
		existingRRS := &route53v1alpha1.ResourceRecordSet{
			ObjectMeta: metav1.ObjectMeta{
				Name: overrideRecordName,
			},
		}
		fakeCli := fake.NewClientBuilder().WithScheme(s).WithObjects(existingVes, existingRRS).Build()

		ctl := gomock.NewController(t)
		plsrw := mockprivatelinkservice.NewMockPrivateLinkServiceReaderWriter(ctl)
		plssvc := NewPrivateLinkServiceService(fakeCli, plsrw, nil, nil, config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeEKS,
			},
		})
		plsrw.EXPECT().Get(context.Background(), client.ObjectKey{Name: "test-pls", Namespace: namer.GetNamespace()}, gomock.Any()).
			DoAndReturn(func(_ context.Context, _ client.ObjectKey, obj *v1alpha1.PrivateLinkService) error {
				obj.Labels = map[string]string{v1alpha1.ClusterIDKey: "test-cluster"}
				obj.Name = "test-pls"
				obj.Namespace = namer.GetNamespace()
				// The status PrivateDNSName differs from the annotation override; the annotation
				// should win when looking up the ResourceRecordSet.
				obj.Status.EndpointService.PrivateDNSName = "ignored.example.com"
				obj.Annotations = map[string]string{
					v1alpha1.AnnotationKeyOwnCrossplaneNameResourceRecordSet: overrideRecordName,
				}
				return nil
			})
		plsrw.EXPECT().Update(context.Background(), gomock.Any()).Return(nil)
		err := plssvc.PausePrivateLinkService(context.Background(), &regionalapiv1.PausePrivateLinkServiceRequest{
			ClusterId:            "test-cluster",
			PrivateLinkServiceId: "test-pls",
		})
		assert.NoError(t, err)

		// The overridden ResourceRecordSet should be annotated as paused.
		updatedRRS := &route53v1alpha1.ResourceRecordSet{}
		err = fakeCli.Get(context.Background(), client.ObjectKey{Name: overrideRecordName}, updatedRRS)
		assert.NoError(t, err)
		assert.Equal(t, "true", updatedRRS.GetAnnotations()[meta.AnnotationKeyReconciliationPaused])
	})
}

func TestEndpointServiceName(t *testing.T) {
	t.Run("with annotation", func(t *testing.T) {
		pls := &v1alpha1.PrivateLinkService{
			ObjectMeta: metav1.ObjectMeta{
				Annotations: map[string]string{
					v1alpha1.AnnotationKeyOwnCrossplaneNameVPCEService: "custom-name",
				},
				Name:      "test-pls",
				Namespace: "test-ns",
			},
		}
		assert.Equal(t, "custom-name", utils.GetEndpointServiceName(pls))
	})

	t.Run("without annotation", func(t *testing.T) {
		pls := &v1alpha1.PrivateLinkService{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "test-pls",
				Namespace: "test-ns",
			},
		}
		assert.Equal(t, "test-ns-test-pls", utils.GetEndpointServiceName(pls))
	})
}