package action

import (
	"context"
	"os"
	"path/filepath"
	"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"
	dpapi "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	repomocks "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"
	"google.golang.org/grpc"
	"google.golang.org/protobuf/encoding/protojson"
	appsv1 "k8s.io/api/apps/v1"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/yaml"
)

func newCassandraProxyTestComp() *entity.Component {
	return &entity.Component{
		Id:            "1234567890",
		ComponentType: apiv1.ComponentType_CASSANDRA_PROXY,
		Phase:         apiv1.Component_CREATING,
		Version:       "v0.5.2",
		NodeQuota: &apiv1.NodeQuota{
			NodeSpec: &apiv1.NodeSpec{
				Cpu:      8,
				MemoryGi: 16,
			},
			NodeCount:    2,
			MinNodeCount: 2,
			MaxNodeCount: 6,
		},
		Labels: map[string]string{
			cassandraProxyHostIdLabel:   "11111111-2222-3333-4444-555555555555",
			cassandraProxyNodeSpecLabel: "8C16G",
		},
	}
}

func newCassandraProxyTestState(image string) *entity.CassandraProxyState {
	return &entity.CassandraProxyState{
		ClusterID: "1234567890",
		HostID:    "11111111-2222-3333-4444-555555555555",
		Image:     image,
	}
}

func newCassandraProxyTestSP() *entity.SharedPool {
	sp := &entity.SharedPool{}
	sp.Id = "9999"
	return sp
}

func newCassandraProxyTestCluster() *entity.TiDBCluster {
	return &entity.TiDBCluster{
		Id:              "1234567890",
		TenantId:        "tenant-x",
		GatewayIdentity: "tidb-1234567890",
	}
}

func newCassandraProxyTestDataPlane(withPCA bool) *dpapi.DataPlane {
	dp := &dpapi.DataPlane{
		Domain: "dev.tidbcloud.com",
	}
	if withPCA {
		dp.Pca = &dpapi.DataPlane_PCA{}
	}
	return dp
}

func TestBuildCassandraProxyTempArgs_HappyPath(t *testing.T) {
	comp := newCassandraProxyTestComp()
	sp := newCassandraProxyTestSP()
	cl := newCassandraProxyTestCluster()
	dp := newCassandraProxyTestDataPlane(false)

	args, err := buildCassandraProxyTempArgs("ghcr.io/pingcap/cassandra-proxy:v0.5.2", comp, sp, cl, dp, "host-uuid", "8C16G", "1.32")
	require.NoError(t, err)

	for _, k := range []string{
		"Namespace", "ClusterId", "DeploymentName",
		"ConfigMapName", "CertSecretName", "HpaName", "Image", "Version",
		"MinReplicas", "MaxReplicas", "NodeSpec", "NodeVersion",
		"HostId", "TiProxyHost", "PartitionDefaultSize",
		"CertIssuerName", "CertIssuerKind", "CertIssuerGroup",
		"CertDNSNamesYAML", "CostTagsYAML",
		"CPU", "Memory", "RequestCPU", "RequestMemory",
	} {
		assert.NotEmpty(t, args[k], "expected %q to be set", k)
	}

	assert.Equal(t, "tidb9999", args["Namespace"])
	assert.Equal(t, "1234567890", args["ClusterId"])
	assert.Equal(t, "cassandra-1234567890", args["DeploymentName"])
	assert.Equal(t, "cassandra-1234567890", args["ConfigMapName"])
	assert.Equal(t, "cassandra-1234567890", args["CertSecretName"])
	assert.Equal(t, "cassandra-1234567890", args["HpaName"])
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:v0.5.2", args["Image"])
	assert.Equal(t, "host-uuid", args["HostId"])
	assert.Equal(t, "8C16G", args["NodeSpec"])
	assert.Equal(t, "1.32", args["NodeVersion"])
	assert.Equal(t, "8", args["CPU"])
	assert.Equal(t, "4", args["RequestCPU"])
	assert.Equal(t, "16Gi", args["Memory"])
	assert.Equal(t, "8Gi", args["RequestMemory"])
	assert.Equal(t, "db-1234567890-tiproxy.tidb9999.svc.cluster.local:4000", args["TiProxyHost"])
	assert.Equal(t, "40", args["PartitionDefaultSize"])
	// dev dataplane (no PCA) -> per-namespace self-signed Issuer (matches tiproxy)
	assert.Equal(t, "tidb-components-self-signed-issuer", args["CertIssuerName"])
	assert.Equal(t, "Issuer", args["CertIssuerKind"])
	// SAN list ends with the wildcard covering the customer-facing PL DNS.
	assert.Contains(t, args["CertDNSNamesYAML"], "*.tidb-1234567890.dev.tidbcloud.com")
}

func TestBuildCassandraProxyTempArgs_PremIssuer(t *testing.T) {
	args, err := buildCassandraProxyTempArgs(
		"img",
		newCassandraProxyTestComp(),
		newCassandraProxyTestSP(),
		newCassandraProxyTestCluster(),
		newCassandraProxyTestDataPlane(true),
		"host", "8C16G", "1.32",
	)
	require.NoError(t, err)
	assert.Equal(t, "cluster-issuer", args["CertIssuerName"])
	assert.Equal(t, "AWSPCAClusterIssuer", args["CertIssuerKind"])
	assert.Equal(t, "awspca.cert-manager.io", args["CertIssuerGroup"])
}

func TestResolveCassandraProxyStateImage(t *testing.T) {
	state := newCassandraProxyTestState(" ghcr.io/pingcap/cassandra-proxy:state ")
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:state", resolveCassandraProxyStateImage(
		state,
		&config.ImageConfig{CassandraProxyImage: "ghcr.io/pingcap/cassandra-proxy:executor"},
	))

	state.Image = ""
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:executor", resolveCassandraProxyStateImage(
		state,
		&config.ImageConfig{CassandraProxyImage: " ghcr.io/pingcap/cassandra-proxy:executor "},
	))
	assert.Empty(t, resolveCassandraProxyStateImage(state, nil))
}

func TestRenderYAMLDashList(t *testing.T) {
	out := renderYAMLDashList([]string{"a", "b", "c"})
	expected := "- a\n- b\n- c"
	assert.Equal(t, expected, out)
	assert.Empty(t, renderYAMLDashList(nil))
}

func TestRenderYAMLLabelMap_Stable(t *testing.T) {
	in := map[string]string{"b": "2", "a": "1"}
	out := renderYAMLLabelMap(in)
	expected := "a: \"1\"\nb: \"2\""
	assert.Equal(t, expected, out)
}

// TestApplyCassandraProxy_RenderTemplate verifies that each per-kind template
// renders into a single valid K8s YAML document. Splitting the cassandra_proxy
// template into cert / configmap / deployment files is required because
// k8s.GenerateUnstructuredFromYaml only decodes the first document of a
// multi-doc YAML and silently drops the rest.
func TestApplyCassandraProxy_RenderTemplate(t *testing.T) {
	args, err := buildCassandraProxyTempArgs(
		"ghcr.io/pingcap/cassandra-proxy:v0.5.2",
		newCassandraProxyTestComp(),
		newCassandraProxyTestSP(),
		newCassandraProxyTestCluster(),
		newCassandraProxyTestDataPlane(false),
		"host-uuid",
		"8C16G",
		"1.32",
	)
	require.NoError(t, err)

	tl := initCassandraProxyTemplateLib(t)

	wantKinds := map[string]string{
		"cassandra_proxy_cert":      "Certificate",
		"cassandra_proxy_configmap": "ConfigMap",
		"cassandra_proxy":           "Deployment",
		"cassandra_proxy_hpa":       "HorizontalPodAutoscaler",
	}
	wantName := entity.GetCassandraProxyResourceName("1234567890")
	for name, wantKind := range wantKinds {
		out, err := tl.Render(name, args)
		require.NoError(t, err, "%s render failed", name)
		require.NotEmpty(t, out, "%s render empty", name)
		// Each .templ must hold exactly one YAML doc (no `---`); a stray
		// separator would silently lose a kind in the apply path.
		assert.NotContains(t, out, "\n---", "%s contains a `---` doc separator", name)
		var obj map[string]any
		require.NoError(t, yaml.Unmarshal([]byte(out), &obj), "%s yaml parse failed", name)
		assert.Equal(t, wantKind, obj["kind"], "%s rendered the wrong kind", name)
		metadata, ok := obj["metadata"].(map[string]any)
		require.True(t, ok, "%s metadata missing", name)
		assert.Equal(t, wantName, metadata["name"], "%s rendered the wrong resource name", name)
	}

	out, err := tl.Render("cassandra_proxy", args)
	require.NoError(t, err)
	var deployment appsv1.Deployment
	require.NoError(t, yaml.Unmarshal([]byte(out), &deployment))
	assert.Nil(t, deployment.Spec.Replicas)
	require.Len(t, deployment.Spec.Template.Spec.Containers, 1)
	container := deployment.Spec.Template.Spec.Containers[0]
	assert.Equal(t, "8C16G", deployment.Spec.Template.Spec.NodeSelector["node-spec"])
	assert.Equal(t, "8", container.Resources.Limits.Cpu().String())
	assert.Equal(t, "16Gi", container.Resources.Limits.Memory().String())
	assert.Equal(t, "4", container.Resources.Requests.Cpu().String())
	assert.Equal(t, "8Gi", container.Resources.Requests.Memory().String())

	out, err = tl.Render("cassandra_proxy_hpa", args)
	require.NoError(t, err)
	var hpa autoscalingv2.HorizontalPodAutoscaler
	require.NoError(t, yaml.Unmarshal([]byte(out), &hpa))
	require.NotNil(t, hpa.Spec.MinReplicas)
	assert.Equal(t, int32(2), *hpa.Spec.MinReplicas)
	assert.Equal(t, int32(6), hpa.Spec.MaxReplicas)
}

func TestApplyCassandraProxyAction_InvokeDoesNotMarkStateAvailable(t *testing.T) {
	sharedPool := newCassandraProxyTestSP()
	sharedPool.DataplaneName = "dp-1"
	sharedPool.Version = "v8.5.0"
	sharedPool.ComponentGroups = []*entity.Component{newCassandraProxyTestComp()}
	sharedPoolBytes, err := protojson.Marshal(sharedPool)
	require.NoError(t, err)

	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	require.NoError(t, appsv1.AddToScheme(scheme))
	require.NoError(t, autoscalingv2.AddToScheme(scheme))
	require.NoError(t, certmanagerv1.AddToScheme(scheme))
	existingReplicas := int32(5)
	existingDeployment := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: entity.GetDataPlaneNamespace(sharedPool.GetId()),
			Name:      entity.GetCassandraProxyResourceName("1234567890"),
		},
		Spec: appsv1.DeploymentSpec{Replicas: &existingReplicas},
	}
	remoteClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(existingDeployment).Build()

	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            sharedPool.GetId(),
			DataplaneName: sharedPool.GetDataplaneName(),
		},
	}, nil).Once()

	cassandraStateRepo := repomocks.NewCassandraProxyStateRepo(t)
	cassandraStateRepo.EXPECT().LoadByClusterID(mock.Anything, "1234567890", false).
		Return(newCassandraProxyTestState("ghcr.io/pingcap/cassandra-proxy:state"), nil).Once()

	tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
	tidbClusterRepo.EXPECT().Load(mock.Anything, "1234567890", false).Return(newCassandraProxyTestCluster(), nil).Once()

	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.EXPECT().GetClient("dp-1").Return(remoteClient, nil).Once()

	templateRepo := repomocks.NewTemplateRegistryRepo(t)
	registerCassandraProxyTemplateMocks(t, templateRepo)
	templateVersionSvc := templateversion.NewService(templateRepo, 3)

	action := NewApplyCassandraProxyAction(
		sharedPoolRepo,
		cassandraStateRepo,
		tidbClusterRepo,
		dataplaneManager,
		&stubCassandraProxyDataPlaneClient{},
		templateVersionSvc,
		&config.ImageConfig{CassandraProxyImage: "ghcr.io/pingcap/cassandra-proxy:executor"},
	)

	err = action.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:   actions.ActionArgString(sharedPool.GetId()),
		workflow.SharedPoolKey:     actions.ActionArgString(string(sharedPoolBytes)),
		workflow.ComponentIdsKey:   actions.ActionArgString("1234567890"),
		workflow.ComponentTypesKey: actions.ActionArgString(entity.GetComponentTypeName(newCassandraProxyTestComp())),
	})
	require.NoError(t, err)

	var deployment appsv1.Deployment
	require.NoError(t, remoteClient.Get(context.Background(), client.ObjectKey{
		Namespace: entity.GetDataPlaneNamespace(sharedPool.GetId()),
		Name:      entity.GetCassandraProxyResourceName("1234567890"),
	}, &deployment))
	require.NotNil(t, deployment.Spec.Replicas)
	assert.Equal(t, int32(5), *deployment.Spec.Replicas)
	require.Len(t, deployment.Spec.Template.Spec.Containers, 1)
	assert.Equal(t, "ghcr.io/pingcap/cassandra-proxy:state", deployment.Spec.Template.Spec.Containers[0].Image)
}

func TestApplyCassandraProxyAction_InvokeNoTargetNoopsBeforeImageCheck(t *testing.T) {
	sharedPool := newCassandraProxyTestSP()
	sharedPool.DataplaneName = "dp-1"
	sharedPool.Version = "v8.5.0"
	tiproxyComp := &entity.Component{
		Id:            "1234567890",
		ComponentType: apiv1.ComponentType_TIPROXY,
		Phase:         apiv1.Component_DELETING,
	}
	sharedPool.ComponentGroups = []*entity.Component{tiproxyComp}
	sharedPoolBytes, err := protojson.Marshal(sharedPool)
	require.NoError(t, err)

	action := NewApplyCassandraProxyAction(
		repomocks.NewSharedPoolRepo(t),
		repomocks.NewCassandraProxyStateRepo(t),
		repomocks.NewTiDBClusterRepo(t),
		dataplanemocks.NewManager(t),
		&stubCassandraProxyDataPlaneClient{},
		nil,
		&config.ImageConfig{},
	)

	err = action.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:   actions.ActionArgString(sharedPool.GetId()),
		workflow.SharedPoolKey:     actions.ActionArgString(string(sharedPoolBytes)),
		workflow.ComponentIdsKey:   actions.ActionArgString(tiproxyComp.GetId()),
		workflow.ComponentTypesKey: actions.ActionArgString(entity.GetComponentTypeName(tiproxyComp)),
	})
	require.NoError(t, err)
}

func TestApplyCassandraProxyAction_InvokeDeleteIgnoresMissingResources(t *testing.T) {
	sharedPool := newCassandraProxyTestSP()
	sharedPool.DataplaneName = "dp-1"
	sharedPool.Version = "v8.5.0"
	comp := newCassandraProxyTestComp()
	comp.Phase = apiv1.Component_DELETING
	sharedPool.ComponentGroups = []*entity.Component{comp}
	sharedPoolBytes, err := protojson.Marshal(sharedPool)
	require.NoError(t, err)

	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	require.NoError(t, appsv1.AddToScheme(scheme))
	require.NoError(t, autoscalingv2.AddToScheme(scheme))
	require.NoError(t, certmanagerv1.AddToScheme(scheme))
	certSecret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      entity.GetCassandraProxyResourceName("1234567890"),
			Namespace: entity.GetDataPlaneNamespace(sharedPool.GetId()),
		},
	}
	remoteClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(certSecret).Build()

	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            sharedPool.GetId(),
			DataplaneName: sharedPool.GetDataplaneName(),
		},
	}, nil).Once()

	cassandraStateRepo := repomocks.NewCassandraProxyStateRepo(t)
	cassandraStateRepo.EXPECT().Delete(mock.Anything, "1234567890", "11111111-2222-3333-4444-555555555555").Return(nil).Once()

	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.EXPECT().GetClient("dp-1").Return(remoteClient, nil).Once()

	action := NewApplyCassandraProxyAction(
		sharedPoolRepo,
		cassandraStateRepo,
		repomocks.NewTiDBClusterRepo(t),
		dataplaneManager,
		&stubCassandraProxyDataPlaneClient{},
		nil,
		&config.ImageConfig{},
	)

	err = action.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:   actions.ActionArgString(sharedPool.GetId()),
		workflow.SharedPoolKey:     actions.ActionArgString(string(sharedPoolBytes)),
		workflow.ComponentIdsKey:   actions.ActionArgString("1234567890"),
		workflow.ComponentTypesKey: actions.ActionArgString(entity.GetComponentTypeName(comp)),
	})
	require.NoError(t, err)

	err = remoteClient.Get(context.Background(), client.ObjectKeyFromObject(certSecret), &corev1.Secret{})
	assert.True(t, apierrors.IsNotFound(err), "expected cassandra-proxy cert secret to be deleted")
}

func TestApplyCassandraProxyAction_InvokeDeleteSkipsTemplateArgCompletion(t *testing.T) {
	sharedPool := newCassandraProxyTestSP()
	sharedPool.DataplaneName = "dp-1"
	sharedPool.Version = "v8.5.0"
	comp := newCassandraProxyTestComp()
	comp.Phase = apiv1.Component_DELETING
	comp.NodeQuota.NodeSpec = &apiv1.NodeSpec{Cpu: 2, MemoryGi: 4}
	comp.Labels[cassandraProxyNodeSpecLabel] = "2C4G"
	sharedPool.ComponentGroups = []*entity.Component{comp}
	sharedPoolBytes, err := protojson.Marshal(sharedPool)
	require.NoError(t, err)

	name := entity.GetCassandraProxyResourceName(comp.GetId())
	namespace := entity.GetDataPlaneNamespace(sharedPool.GetId())
	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	require.NoError(t, appsv1.AddToScheme(scheme))
	require.NoError(t, autoscalingv2.AddToScheme(scheme))
	require.NoError(t, certmanagerv1.AddToScheme(scheme))
	objects := []client.Object{
		&certmanagerv1.Certificate{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}},
		&corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}},
		&appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}},
		&autoscalingv2.HorizontalPodAutoscaler{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}},
		&corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}},
	}
	remoteClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build()

	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            sharedPool.GetId(),
			DataplaneName: sharedPool.GetDataplaneName(),
		},
	}, nil).Once()

	cassandraStateRepo := repomocks.NewCassandraProxyStateRepo(t)
	cassandraStateRepo.EXPECT().Delete(mock.Anything, comp.GetId(), comp.GetLabels()[cassandraProxyHostIdLabel]).Return(nil).Once()

	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.EXPECT().GetClient("dp-1").Return(remoteClient, nil).Once()

	action := NewApplyCassandraProxyAction(
		sharedPoolRepo,
		cassandraStateRepo,
		repomocks.NewTiDBClusterRepo(t),
		dataplaneManager,
		&stubCassandraProxyDataPlaneClient{},
		nil,
		&config.ImageConfig{},
	)

	err = action.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:   actions.ActionArgString(sharedPool.GetId()),
		workflow.SharedPoolKey:     actions.ActionArgString(string(sharedPoolBytes)),
		workflow.ComponentIdsKey:   actions.ActionArgString(comp.GetId()),
		workflow.ComponentTypesKey: actions.ActionArgString(entity.GetComponentTypeName(comp)),
	})
	require.NoError(t, err)

	for _, obj := range objects {
		err = remoteClient.Get(context.Background(), client.ObjectKeyFromObject(obj), obj)
		assert.True(t, apierrors.IsNotFound(err), "expected %T to be deleted", obj)
	}
}

func TestApplyCassandraProxyAction_InvokePreLoopFailureMarksUnavailable(t *testing.T) {
	sharedPool := newCassandraProxyTestSP()
	sharedPool.DataplaneName = "dp-1"
	sharedPool.Version = "v8.5.0"
	sharedPool.ComponentGroups = []*entity.Component{newCassandraProxyTestComp()}
	sharedPoolBytes, err := protojson.Marshal(sharedPool)
	require.NoError(t, err)

	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            sharedPool.GetId(),
			DataplaneName: sharedPool.GetDataplaneName(),
		},
	}, nil).Once()
	cassandraStateRepo := repomocks.NewCassandraProxyStateRepo(t)
	cassandraStateRepo.EXPECT().LoadByClusterID(mock.Anything, "1234567890", false).
		Return(newCassandraProxyTestState(""), nil).Once()
	cassandraStateRepo.EXPECT().UpdatePhase(
		mock.Anything,
		"1234567890",
		"11111111-2222-3333-4444-555555555555",
		apiv1.Component_UNAVAILABLE,
	).Return(nil).Once()
	tidbClusterRepo := repomocks.NewTiDBClusterRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.EXPECT().GetClient("dp-1").Return(fake.NewClientBuilder().Build(), nil).Once()

	action := NewApplyCassandraProxyAction(
		sharedPoolRepo,
		cassandraStateRepo,
		tidbClusterRepo,
		dataplaneManager,
		&stubCassandraProxyDataPlaneClient{},
		nil,
		&config.ImageConfig{},
	)

	err = action.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:   actions.ActionArgString(sharedPool.GetId()),
		workflow.SharedPoolKey:     actions.ActionArgString(string(sharedPoolBytes)),
		workflow.ComponentIdsKey:   actions.ActionArgString("1234567890"),
		workflow.ComponentTypesKey: actions.ActionArgString(entity.GetComponentTypeName(newCassandraProxyTestComp())),
	})
	require.Error(t, err)
	assert.Contains(t, err.Error(), "cassandra-proxy image is not configured")
}

func TestApplyCassandraProxyAction_InvokeDeletesState(t *testing.T) {
	sharedPool := newCassandraProxyTestSP()
	sharedPool.DataplaneName = "dp-1"
	sharedPool.Version = "v8.5.0"
	comp := newCassandraProxyTestComp()
	comp.Phase = apiv1.Component_DELETING
	sharedPool.ComponentGroups = []*entity.Component{comp}
	sharedPoolBytes, err := protojson.Marshal(sharedPool)
	require.NoError(t, err)

	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	require.NoError(t, appsv1.AddToScheme(scheme))
	require.NoError(t, autoscalingv2.AddToScheme(scheme))
	require.NoError(t, certmanagerv1.AddToScheme(scheme))
	remoteClient := fake.NewClientBuilder().WithScheme(scheme).Build()

	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            sharedPool.GetId(),
			DataplaneName: sharedPool.GetDataplaneName(),
		},
	}, nil).Once()

	cassandraStateRepo := repomocks.NewCassandraProxyStateRepo(t)
	cassandraStateRepo.EXPECT().Delete(mock.Anything, "1234567890", "11111111-2222-3333-4444-555555555555").Return(nil).Once()

	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.EXPECT().GetClient("dp-1").Return(remoteClient, nil).Once()

	action := NewApplyCassandraProxyAction(
		sharedPoolRepo,
		cassandraStateRepo,
		repomocks.NewTiDBClusterRepo(t),
		dataplaneManager,
		&stubCassandraProxyDataPlaneClient{},
		nil,
		&config.ImageConfig{},
	)

	err = action.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:   actions.ActionArgString(sharedPool.GetId()),
		workflow.SharedPoolKey:     actions.ActionArgString(string(sharedPoolBytes)),
		workflow.ComponentIdsKey:   actions.ActionArgString("1234567890"),
		workflow.ComponentTypesKey: actions.ActionArgString(entity.GetComponentTypeName(comp)),
	})
	require.NoError(t, err)
}

func TestWaitCassandraProxyReadyAction_ReadyMarksAvailable(t *testing.T) {
	sharedPool := newCassandraProxyTestSP()
	sharedPool.DataplaneName = "dp-1"
	sharedPool.ComponentGroups = []*entity.Component{newCassandraProxyTestComp()}
	sharedPoolBytes, err := protojson.Marshal(sharedPool)
	require.NoError(t, err)

	replicas := int32(2)
	deploy := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:       entity.GetCassandraProxyResourceName("1234567890"),
			Namespace:  entity.GetDataPlaneNamespace(sharedPool.GetId()),
			Generation: 3,
		},
		Spec: appsv1.DeploymentSpec{Replicas: &replicas},
		Status: appsv1.DeploymentStatus{
			ObservedGeneration: 3,
			UpdatedReplicas:    2,
			AvailableReplicas:  2,
			ReadyReplicas:      2,
		},
	}
	scheme := runtime.NewScheme()
	require.NoError(t, appsv1.AddToScheme(scheme))
	remoteClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(deploy).Build()

	cassandraStateRepo := repomocks.NewCassandraProxyStateRepo(t)
	cassandraStateRepo.EXPECT().UpdatePhase(mock.Anything, "1234567890", "11111111-2222-3333-4444-555555555555", apiv1.Component_AVAILABLE).Return(nil).Once()

	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.EXPECT().GetClient("dp-1").Return(remoteClient, nil).Once()

	action := NewWaitCassandraProxyReadyAction(dataplaneManager, cassandraStateRepo).(*WaitCassandraProxyReadyAction)
	action.waitInterval = time.Millisecond
	action.timeout = 50 * time.Millisecond

	err = action.Invoke(actions.ActionArgs{
		workflow.SharedPoolKey:     actions.ActionArgString(string(sharedPoolBytes)),
		workflow.ComponentIdsKey:   actions.ActionArgString("1234567890"),
		workflow.ComponentTypesKey: actions.ActionArgString(entity.GetComponentTypeName(newCassandraProxyTestComp())),
	})
	require.NoError(t, err)
}

func TestWaitCassandraProxyReadyAction_NotReadyTimeoutMarksUnavailable(t *testing.T) {
	sharedPool := newCassandraProxyTestSP()
	sharedPool.DataplaneName = "dp-1"
	sharedPool.ComponentGroups = []*entity.Component{newCassandraProxyTestComp()}
	sharedPoolBytes, err := protojson.Marshal(sharedPool)
	require.NoError(t, err)

	replicas := int32(2)
	deploy := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:       entity.GetCassandraProxyResourceName("1234567890"),
			Namespace:  entity.GetDataPlaneNamespace(sharedPool.GetId()),
			Generation: 3,
		},
		Spec: appsv1.DeploymentSpec{Replicas: &replicas},
		Status: appsv1.DeploymentStatus{
			ObservedGeneration: 3,
			UpdatedReplicas:    1,
			AvailableReplicas:  1,
			ReadyReplicas:      1,
		},
	}
	scheme := runtime.NewScheme()
	require.NoError(t, appsv1.AddToScheme(scheme))
	remoteClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(deploy).Build()

	cassandraStateRepo := repomocks.NewCassandraProxyStateRepo(t)
	cassandraStateRepo.EXPECT().UpdatePhase(mock.Anything, "1234567890", "11111111-2222-3333-4444-555555555555", apiv1.Component_UNAVAILABLE).Return(nil).Once()

	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.EXPECT().GetClient("dp-1").Return(remoteClient, nil).Once()

	action := NewWaitCassandraProxyReadyAction(dataplaneManager, cassandraStateRepo).(*WaitCassandraProxyReadyAction)
	action.waitInterval = time.Millisecond
	action.timeout = 5 * time.Millisecond

	err = action.Invoke(actions.ActionArgs{
		workflow.SharedPoolKey:     actions.ActionArgString(string(sharedPoolBytes)),
		workflow.ComponentIdsKey:   actions.ActionArgString("1234567890"),
		workflow.ComponentTypesKey: actions.ActionArgString(entity.GetComponentTypeName(newCassandraProxyTestComp())),
	})
	require.Error(t, err)
	assert.Contains(t, err.Error(), "timed out waiting")
}

func initCassandraProxyTemplateLib(t *testing.T) *render.TemplateLibrary {
	t.Helper()
	tplPath := "../../../deployments/templates"
	lib := render.NewTemplateLibrary()
	for _, name := range []string{
		"cassandra_proxy_cert",
		"cassandra_proxy_configmap",
		"cassandra_proxy",
		"cassandra_proxy_hpa",
	} {
		data, err := os.ReadFile(filepath.Join(tplPath, name+".templ"))
		require.NoError(t, err, "read template %s", name)
		lib.SafeNewTemplate(name, string(data))
	}
	return lib
}

func registerCassandraProxyTemplateMocks(t *testing.T, templateRepo *repomocks.TemplateRegistryRepo) {
	t.Helper()
	tplPath := "../../../deployments/templates"
	for _, name := range []string{
		"cassandra_proxy_cert",
		"cassandra_proxy_configmap",
		"cassandra_proxy",
		"cassandra_proxy_hpa",
	} {
		data, err := os.ReadFile(filepath.Join(tplPath, name+".templ"))
		require.NoError(t, err, "read template %s", name)
		content := templateversion.NormalizeTemplateContent(string(data))
		version := templateversion.ComputeTemplateVersion(content)
		templateRepo.EXPECT().ListByTemplateName(mock.Anything, name).Return([]*entity.TemplateRegistry{{
			TemplateVersionInfo: entity.TemplateVersionInfo{
				TemplateName: name,
				Version:      version,
				Status:       apiv1.TemplateVersionStatus_TEMPLATE_VERSION_STATUS_AVAILABLE,
			},
			Content: content,
		}}, nil).Once()
	}
}

type stubCassandraProxyDataPlaneClient struct {
	dpapi.DataPlaneServiceClient
}

func (s *stubCassandraProxyDataPlaneClient) ListNodePools(ctx context.Context, in *dpapi.ListNodePoolsRequest, opts ...grpc.CallOption) (*dpapi.ListNodePoolsResponse, error) {
	return &dpapi.ListNodePoolsResponse{NodePools: []*dpapi.NodePool{{
		CpuCore:    8,
		MemoryGi:   16,
		K8SVersion: "1.32",
	}}}, nil
}

func (s *stubCassandraProxyDataPlaneClient) GetDataPlane(ctx context.Context, in *dpapi.GetDataPlaneRequest, opts ...grpc.CallOption) (*dpapi.DataPlane, error) {
	return newCassandraProxyTestDataPlane(false), nil
}