package workflow

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/samber/lo"
	"github.com/stretchr/testify/assert"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	"github.com/tidbcloud/eks-provider/controllers/workflow/action"
	"github.com/tidbcloud/eks-provider/tests/e2e/scheme"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	utiltime "github.com/tidbcloud/infra-provider/pkg/util/time"
)

func Test_handeleRunning(t *testing.T) {
	// Setup workflow
	wf := &cloudv1.Workflow{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "test-workflow",
		},
		Spec: cloudv1.WorkflowSpec{
			Steps: []cloudv1.WorkflowStep{
				{
					Name: "step1",
					Uses: "mockActionWithPost",
				},
				{
					Name: "step2",
					Uses: "mockAction",
				},
			},
			RetryPolicy: &cloudv1.WorkflowRetryPolicy{
				Type:    cloudv1.RetryPolicyTypeAlways,
				Backoff: lo.ToPtr(metav1.Duration{Duration: time.Millisecond}),
			},
		},
		Status: cloudv1.WorkflowStatus{
			Steps: []*cloudv1.WorkflowStepStatus{
				{
					Name:    "step1",
					Phase:   cloudv1.WorkflowStepPhasePending,
					BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
				},
				{
					Name:    "step2",
					Phase:   cloudv1.WorkflowStepPhasePending,
					BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
				},
			},
			Phase:   cloudv1.WorkflowPhaseRunning,
			BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
		},
	}
	wf.Default()
	registry := &action.Registry{
		Actions: map[cloudv1.WorkflowActionName]action.Action{
			"mockAction": &mockAction{},
			"mockActionWithPost": &mockActionWithPost{
				postHandleFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					stepStatus, _ := wf.StepStatus(ctx, step.Name)
					if stepStatus.EndingCount < 10 {
						return fmt.Errorf("endingCount %d is less than 10", stepStatus.EndingCount)
					}
					return nil
				},
			},
		},
	}
	cli := newFakeClient([]client.Object{wf})
	ctx := context.Background()
	r := &WorkflowReconciler{
		clock:     utiltime.RealClock{},
		registry:  registry,
		scheduler: NewSequenceScheduler(registry),
		client:    cli,
	}

	assert.Eventually(t, func() bool {
		// refresh wf from fake client
		assert.NoError(t, cli.Get(ctx, client.ObjectKeyFromObject(wf), wf))
		fmt.Printf("phase: %s, finishedSteps: %v, runningSteps: %v, endingSteps: %v\n", wf.Status.Phase, wf.Status.FinishedSteps, wf.Status.RunningSteps, wf.Status.EndingSteps)

		result, err := r.handleRunning(ctx, wf)
		fmt.Printf("result: %+v, err: %v\n", result, err)
		return wf.Status.Phase == cloudv1.WorkflowPhaseCompleted && !result.Requeue && err == nil
	}, 30*time.Second, 100*time.Millisecond)

	assert.Equalf(t, cloudv1.WorkflowPhaseCompleted, wf.Status.Phase, "phase")
	assert.Equalf(t, []string{"step1", "step2"}, wf.Status.FinishedSteps, "finished steps")
	assert.Emptyf(t, wf.Status.RunningSteps, "running steps")
	assert.Emptyf(t, wf.Status.EndingSteps, "ending steps")
}

func Test_executeOneStep(t *testing.T) {
	tests := []struct {
		name                 string
		stepName             string
		stepPhase            cloudv1.WorkflowStepPhase
		action               action.Action
		inputs               map[string]interface{}
		wfModifier           func(wf *cloudv1.Workflow)
		expectStepPhase      cloudv1.WorkflowStepPhase
		errContainsSubString string
	}{
		{
			name:                 "unknown step should return error",
			stepName:             "unknownStep",
			stepPhase:            cloudv1.WorkflowStepPhaseRunning,
			errContainsSubString: "unknown step",
		},
		{
			name:                 "unknown action should return error",
			stepName:             "step1",
			stepPhase:            cloudv1.WorkflowStepPhaseRunning,
			errContainsSubString: "unknown action refereced by step",
		},
		{
			name:      "inflate inputs error should return error",
			stepName:  "step1",
			stepPhase: cloudv1.WorkflowStepPhaseRunning,
			action:    &mockAction{},
			inputs: map[string]interface{}{
				"invalid": "$invalid.variable",
			},
			errContainsSubString: "inflate step",
		},
		{
			name:      "step in Running phase should call Handle",
			stepName:  "step1",
			stepPhase: cloudv1.WorkflowStepPhaseRunning,
			action: &mockAction{
				handleFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return nil
				},
			},
			expectStepPhase:      cloudv1.WorkflowStepPhaseEnding,
			errContainsSubString: "",
		},
		{
			name:      "step in Ending phase should call PostHandle",
			stepName:  "step1",
			stepPhase: cloudv1.WorkflowStepPhaseEnding,
			action: &mockActionWithPost{
				mockAction: mockAction{},
				postHandleFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return nil
				},
			},
			expectStepPhase:      cloudv1.WorkflowStepPhaseCompleted,
			errContainsSubString: "",
		},
		{
			name:      "step in Ending phase should call PostHandle and PostHandle return error",
			stepName:  "step1",
			stepPhase: cloudv1.WorkflowStepPhaseEnding,
			action: &mockActionWithPost{
				mockAction: mockAction{},
				postHandleFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return fmt.Errorf("post handle error")
				},
			},
			expectStepPhase:      cloudv1.WorkflowStepPhaseEnding,
			errContainsSubString: "", // PostHandle error should be ignored and remain in Ending phase
		},
		{
			name:      "step in Ending phase without PostHandle should complete",
			stepName:  "step1",
			stepPhase: cloudv1.WorkflowStepPhaseEnding,
			action:    &mockAction{},
			wfModifier: func(wf *cloudv1.Workflow) {
				wf.Status.Steps[0].Phase = cloudv1.WorkflowStepPhaseEnding
			},
			expectStepPhase:      cloudv1.WorkflowStepPhaseCompleted,
			errContainsSubString: "",
		},
		{
			name:                 "step in invalid phase should return error",
			stepName:             "step1",
			stepPhase:            cloudv1.WorkflowStepPhaseCompleted,
			action:               &mockAction{},
			errContainsSubString: "step is not in Running, Ending, Cancelling, or Cancelled phase",
		},
		{
			name:      "step in Cancelling phase with same group completed step should continue normal execution",
			stepName:  "step1",
			stepPhase: cloudv1.WorkflowStepPhaseCancelling,
			action: &mockAction{
				handleFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return nil
				},
			},
			wfModifier: func(wf *cloudv1.Workflow) {
				wf.Status.Phase = cloudv1.WorkflowPhaseCancelling
				wf.Status.Steps[0].Phase = cloudv1.WorkflowStepPhaseCancelling
				wf.Status.CancellingSteps = []string{"step1"}
				wf.Spec.Steps[0].StepGroup = "group1"
				// Add a completed step in the same group
				wf.Spec.Steps = append(wf.Spec.Steps, cloudv1.WorkflowStep{
					Name:      "step2",
					StepGroup: "group1",
					Uses:      "mockAction",
				})
				wf.Status.Steps = append(wf.Status.Steps, &cloudv1.WorkflowStepStatus{
					Name:    "step2",
					Phase:   cloudv1.WorkflowStepPhaseCompleted,
					BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
				})
			},
			// Note: According to current implementation, if action doesn't support cancel,
			// it will return an error when step is in Cancelling phase.
			expectStepPhase:      cloudv1.WorkflowStepPhaseCancelling,
			errContainsSubString: "action supports cancellation, but step",
		},
		{
			name:      "step in Cancelling phase with action supporting cancel should call HandleCancel",
			stepName:  "step1",
			stepPhase: cloudv1.WorkflowStepPhaseCancelling,
			action: &mockActionWithCancel{
				mockAction: mockAction{},
				handleCancelFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return nil
				},
			},
			wfModifier: func(wf *cloudv1.Workflow) {
				wf.Status.Phase = cloudv1.WorkflowPhaseCancelling
				wf.Status.Steps[0].Phase = cloudv1.WorkflowStepPhaseCancelling
				wf.Spec.Steps[0].StepGroup = "group1"
			},
			expectStepPhase:      cloudv1.WorkflowStepPhaseCancelEnding,
			errContainsSubString: "",
		},
		{
			name:      "step in Cancelling phase without action supporting cancel should continue normal execution",
			stepName:  "step1",
			stepPhase: cloudv1.WorkflowStepPhaseCancelling,
			action: &mockAction{
				handleFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return nil
				},
			},
			wfModifier: func(wf *cloudv1.Workflow) {
				wf.Status.Phase = cloudv1.WorkflowPhaseCancelling
				wf.Status.Steps[0].Phase = cloudv1.WorkflowStepPhaseCancelling
				wf.Status.CancellingSteps = []string{"step1"}
				wf.Spec.Steps[0].StepGroup = "group1"
			},
			// Note: According to current implementation, if action doesn't support cancel,
			// it will return an error when step is in Cancelling phase.
			expectStepPhase:      cloudv1.WorkflowStepPhaseCancelling,
			errContainsSubString: "action supports cancellation, but step",
		},
		{
			name:      "step in CancelEnding phase with action supporting cancel should call PostHandleCancel",
			stepName:  "step1",
			stepPhase: cloudv1.WorkflowStepPhaseCancelEnding,
			action: &mockActionWithCancel{
				mockAction: mockAction{},
				postHandleCancelFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return nil
				},
			},
			wfModifier: func(wf *cloudv1.Workflow) {
				wf.Status.Phase = cloudv1.WorkflowPhaseCancelling
				wf.Status.Steps[0].Phase = cloudv1.WorkflowStepPhaseCancelEnding
				wf.Status.CancelEndingSteps = []string{"step1"}
			},
			expectStepPhase:      cloudv1.WorkflowStepPhaseCancelled,
			errContainsSubString: "",
		},
		{
			name:      "step in CancelEnding phase without action supporting cancel but with PostHandle should call PostHandle",
			stepName:  "step1",
			stepPhase: cloudv1.WorkflowStepPhaseCancelEnding,
			action: &mockActionWithPost{
				mockAction: mockAction{},
				postHandleFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return nil
				},
			},
			wfModifier: func(wf *cloudv1.Workflow) {
				wf.Status.Phase = cloudv1.WorkflowPhaseCancelling
				wf.Status.Steps[0].Phase = cloudv1.WorkflowStepPhaseCancelEnding
				wf.Status.CancelEndingSteps = []string{"step1"}
			},
			// Note: According to current implementation, if action doesn't support cancel,
			// it will return an error when step is in CancelEnding phase, even if PostHandle is available.
			expectStepPhase:      cloudv1.WorkflowStepPhaseCancelEnding,
			errContainsSubString: "action supports cancellation, but step",
		},
		{
			name:      "step in CancelEnding phase without action supporting cancel and no PostHandle should return error",
			stepName:  "step1",
			stepPhase: cloudv1.WorkflowStepPhaseCancelEnding,
			action:    &mockAction{},
			wfModifier: func(wf *cloudv1.Workflow) {
				wf.Status.Phase = cloudv1.WorkflowPhaseCancelling
				wf.Status.Steps[0].Phase = cloudv1.WorkflowStepPhaseCancelEnding
				wf.Status.CancelEndingSteps = []string{"step1"}
			},
			// Note: According to current implementation, if action doesn't support cancel,
			// it will return an error when step is in CancelEnding phase.
			expectStepPhase:      cloudv1.WorkflowStepPhaseCancelEnding,
			errContainsSubString: "action supports cancellation, but step",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			// Setup workflow
			wf := &cloudv1.Workflow{
				Spec: cloudv1.WorkflowSpec{
					Steps: []cloudv1.WorkflowStep{
						{
							Name:   "step1",
							Uses:   "mockAction",
							Inputs: cloudv1.InputsWrapper{Object: tt.inputs},
						},
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase: cloudv1.WorkflowPhaseRunning,
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Phase:   tt.stepPhase,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
					},
					BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
				},
			}
			if tt.wfModifier != nil {
				tt.wfModifier(wf)
			}

			// Setup reconciler with mock action
			r := &WorkflowReconciler{
				registry: &action.Registry{
					Actions: make(map[cloudv1.WorkflowActionName]action.Action),
				},
			}
			if tt.action != nil {
				r.registry.Actions["mockAction"] = tt.action
			}

			// Execute
			err := r.executeOneStep(context.Background(), wf, tt.stepName)

			// Assert
			if tt.errContainsSubString != "" {
				assert.ErrorContains(t, err, tt.errContainsSubString)
			} else {
				assert.NoError(t, err)
				if tt.expectStepPhase != "" {
					status, _ := wf.StepStatus(context.Background(), tt.stepName)
					assert.Equal(t, tt.expectStepPhase, status.Phase)
				}
			}
		})
	}
}

func Test_WorkflowReconciler_parseAndSetNextExecuteTime(t *testing.T) {
	now := time.Now()
	tests := map[string]struct {
		retryPolicyType      cloudv1.RetryPolicyType
		backoffDuration      time.Duration
		stepStatuses         []*cloudv1.WorkflowStepStatus
		expectRequeue        bool
		expectRequeueAfter   time.Duration
		expectErr            bool
		expectNextExecuteSet bool
	}{
		"RetryPolicyTypeNever should not requeue": {
			retryPolicyType: cloudv1.RetryPolicyTypeNever,
			backoffDuration: time.Second,
			stepStatuses: []*cloudv1.WorkflowStepStatus{
				{Name: "step1", Phase: cloudv1.WorkflowStepPhaseCompleted},
			},
			expectRequeue:        false,
			expectRequeueAfter:   0,
			expectErr:            false,
			expectNextExecuteSet: false,
		},
		"RetryPolicyTypeAlways, no error, should requeue immediately": {
			retryPolicyType: cloudv1.RetryPolicyTypeAlways,
			backoffDuration: time.Second,
			stepStatuses: []*cloudv1.WorkflowStepStatus{
				{Name: "step1", Phase: cloudv1.WorkflowStepPhaseRunning, LastError: ""},
				{Name: "step2", Phase: cloudv1.WorkflowStepPhasePending, LastError: ""},
			},
			expectRequeue:        true,
			expectRequeueAfter:   0,
			expectErr:            false,
			expectNextExecuteSet: false,
		},
		"RetryPolicyTypeAlways, error occurred, should set NextExecuteTime and requeue after backoff": {
			retryPolicyType: cloudv1.RetryPolicyTypeAlways,
			backoffDuration: 2 * time.Second,
			stepStatuses: []*cloudv1.WorkflowStepStatus{
				{Name: "step1", Phase: cloudv1.WorkflowStepPhaseRunning, LastError: "some error"},
				{Name: "step2", Phase: cloudv1.WorkflowStepPhasePending, LastError: ""},
			},
			expectRequeue:        false,
			expectRequeueAfter:   2 * time.Second,
			expectErr:            false,
			expectNextExecuteSet: true,
		},
	}
	for name, tt := range tests {
		t.Run(name, func(t *testing.T) {
			// Setup workflow
			wf := &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test1",
				},
				Spec: cloudv1.WorkflowSpec{
					RetryPolicy: &cloudv1.WorkflowRetryPolicy{
						Type:    tt.retryPolicyType,
						Backoff: &metav1.Duration{Duration: tt.backoffDuration},
					},
				},
				Status: cloudv1.WorkflowStatus{
					Steps: tt.stepStatuses,
				},
			}
			// Setup fake client
			cli := fake.NewClientBuilder().
				WithScheme(scheme.Scheme).
				WithObjects(wf).
				WithStatusSubresource(wf).
				Build()

			r := &WorkflowReconciler{
				client: cli,
				clock:  &fakeClock{now: now},
			}

			res, err := r.parseAndSetNextExecuteTime(context.Background(), wf)
			if tt.expectErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
			if tt.expectRequeue {
				assert.True(t, res.Requeue)
				assert.Zero(t, res.RequeueAfter)
			} else if tt.expectRequeueAfter > 0 {
				assert.InDelta(t, tt.expectRequeueAfter.Seconds(), res.RequeueAfter.Seconds(), 0.1)
			} else {
				assert.False(t, res.Requeue)
				assert.Zero(t, res.RequeueAfter)
			}
			if tt.expectNextExecuteSet {
				assert.NotNil(t, wf.Status.NextExecuteTime)
				assert.WithinDuration(t, now.Add(tt.backoffDuration), wf.Status.NextExecuteTime.Time, time.Second)
			} else {
				assert.Nil(t, wf.Status.NextExecuteTime)
			}
		})
	}
}

// fakeClock implements utiltime.Clock for testing
type fakeClock struct {
	now time.Time
}

func (f *fakeClock) Now() time.Time {
	return f.now
}

// mockAction implements action.Action interface
type mockAction struct {
	handleFunc   func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error
	preCheckFunc func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error
}

func (m *mockAction) Name() cloudv1.WorkflowActionName {
	return "mockAction"
}

func (m *mockAction) Handle(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
	if m.handleFunc != nil {
		return m.handleFunc(ctx, wf, step)
	}
	return nil
}

func (m *mockAction) PreCheck(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
	if m.preCheckFunc != nil {
		return m.preCheckFunc(ctx, wf, step)
	}
	return nil
}

// mockActionWithPost implements both action.Action and action.ActionPost interfaces
type mockActionWithPost struct {
	mockAction
	postHandleFunc func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error
}

func (m *mockActionWithPost) PostHandle(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
	if m.postHandleFunc != nil {
		return m.postHandleFunc(ctx, wf, step)
	}
	return nil
}

// mockActionWithCancel implements both action.Action and action.ActionCancel interfaces
type mockActionWithCancel struct {
	mockAction
	handleCancelFunc     func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error
	postHandleCancelFunc func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error
}

func (m *mockActionWithCancel) Cancel(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep, dryRun bool) error {
	if m.handleCancelFunc != nil {
		return m.handleCancelFunc(ctx, wf, step)
	}
	return nil
}

func (m *mockActionWithCancel) PostCancel(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
	if m.postHandleCancelFunc != nil {
		return m.postHandleCancelFunc(ctx, wf, step)
	}
	return nil
}

func Test_inflateStepInputs(t *testing.T) {
	type args struct {
		inputs     map[string]interface{}
		outputs    map[string]string
		wfModifier func(wf *cloudv1.Workflow)
	}
	tests := []struct {
		name                 string
		args                 args
		want                 map[string]interface{}
		errContainsSubString string
	}{
		{
			name: "variable is invalid should return error",
			args: args{
				inputs: map[string]interface{}{
					"a": "$$$",
				},
			},
			want:                 nil,
			errContainsSubString: "invalid step input variable",
		},
		{
			name: "fromCategory is not 'steps' should return error",
			args: args{
				inputs: map[string]interface{}{
					"a": "$steps111.fromStepName.outputs.key",
				},
			},
			want:                 nil,
			errContainsSubString: "unknown variable prefix",
		},
		{
			name: "fromWhere is not 'outputs' should return error",
			args: args{
				inputs: map[string]interface{}{
					"a": "$steps.fromStepName.outputs1111.key",
				},
			},
			want:                 nil,
			errContainsSubString: "fromWhere must be",
		},
		{
			name: "fromStep is not found should return error",
			args: args{
				inputs: map[string]interface{}{
					"a": "$steps.step3.outputs.key",
				},
			},
			want:                 nil,
			errContainsSubString: "fromStep 'step3' is not found",
		},
		{
			name: "fromStep is not Completed should return error",
			args: args{
				inputs: map[string]interface{}{
					"a": "$steps.step1.outputs.key",
				},
				wfModifier: func(wf *cloudv1.Workflow) {
					wf.Status.Steps[0].Phase = cloudv1.WorkflowStepPhaseRunning
				},
			},
			want:                 nil,
			errContainsSubString: "is not Completed",
		},
		{
			name: "key is not found in fromStep's outputs should return error",
			args: args{
				inputs: map[string]interface{}{
					"a": "$steps.step1.outputs.notFound",
				},
			},
			want:                 nil,
			errContainsSubString: "not found in outputs of fromStep",
		},
		{
			name: "all is fine",
			args: args{
				inputs: map[string]interface{}{
					"inflated": "$steps.step1.outputs.output1",
					"plain":    "abc",
				},
				outputs: map[string]string{
					"output1": "123",
				},
			},
			want: map[string]interface{}{
				"inflated": "123",
				"plain":    "abc",
			},
			errContainsSubString: "",
		}, {
			name: "all is fine, non-string value should be untouched",
			args: args{
				inputs: map[string]interface{}{
					"inflated": "$steps.step1.outputs.output1",
					"plain":    "abc",
					"labels":   map[string]string{"a": "b", "b": "c"},
					"count":    42,
				},
				outputs: map[string]string{
					"output1": "123",
				},
			},
			want: map[string]interface{}{
				"inflated": "123",
				"plain":    "abc",
				"count":    42,
				"labels":   map[string]string{"a": "b", "b": "c"},
			},
			errContainsSubString: "",
		}, {
			name: "status variable, beginAt is nil should return empty string",
			args: args{
				inputs: map[string]interface{}{
					"inflated": "$steps.step1.status.beginTime",
					"plain":    "abc",
				},
				wfModifier: func(wf *cloudv1.Workflow) {
					lo.Must(wf.StepStatus(context.Background(), "step1")).BeginAt = nil
				},
			},
			want: map[string]interface{}{
				"inflated": "",
				"plain":    "abc",
			},
			errContainsSubString: "",
		}, {
			name: "status variable, beginAt is not nil should return string",
			args: args{
				inputs: map[string]interface{}{
					"inflated": "$steps.step1.status.beginTime",
					"plain":    "abc",
				},
				wfModifier: func(wf *cloudv1.Workflow) {
					status := lo.Must(wf.StepStatus(context.Background(), "step1"))
					status.BeginAt = &metav1.Time{Time: lo.Must(time.Parse(time.RFC3339, "2022-01-01T00:00:00Z"))}
				},
			},
			want: map[string]interface{}{
				"inflated": "2022-01-01T00:00:00Z",
				"plain":    "abc",
			},
			errContainsSubString: "",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			wf := &cloudv1.Workflow{
				Spec: cloudv1.WorkflowSpec{
					Steps: []cloudv1.WorkflowStep{
						{
							Name: "step1",
						},
						{
							Name: "step2",
						},
					},
				},
				Status: cloudv1.WorkflowStatus{
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Outputs: tt.args.outputs,
							Phase:   cloudv1.WorkflowStepPhaseCompleted,
						},
					},
				},
			}
			if tt.args.wfModifier != nil {
				tt.args.wfModifier(wf)
			}
			got, err := inflateStepInputs(context.Background(), wf, tt.args.inputs)
			if tt.errContainsSubString != "" {
				assert.ErrorContains(t, err, tt.errContainsSubString)
			} else {
				assert.NoError(t, err)
			}
			assert.Equalf(t, tt.want, got, "compare inflated inputs")
		})
	}
}

func Test_inflateFromWorkflow(t *testing.T) {
	type args struct {
		inputs     map[string]interface{}
		wfModifier func(wf *cloudv1.Workflow)
	}
	tests := map[string]struct {
		args                 args
		want                 interface{}
		errContainsSubString string
	}{
		"invalid workflow input variable": {
			args: args{
				inputs: map[string]interface{}{
					"a": "$wf.spec.inputs",
				},
			},
			want:                 nil,
			errContainsSubString: "invalid workflow input variable",
		},
		"fromCategory is not wf": {
			args: args{
				inputs: map[string]interface{}{
					"a": "$notwf.spec.inputs.key",
				},
			},
			want:                 nil,
			errContainsSubString: "fromCategory must be 'wf'",
		},
		"fromWhere is not inputs": {
			args: args{
				inputs: map[string]interface{}{
					"a": "$wf.spec.notinputs.key",
				},
			},
			want:                 nil,
			errContainsSubString: "fromWhere must be 'inputs'",
		},
		"key not found in workflow inputs": {
			args: args{
				inputs: map[string]interface{}{
					"a": "$wf.spec.inputs.notfound",
				},
			},
			want:                 nil,
			errContainsSubString: "is not found in inputs of workflow",
		},
		"all is fine": {
			args: args{
				inputs: map[string]interface{}{
					"a": "$wf.spec.inputs.key1",
				},
				wfModifier: func(wf *cloudv1.Workflow) {
					wf.Spec.Inputs = map[string]string{
						"key1": "value1",
					}
				},
			},
			want:                 "value1",
			errContainsSubString: "",
		},
		"plain text": {
			args: args{
				inputs: map[string]interface{}{
					"a": "plain text",
				},
			},
			want:                 "plain text",
			errContainsSubString: "",
		},
		"non-string value in workflow inputs": {
			args: args{
				inputs: map[string]interface{}{
					"a": "$wf.spec.inputs.key2",
				},
				wfModifier: func(wf *cloudv1.Workflow) {
					wf.Spec.Inputs = map[string]string{
						"key2": "42",
					}
				},
			},
			want:                 "42",
			errContainsSubString: "",
		},
	}
	for name, tt := range tests {
		t.Run(name, func(t *testing.T) {
			wf := &cloudv1.Workflow{
				Spec: cloudv1.WorkflowSpec{
					Inputs: map[string]string{},
				},
			}
			if tt.args.wfModifier != nil {
				tt.args.wfModifier(wf)
			}
			for k, v := range tt.args.inputs {
				got, err := inflateFromWorkflow(context.Background(), wf, k, v.(string))
				if tt.errContainsSubString != "" {
					assert.ErrorContains(t, err, tt.errContainsSubString)
				} else {
					assert.NoError(t, err)
					assert.Equalf(t, tt.want, got, "compare inflated workflow input")
				}
			}
		})
	}
}

func Test_getValueFromStepStatus(t *testing.T) {
	type args struct {
		statusModifier func(status *cloudv1.WorkflowStepStatus)
		key            string
	}
	tests := []struct {
		name                 string
		args                 args
		want                 string
		errContainsSubString string
	}{
		{
			name: "extra from beginTime",
			args: args{
				key: "beginTime",
			},
			want: "2022-01-01T00:00:00Z",
		},
		{
			name: "extra from endTime",
			args: args{
				key: "endTime",
			},
			want: "2022-01-01T01:00:00Z",
		},
		{
			name: "extra from phase is not supported",
			args: args{
				key: "phase",
			},
			want:                 "",
			errContainsSubString: "unsupported key",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			status := &cloudv1.WorkflowStepStatus{
				Name:      "step1",
				Phase:     cloudv1.WorkflowStepPhaseCompleted,
				Reason:    "success",
				LastError: "",
				Count:     1,
				BeginAt:   &metav1.Time{Time: lo.Must(time.Parse(time.RFC3339, "2022-01-01T00:00:00Z"))},
				EndAt:     &metav1.Time{Time: lo.Must(time.Parse(time.RFC3339, "2022-01-01T01:00:00Z"))},
				Duration:  "1h",
				Outputs:   map[string]string{},
			}
			if tt.args.statusModifier != nil {
				tt.args.statusModifier(status)
			}
			got, err := getValueFromStepStatus(status, tt.args.key)
			if tt.errContainsSubString != "" {
				assert.ErrorContains(t, err, tt.errContainsSubString)
			} else {
				assert.NoError(t, err)
			}
			assert.Equalf(t, tt.want, got, "compare result")
		})
	}
}

func Test_handleCancelling(t *testing.T) {
	tests := []struct {
		name                    string
		wf                      *cloudv1.Workflow
		expectPhase             cloudv1.WorkflowPhase
		expectCancellingSteps   []string
		expectCancelEndingSteps []string
		errContainsSubString    string
	}{
		{
			name: "should schedule pending steps to even if it doesn't implment ActionCancel",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
				},
				Spec: cloudv1.WorkflowSpec{
					Steps: []cloudv1.WorkflowStep{
						{
							Name:      "step1",
							StepGroup: "group1",
							Uses:      "mockAction",
						},
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase: cloudv1.WorkflowPhaseCancelling,
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Phase:   cloudv1.WorkflowStepPhasePending,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
					},
				},
			},
			expectPhase: cloudv1.WorkflowPhaseCancelled,
		},
		{
			name: "should schedule pending steps to cancelled",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
				},
				Spec: cloudv1.WorkflowSpec{
					Steps: []cloudv1.WorkflowStep{
						{
							Name:      "step1",
							StepGroup: "group1",
							Uses:      "mockActionWithCancel",
						},
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase: cloudv1.WorkflowPhaseCancelling,
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Phase:   cloudv1.WorkflowStepPhasePending,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
					},
				},
			},
			// When all steps are cancelled, the workflow transitions to Cancelled phase
			expectPhase: cloudv1.WorkflowPhaseCancelled,
		},
		{
			name: "should wait for NextExecuteTime if set",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
				},
				Spec: cloudv1.WorkflowSpec{
					Steps: []cloudv1.WorkflowStep{
						{
							Name: "step1",
							Uses: "mockAction",
						},
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase: cloudv1.WorkflowPhaseCancelling,
					NextExecuteTime: &metav1.Time{
						Time: time.Now().Add(1 * time.Second),
					},
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Phase:   cloudv1.WorkflowStepPhasePending,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
					},
				},
			},
			expectPhase: cloudv1.WorkflowPhaseCancelling,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tt.wf.Default()
			registry := &action.Registry{
				Actions: map[cloudv1.WorkflowActionName]action.Action{
					"mockAction":           &mockAction{},
					"mockActionWithCancel": &mockActionWithCancel{},
				},
			}
			cli := newFakeClient([]client.Object{tt.wf})
			ctx := context.Background()
			r := &WorkflowReconciler{
				clock:     &fakeClock{now: time.Now()},
				registry:  registry,
				scheduler: NewSequenceScheduler(registry),
				client:    cli,
			}

			result, err := r.handleCancelling(ctx, tt.wf)

			if tt.errContainsSubString != "" {
				assert.ErrorContains(t, err, tt.errContainsSubString)
			} else {
				assert.NoError(t, err)
				if tt.expectPhase != "" {
					assert.Equal(t, tt.expectPhase, tt.wf.Status.Phase)
				}
				if len(tt.expectCancellingSteps) > 0 {
					assert.Equal(t, tt.expectCancellingSteps, tt.wf.Status.CancellingSteps)
				}
				if len(tt.expectCancelEndingSteps) > 0 {
					assert.Equal(t, tt.expectCancelEndingSteps, tt.wf.Status.CancelEndingSteps)
				}
				// If NextExecuteTime is set, should requeue
				if tt.wf.Status.NextExecuteTime != nil && time.Now().Before(tt.wf.Status.NextExecuteTime.Time) {
					assert.True(t, result.RequeueAfter > 0)
				}
			}
		})
	}
}

func Test_executeCancellingSteps(t *testing.T) {
	tests := []struct {
		name                 string
		wf                   *cloudv1.Workflow
		action               action.Action
		expectPhase          cloudv1.WorkflowStepPhase
		errContainsSubString string
	}{
		{
			name: "should execute cancelling steps successfully",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
				},
				Spec: cloudv1.WorkflowSpec{
					Steps: []cloudv1.WorkflowStep{
						{
							Name: "step1",
							Uses: "mockActionWithCancel",
						},
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase:           cloudv1.WorkflowPhaseCancelling,
					CancellingSteps: []string{"step1"},
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Phase:   cloudv1.WorkflowStepPhaseCancelling,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
					},
				},
			},
			action: &mockActionWithCancel{
				mockAction: mockAction{},
				handleCancelFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return nil
				},
			},
			expectPhase: cloudv1.WorkflowStepPhaseCancelEnding,
		},
		{
			name: "should handle error in cancelling step and record it in step status",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
				},
				Spec: cloudv1.WorkflowSpec{
					Steps: []cloudv1.WorkflowStep{
						{
							Name: "step1",
							Uses: "mockActionWithCancel",
						},
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase:           cloudv1.WorkflowPhaseCancelling,
					CancellingSteps: []string{"step1"},
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Phase:   cloudv1.WorkflowStepPhaseCancelling,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
					},
				},
			},
			action: &mockActionWithCancel{
				mockAction: mockAction{},
				handleCancelFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return fmt.Errorf("cancel error")
				},
			},
			expectPhase: cloudv1.WorkflowStepPhaseCancelling, // Error is recorded but step remains in Cancelling phase
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tt.wf.Default()
			registry := &action.Registry{
				Actions: map[cloudv1.WorkflowActionName]action.Action{
					"mockActionWithCancel": tt.action,
				},
			}
			cli := newFakeClient([]client.Object{tt.wf})
			ctx := context.Background()
			r := &WorkflowReconciler{
				registry: registry,
				client:   cli,
			}

			err := r.executeCancellingSteps(ctx, tt.wf)

			if tt.errContainsSubString != "" {
				assert.ErrorContains(t, err, tt.errContainsSubString)
			} else {
				assert.NoError(t, err)
				if tt.expectPhase != "" {
					status, ok := tt.wf.StepStatus(ctx, "step1")
					assert.True(t, ok, "step1 should exist")
					assert.NotNil(t, status, "step status should not be nil")
					assert.Equal(t, tt.expectPhase, status.Phase)
					// If error was returned, it should be recorded in LastError
					if status.Phase == cloudv1.WorkflowStepPhaseCancelling && status.LastError != "" {
						assert.Contains(t, status.LastError, "cancel error")
					}
				}
			}
		})
	}
}

func Test_executeCancelEndingSteps(t *testing.T) {
	tests := []struct {
		name                 string
		wf                   *cloudv1.Workflow
		action               action.Action
		expectPhase          cloudv1.WorkflowStepPhase
		errContainsSubString string
	}{
		{
			name: "should execute cancel ending steps successfully",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
				},
				Spec: cloudv1.WorkflowSpec{
					Steps: []cloudv1.WorkflowStep{
						{
							Name: "step1",
							Uses: "mockActionWithCancel",
						},
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase:             cloudv1.WorkflowPhaseCancelling,
					CancelEndingSteps: []string{"step1"},
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Phase:   cloudv1.WorkflowStepPhaseCancelEnding,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
					},
				},
			},
			action: &mockActionWithCancel{
				mockAction: mockAction{},
				postHandleCancelFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return nil
				},
			},
			expectPhase: cloudv1.WorkflowStepPhaseCancelled,
		},
		{
			name: "should handle error in cancel ending step and record it in step status",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
				},
				Spec: cloudv1.WorkflowSpec{
					Steps: []cloudv1.WorkflowStep{
						{
							Name: "step1",
							Uses: "mockActionWithCancel",
						},
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase:             cloudv1.WorkflowPhaseCancelling,
					CancelEndingSteps: []string{"step1"},
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Phase:   cloudv1.WorkflowStepPhaseCancelEnding,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
					},
				},
			},
			action: &mockActionWithCancel{
				mockAction: mockAction{},
				postHandleCancelFunc: func(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
					return fmt.Errorf("post cancel error")
				},
			},
			expectPhase: cloudv1.WorkflowStepPhaseCancelEnding, // Error is recorded but step remains in CancelEnding phase
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tt.wf.Default()
			registry := &action.Registry{
				Actions: map[cloudv1.WorkflowActionName]action.Action{
					"mockActionWithCancel": tt.action,
				},
			}
			cli := newFakeClient([]client.Object{tt.wf})
			ctx := context.Background()
			r := &WorkflowReconciler{
				registry: registry,
				client:   cli,
			}

			err := r.executeCancelEndingSteps(ctx, tt.wf)

			if tt.errContainsSubString != "" {
				assert.ErrorContains(t, err, tt.errContainsSubString)
			} else {
				assert.NoError(t, err)
				if tt.expectPhase != "" {
					status, ok := tt.wf.StepStatus(ctx, "step1")
					assert.True(t, ok, "step1 should exist")
					assert.NotNil(t, status, "step status should not be nil")
					assert.Equal(t, tt.expectPhase, status.Phase)
					// If error was returned, it should be recorded in LastError
					if status.Phase == cloudv1.WorkflowStepPhaseCancelEnding && status.LastError != "" {
						assert.Contains(t, status.LastError, "post cancel error")
					}
				}
			}
		})
	}
}

func Test_handleCancelAnnotation(t *testing.T) {
	tests := []struct {
		name                 string
		wf                   *cloudv1.Workflow
		expectPhase          cloudv1.WorkflowPhase
		expectStepPhases     map[string]cloudv1.WorkflowStepPhase
		errContainsSubString string
	}{
		{
			name: "should handle cancel annotation for Pending workflow",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
					Annotations: map[string]string{
						"tidbcloud.com/cancel-workflow": "true",
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase: cloudv1.WorkflowPhasePending,
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Phase:   cloudv1.WorkflowStepPhasePending,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
					},
				},
			},
			expectPhase: cloudv1.WorkflowPhaseCancelled,
			expectStepPhases: map[string]cloudv1.WorkflowStepPhase{
				"step1": cloudv1.WorkflowStepPhaseCancelled,
			},
		},
		{
			name: "should handle cancel annotation for None phase workflow",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
					Annotations: map[string]string{
						"tidbcloud.com/cancel-workflow": "true",
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase: cloudv1.WorkflowPhaseNone,
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Phase:   cloudv1.WorkflowStepPhasePending,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
					},
				},
			},
			expectPhase: cloudv1.WorkflowPhaseCancelled,
			expectStepPhases: map[string]cloudv1.WorkflowStepPhase{
				"step1": cloudv1.WorkflowStepPhaseCancelled,
			},
		},
		{
			name: "should handle cancel annotation for Running workflow",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
					Annotations: map[string]string{
						"tidbcloud.com/cancel-workflow": "true",
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase:        cloudv1.WorkflowPhaseRunning,
					RunningSteps: []string{"step1"},
					Steps: []*cloudv1.WorkflowStepStatus{
						{
							Name:    "step1",
							Phase:   cloudv1.WorkflowStepPhaseRunning,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
						{
							Name:    "step2",
							Phase:   cloudv1.WorkflowStepPhasePending,
							BeginAt: lo.ToPtr(metav1.NewTime(time.Now())),
						},
					},
				},
				Spec: cloudv1.WorkflowSpec{
					Steps: []cloudv1.WorkflowStep{
						{Name: "step1", StepGroup: "group1", Uses: "mockAction"},
						{Name: "step2", StepGroup: "group2", Uses: "mockAction"},
					},
				},
			},
			expectPhase: cloudv1.WorkflowPhaseCancelling,
			expectStepPhases: map[string]cloudv1.WorkflowStepPhase{
				// Note: handleCancelAnnotation only changes the workflow phase to Cancelling.
				// It does NOT convert steps. Step conversion is handled by CancelWfScheduler
				// in handleCancelling during the next reconcile.
				// So step1 remains Running and step2 remains Pending after handleCancelAnnotation.
				"step1": cloudv1.WorkflowStepPhaseRunning,
				"step2": cloudv1.WorkflowStepPhasePending,
			},
		},
		{
			name: "should ignore cancel annotation for already cancelled workflow",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
					Annotations: map[string]string{
						"tidbcloud.com/cancel-workflow": "true",
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase: cloudv1.WorkflowPhaseCancelled,
				},
			},
			expectPhase: cloudv1.WorkflowPhaseCancelled,
		},
		{
			name: "should ignore cancel annotation for completed workflow",
			wf: &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: "default",
					Name:      "test-workflow",
					Annotations: map[string]string{
						"tidbcloud.com/cancel-workflow": "true",
					},
				},
				Status: cloudv1.WorkflowStatus{
					Phase: cloudv1.WorkflowPhaseCompleted,
				},
			},
			expectPhase: cloudv1.WorkflowPhaseCompleted,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tt.wf.Default()
			cli := newFakeClient([]client.Object{tt.wf})
			ctx := context.Background()
			r := &WorkflowReconciler{
				client: cli,
			}

			err := r.handleCancelAnnotation(ctx, tt.wf)

			if tt.errContainsSubString != "" {
				assert.ErrorContains(t, err, tt.errContainsSubString)
			} else {
				assert.NoError(t, err)
				if tt.expectPhase != "" {
					assert.Equal(t, tt.expectPhase, tt.wf.Status.Phase)
				}
				if len(tt.expectStepPhases) > 0 {
					for stepName, expectedPhase := range tt.expectStepPhases {
						status, ok := tt.wf.StepStatus(ctx, stepName)
						assert.True(t, ok, "step %s should exist", stepName)
						assert.NotNil(t, status, "step %s status should not be nil", stepName)
						assert.Equal(t, expectedPhase, status.Phase, "step %s phase", stepName)
					}
				}
			}
		})
	}
}

func newFakeClient(initObjs []client.Object) client.Client {
	cli := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(initObjs...).
		WithStatusSubresource(initObjs...).
		Build()
	return cli
}