package action

import (
	"context"
	"os"
	"path/filepath"
	"slices"
	"testing"
	"time"

	certmanagerv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/cert"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/template/render"
	templateversion "github.com/tidbcloud/cluster-service-ng/pkg/template/version"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/intstr"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/yaml"
)

var clusterAgentTestTemplArgs = map[string]string{
	"ClusterId": "1233438346465886",
}

var clusterAgentDeploymentTestTemplArgs = map[string]string{
	"ClusterId":              "1233438346465886",
	"Provider":               "aws",
	"Region":                 "us-west-2",
	"ClusterAgentImage":      "cluster-agent:v1.0.0",
	"ReplaceNodeWarmUpImage": "registry.example.com/third-party/pause:3.9",
	"AuditLogImage":          "audit-ng:v1.0.0",
	"TiKVImage":              "tikv:v8.0.0",
	"PDImage":                "pd:v8.0.0",
	"ImagePullPolicy":        "IfNotPresent",
}

func Test_RenderClusterAgentNetworkPolicy(t *testing.T) {
	templateLib := initClusterAgentTemplate(t, "cluster_agent_network_policy")
	yamlStr, err := templateLib.Render("cluster_agent_network_policy", clusterAgentTestTemplArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)
	assert.YAMLEq(t, `apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  annotations:
    pingcap.com/component: cluster-agent
  labels:
    app.kubernetes.io/instance: cluster-agent
    app.kubernetes.io/name: cluster-agent
  name: cluster-agent-network-policy
  namespace: tidb1233438346465886
spec:
  podSelector:
    matchLabels:
      app.kubernetes.io/name: cluster-agent
  policyTypes:
  - Ingress
  ingress:
  - from:
    - namespaceSelector:
        matchLabels:
          kubernetes.io/metadata.name: tidb-admin
      podSelector:
        matchLabels:
          app.kubernetes.io/name: kong
    ports:
    - protocol: TCP
      port: 8080
`, yamlStr)
}

func Test_RenderClusterAgentRoleIncludesNodeUpdateVerb(t *testing.T) {
	templateLib := initClusterAgentTemplate(t, "cluster_agent_role")
	yamlStr, err := templateLib.Render("cluster_agent_role", clusterAgentTestTemplArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	var role struct {
		Rules []struct {
			Resources []string `yaml:"resources"`
			Verbs     []string `yaml:"verbs"`
		} `yaml:"rules"`
	}
	require.NoError(t, yaml.Unmarshal([]byte(yamlStr), &role))

	for _, rule := range role.Rules {
		if slices.Contains(rule.Resources, "nodes") {
			assert.Contains(t, rule.Verbs, "update")
			return
		}
	}

	t.Fatal("expected a ClusterRole rule for nodes")
}

func Test_RenderClusterAgentDeploymentTLSConsistency(t *testing.T) {
	templateLib := initClusterAgentTemplate(t, "cluster_agent")
	yamlStr, err := templateLib.Render("cluster_agent", clusterAgentDeploymentTestTemplArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	var deployment appsv1.Deployment
	err = yaml.Unmarshal([]byte(yamlStr), &deployment)
	require.NoError(t, err)

	require.Len(t, deployment.Spec.Template.Spec.Containers, 1)
	assert.Equal(t, appsv1.RecreateDeploymentStrategyType, deployment.Spec.Strategy.Type)
	container := deployment.Spec.Template.Spec.Containers[0]
	assert.Equal(t, []string{
		"/cluster-agent",
		"--provider",
		"aws",
		"--region",
		"us-west-2",
		"--tls-cert-file",
		"/server-ssl/tls.crt",
		"--tls-key-file",
		"/server-ssl/tls.key",
	}, container.Command)

	for _, arg := range []string{"--audit-enabled", "--audit-file-path", "--audit-max-body-size"} {
		assert.NotContains(t, container.Command, arg)
	}

	for _, mount := range container.VolumeMounts {
		assert.NotEqual(t, "audit-log", mount.Name)
	}
	assert.Contains(t, container.Env, corev1.EnvVar{
		Name:  "REPLACE_NODE_WARMUP_IMAGE",
		Value: "registry.example.com/third-party/pause:3.9",
	})

	for _, volume := range deployment.Spec.Template.Spec.Volumes {
		assert.NotEqual(t, "audit-log", volume.Name)
	}

	tests := []struct {
		name  string
		probe *corev1.Probe
	}{
		{
			name:  "liveness probe uses https endpoint",
			probe: container.LivenessProbe,
		},
		{
			name:  "readiness probe uses https endpoint",
			probe: container.ReadinessProbe,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			require.NotNil(t, tt.probe)
			require.NotNil(t, tt.probe.HTTPGet)
			assert.Equal(t, corev1.URISchemeHTTPS, tt.probe.HTTPGet.Scheme)
			assert.Equal(t, intstr.FromString("https"), tt.probe.HTTPGet.Port)
		})
	}
}

func Test_RenderClusterAgentDeploymentAuditSettings(t *testing.T) {
	templateLib := initClusterAgentTemplate(t, "cluster_agent")
	templArgs := copyMap(clusterAgentDeploymentTestTemplArgs)
	templArgs["AuditEnabled"] = "true"
	templArgs["AuditLogEnabled"] = "true"
	templArgs["AuditStorageClass"] = "cloud-ssd"
	templArgs["AuditStorageSize"] = "1Gi"
	auditConfig, err := genClusterAgentAuditSidecarConfig(&clusterAgentKongAuditConfig{
		EnableLog:   true,
		BucketURL:   "s3://bucket/dataplane-id",
		Region:      "us-west-2",
		MaxSize:     20,
		MaxLifetime: 3600,
		AWS: clusterAgentAuditAWS{
			RoleARN:    "arn:aws:iam::123456789012:role/test-role",
			ExternalID: "ext-123",
		},
	})
	require.NoError(t, err)
	templArgs["AuditSidecarConfig"] = auditConfig
	yamlStr, err := templateLib.Render("cluster_agent", templArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	var deployment appsv1.Deployment
	err = yaml.Unmarshal([]byte(yamlStr), &deployment)
	require.NoError(t, err)

	require.Len(t, deployment.Spec.Template.Spec.Containers, 2)
	container := deployment.Spec.Template.Spec.Containers[0]

	require.Contains(t, container.Command, "--audit-enabled")
	require.Contains(t, container.Command, "--audit-file-path")
	require.Contains(t, container.Command, "/var/log/cluster-agent/ops-audit.log")
	require.Contains(t, container.Command, "--audit-max-body-size")
	require.Contains(t, container.Command, "0")

	require.Contains(t, deployment.Spec.Template.Annotations, "tidbcloud.com/audit-config")
	annotation := deployment.Spec.Template.Annotations["tidbcloud.com/audit-config"]
	require.Contains(t, annotation, "mode: agent")
	require.Contains(t, annotation, "subPath: cluster-agent")
	require.Contains(t, annotation, "bucketURL: s3://bucket/dataplane-id")
	require.Contains(t, annotation, "region: us-west-2")
	require.Contains(t, annotation, "maxLifetime: 3600")

	require.NotNil(t, container.SecurityContext)
	require.NotNil(t, container.SecurityContext.ReadOnlyRootFilesystem)
	assert.True(t, *container.SecurityContext.ReadOnlyRootFilesystem)

	require.Len(t, deployment.Spec.Template.Spec.Containers, 2)
	var auditContainer *corev1.Container
	for i := range deployment.Spec.Template.Spec.Containers {
		if deployment.Spec.Template.Spec.Containers[i].Name == "auditlog" {
			auditContainer = &deployment.Spec.Template.Spec.Containers[i]
			break
		}
	}
	require.NotNil(t, auditContainer)
	assert.Equal(t, []string{
		"/audit-ng",
		"server",
		"--config=/etc/podinfo/audit-config.yaml",
	}, auditContainer.Command)

	foundAuditMount := false
	for _, mount := range container.VolumeMounts {
		if mount.Name == "audit-log" {
			foundAuditMount = true
			assert.Equal(t, "/var/log/cluster-agent", mount.MountPath)
		}
	}
	assert.True(t, foundAuditMount)

	foundAuditVolume := false
	foundPodInfoVolume := false
	for _, volume := range deployment.Spec.Template.Spec.Volumes {
		if volume.Name == "audit-log" {
			foundAuditVolume = true
			require.NotNil(t, volume.Ephemeral)
			require.NotNil(t, volume.Ephemeral.VolumeClaimTemplate)
			require.NotNil(t, volume.Ephemeral.VolumeClaimTemplate.Spec.StorageClassName)
			require.Equal(t, "cloud-ssd", *volume.Ephemeral.VolumeClaimTemplate.Spec.StorageClassName)
			require.Len(t, volume.Ephemeral.VolumeClaimTemplate.Spec.AccessModes, 1)
			assert.Equal(t, corev1.ReadWriteOnce, volume.Ephemeral.VolumeClaimTemplate.Spec.AccessModes[0])
			assert.Equal(t, "1Gi", volume.Ephemeral.VolumeClaimTemplate.Spec.Resources.Requests.Storage().String())
		}
		if volume.Name == "podinfo" {
			foundPodInfoVolume = true
			require.NotNil(t, volume.DownwardAPI)
		}
	}
	assert.True(t, foundAuditVolume)
	assert.True(t, foundPodInfoVolume)
}

func Test_RenderClusterAgentDeploymentAuditInfraWithoutAppLogging(t *testing.T) {
	templateLib := initClusterAgentTemplate(t, "cluster_agent")
	templArgs := copyMap(clusterAgentDeploymentTestTemplArgs)
	templArgs["AuditEnabled"] = "true"
	templArgs["AuditStorageClass"] = "cloud-ssd"
	templArgs["AuditStorageSize"] = "1Gi"
	auditConfig, err := genClusterAgentAuditSidecarConfig(&clusterAgentKongAuditConfig{
		EnableLog: false,
	})
	require.NoError(t, err)
	templArgs["AuditSidecarConfig"] = auditConfig

	yamlStr, err := templateLib.Render("cluster_agent", templArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	var deployment appsv1.Deployment
	err = yaml.Unmarshal([]byte(yamlStr), &deployment)
	require.NoError(t, err)

	require.Len(t, deployment.Spec.Template.Spec.Containers, 2)
	mainContainer := deployment.Spec.Template.Spec.Containers[0]
	for _, arg := range []string{"--audit-enabled", "--audit-file-path", "--audit-max-body-size"} {
		assert.NotContains(t, mainContainer.Command, arg)
	}
	require.Contains(t, deployment.Spec.Template.Annotations, "tidbcloud.com/audit-config")
}

func Test_RenderClusterAgentDeploymentPrivilegedTiDBDeactivationWait(t *testing.T) {
	templateLib := initClusterAgentTemplate(t, "cluster_agent")

	t.Run("flag rendered when arg set", func(t *testing.T) {
		templArgs := copyMap(clusterAgentDeploymentTestTemplArgs)
		templArgs["PrivilegedTiDBDeactivationWait"] = "5s"
		yamlStr, err := templateLib.Render("cluster_agent", templArgs)
		require.NoError(t, err)
		var deployment appsv1.Deployment
		require.NoError(t, yaml.Unmarshal([]byte(yamlStr), &deployment))
		require.Len(t, deployment.Spec.Template.Spec.Containers, 1)
		container := deployment.Spec.Template.Spec.Containers[0]
		require.Contains(t, container.Command, "--privileged-tidb-deactivation-wait")
		require.Contains(t, container.Command, "5s")
	})

	t.Run("flag absent when arg unset", func(t *testing.T) {
		yamlStr, err := templateLib.Render("cluster_agent", clusterAgentDeploymentTestTemplArgs)
		require.NoError(t, err)
		var deployment appsv1.Deployment
		require.NoError(t, yaml.Unmarshal([]byte(yamlStr), &deployment))
		container := deployment.Spec.Template.Spec.Containers[0]
		assert.NotContains(t, container.Command, "--privileged-tidb-deactivation-wait")
	})
}

func TestTemplateExecuteWithVersionResolverUpdatesExistingClusterAgentDeploymentSpec(t *testing.T) {
	ctx := context.Background()
	namespace := "tidb" + clusterAgentDeploymentTestTemplArgs["ClusterId"]
	existing := &unstructured.Unstructured{}
	existing.SetAPIVersion("apps/v1")
	existing.SetKind("Deployment")
	existing.SetName("cluster-agent")
	existing.SetNamespace(namespace)
	existing.Object["spec"] = map[string]any{
		"replicas": int64(1),
		"selector": map[string]any{
			"matchLabels": map[string]any{
				"app.kubernetes.io/instance": "cluster-agent",
				"app.kubernetes.io/name":     "cluster-agent",
			},
		},
		"template": map[string]any{
			"metadata": map[string]any{
				"labels": map[string]any{
					"app.kubernetes.io/instance": "cluster-agent",
					"app.kubernetes.io/name":     "cluster-agent",
				},
			},
			"spec": map[string]any{
				"containers": []any{
					map[string]any{
						"name":  "cluster-agent",
						"image": "old-cluster-agent:v0.0.1",
					},
				},
			},
		},
	}
	remoteClient := fake.NewClientBuilder().WithObjects(existing).Build()

	err := templateExecuteWithVersionResolver(
		ctx,
		remoteClient,
		"cluster_agent",
		"",
		clusterAgentDeploymentTestTemplArgs,
		false,
		newClusterAgentTemplateVersionSvc(t),
	)
	require.NoError(t, err)

	var got unstructured.Unstructured
	got.SetAPIVersion("apps/v1")
	got.SetKind("Deployment")
	require.NoError(t, remoteClient.Get(ctx, types.NamespacedName{Name: "cluster-agent", Namespace: namespace}, &got))
	strategyType, found, err := unstructured.NestedString(got.Object, "spec", "strategy", "type")
	require.NoError(t, err)
	require.True(t, found)
	assert.Equal(t, string(appsv1.RecreateDeploymentStrategyType), strategyType)
}

func copyMap(in map[string]string) map[string]string {
	out := make(map[string]string, len(in))
	for k, v := range in {
		out[k] = v
	}
	return out
}

func TestPopulateClusterAgentOptionalTemplateArgs(t *testing.T) {
	tests := []struct {
		name             string
		bizConfig        *config.BizConfig
		wantKong         string
		wantAuditFlag    bool
		wantStorageClass string
		wantStorageSize  string
	}{
		{
			name:             "nil biz config",
			bizConfig:        nil,
			wantKong:         "",
			wantAuditFlag:    false,
			wantStorageClass: "",
			wantStorageSize:  "",
		},
		{
			name:             "empty biz config",
			bizConfig:        &config.BizConfig{},
			wantKong:         "",
			wantAuditFlag:    false,
			wantStorageClass: "",
			wantStorageSize:  "",
		},
		{
			name: "kong plugins only",
			bizConfig: &config.BizConfig{
				KongDataplanePlugins: "plugin-a,plugin-b",
			},
			wantKong:         "plugin-a,plugin-b",
			wantAuditFlag:    false,
			wantStorageClass: "",
			wantStorageSize:  "",
		},
		{
			name: "audit enabled only",
			bizConfig: &config.BizConfig{
				ClusterAgentAudit: config.ClusterAgentAuditConfig{
					Enabled: true,
				},
			},
			wantKong:         "",
			wantAuditFlag:    true,
			wantStorageClass: "cloud-ssd",
			wantStorageSize:  "1Gi",
		},
		{
			name: "both kong plugins and audit enabled",
			bizConfig: &config.BizConfig{
				KongDataplanePlugins: "plugin-c",
				ClusterAgentAudit: config.ClusterAgentAuditConfig{
					Enabled:      true,
					StorageClass: "local-ssd",
					StorageSize:  "2Gi",
				},
			},
			wantKong:         "plugin-c",
			wantAuditFlag:    true,
			wantStorageClass: "local-ssd",
			wantStorageSize:  "2Gi",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			templArgs := map[string]string{}
			populateClusterAgentOptionalTemplateArgs(templArgs, tt.bizConfig)

			if tt.wantKong == "" {
				_, exists := templArgs["KongPlugins"]
				require.False(t, exists)
			} else {
				require.Equal(t, tt.wantKong, templArgs["KongPlugins"])
			}

			_, auditExists := templArgs["AuditEnabled"]
			require.Equal(t, tt.wantAuditFlag, auditExists)
			require.Equal(t, tt.wantStorageClass, templArgs["AuditStorageClass"])
			require.Equal(t, tt.wantStorageSize, templArgs["AuditStorageSize"])
		})
	}
}

func TestPopulateClusterAgentPrivilegedTiDBArgs(t *testing.T) {
	tests := []struct {
		name      string
		bizConfig *config.BizConfig
		wantSet   bool
		wantValue string
	}{
		{name: "nil biz config", bizConfig: nil, wantSet: false},
		{name: "zero wait", bizConfig: &config.BizConfig{}, wantSet: false},
		{name: "negative wait", bizConfig: &config.BizConfig{PrivilegedTiDBDeactivationWait: -1 * time.Second}, wantSet: false},
		{
			name:      "5s wait",
			bizConfig: &config.BizConfig{PrivilegedTiDBDeactivationWait: 5 * time.Second},
			wantSet:   true,
			wantValue: "5s",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			templArgs := map[string]string{}
			populateClusterAgentPrivilegedTiDBArgs(templArgs, tt.bizConfig)
			got, ok := templArgs["PrivilegedTiDBDeactivationWait"]
			assert.Equal(t, tt.wantSet, ok)
			if tt.wantSet {
				assert.Equal(t, tt.wantValue, got)
			}
		})
	}
}

func TestPopulateClusterAgentAuditTemplateArgs(t *testing.T) {
	tests := []struct {
		name               string
		kongCfg            *clusterAgentKongAuditConfig
		wantAuditLogEnable bool
		wantAnnotationText string
	}{
		{
			name: "audit log disabled in configmap",
			kongCfg: &clusterAgentKongAuditConfig{
				EnableLog: false,
			},
			wantAuditLogEnable: false,
			wantAnnotationText: "bucketURL: \"\"",
		},
		{
			name: "audit log enabled in configmap",
			kongCfg: &clusterAgentKongAuditConfig{
				EnableLog:   true,
				BucketURL:   "s3://bucket/dataplane-id",
				Region:      "us-west-2",
				MaxSize:     15,
				MaxLifetime: 7200,
				AWS: clusterAgentAuditAWS{
					RoleARN:    "arn:aws:iam::123456789012:role/test-role",
					ExternalID: "ext-123",
				},
			},
			wantAuditLogEnable: true,
			wantAnnotationText: "maxLifetime: 7200",
		},
		{
			name: "audit log enabled but maxSize and maxLifetime absent keeps defaults",
			kongCfg: &clusterAgentKongAuditConfig{
				EnableLog: true,
				BucketURL: "s3://bucket/dp-id",
				Region:    "us-east-1",
				// MaxSize and MaxLifetime are zero (absent from YAML).
				// The sidecar config must retain the default maxSize (10)
				// instead of writing 0, which audit-ng treats as "disable rotation".
			},
			wantAuditLogEnable: true,
			wantAnnotationText: "maxSize: 10",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			templArgs := map[string]string{}
			err := populateClusterAgentAuditTemplateArgs(templArgs, &config.ImageConfig{
				Registry:      "registry.example.com",
				AuditLogImage: "tidbcloud/audit-ng:v1.0.0",
			}, tt.kongCfg)
			require.NoError(t, err)

			assert.Equal(t, "registry.example.com/tidbcloud/audit-ng:v1.0.0", templArgs["AuditLogImage"])
			require.Contains(t, templArgs["AuditSidecarConfig"], tt.wantAnnotationText)

			_, exists := templArgs["AuditLogEnabled"]
			assert.Equal(t, tt.wantAuditLogEnable, exists)
		})
	}
}

func TestGetClusterAgentKongAuditConfig(t *testing.T) {
	tests := map[string]struct {
		existingConfigMaps []*corev1.ConfigMap
		expectBucketURL    string
		expectRegion       string
		expectRoleARN      string
		expectExternalID   string
		expectEnableLog    bool
		expectError        bool
		expectErrorContain string
	}{
		"configmap exists with valid data": {
			existingConfigMaps: []*corev1.ConfigMap{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      clusterAgentAuditConfigMapName,
						Namespace: clusterAgentAuditConfigNamespace,
					},
					Data: map[string]string{
						clusterAgentAuditConfigKey: "enableLog: true\nbucketURL: s3://test-bucket\nregion: us-west-2\nmaxSize: 10\nmaxLifetime: 86400",
					},
				},
			},
			expectEnableLog: true,
			expectBucketURL: "s3://test-bucket",
			expectRegion:    "us-west-2",
		},
		"configmap exists with aws config": {
			existingConfigMaps: []*corev1.ConfigMap{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      clusterAgentAuditConfigMapName,
						Namespace: clusterAgentAuditConfigNamespace,
					},
					Data: map[string]string{
						clusterAgentAuditConfigKey: "enableLog: true\nbucketURL: s3://test-bucket\nregion: us-west-2\naws:\n  roleARN: arn:aws:iam::123456789012:role/test-role\n  externalID: ext-123",
					},
				},
			},
			expectEnableLog:  true,
			expectBucketURL:  "s3://test-bucket",
			expectRegion:     "us-west-2",
			expectRoleARN:    "arn:aws:iam::123456789012:role/test-role",
			expectExternalID: "ext-123",
		},
		"configmap exists with whitespace to trim": {
			existingConfigMaps: []*corev1.ConfigMap{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      clusterAgentAuditConfigMapName,
						Namespace: clusterAgentAuditConfigNamespace,
					},
					Data: map[string]string{
						clusterAgentAuditConfigKey: "enableLog: true\nbucketURL:   s3://trimmed-bucket  \nregion:  ap-northeast-1  ",
					},
				},
			},
			expectEnableLog: true,
			expectBucketURL: "s3://trimmed-bucket",
			expectRegion:    "ap-northeast-1",
		},
		"configmap not found": {
			expectError:        true,
			expectErrorContain: "not found",
		},
		"configmap missing key": {
			existingConfigMaps: []*corev1.ConfigMap{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      clusterAgentAuditConfigMapName,
						Namespace: clusterAgentAuditConfigNamespace,
					},
					Data: map[string]string{
						"other-key": "some-value",
					},
				},
			},
			expectError:        true,
			expectErrorContain: "does not contain key",
		},
		"configmap invalid yaml": {
			existingConfigMaps: []*corev1.ConfigMap{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      clusterAgentAuditConfigMapName,
						Namespace: clusterAgentAuditConfigNamespace,
					},
					Data: map[string]string{
						clusterAgentAuditConfigKey: "invalid: yaml: content: [",
					},
				},
			},
			expectError:        true,
			expectErrorContain: "failed to unmarshal",
		},
	}

	for name, tt := range tests {
		t.Run(name, func(t *testing.T) {
			scheme := runtime.NewScheme()
			require.NoError(t, corev1.AddToScheme(scheme))

			fakeClientBuilder := fake.NewClientBuilder().WithScheme(scheme)
			if len(tt.existingConfigMaps) > 0 {
				objs := make([]runtime.Object, 0, len(tt.existingConfigMaps))
				for _, cm := range tt.existingConfigMaps {
					objs = append(objs, cm)
				}
				fakeClientBuilder = fakeClientBuilder.WithRuntimeObjects(objs...)
			}

			cfg, err := getClusterAgentKongAuditConfig(context.Background(), fakeClientBuilder.Build())
			if tt.expectError {
				require.Error(t, err)
				require.Contains(t, err.Error(), tt.expectErrorContain)
				return
			}

			require.NoError(t, err)
			require.NotNil(t, cfg)
			assert.Equal(t, tt.expectEnableLog, cfg.EnableLog)
			assert.Equal(t, tt.expectBucketURL, cfg.BucketURL)
			assert.Equal(t, tt.expectRegion, cfg.Region)
			assert.Equal(t, tt.expectRoleARN, cfg.AWS.RoleARN)
			assert.Equal(t, tt.expectExternalID, cfg.AWS.ExternalID)
		})
	}
}

func initClusterAgentTemplate(t *testing.T, templateNames ...string) *render.TemplateLibrary {
	t.Helper()
	tplPath := "../../../deployments/templates"
	templateLib := render.NewTemplateLibrary()

	for _, tpl := range templateNames {
		data, err := os.ReadFile(filepath.Join(tplPath, tpl+".templ"))
		require.NoError(t, err, "Failed to read template file: %s", tpl)
		templateLib.SafeNewTemplate(tpl, string(data))
	}
	return templateLib
}

func newClusterAgentTemplateVersionSvc(t *testing.T) *templateversion.Service {
	t.Helper()
	templateMap, err := templateversion.LoadTemplatesFromPath("../../../deployments/templates")
	require.NoError(t, err)

	mockRepo := repomnocks.NewTemplateRegistryRepo(t)
	mockRepo.On("ListByTemplateName", mock.Anything, mock.Anything).Return(
		func(_ context.Context, name string) []*entity.TemplateRegistry {
			content, ok := templateMap[name]
			if !ok {
				return nil
			}
			return []*entity.TemplateRegistry{{
				TemplateVersionInfo: apiv1.TemplateVersionInfo{
					TemplateName: name,
					Version:      "test-sha256",
					Status:       apiv1.TemplateVersionStatus_TEMPLATE_VERSION_STATUS_AVAILABLE,
				},
				Content: content,
			}}
		},
		func(_ context.Context, _ string) error {
			return nil
		},
	)

	return templateversion.NewService(mockRepo, 3)
}

func TestApplyClusterAgentActionInvokeFailsWhenAuditConfigMissing(t *testing.T) {
	sharedPool := newClusterAgentSharedPoolForTest()

	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))

	remoteClient := fake.NewClientBuilder().WithScheme(scheme).Build()

	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil).Once()

	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.EXPECT().GetClient(sharedPool.GetDataplaneName()).Return(remoteClient, nil).Once()

	a := NewApplyClusterAgentAction(
		dataplaneManager,
		sharedPoolRepo,
		nil,
		&config.ImageConfig{
			Registry:          "registry.example.com",
			RepoPrefix:        "pingcap",
			ClusterAgentImage: "cluster-agent:v1.0.0",
			AuditLogImage:     "audit-ng:v1.0.0",
		},
		&config.BizConfig{
			ClusterAgentAudit: config.ClusterAgentAuditConfig{
				Enabled: true,
			},
		},
	)

	err := a.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey: actions.ActionArgString(sharedPool.GetId()),
		utils.TraceIdKey:         actions.ActionArgString(""),
	})
	require.Error(t, err)
	require.Contains(t, err.Error(), "failed to get cluster agent audit config")
}

func TestApplyClusterAgentActionInvokeCreateServerCert(t *testing.T) {
	tests := []struct {
		name            string
		preExistingCert bool
	}{
		{
			name:            "create cert when not exists",
			preExistingCert: false,
		},
		{
			name:            "reuse cert when already exists",
			preExistingCert: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sharedPool := newClusterAgentSharedPoolForTest()
			expectedCert := cert.ClusterAgentServerCert(sharedPool.GetDataPlaneNamespace())

			scheme := runtime.NewScheme()
			require.NoError(t, certmanagerv1.AddToScheme(scheme))

			fakeClientBuilder := fake.NewClientBuilder().WithScheme(scheme)
			if tt.preExistingCert {
				fakeClientBuilder = fakeClientBuilder.WithObjects(expectedCert.DeepCopy())
			}
			remoteClient := fakeClientBuilder.Build()

			sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
			sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil).Once()

			dataplaneManager := dataplanemocks.NewManager(t)
			dataplaneManager.EXPECT().GetClient(sharedPool.GetDataplaneName()).Return(remoteClient, nil).Once()

			a := NewApplyClusterAgentAction(
				dataplaneManager,
				sharedPoolRepo,
				nil, // templateVersionSvc
				&config.ImageConfig{
					Registry:          "registry.example.com",
					RepoPrefix:        "pingcap",
					ClusterAgentImage: "cluster-agent:v1.0.0",
				},
				&config.BizConfig{},
			)

			err := a.Invoke(actions.ActionArgs{
				workflow.SharedPoolIdKey: actions.ActionArgString(sharedPool.GetId()),
				utils.TraceIdKey:         actions.ActionArgString(""),
			})
			require.Error(t, err)

			actualCert := &certmanagerv1.Certificate{}
			err = remoteClient.Get(context.Background(), types.NamespacedName{
				Name:      expectedCert.GetName(),
				Namespace: expectedCert.GetNamespace(),
			}, actualCert)
			require.NoError(t, err)
			assert.Equal(t, expectedCert.Spec.SecretName, actualCert.Spec.SecretName)
		})
	}
}

func newClusterAgentSharedPoolForTest() *entity.SharedPool {
	return &entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            "sp-cluster-agent-test",
			DataplaneName: "dataplane-test",
			Provider: &apiv1.Provider{
				Type:   apiv1.ProviderType_AWS,
				Region: "us-west-2",
			},
			ComponentGroups: []*apiv1.Component{
				{
					Id:            "tikv-1",
					ComponentType: apiv1.ComponentType_TIKV,
					Configuration: &apiv1.Component_Configuration{
						ImageRepo: "pingcap/tikv",
						ImageTag:  "v8.0.0",
					},
				},
				{
					Id:            "pd-1",
					ComponentType: apiv1.ComponentType_PD,
					Configuration: &apiv1.Component_Configuration{
						ImageRepo: "pingcap/pd",
						ImageTag:  "v8.0.0",
					},
				},
			},
		},
	}
}