package alicloud

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"

	openapi "github.com/alibabacloud-go/darabonba-openapi/v2/client"
	privatelink "github.com/alibabacloud-go/privatelink-20200415/v5/client"
	aliplv1alpha1 "github.com/crossplane-contrib/provider-upjet-alibabacloud/apis/privatelink/v1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/stretchr/testify/assert"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	agentv2 "github.com/tidbcloud/observability/agent-control-server/api/v2"
	pkgalicloud "github.com/tidbcloud/observability/agent-control-server/pkg/cloud/alicloud"
	"github.com/tidbcloud/observability/agent-control-server/pkg/config"
	"github.com/tidbcloud/observability/agent-control-server/pkg/consts"
	"github.com/tidbcloud/observability/agent-control-server/pkg/deploy"
	"github.com/tidbcloud/observability/agent-control-server/pkg/reconciler"
	"github.com/tidbcloud/observability/agent-control-server/pkg/testkits"
	"github.com/tidbcloud/observability/agent-control-server/pkg/util"
)

func makeKubernetesAgent() *agentv2.KubernetesAgent {
	return &agentv2.KubernetesAgent{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "test-agent",
			Namespace: "default",
			Annotations: map[string]string{
				consts.AnnotationKeyO11yEndpointServiceName:  "epsrv-mock",
				consts.AnnotationKeyO11yServiceOriginDNSName: "test.service.com",
			},
			Labels: map[string]string{
				agentv2.TagKeyTenant: "tenant-1",
			},
		},
		Spec: agentv2.KubernetesAgentSpec{
			KubernetesClusterName: "cluster-1",
			ACK: &agentv2.ACKConfig{
				VpcID:           "vpc-123",
				AssumeRoleArn:   "arn:aliyun:ram::123:role/role",
				SecurityGroupID: "sg-123",
			},
		},
	}
}

type runTestOpts struct {
	handler        http.HandlerFunc
	mutateAgent    func(*agentv2.KubernetesAgent)
	setupResources func(client.Client)
	serverless     bool
	region         string
	errMsg         string
}

func runTest(opts runTestOpts, testFn func(*reconciler.Context[*agentv2.KubernetesAgent], *agentv2.KubernetesAgent)) {
	var serverURL string
	if opts.handler != nil {
		ts := httptest.NewServer(opts.handler)
		defer ts.Close()
		serverURL = ts.Listener.Addr().String()
	}

	agent := makeKubernetesAgent()
	if opts.mutateAgent != nil {
		opts.mutateAgent(agent)
	}

	fakeCli := testkits.BuildClient(agent)
	if opts.setupResources != nil {
		opts.setupResources(fakeCli)
	}

	cfg := &config.RootConfig{
		O11yConfig: config.ObservabilityInfra{
			Version: 1,
			Region:  opts.region,
		},
	}
	if opts.serverless {
		cfg.Mode = config.ModeServerless
	}

	ctx := &reconciler.Context[*agentv2.KubernetesAgent]{
		Agent:    agent,
		Local:    fakeCli,
		AliCloud: testkits.NewAliCloudClient(serverURL),
		Config:   cfg,
	}

	testFn(ctx, agent)
}

func defaultResponses() map[string]any {
	return map[string]any{
		"ListVpcEndpointServicesByEndUser": map[string]any{
			"Services": []map[string]any{
				{
					"ServiceId": "epsrv-mock",
					"Zones":     []string{"zone-a", "zone-b"},
				},
			},
			"RequestId": "req-id-1",
		},
		"DescribeVSwitches": map[string]any{
			"VSwitches": map[string]any{
				"VSwitch": []map[string]any{
					{
						"VSwitchId": "vsw-mock",
						"ZoneId":    "zone-a",
					},
				},
			},
			"RequestId": "req-id-2",
		},
	}
}

func makeHandler(responses map[string]any) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		action := r.Header.Get("X-Acs-Action")
		if resp, ok := responses[action]; ok {
			json.NewEncoder(w).Encode(resp)
		} else {
			w.WriteHeader(http.StatusBadRequest)
		}
	}
}

func makeServerlessHandler() http.HandlerFunc {
	responses := defaultResponses()
	responses["ListVpcEndpoints"] = map[string]any{
		"Endpoints": []any{},
		"RequestId": "req-id-1",
	}
	responses["CreateVpcEndpoint"] = map[string]any{
		"EndpointId":     "ep-created-123",
		"EndpointDomain": "ep-created-123.epsrv-mock.cn-hangzhou.privatelink.aliyuncs.com",
		"RequestId":      "req-id-4",
	}
	return makeHandler(responses)
}

func makeEndpointHandler() http.HandlerFunc {
	responses := defaultResponses()
	responses["ListVpcEndpoints"] = map[string]any{
		"Endpoints": []map[string]any{
			{
				"EndpointId":     "ep-existing-123",
				"ServiceName":    "epsrv-mock",
				"VpcId":          "vpc-test",
				"EndpointType":   "Interface",
				"EndpointStatus": "Active",
			},
		},
		"RequestId": "req-id-1",
	}
	responses["GetVpcEndpointAttribute"] = map[string]any{
		"EndpointDomain": "ep-existing-123.epsrv-mock.cn-hangzhou.privatelink.aliyuncs.com",
		"RequestId":      "req-id-2",
	}
	responses["TagResources"] = map[string]any{
		"RequestId": "req-id-3",
	}
	return makeHandler(responses)
}

func makeClusterTagHandler() http.HandlerFunc {
	responses := defaultResponses()
	responses["ListVpcEndpoints"] = map[string]any{
		"Endpoints": []map[string]any{
			{
				"EndpointId":     "ep-existing-with-tag",
				"ServiceName":    "epsrv-mock",
				"VpcId":          "vpc-test",
				"EndpointType":   "Interface",
				"EndpointStatus": "Active",
				"Tags": []map[string]any{
					{
						"Key":   consts.LabelKeyRelatedCluster,
						"Value": "cluster-1",
					},
				},
			},
		},
		"RequestId": "req-id-1",
	}
	responses["GetVpcEndpointAttribute"] = map[string]any{
		"EndpointDomain": "ep-existing-with-tag.epsrv-mock.cn-hangzhou.privatelink.aliyuncs.com",
		"RequestId":      "req-id-2",
	}
	return makeHandler(responses)
}

func makeDeleteHandler() http.HandlerFunc {
	return makeHandler(map[string]any{
		"DeleteVpcEndpoint": map[string]any{
			"RequestId": "req-id-1",
		},
	})
}

func TestPrivateLinkTask_Create(t *testing.T) {
	tests := []struct {
		name           string
		serverless     bool
		handler        http.HandlerFunc
		mutateAgent    func(*agentv2.KubernetesAgent)
		setupResources func(client.Client)
		verify         func(*testing.T, client.Client, *agentv2.KubernetesAgent)
		errMsg         string
	}{
		{
			name:       "serverless: create new endpoint successfully",
			serverless: true,
			handler:    makeServerlessHandler(),
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				util.SetAnnotation(agent, consts.AnnotationKeyO11yServiceOriginDNSName, "test.o11y.aliyuncs.com")
				agent.Spec.ACK.VpcID = "vpc-test"
				agent.Spec.ACK.SecurityGroupID = "sg-test"
			},
			verify: func(t *testing.T, _ client.Client, agent *agentv2.KubernetesAgent) {
				endpointID := util.GetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName)
				assert.Equal(t, "ep-created-123", endpointID)
				domain := util.GetAnnotation(agent, consts.AnnotationKeyO11yServiceDNSName)
				assert.Equal(t, "ep-created-123.epsrv-mock.cn-hangzhou.privatelink.aliyuncs.com", domain)
			},
		},
		{
			name:       "serverless: reuse existing endpoint",
			serverless: true,
			handler:    makeEndpointHandler(),
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				util.SetAnnotation(agent, consts.AnnotationKeyO11yServiceOriginDNSName, "test.o11y.aliyuncs.com")
				agent.Spec.ACK.VpcID = "vpc-test"
				agent.Spec.ACK.SecurityGroupID = "sg-test"
			},
			verify: func(t *testing.T, _ client.Client, agent *agentv2.KubernetesAgent) {
				endpointID := util.GetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName)
				assert.Equal(t, "ep-existing-123", endpointID)
				domain := util.GetAnnotation(agent, consts.AnnotationKeyO11yServiceDNSName)
				assert.Equal(t, "ep-existing-123.epsrv-mock.cn-hangzhou.privatelink.aliyuncs.com", domain)
			},
		},
		{
			name:       "serverless: reuse existing endpoint with cluster tag (skip tagging)",
			serverless: true,
			handler:    makeClusterTagHandler(),
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				util.SetAnnotation(agent, consts.AnnotationKeyO11yServiceOriginDNSName, "test.o11y.aliyuncs.com")
				agent.Spec.ACK.VpcID = "vpc-test"
				agent.Spec.ACK.SecurityGroupID = "sg-test"
			},
			verify: func(t *testing.T, _ client.Client, agent *agentv2.KubernetesAgent) {
				endpointID := util.GetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName)
				assert.Equal(t, "ep-existing-with-tag", endpointID)
				domain := util.GetAnnotation(agent, consts.AnnotationKeyO11yServiceDNSName)
				assert.Equal(t, "ep-existing-with-tag.epsrv-mock.cn-hangzhou.privatelink.aliyuncs.com", domain)
			},
		},
		{
			name:       "serverless: skip if endpoint already exists and version not changed",
			serverless: true,
			handler:    makeServerlessHandler(),
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				util.SetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName, "ep-existing")
				util.SetAnnotation(agent, consts.AnnotationKeyO11yVersion, "1")
			},
			verify: func(t *testing.T, _ client.Client, agent *agentv2.KubernetesAgent) {
				assert.Equal(t, "ep-existing", util.GetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName))
			},
		},
		{
			name:       "serverless: endpoint service name not found",
			serverless: true,
			handler:    makeServerlessHandler(),
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				delete(agent.Annotations, consts.AnnotationKeyO11yEndpointServiceName)
				agent.Spec.ACK.SecurityGroupID = "sg-test"
			},
			errMsg: "endpoint service name not found",
		},
		{
			name:       "serverless: service dns name not found",
			serverless: true,
			handler:    makeServerlessHandler(),
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				delete(agent.Annotations, consts.AnnotationKeyO11yServiceOriginDNSName)
				agent.Spec.ACK.SecurityGroupID = "sg-test"
			},
			errMsg: "service dns name not found",
		},
		{
			name:    "upbound: success create new endpoint",
			handler: makeHandler(defaultResponses()),
			verify: func(t *testing.T, c client.Client, a *agentv2.KubernetesAgent) {
				epName := util.GetAnnotation(a, consts.AnnotationKeyO11ySvcInterfaceEndpointName)
				assert.NotEmpty(t, epName)

				var ep aliplv1alpha1.VpcEndpoint
				err := c.Get(context.Background(), client.ObjectKey{Name: epName}, &ep)
				assert.NoError(t, err)
				assert.Equal(t, "epsrv-mock", *ep.Spec.ForProvider.ServiceName)

				var epZone aliplv1alpha1.VpcEndpointZone
				err = c.Get(context.Background(), client.ObjectKey{Name: epName}, &epZone)
				assert.NoError(t, err)
				assert.Equal(t, "zone-a", *epZone.Spec.ForProvider.ZoneID)
			},
		},
		{
			name: "upbound: skip if endpoint already exists and version not changed",
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				util.SetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName, "existing-ep")
				util.SetAnnotation(agent, consts.AnnotationKeyO11yVersion, "1")
			},
			verify: func(t *testing.T, _ client.Client, a *agentv2.KubernetesAgent) {
				assert.Equal(t, "existing-ep", util.GetAnnotation(a, consts.AnnotationKeyO11ySvcInterfaceEndpointName))
			},
		},
		{
			name: "upbound: reuse existing endpoint",
			setupResources: func(c client.Client) {
				ep := &aliplv1alpha1.VpcEndpoint{
					ObjectMeta: metav1.ObjectMeta{
						Name: "reusable-ep",
					},
					Spec: aliplv1alpha1.VpcEndpointSpec{
						ForProvider: aliplv1alpha1.VpcEndpointParameters{
							ServiceName:  ptr.To("epsrv-mock"),
							VPCID:        ptr.To("vpc-123"),
							EndpointType: ptr.To(pkgalicloud.VpcEndpointTypeInterface),
						},
					},
				}
				ep.Status.SetConditions(xpv1.Available())
				_ = c.Create(context.Background(), ep)
			},
			verify: func(t *testing.T, _ client.Client, a *agentv2.KubernetesAgent) {
				assert.Equal(t, "reusable-ep", util.GetAnnotation(a, consts.AnnotationKeyO11ySvcInterfaceEndpointName))
			},
		},
		{
			name: "upbound: endpoint service name not found",
			mutateAgent: func(a *agentv2.KubernetesAgent) {
				delete(a.Annotations, consts.AnnotationKeyO11yEndpointServiceName)
			},
			errMsg: "endpoint service name not found",
		},
		{
			name: "upbound: service dns name not found",
			mutateAgent: func(a *agentv2.KubernetesAgent) {
				delete(a.Annotations, consts.AnnotationKeyO11yServiceOriginDNSName)
			},
			errMsg: "service dns name not found",
		},
		{
			name: "upbound: failed to create interface endpoint",
			handler: func(w http.ResponseWriter, r *http.Request) {
				w.WriteHeader(http.StatusInternalServerError)
			},
			errMsg: "failed to create interface endpoint",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			opts := runTestOpts{
				handler:        tt.handler,
				mutateAgent:    tt.mutateAgent,
				setupResources: tt.setupResources,
				serverless:     tt.serverless,
				region:         "cn-hangzhou",
				errMsg:         tt.errMsg,
			}
			runTest(opts, func(ctx *reconciler.Context[*agentv2.KubernetesAgent], agent *agentv2.KubernetesAgent) {
				task := NewPrivateLinkTask(ctx)
				assert.Equal(t, "PrivateLink", task.Name())
				assert.Equal(t, "cluster-1", task.ClusterMeta().K8sClusterName)

				res := task.Create(context.Background())
				if tt.errMsg != "" {
					assert.Contains(t, res.Error().Error(), tt.errMsg)
				} else {
					assert.Equal(t, deploy.Done(), res)
				}

				if tt.verify != nil {
					tt.verify(t, ctx.Local, agent)
				}
			})
		})
	}
}

func TestPrivateLinkTask_Delete(t *testing.T) {
	tests := []struct {
		name           string
		serverless     bool
		handler        http.HandlerFunc
		mutateAgent    func(*agentv2.KubernetesAgent)
		setupResources func(client.Client)
		errMsg         string
		verify         func(*testing.T, client.Client, *agentv2.KubernetesAgent)
	}{
		{
			name:       "serverless: delete endpoint successfully",
			serverless: true,
			handler:    makeDeleteHandler(),
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				util.SetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName, "ep-to-delete")
			},
			verify: func(t *testing.T, _ client.Client, agent *agentv2.KubernetesAgent) {
				endpointID := util.GetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName)
				assert.Equal(t, "", endpointID) // should be cleared
			},
		},
		{
			name:       "serverless: skip delete when endpoint id is empty",
			serverless: true,
			handler:    makeDeleteHandler(),
			verify: func(t *testing.T, _ client.Client, agent *agentv2.KubernetesAgent) {
				endpointID := util.GetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName)
				assert.Equal(t, "", endpointID)
			},
		},
		{
			name: "upbound: success delete",
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				// Keep default dnsName from makeKubernetesAgent
			},
			setupResources: func(c client.Client) {
				agent := makeKubernetesAgent()
				epName := util.InterfaceEndpointName(agent.ObjectMeta, "test.service.com", consts.O11yEndpointNamePrefixDataplaneNextGen)
				ep := &aliplv1alpha1.VpcEndpoint{
					ObjectMeta: metav1.ObjectMeta{Name: epName},
				}
				epZone := &aliplv1alpha1.VpcEndpointZone{
					ObjectMeta: metav1.ObjectMeta{Name: epName},
				}
				assert.NoError(t, c.Create(context.Background(), ep))
				assert.NoError(t, c.Create(context.Background(), epZone))
			},
			verify: func(t *testing.T, c client.Client, a *agentv2.KubernetesAgent) {
				epName := util.InterfaceEndpointName(a.ObjectMeta, "test.service.com", consts.O11yEndpointNamePrefixDataplaneNextGen)

				var ep aliplv1alpha1.VpcEndpoint
				err := c.Get(context.Background(), client.ObjectKey{Name: epName}, &ep)
				assert.True(t, apierrors.IsNotFound(err))

				var epZone aliplv1alpha1.VpcEndpointZone
				err = c.Get(context.Background(), client.ObjectKey{Name: epName}, &epZone)
				assert.True(t, apierrors.IsNotFound(err))
			},
		},
		{
			name: "upbound: resources already deleted",
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				// Keep default dnsName from makeKubernetesAgent, but don't setup resources
			},
		},
		{
			name: "upbound: no dns name configured",
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				delete(agent.Annotations, consts.AnnotationKeyO11yServiceOriginDNSName)
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			opts := runTestOpts{
				handler:        tt.handler,
				mutateAgent:    tt.mutateAgent,
				setupResources: tt.setupResources,
				serverless:     tt.serverless,
				region:         "ap-southeast-1",
				errMsg:         tt.errMsg,
			}
			runTest(opts, func(ctx *reconciler.Context[*agentv2.KubernetesAgent], agent *agentv2.KubernetesAgent) {
				task := NewPrivateLinkTask(ctx)
				res := task.Delete(context.Background())
				if tt.errMsg != "" {
					assert.Contains(t, res.Error().Error(), tt.errMsg)
				} else {
					assert.Equal(t, deploy.Done(), res)
				}

				if tt.verify != nil {
					tt.verify(t, ctx.Local, agent)
				}
			})
		})
	}
}

func TestPrivateLinkTask_CreateInterfaceEndpoint(t *testing.T) {
	tests := []struct {
		name        string
		region      string
		handler     func(w http.ResponseWriter, r *http.Request)
		mutateAgent func(a *agentv2.KubernetesAgent)
		verify      func(t *testing.T, c client.Client)
		errMsg      string
	}{
		{
			name:    "success",
			region:  "cn-hangzhou",
			handler: makeHandler(defaultResponses()),
			verify: func(t *testing.T, c client.Client) {
				var ep aliplv1alpha1.VpcEndpoint
				err := c.Get(context.Background(), client.ObjectKey{Name: "ep-test"}, &ep)
				assert.NoError(t, err)
				assert.Equal(t, "epsrv-mock", *ep.Spec.ForProvider.ServiceName)
				assert.Equal(t, "cn-hangzhou", *ep.Spec.ForProvider.Region)
				assert.Equal(t, "vpc-123", *ep.Spec.ForProvider.VPCID)
				assert.Equal(t, "sg-123", *ep.Spec.ForProvider.SecurityGroupIds[0])

				// Check tags
				assert.Equal(t, consts.CrossplaneValueVendor, *ep.Spec.ForProvider.Tags[consts.CrossplaneKeyVendor])
			},
		},
		{
			name:   "no region configured",
			errMsg: "AliCloud region is not configured",
		},
		{
			name:   "no endpoint service returned",
			region: "cn-hangzhou",
			handler: makeHandler(map[string]any{
				"ListVpcEndpointServicesByEndUser": map[string]any{
					"Services":  []any{},
					"RequestId": "req-id-1",
				},
			}),
			errMsg: "no endpoint service returned",
		},
		{
			name:   "no zones found for the endpoint service",
			region: "cn-hangzhou",
			handler: makeHandler(map[string]any{
				"ListVpcEndpointServicesByEndUser": map[string]any{
					"Services": []map[string]any{
						{
							"ServiceId": "epsrv-mock",
							"Zones":     []string{},
						},
					},
					"RequestId": "req-id-1",
				},
			}),
			errMsg: "no zones found for the endpoint service",
		},
		{
			name:   "no vswitch found",
			region: "cn-hangzhou",
			handler: makeHandler(map[string]any{
				"ListVpcEndpointServicesByEndUser": map[string]any{
					"Services": []map[string]any{
						{
							"ServiceId": "epsrv-mock",
							"Zones":     []string{"zone-a"},
						},
					},
					"RequestId": "req-id-1",
				},
				"DescribeVSwitches": map[string]any{
					"VSwitches": map[string]any{
						"VSwitch": []any{},
					},
					"RequestId": "req-id-2",
				},
			}),
			errMsg: "no vswitch found in vpc vpc-123",
		},
		{
			name:   "api error list services",
			region: "cn-hangzhou",
			handler: func(w http.ResponseWriter, r *http.Request) {
				w.WriteHeader(http.StatusInternalServerError)
			},
			errMsg: "failed to list vpc endpoint services",
		},
		{
			name:   "api error describe vswitches",
			region: "cn-hangzhou",
			handler: func(w http.ResponseWriter, r *http.Request) {
				action := r.Header.Get("X-Acs-Action")
				if action == "ListVpcEndpointServicesByEndUser" {
					json.NewEncoder(w).Encode(map[string]any{
						"Services": []map[string]any{
							{
								"ServiceId": "epsrv-mock",
								"Zones":     []string{"zone-a"},
							},
						},
						"RequestId": "req-id-1",
					})
				} else {
					w.WriteHeader(http.StatusInternalServerError)
				}
			},
			errMsg: "failed to describe vswitches",
		},
		{
			name:    "security group id is empty",
			region:  "cn-hangzhou",
			handler: makeHandler(defaultResponses()),
			mutateAgent: func(a *agentv2.KubernetesAgent) {
				a.Spec.ACK.SecurityGroupID = ""
			},
			errMsg: "security group id is empty",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var serverURL string
			if tt.handler != nil {
				ts := httptest.NewServer(http.HandlerFunc(tt.handler))
				defer ts.Close()
				serverURL = ts.Listener.Addr().String()
			}

			aliCloudCli := testkits.NewAliCloudClient(serverURL)
			agent := makeKubernetesAgent()
			if tt.mutateAgent != nil {
				tt.mutateAgent(agent)
			}
			fakeCli := testkits.BuildClient(agent)

			ctx := &reconciler.Context[*agentv2.KubernetesAgent]{
				Agent:    agent,
				Local:    fakeCli,
				AliCloud: aliCloudCli,
				Config: &config.RootConfig{
					O11yConfig: config.ObservabilityInfra{
						Region: tt.region,
					},
				},
			}

			task := NewPrivateLinkTask(ctx)
			err := task.createInterfaceEndpointUpbound(context.Background(), "epsrv-mock", "ep-test")
			if tt.errMsg != "" {
				assert.True(t, strings.Contains(err.Error(), tt.errMsg))
			} else {
				assert.NoError(t, err)
			}
			if tt.verify != nil {
				tt.verify(t, fakeCli)
			}
		})
	}
}

func TestPrivateLinkTask_PostCreate(t *testing.T) {
	tests := []struct {
		name           string
		serverless     bool
		mutateAgent    func(*agentv2.KubernetesAgent)
		setupResources func(client.Client)
		errMsg         string
		verify         func(*testing.T, *agentv2.KubernetesAgent)
	}{
		{
			name:       "serverless: post create should return done immediately",
			serverless: true,
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				util.SetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName, "ep-test")
			},
		},
		{
			name: "upbound: success",
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				util.SetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName, "test-ep-name")
			},
			setupResources: func(c client.Client) {
				ep := &aliplv1alpha1.VpcEndpoint{
					ObjectMeta: metav1.ObjectMeta{Name: "test-ep-name"},
					Status: aliplv1alpha1.VpcEndpointStatus{
						AtProvider: aliplv1alpha1.VpcEndpointObservation{
							EndpointDomain: ptr.To("ep-domain.com"),
						},
					},
				}
				ep.Status.SetConditions(xpv1.Available())
				assert.NoError(t, c.Create(context.Background(), ep))
			},
			verify: func(t *testing.T, agent *agentv2.KubernetesAgent) {
				assert.Equal(t, "ep-domain.com", util.GetAnnotation(agent, consts.AnnotationKeyO11yServiceDNSName))
			},
		},
		{
			name: "upbound: vpc endpoint not found",
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				util.SetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName, "test-ep-name")
			},
			errMsg: "failed to get vpc endpoint",
		},
		{
			name: "upbound: vpc endpoint not ready",
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				util.SetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName, "test-ep-name")
			},
			setupResources: func(c client.Client) {
				ep := &aliplv1alpha1.VpcEndpoint{
					ObjectMeta: metav1.ObjectMeta{Name: "test-ep-name"},
				}
				ep.Status.SetConditions(xpv1.Creating())
				assert.NoError(t, c.Create(context.Background(), ep))
			},
			errMsg: "vpc endpoint is not ready",
		},
		{
			name: "upbound: vpc endpoint domain is nil",
			mutateAgent: func(agent *agentv2.KubernetesAgent) {
				util.SetAnnotation(agent, consts.AnnotationKeyO11ySvcInterfaceEndpointName, "test-ep-name")
			},
			setupResources: func(c client.Client) {
				ep := &aliplv1alpha1.VpcEndpoint{
					ObjectMeta: metav1.ObjectMeta{Name: "test-ep-name"},
					Status: aliplv1alpha1.VpcEndpointStatus{
						AtProvider: aliplv1alpha1.VpcEndpointObservation{
							EndpointDomain: nil,
						},
					},
				}
				ep.Status.SetConditions(xpv1.Available())
				assert.NoError(t, c.Create(context.Background(), ep))
			},
			errMsg: "vpc endpoint domain is nil",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			opts := runTestOpts{
				mutateAgent:    tt.mutateAgent,
				setupResources: tt.setupResources,
				serverless:     tt.serverless,
				region:         "ap-southeast-1",
				errMsg:         tt.errMsg,
			}
			runTest(opts, func(ctx *reconciler.Context[*agentv2.KubernetesAgent], agent *agentv2.KubernetesAgent) {
				task := NewPrivateLinkTask(ctx)
				res := task.PostCreate(context.Background())
				if tt.errMsg != "" {
					assert.Contains(t, res.Error().Error(), tt.errMsg)
				} else {
					assert.Equal(t, deploy.Done(), res)
				}

				if tt.verify != nil {
					tt.verify(t, agent)
				}
			})
		})
	}
}

func TestCreateVpcEndpointNative(t *testing.T) {
	tests := []struct {
		name           string
		handler        func(w http.ResponseWriter, r *http.Request)
		expectError    bool
		expectedErrMsg string
		verifyResponse func(t *testing.T, endpointID, endpointDomain string)
	}{
		{
			name: "success_create_endpoint",
			handler: func(w http.ResponseWriter, r *http.Request) {
				// Verify request parameters
				action := r.Header.Get("X-Acs-Action")
				assert.Equal(t, "CreateVpcEndpoint", action)

				// Return successful response
				resp := map[string]any{
					"EndpointId":     "ep-abc123",
					"EndpointDomain": "ep-abc123.epsrv-test.cn-hangzhou.privatelink.aliyuncs.com",
					"RequestId":      "req-123",
				}
				w.Header().Set("Content-Type", "application/json")
				json.NewEncoder(w).Encode(resp)
			},
			expectError: false,
			verifyResponse: func(t *testing.T, endpointID, endpointDomain string) {
				assert.Equal(t, "ep-abc123", endpointID)
				assert.Equal(t, "ep-abc123.epsrv-test.cn-hangzhou.privatelink.aliyuncs.com", endpointDomain)
			},
		},
		{
			name: "missing_endpoint_id_in_response",
			handler: func(w http.ResponseWriter, r *http.Request) {
				resp := map[string]any{
					"EndpointDomain": "ep-abc123.epsrv-test.cn-hangzhou.privatelink.aliyuncs.com",
					"RequestId":      "req-123",
				}
				json.NewEncoder(w).Encode(resp)
			},
			expectError:    true,
			expectedErrMsg: "vpc endpoint id not found in response",
		},
		{
			name: "missing_endpoint_domain_in_response",
			handler: func(w http.ResponseWriter, r *http.Request) {
				resp := map[string]any{
					"EndpointId": "ep-abc123",
					"RequestId":  "req-123",
				}
				json.NewEncoder(w).Encode(resp)
			},
			expectError:    true,
			expectedErrMsg: "vpc endpoint domain not found in response",
		},
		{
			name: "api_returns_error",
			handler: func(w http.ResponseWriter, r *http.Request) {
				w.WriteHeader(http.StatusBadRequest)
				resp := map[string]any{
					"Code":      "InvalidParameter",
					"Message":   "The specified parameter is invalid.",
					"RequestId": "req-123",
				}
				json.NewEncoder(w).Encode(resp)
			},
			expectError:    true,
			expectedErrMsg: "failed to create vpc endpoint",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			server := httptest.NewServer(http.HandlerFunc(tt.handler))
			defer server.Close()

			cfg := &openapi.Config{
				Endpoint:        ptr.To(strings.ReplaceAll(server.URL, "http://", "")),
				AccessKeyId:     ptr.To("fake-ak"),
				AccessKeySecret: ptr.To("fake-sk"),
				Protocol:        ptr.To("HTTP"),
			}
			cli, err := privatelink.NewClient(cfg)
			assert.NoError(t, err)

			agent := makeKubernetesAgent()
			agent.Spec.ACK.VpcID = "vpc-123456"
			agent.Spec.ACK.SecurityGroupID = "sg-abc123"

			fakeCli := testkits.BuildClient(agent)
			ctx := &reconciler.Context[*agentv2.KubernetesAgent]{
				Agent: agent,
				Local: fakeCli,
				Config: &config.RootConfig{
					O11yConfig: config.ObservabilityInfra{
						Region: "cn-hangzhou",
					},
				},
			}
			task := NewPrivateLinkTask(ctx)

			endpointID, endpointDomain, err := task.createVpcEndpointNativeAPI(
				cli,
				"cn-hangzhou",
				"com.aliyuncs.privatelink.cn-hangzhou.epsrv-test",
				"ep-test-endpoint",
				map[string]*string{
					"tag-key":   ptr.To("tag-value"),
					"tag-key-2": ptr.To("tag-value-2"),
				},
				"cn-hangzhou-i",
				"vsw-xyz789",
			)

			if tt.expectError {
				assert.Error(t, err)
				if tt.expectedErrMsg != "" {
					assert.Contains(t, err.Error(), tt.expectedErrMsg)
				}
			} else {
				assert.NoError(t, err)
				if tt.verifyResponse != nil {
					tt.verifyResponse(t, endpointID, endpointDomain)
				}
			}
		})
	}
}