package service

import (
	"context"
	"errors"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/cluster-agent/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
)

const (
	replaceNodeTestClusterID   = "cluster123"
	replaceNodeTestNamespace   = "tidbpool123"
	replaceNodeTestNodeName    = "node-1"
	replaceNodeTestWarmUpNode  = "node-2"
	replaceNodeTestOperationID = "op001"
	replaceNodeTestWarmUpImage = "registry.example.com/third-party/pause:3.9"
)

func TestAgentService_PrepareReplaceNode(t *testing.T) {
	for _, kind := range []string{"PD", "TiKV", "TiDB", "TiFlash", "TiProxy", "TSO", "Scheduling", "TiKVWorker"} {
		t.Run(kind, func(t *testing.T) {
			t.Setenv("NAMESPACE", replaceNodeTestNamespace)
			t.Setenv(replaceNodeWarmUpImageEnvKey, replaceNodeTestWarmUpImage)
			pod := makeReadyOperatorPod("target-pod", kind, "instance-"+kind, replaceNodeTestNodeName, "old-uid")
			pod.Spec.NodeSelector = map[string]string{"pool": "standard"}
			pod.Spec.Tolerations = []corev1.Toleration{{Key: "dedicated", Operator: corev1.TolerationOpExists}}
			pod.Spec.SchedulerName = "custom-scheduler"
			pod.Spec.PriorityClassName = "high"
			pod.Spec.TopologySpreadConstraints = []corev1.TopologySpreadConstraint{{
				MaxSkew:           1,
				TopologyKey:       corev1.LabelTopologyZone,
				WhenUnsatisfiable: corev1.DoNotSchedule,
			}}
			pod.Spec.Containers[0].Resources.Requests = corev1.ResourceList{
				corev1.ResourceCPU:    resource.MustParse("2"),
				corev1.ResourceMemory: resource.MustParse("4Gi"),
			}
			cli := newReplaceNodeTestClient(t,
				pod,
				makeOperatorInstance(t, kind, "instance-"+kind, nil, nil, false),
				makeNode(replaceNodeTestNodeName),
			)
			svc := NewAgentService(cli, nil)

			resp, err := svc.PrepareReplaceNode(context.Background(), makeReplaceNodeReqByPod("target-pod"))

			require.NoError(t, err)
			assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_PREPARING, resp.Phase)
			assert.Equal(t, "waiting for warm-up pod to run", resp.Message)
			assert.Equal(t, "target-pod", resp.PodName)
			assert.Equal(t, kind, resp.ComponentKind)
			assert.Equal(t, replaceNodeTestNodeName, resp.OldNode)
			assert.NotEmpty(t, resp.WarmUpPod)
			assertNodeCordoned(t, cli, replaceNodeTestNodeName, true)

			var warmUp corev1.Pod
			require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Namespace: replaceNodeTestNamespace, Name: warmUpPodName(replaceNodeTestOperationID)}, &warmUp))
			assert.Equal(t, replaceNodeTestWarmUpImage, warmUp.Spec.Containers[0].Image)
			assert.Equal(t, pod.Spec.NodeSelector, warmUp.Spec.NodeSelector)
			assert.Equal(t, pod.Spec.Tolerations, warmUp.Spec.Tolerations)
			assert.Equal(t, pod.Spec.SchedulerName, warmUp.Spec.SchedulerName)
			assert.Equal(t, pod.Spec.PriorityClassName, warmUp.Spec.PriorityClassName)
			assert.NotEqual(t, pod.Spec.TopologySpreadConstraints, warmUp.Spec.TopologySpreadConstraints)
			assert.Empty(t, warmUp.Spec.TopologySpreadConstraints)
			require.NotNil(t, warmUp.Spec.AutomountServiceAccountToken)
			assert.False(t, *warmUp.Spec.AutomountServiceAccountToken)
			assert.Equal(t, resource.MustParse("2"), warmUp.Spec.Containers[0].Resources.Requests[corev1.ResourceCPU])
			assert.Equal(t, resource.MustParse("4Gi"), warmUp.Spec.Containers[0].Resources.Requests[corev1.ResourceMemory])
			assert.Empty(t, warmUp.Spec.Volumes)
			assert.Empty(t, warmUp.Spec.ServiceAccountName)
		})
	}
}

func TestAgentService_PrepareReplaceNodeRejectsOperationTargetRebind(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	t.Setenv(replaceNodeWarmUpImageEnvKey, replaceNodeTestWarmUpImage)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}
	cli := newReplaceNodeTestClient(t,
		makeReplaceNodeStateConfigMap(state),
		makeReadyOperatorPod("other-pod", "TiDB", "tidb-1", "node-2", "other-uid"),
		makeCordonedNode(replaceNodeTestNodeName),
		makeNode("node-2"),
	)
	svc := NewAgentService(cli, nil)

	_, err := svc.PrepareReplaceNode(context.Background(), makeReplaceNodeReqByPod("other-pod"))

	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, true)
	assertNodeCordoned(t, cli, "node-2", false)
	var stateConfigMap corev1.ConfigMap
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{
		Namespace: replaceNodeTestNamespace,
		Name:      replaceNodeStateConfigMapName(replaceNodeTestOperationID),
	}, &stateConfigMap))
	assert.Equal(t, "tidb-pod", stateConfigMap.Data[replaceNodeStatePodNameKey])
	assert.Equal(t, replaceNodeTestNodeName, stateConfigMap.Data[replaceNodeStateOldNodeKey])
}

func TestAgentService_PrepareReplaceNodeReturnsAbortedOperationState(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	t.Setenv(replaceNodeWarmUpImageEnvKey, replaceNodeTestWarmUpImage)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "missing-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
		AbortedAt:       v1.Now().Format("2006-01-02T15:04:05Z07:00"),
	}
	cli := newReplaceNodeTestClient(t, makeReplaceNodeStateConfigMap(state), makeNode(replaceNodeTestNodeName))
	svc := NewAgentService(cli, nil)

	resp, err := svc.PrepareReplaceNode(context.Background(), makeReplaceNodeReqByPod("missing-pod"))

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_ABORTED, resp.Phase)
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, false)
	var warmUp corev1.Pod
	err = cli.Get(context.Background(), client.ObjectKey{Namespace: replaceNodeTestNamespace, Name: warmUpPodName(replaceNodeTestOperationID)}, &warmUp)
	assert.True(t, apierrors.IsNotFound(err))
}

func TestAgentService_PrepareReplaceNodeValidation(t *testing.T) {
	tests := []struct {
		name                  string
		objects               []client.Object
		req                   *apiv1.PrepareReplaceNodeReq
		wantCode              codes.Code
		wantNoCordonAndWarmUp bool
	}{
		{
			name:     "target is required",
			req:      makeReplaceNodeReqWithoutTarget(),
			wantCode: codes.InvalidArgument,
		},
		{
			name: "pod target rejects non operator pod",
			objects: []client.Object{
				makePlainPod("plain-pod", replaceNodeTestNodeName),
			},
			req:      makeReplaceNodeReqByPod("plain-pod"),
			wantCode: codes.FailedPrecondition,
		},
		{
			name: "pod target rejects local ssd",
			objects: []client.Object{
				makeReadyOperatorPodWithVolumes("tikv-pod", "TiKV", "tikv-0", replaceNodeTestNodeName, "old-uid", []corev1.Volume{makePVCVolume("data")}),
				makePVC("data", localSSDStorageClassName),
			},
			req:      makeReplaceNodeReqByPod("tikv-pod"),
			wantCode: codes.FailedPrecondition,
		},
		{
			name: "pod target rejects non ready pod",
			objects: []client.Object{
				makeOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
				makeNode(replaceNodeTestNodeName),
			},
			req:                   makeReplaceNodeReqByPod("tidb-pod"),
			wantCode:              codes.FailedPrecondition,
			wantNoCordonAndWarmUp: true,
		},
		{
			name: "instance id target rejects multiple operator pods",
			objects: []client.Object{
				makeNodeWithProviderID(replaceNodeTestNodeName, "aws:///us-west-2a/i-node123"),
				makeOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "tidb-uid"),
				makeOperatorPod("tikv-pod", "TiKV", "tikv-0", replaceNodeTestNodeName, "tikv-uid"),
			},
			req:      makeReplaceNodeReqByInstanceID("i-node123"),
			wantCode: codes.FailedPrecondition,
		},
		{
			name: "pod target rejects unscheduled pod",
			objects: []client.Object{
				makeOperatorPod("tidb-pod", "TiDB", "tidb-0", "", "old-uid"),
			},
			req:      makeReplaceNodeReqByPod("tidb-pod"),
			wantCode: codes.FailedPrecondition,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Setenv("NAMESPACE", replaceNodeTestNamespace)
			t.Setenv(replaceNodeWarmUpImageEnvKey, replaceNodeTestWarmUpImage)
			cli := newReplaceNodeTestClient(t, tt.objects...)
			svc := NewAgentService(cli, nil)

			_, err := svc.PrepareReplaceNode(context.Background(), tt.req)

			require.Error(t, err)
			assert.Equal(t, tt.wantCode, status.Code(err))
			if tt.wantNoCordonAndWarmUp {
				assertNodeCordoned(t, cli, replaceNodeTestNodeName, false)
				var warmUp corev1.Pod
				err = cli.Get(context.Background(), client.ObjectKey{Namespace: replaceNodeTestNamespace, Name: warmUpPodName(replaceNodeTestOperationID)}, &warmUp)
				assert.True(t, apierrors.IsNotFound(err))
			}
		})
	}
}

func TestAgentService_PrepareReplaceNodeForceAllowsNonReadyPod(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	t.Setenv(replaceNodeWarmUpImageEnvKey, replaceNodeTestWarmUpImage)
	cli := newReplaceNodeTestClient(t,
		makeOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
		makeNode(replaceNodeTestNodeName),
	)
	svc := NewAgentService(cli, nil)
	req := makeReplaceNodeReqByPod("tidb-pod")
	req.Force = true

	resp, err := svc.PrepareReplaceNode(context.Background(), req)

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_PREPARING, resp.GetPhase())
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, true)
}

func TestAgentService_PrepareReplaceNodeByInstanceID(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	t.Setenv(replaceNodeWarmUpImageEnvKey, replaceNodeTestWarmUpImage)
	cli := newReplaceNodeTestClient(t,
		makeNodeWithProviderIDAndLabels(replaceNodeTestNodeName, "ap-southeast-1.i-t4nfbkbahkd0emdq4jzw", map[string]string{
			alicloudECSInstanceIDLabelKey: "i-t4nfbkbahkd0emdq4jzw",
		}),
		makeReadyOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
	)
	svc := NewAgentService(cli, nil)

	resp, err := svc.PrepareReplaceNode(context.Background(), makeReplaceNodeReqByInstanceID("i-t4nfbkbahkd0emdq4jzw"))

	require.NoError(t, err)
	assert.Equal(t, "tidb-pod", resp.PodName)
	assert.Equal(t, "i-t4nfbkbahkd0emdq4jzw", resp.InstanceId)
	assert.Equal(t, "TiDB", resp.ComponentKind)
}

func TestAgentService_PrepareReplaceNodeWarmUpImageMissingDoesNotCordon(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	cli := newReplaceNodeTestClient(t,
		makeReadyOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
		makeNode(replaceNodeTestNodeName),
	)
	svc := NewAgentService(cli, nil)

	_, err := svc.PrepareReplaceNode(context.Background(), makeReplaceNodeReqByPod("tidb-pod"))

	require.Error(t, err)
	require.ErrorContains(t, err, replaceNodeWarmUpImageEnvKey)
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, false)
	var stateConfigMap corev1.ConfigMap
	err = cli.Get(context.Background(), client.ObjectKey{
		Namespace: replaceNodeTestNamespace,
		Name:      replaceNodeStateConfigMapName(replaceNodeTestOperationID),
	}, &stateConfigMap)
	assert.True(t, apierrors.IsNotFound(err))
}

func TestAgentService_PrepareReplaceNodeWarmUpCreateFailureCanAbort(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	t.Setenv(replaceNodeWarmUpImageEnvKey, replaceNodeTestWarmUpImage)
	createErr := errors.New("warm-up create failed")
	cli := newReplaceNodeTestClientWithInterceptor(t, interceptor.Funcs{
		Create: func(ctx context.Context, cli client.WithWatch, obj client.Object, opts ...client.CreateOption) error {
			if _, ok := obj.(*corev1.Pod); ok && obj.GetName() == warmUpPodName(replaceNodeTestOperationID) {
				return createErr
			}
			return cli.Create(ctx, obj, opts...)
		},
	},
		makeReadyOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
		makeNode(replaceNodeTestNodeName),
	)
	svc := NewAgentService(cli, nil)

	_, err := svc.PrepareReplaceNode(context.Background(), makeReplaceNodeReqByPod("tidb-pod"))

	require.Error(t, err)
	require.ErrorIs(t, err, createErr)
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, true)
	var stateConfigMap corev1.ConfigMap
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{
		Namespace: replaceNodeTestNamespace,
		Name:      replaceNodeStateConfigMapName(replaceNodeTestOperationID),
	}, &stateConfigMap))

	resp, err := svc.AbortReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_ABORTED, resp.Phase)
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, false)
	err = cli.Get(context.Background(), client.ObjectKey{
		Namespace: replaceNodeTestNamespace,
		Name:      replaceNodeStateConfigMapName(replaceNodeTestOperationID),
	}, &stateConfigMap)
	require.NoError(t, err)
	assert.NotEmpty(t, stateConfigMap.Data[replaceNodeStateAbortedAtKey])
}

func TestAgentService_CommitReplaceNode(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	t.Setenv(replaceNodeWarmUpImageEnvKey, replaceNodeTestWarmUpImage)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
		InstanceID:      "i-node123",
	}
	cli := newReplaceNodeTestClient(t,
		makeOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
		makeReplaceNodeStateConfigMap(state),
		makeRunningWarmUpPod(state),
	)
	svc := NewAgentService(cli, nil)

	resp, err := svc.CommitReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_COMMITTED, resp.Phase)
	assert.Equal(t, replaceNodeTestWarmUpNode, resp.WarmUpNode)

	var warmUp corev1.Pod
	err = cli.Get(context.Background(), client.ObjectKey{Namespace: replaceNodeTestNamespace, Name: warmUpPodName(replaceNodeTestOperationID)}, &warmUp)
	assert.True(t, apierrors.IsNotFound(err))

	instance := makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false)
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Namespace: replaceNodeTestNamespace, Name: "tidb-0"}, instance))
	assert.Equal(t, replaceNodeTestOperationID, instance.GetLabels()[replaceNodeOperationIDLabelKey])
	assert.Empty(t, instance.GetAnnotations())

	var stateConfigMap corev1.ConfigMap
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{
		Namespace: replaceNodeTestNamespace,
		Name:      replaceNodeStateConfigMapName(replaceNodeTestOperationID),
	}, &stateConfigMap))
	assert.Equal(t, replaceNodeTestWarmUpNode, stateConfigMap.Data[replaceNodeStateWarmUpNodeKey])
	assert.Equal(t, "old-uid", stateConfigMap.Data[replaceNodeStateOldPodUIDKey])
}

func TestAgentService_CommitReplaceNodeRetryDeletesWarmUpAfterCommittedDeleteFailure(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	t.Setenv(replaceNodeWarmUpImageEnvKey, replaceNodeTestWarmUpImage)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}
	deleteErr := errors.New("delete warm-up failed")
	deleteCalls := 0
	cli := newReplaceNodeTestClientWithInterceptor(t, interceptor.Funcs{
		Delete: func(ctx context.Context, cli client.WithWatch, obj client.Object, opts ...client.DeleteOption) error {
			if pod, ok := obj.(*corev1.Pod); ok && pod.Name == warmUpPodName(replaceNodeTestOperationID) {
				deleteCalls++
				if deleteCalls == 1 {
					return deleteErr
				}
			}
			return cli.Delete(ctx, obj, opts...)
		},
	},
		makeOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
		makeReplaceNodeStateConfigMap(state),
		makeRunningWarmUpPod(state),
	)
	svc := NewAgentService(cli, nil)

	_, err := svc.CommitReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.Error(t, err)
	require.ErrorIs(t, err, deleteErr)
	instance := makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false)
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Namespace: replaceNodeTestNamespace, Name: "tidb-0"}, instance))
	assert.Equal(t, replaceNodeTestOperationID, instance.GetLabels()[replaceNodeOperationIDLabelKey])
	var warmUp corev1.Pod
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Namespace: replaceNodeTestNamespace, Name: warmUpPodName(replaceNodeTestOperationID)}, &warmUp))

	resp, err := svc.CommitReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_COMMITTED, resp.Phase)
	assert.Equal(t, 2, deleteCalls)
	err = cli.Get(context.Background(), client.ObjectKey{Namespace: replaceNodeTestNamespace, Name: warmUpPodName(replaceNodeTestOperationID)}, &warmUp)
	assert.True(t, apierrors.IsNotFound(err))
}

func TestAgentService_CommitReplaceNodeWithWarmUpSkipsTargetResolution(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
		InstanceID:      "i-node123",
	}
	listErr := errors.New("target resolution should not list")
	cli := newReplaceNodeTestClientWithInterceptor(t, interceptor.Funcs{
		List: func(context.Context, client.WithWatch, client.ObjectList, ...client.ListOption) error {
			return listErr
		},
	},
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
		makeReplaceNodeStateConfigMap(state),
		makeRunningWarmUpPod(state),
	)
	svc := NewAgentService(cli, nil)

	resp, err := svc.CommitReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_COMMITTED, resp.Phase)
	assert.Equal(t, "tidb-pod", resp.PodName)
	assert.Equal(t, "TiDB", resp.ComponentKind)
	assert.Equal(t, replaceNodeTestNodeName, resp.OldNode)
}

func TestAgentService_CommitReplaceNodeWaitingForWarmUpUsesCommittingPhase(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	t.Setenv(replaceNodeWarmUpImageEnvKey, replaceNodeTestWarmUpImage)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}
	warmUpPod := makeRunningWarmUpPod(state)
	warmUpPod.Spec.NodeName = ""
	warmUpPod.Status.Phase = corev1.PodPending
	cli := newReplaceNodeTestClient(t,
		makeOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
		makeReplaceNodeStateConfigMap(state),
		warmUpPod,
	)
	svc := NewAgentService(cli, nil)

	resp, err := svc.CommitReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_COMMITTING, resp.Phase)
}

func TestAgentService_CommitReplaceNodePersistsWarmUpNodeInState(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	t.Setenv(replaceNodeWarmUpImageEnvKey, replaceNodeTestWarmUpImage)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}
	cli := newReplaceNodeTestClient(t,
		makeOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
		makeReplaceNodeStateConfigMap(state),
		makeRunningWarmUpPod(state),
	)
	svc := NewAgentService(cli, nil)

	resp, err := svc.CommitReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_COMMITTED, resp.Phase)
	var stateConfigMap corev1.ConfigMap
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{
		Namespace: replaceNodeTestNamespace,
		Name:      replaceNodeStateConfigMapName(replaceNodeTestOperationID),
	}, &stateConfigMap))
	assert.Equal(t, replaceNodeTestWarmUpNode, stateConfigMap.Data[replaceNodeStateWarmUpNodeKey])
}

func TestAgentService_CommitReplaceNodeSucceededAfterReplacementReady(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
		InstanceID:      "i-node123",
	}
	cli := newReplaceNodeTestClient(t,
		makeReadyOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestWarmUpNode, "new-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", map[string]string{
			replaceNodeOperationIDLabelKey: replaceNodeTestOperationID,
		}, nil, true),
		makeReplaceNodeStateConfigMap(state),
	)
	svc := NewAgentService(cli, nil)

	resp, err := svc.CommitReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_SUCCEEDED, resp.Phase)
	assert.Equal(t, replaceNodeTestWarmUpNode, resp.WarmUpNode)
	assert.Equal(t, replaceNodeTestWarmUpNode, resp.ReplacementNode)
}

func TestAgentService_CommitReplaceNodeAllowsDifferentReplacementNode(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}
	cli := newReplaceNodeTestClient(t,
		makeReadyOperatorPod("tidb-pod", "TiDB", "tidb-0", "other-node", "new-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", map[string]string{
			replaceNodeOperationIDLabelKey: replaceNodeTestOperationID,
		}, nil, true),
		makeReplaceNodeStateConfigMap(state),
	)
	svc := NewAgentService(cli, nil)

	resp, err := svc.CommitReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_SUCCEEDED, resp.Phase)
	assert.Equal(t, replaceNodeTestWarmUpNode, resp.WarmUpNode)
	assert.Equal(t, "other-node", resp.ReplacementNode)
	assert.Contains(t, resp.Message, "instead of warm-up node")
}

func TestAgentService_CommitReplaceNodeRejectsPreCommitStateWithoutWarmUp(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}
	cli := newReplaceNodeTestClient(t,
		makeOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
		makeReplaceNodeStateConfigMap(state),
	)
	svc := NewAgentService(cli, nil)

	_, err := svc.CommitReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))

	instance := makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false)
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Namespace: replaceNodeTestNamespace, Name: "tidb-0"}, instance))
	assert.Empty(t, instance.GetLabels()[replaceNodeOperationIDLabelKey])
}

func TestAgentService_CommitReplaceNodeRequiresStateConfigMap(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	cli := newReplaceNodeTestClient(t,
		makeOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
	)
	svc := NewAgentService(cli, nil)

	_, err := svc.CommitReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.Error(t, err)
	assert.Equal(t, codes.NotFound, status.Code(err))
}

func TestAgentService_GetReplaceNodeOperation(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	baseState := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}

	t.Run("pre commit warm-up pending", func(t *testing.T) {
		warmUpPod := makeRunningWarmUpPod(baseState)
		warmUpPod.Spec.NodeName = ""
		warmUpPod.Status.Phase = corev1.PodPending
		cli := newReplaceNodeTestClient(t,
			makeReplaceNodeStateConfigMap(baseState),
			makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
			warmUpPod,
		)
		svc := NewAgentService(cli, nil)

		resp, err := svc.GetReplaceNodeOperation(context.Background(), makeReplaceNodeOperationReq())

		require.NoError(t, err)
		assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_PREPARING, resp.Phase)
	})

	t.Run("pre commit warm-up running", func(t *testing.T) {
		cli := newReplaceNodeTestClient(t,
			makeReplaceNodeStateConfigMap(baseState),
			makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
			makeRunningWarmUpPod(baseState),
		)
		svc := NewAgentService(cli, nil)

		resp, err := svc.GetReplaceNodeOperation(context.Background(), makeReplaceNodeOperationReq())

		require.NoError(t, err)
		assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_PREPARED, resp.Phase)
		assert.Equal(t, replaceNodeTestWarmUpNode, resp.WarmUpNode)
	})

	t.Run("committed waits for replacement uid change", func(t *testing.T) {
		cli := newReplaceNodeTestClient(t,
			makeReplaceNodeStateConfigMap(baseState),
			makeOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestNodeName, "old-uid"),
			makeOperatorInstance(t, "TiDB", "tidb-0", map[string]string{
				replaceNodeOperationIDLabelKey: replaceNodeTestOperationID,
			}, nil, true),
		)
		svc := NewAgentService(cli, nil)

		resp, err := svc.GetReplaceNodeOperation(context.Background(), makeReplaceNodeOperationReq())

		require.NoError(t, err)
		assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_COMMITTED, resp.Phase)
		assert.Contains(t, resp.Message, "uid to change")
	})

	t.Run("committed succeeds after replacement ready", func(t *testing.T) {
		cli := newReplaceNodeTestClient(t,
			makeReplaceNodeStateConfigMap(baseState),
			makeReadyOperatorPod("tidb-pod", "TiDB", "tidb-0", replaceNodeTestWarmUpNode, "new-uid"),
			makeOperatorInstance(t, "TiDB", "tidb-0", map[string]string{
				replaceNodeOperationIDLabelKey: replaceNodeTestOperationID,
			}, nil, true),
		)
		svc := NewAgentService(cli, nil)

		resp, err := svc.GetReplaceNodeOperation(context.Background(), makeReplaceNodeOperationReq())

		require.NoError(t, err)
		assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_SUCCEEDED, resp.Phase)
	})

	t.Run("committed succeeds when replacement node differs", func(t *testing.T) {
		cli := newReplaceNodeTestClient(t,
			makeReplaceNodeStateConfigMap(baseState),
			makeReadyOperatorPod("tidb-pod", "TiDB", "tidb-0", "other-node", "new-uid"),
			makeOperatorInstance(t, "TiDB", "tidb-0", map[string]string{
				replaceNodeOperationIDLabelKey: replaceNodeTestOperationID,
			}, nil, true),
		)
		svc := NewAgentService(cli, nil)

		resp, err := svc.GetReplaceNodeOperation(context.Background(), makeReplaceNodeOperationReq())

		require.NoError(t, err)
		assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_SUCCEEDED, resp.Phase)
		assert.Equal(t, "other-node", resp.ReplacementNode)
		assert.Contains(t, resp.Message, "instead of warm-up node")
	})

	t.Run("aborted", func(t *testing.T) {
		state := baseState
		state.AbortedAt = "2026-05-21T00:00:00Z"
		cli := newReplaceNodeTestClient(t, makeReplaceNodeStateConfigMap(state))
		svc := NewAgentService(cli, nil)

		resp, err := svc.GetReplaceNodeOperation(context.Background(), makeReplaceNodeOperationReq())

		require.NoError(t, err)
		assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_ABORTED, resp.Phase)
	})

	t.Run("missing state", func(t *testing.T) {
		cli := newReplaceNodeTestClient(t)
		svc := NewAgentService(cli, nil)

		_, err := svc.GetReplaceNodeOperation(context.Background(), makeReplaceNodeOperationReq())

		require.Error(t, err)
		assert.Equal(t, codes.NotFound, status.Code(err))
	})
}

func TestAgentService_AbortReplaceNode(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}
	cli := newReplaceNodeTestClient(t,
		makeCordonedNode(replaceNodeTestNodeName),
		makeRunningWarmUpPod(state),
		makeReplaceNodeStateConfigMap(state),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
	)
	svc := NewAgentService(cli, nil)

	resp, err := svc.AbortReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_ABORTED, resp.Phase)
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, false)

	var stateConfigMap corev1.ConfigMap
	err = cli.Get(context.Background(), client.ObjectKey{
		Namespace: replaceNodeTestNamespace,
		Name:      replaceNodeStateConfigMapName(replaceNodeTestOperationID),
	}, &stateConfigMap)
	require.NoError(t, err)
	assert.NotEmpty(t, stateConfigMap.Data[replaceNodeStateAbortedAtKey])
}

func TestAgentService_AbortReplaceNodeKeepsWarmUpWhenUncordonFails(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}
	patchErr := errors.New("uncordon failed")
	patchAttempts := 0
	cli := newReplaceNodeTestClientWithInterceptor(t, interceptor.Funcs{
		Patch: func(ctx context.Context, cli client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
			if _, ok := obj.(*corev1.Node); ok && obj.GetName() == replaceNodeTestNodeName {
				patchAttempts++
				if patchAttempts == 1 {
					return patchErr
				}
			}
			return cli.Patch(ctx, obj, patch, opts...)
		},
	},
		makeCordonedNode(replaceNodeTestNodeName),
		makeRunningWarmUpPod(state),
		makeReplaceNodeStateConfigMap(state),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
	)
	svc := NewAgentService(cli, nil)

	_, err := svc.AbortReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.Error(t, err)
	require.ErrorIs(t, err, patchErr)
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, true)
	var warmUp corev1.Pod
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Namespace: replaceNodeTestNamespace, Name: warmUpPodName(replaceNodeTestOperationID)}, &warmUp))
	var stateConfigMap corev1.ConfigMap
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{
		Namespace: replaceNodeTestNamespace,
		Name:      replaceNodeStateConfigMapName(replaceNodeTestOperationID),
	}, &stateConfigMap))

	resp, err := svc.AbortReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_ABORTED, resp.Phase)
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, false)
	err = cli.Get(context.Background(), client.ObjectKey{Namespace: replaceNodeTestNamespace, Name: warmUpPodName(replaceNodeTestOperationID)}, &warmUp)
	assert.True(t, apierrors.IsNotFound(err))
	err = cli.Get(context.Background(), client.ObjectKey{
		Namespace: replaceNodeTestNamespace,
		Name:      replaceNodeStateConfigMapName(replaceNodeTestOperationID),
	}, &stateConfigMap)
	require.NoError(t, err)
	assert.NotEmpty(t, stateConfigMap.Data[replaceNodeStateAbortedAtKey])
}

func TestAgentService_AbortReplaceNodeKeepsStateWhenUncordonFails(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}
	patchErr := errors.New("uncordon failed")
	patchAttempts := 0
	cli := newReplaceNodeTestClientWithInterceptor(t, interceptor.Funcs{
		Patch: func(ctx context.Context, cli client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
			if _, ok := obj.(*corev1.Node); ok && obj.GetName() == replaceNodeTestNodeName {
				patchAttempts++
				if patchAttempts == 1 {
					return patchErr
				}
			}
			return cli.Patch(ctx, obj, patch, opts...)
		},
	},
		makeCordonedNode(replaceNodeTestNodeName),
		makeReplaceNodeStateConfigMap(state),
		makeOperatorInstance(t, "TiDB", "tidb-0", nil, nil, false),
	)
	svc := NewAgentService(cli, nil)

	_, err := svc.AbortReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.Error(t, err)
	require.ErrorIs(t, err, patchErr)
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, true)
	var stateConfigMap corev1.ConfigMap
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{
		Namespace: replaceNodeTestNamespace,
		Name:      replaceNodeStateConfigMapName(replaceNodeTestOperationID),
	}, &stateConfigMap))

	resp, err := svc.AbortReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.NoError(t, err)
	assert.Equal(t, apiv1.ReplaceNodePhase_REPLACE_NODE_PHASE_ABORTED, resp.Phase)
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, false)
	err = cli.Get(context.Background(), client.ObjectKey{
		Namespace: replaceNodeTestNamespace,
		Name:      replaceNodeStateConfigMapName(replaceNodeTestOperationID),
	}, &stateConfigMap)
	require.NoError(t, err)
	assert.NotEmpty(t, stateConfigMap.Data[replaceNodeStateAbortedAtKey])
}

func TestAgentService_AbortReplaceNodeRejectsCommittedOperation(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}
	cli := newReplaceNodeTestClient(t,
		makeCordonedNode(replaceNodeTestNodeName),
		makeRunningWarmUpPod(state),
		makeReplaceNodeStateConfigMap(state),
		makeOperatorInstance(t, "TiDB", "tidb-0", map[string]string{
			replaceNodeOperationIDLabelKey: replaceNodeTestOperationID,
		}, nil, false),
	)
	svc := NewAgentService(cli, nil)

	_, err := svc.AbortReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, true)
}

func TestAgentService_AbortReplaceNodeRejectsCommittedOperationWithoutWarmUpPod(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	state := replaceNodeOperationState{
		OperationID:     replaceNodeTestOperationID,
		PodName:         "tidb-pod",
		OldPodUID:       "old-uid",
		OldNode:         replaceNodeTestNodeName,
		WarmUpNode:      replaceNodeTestWarmUpNode,
		OwnerAPIVersion: CoreGroup + "/" + CoreVersion,
		ComponentKind:   "TiDB",
		OwnerName:       "tidb-0",
	}
	cli := newReplaceNodeTestClient(t,
		makeCordonedNode(replaceNodeTestNodeName),
		makeReplaceNodeStateConfigMap(state),
		makeOperatorInstance(t, "TiDB", "tidb-0", map[string]string{
			replaceNodeOperationIDLabelKey: replaceNodeTestOperationID,
		}, nil, false),
	)
	svc := NewAgentService(cli, nil)

	_, err := svc.AbortReplaceNode(context.Background(), makeReplaceNodeOperationReq())

	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assertNodeCordoned(t, cli, replaceNodeTestNodeName, true)
}

func TestAgentService_PrepareReplaceNodeRequiresNamespace(t *testing.T) {
	cli := newReplaceNodeTestClient(t)
	svc := NewAgentService(cli, nil)

	_, err := svc.PrepareReplaceNode(context.Background(), makeReplaceNodeReqByPod("pod"))

	require.Error(t, err)
	assert.Contains(t, err.Error(), "NAMESPACE is empty")
}

func TestAgentService_ReplaceNodeRejectsNilRequest(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	cli := newReplaceNodeTestClient(t)
	svc := NewAgentService(cli, nil)

	_, err := svc.PrepareReplaceNode(context.Background(), nil)
	require.Error(t, err)
	assert.Equal(t, codes.InvalidArgument, status.Code(err))

	_, err = svc.GetReplaceNodeOperation(context.Background(), nil)
	require.Error(t, err)
	assert.Equal(t, codes.InvalidArgument, status.Code(err))

	_, err = svc.CommitReplaceNode(context.Background(), nil)
	require.Error(t, err)
	assert.Equal(t, codes.InvalidArgument, status.Code(err))

	_, err = svc.AbortReplaceNode(context.Background(), nil)
	require.Error(t, err)
	assert.Equal(t, codes.InvalidArgument, status.Code(err))
}

func TestAgentService_PrepareReplaceNodeRejectsMissingTarget(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	cli := newReplaceNodeTestClient(t)
	svc := NewAgentService(cli, nil)

	_, err := svc.PrepareReplaceNode(context.Background(), makeReplaceNodeReqWithoutTarget())

	require.Error(t, err)
	assert.Equal(t, codes.InvalidArgument, status.Code(err))
}

func TestAgentService_ReplaceNodeOperationRejectsMissingRequiredFields(t *testing.T) {
	t.Setenv("NAMESPACE", replaceNodeTestNamespace)
	cli := newReplaceNodeTestClient(t)
	svc := NewAgentService(cli, nil)
	req := &apiv1.ReplaceNodeOperationReq{ClusterId: replaceNodeTestClusterID}

	_, err := svc.GetReplaceNodeOperation(context.Background(), req)
	require.Error(t, err)
	assert.Equal(t, codes.InvalidArgument, status.Code(err))

	_, err = svc.CommitReplaceNode(context.Background(), req)
	require.Error(t, err)
	assert.Equal(t, codes.InvalidArgument, status.Code(err))

	_, err = svc.AbortReplaceNode(context.Background(), req)
	require.Error(t, err)
	assert.Equal(t, codes.InvalidArgument, status.Code(err))
}

func Test_parseProviderInstanceIDFromNode(t *testing.T) {
	tests := []struct {
		name       string
		providerID string
		labels     map[string]string
		want       string
		wantErr    bool
	}{
		{name: "AWS", providerID: "aws:///us-west-2a/i-1234567890abcdef0", want: "i-1234567890abcdef0"},
		{name: "GCP", providerID: "gce://project-id/us-west1-b/gke-pool-node-abc", want: "gke-pool-node-abc"},
		{name: "Azure", providerID: "azure:///subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/virtualMachineScaleSets/vmss/virtualMachines/3", want: "3"},
		{
			name:       "Alicloud ACK node",
			providerID: "ap-southeast-1.i-t4nfbkbahkd0emdq4jzw",
			labels: map[string]string{
				alicloudECSInstanceIDLabelKey: "i-t4nfbkbahkd0emdq4jzw",
			},
			want: "i-t4nfbkbahkd0emdq4jzw",
		},
		{
			name:       "ACK style providerID without Alicloud label",
			providerID: "ap-southeast-1.i-t4nfbkbahkd0emdq4jzw",
			wantErr:    true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := parseProviderInstanceIDFromNode(makeNodeWithProviderIDAndLabels("node", tt.providerID, tt.labels))

			if tt.wantErr {
				require.Error(t, err)
				return
			}
			require.NoError(t, err)
			assert.Equal(t, tt.want, got)
		})
	}
}

func newReplaceNodeTestClient(t *testing.T, objects ...client.Object) client.Client {
	t.Helper()

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

	return fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build()
}

func newReplaceNodeTestClientWithInterceptor(t *testing.T, funcs interceptor.Funcs, objects ...client.Object) client.Client {
	t.Helper()

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

	return fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).WithInterceptorFuncs(funcs).Build()
}

func makeReplaceNodeReqByPod(podName string) *apiv1.PrepareReplaceNodeReq {
	return &apiv1.PrepareReplaceNodeReq{
		OperationId: replaceNodeTestOperationID,
		ClusterId:   replaceNodeTestClusterID,
		Target:      &apiv1.PrepareReplaceNodeReq_PodName{PodName: podName},
	}
}

func makeReplaceNodeReqByInstanceID(instanceID string) *apiv1.PrepareReplaceNodeReq {
	return &apiv1.PrepareReplaceNodeReq{
		OperationId: replaceNodeTestOperationID,
		ClusterId:   replaceNodeTestClusterID,
		Target:      &apiv1.PrepareReplaceNodeReq_InstanceId{InstanceId: instanceID},
	}
}

func makeReplaceNodeReqWithoutTarget() *apiv1.PrepareReplaceNodeReq {
	return &apiv1.PrepareReplaceNodeReq{OperationId: replaceNodeTestOperationID, ClusterId: replaceNodeTestClusterID}
}

func makeReplaceNodeOperationReq() *apiv1.ReplaceNodeOperationReq {
	return &apiv1.ReplaceNodeOperationReq{OperationId: replaceNodeTestOperationID, ClusterId: replaceNodeTestClusterID}
}

func makeOperatorPod(name, ownerKind, ownerName, nodeName, uid string) *corev1.Pod {
	return makeOperatorPodWithVolumes(name, ownerKind, ownerName, nodeName, uid, nil)
}

func makeReadyOperatorPod(name, ownerKind, ownerName, nodeName, uid string) *corev1.Pod {
	return makeReadyOperatorPodWithVolumes(name, ownerKind, ownerName, nodeName, uid, nil)
}

func makeReadyOperatorPodWithVolumes(name, ownerKind, ownerName, nodeName, uid string, volumes []corev1.Volume) *corev1.Pod {
	pod := makeOperatorPodWithVolumes(name, ownerKind, ownerName, nodeName, uid, volumes)
	pod.Status.Phase = corev1.PodRunning
	pod.Status.Conditions = []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}}
	return pod
}

func makeOperatorPodWithVolumes(name, ownerKind, ownerName, nodeName, uid string, volumes []corev1.Volume) *corev1.Pod {
	return &corev1.Pod{
		ObjectMeta: v1.ObjectMeta{
			Name:      name,
			Namespace: replaceNodeTestNamespace,
			UID:       types.UID(uid),
			Labels: map[string]string{
				tidbOperatorManagedLabelKey: tidbOperatorManagedByValue,
				K8sComponentLabelKey:        componentTypeForOwnerKind(ownerKind),
			},
			OwnerReferences: []v1.OwnerReference{{
				APIVersion: CoreGroup + "/" + CoreVersion,
				Kind:       ownerKind,
				Name:       ownerName,
			}},
		},
		Spec: corev1.PodSpec{
			NodeName: nodeName,
			Volumes:  volumes,
			Containers: []corev1.Container{{
				Name: "main",
				Resources: corev1.ResourceRequirements{Requests: corev1.ResourceList{
					corev1.ResourceCPU:    resource.MustParse("500m"),
					corev1.ResourceMemory: resource.MustParse("512Mi"),
				}},
			}},
		},
	}
}

func makePlainPod(name, nodeName string) *corev1.Pod {
	return &corev1.Pod{
		ObjectMeta: v1.ObjectMeta{Name: name, Namespace: replaceNodeTestNamespace},
		Spec:       corev1.PodSpec{NodeName: nodeName},
	}
}

func makeRunningWarmUpPod(state replaceNodeOperationState) *corev1.Pod {
	pod := &corev1.Pod{
		ObjectMeta: v1.ObjectMeta{
			Name:      warmUpPodName(state.OperationID),
			Namespace: replaceNodeTestNamespace,
			Labels: map[string]string{
				replaceNodeWarmUpLabelKey:      "true",
				replaceNodeOperationIDLabelKey: state.OperationID,
			},
			Annotations: map[string]string{
				replaceNodeTargetPodNameAnnotationKey:   state.PodName,
				replaceNodeTargetPodUIDAnnotationKey:    state.OldPodUID,
				replaceNodeOldNodeAnnotationKey:         state.OldNode,
				replaceNodeOwnerAPIVersionAnnotationKey: state.OwnerAPIVersion,
				replaceNodeOwnerKindAnnotationKey:       state.ComponentKind,
				replaceNodeOwnerNameAnnotationKey:       state.OwnerName,
				replaceNodeInstanceIDAnnotationKey:      state.InstanceID,
			},
		},
		Spec:   corev1.PodSpec{NodeName: state.WarmUpNode},
		Status: corev1.PodStatus{Phase: corev1.PodRunning},
	}
	return pod
}

func makeReplaceNodeStateConfigMap(state replaceNodeOperationState) *corev1.ConfigMap {
	return buildReplaceNodeStateConfigMap(replaceNodeTestNamespace, state)
}

func makeOperatorInstance(t *testing.T, kind, name string, labels, annotations map[string]string, readySynced bool) client.Object {
	t.Helper()

	obj := &unstructured.Unstructured{}
	obj.SetGroupVersionKind(schema.GroupVersionKind{Group: CoreGroup, Version: CoreVersion, Kind: kind})
	obj.SetName(name)
	obj.SetNamespace(replaceNodeTestNamespace)
	obj.SetLabels(labels)
	obj.SetAnnotations(annotations)
	if readySynced {
		obj.Object["status"] = map[string]any{
			"conditions": []any{
				map[string]any{"type": "Ready", "status": "True"},
				map[string]any{"type": "Synced", "status": "True"},
			},
		}
	}
	return obj
}

func makePVCVolume(claimName string) corev1.Volume {
	return corev1.Volume{
		Name: "data",
		VolumeSource: corev1.VolumeSource{
			PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: claimName},
		},
	}
}

func makePVC(name, storageClassName string) *corev1.PersistentVolumeClaim {
	return &corev1.PersistentVolumeClaim{
		ObjectMeta: v1.ObjectMeta{Name: name, Namespace: replaceNodeTestNamespace},
		Spec:       corev1.PersistentVolumeClaimSpec{StorageClassName: &storageClassName},
	}
}

func makeNode(name string) *corev1.Node {
	return &corev1.Node{ObjectMeta: v1.ObjectMeta{Name: name}}
}

func makeNodeWithProviderID(name, providerID string) *corev1.Node {
	node := makeNode(name)
	node.Spec.ProviderID = providerID
	return node
}

func makeNodeWithProviderIDAndLabels(name, providerID string, labels map[string]string) *corev1.Node {
	node := makeNodeWithProviderID(name, providerID)
	node.Labels = labels
	return node
}

func makeCordonedNode(name string) *corev1.Node {
	node := makeNode(name)
	node.Spec.Unschedulable = true
	return node
}

func assertNodeCordoned(t *testing.T, cli client.Client, nodeName string, want bool) {
	t.Helper()

	var node corev1.Node
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Name: nodeName}, &node))
	assert.Equal(t, want, node.Spec.Unschedulable)
}