/*
Copyright 2019 The Volcano Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package job

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"testing"
	"time"

	"github.com/agiledragon/gomonkey/v2"
	v1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	"k8s.io/utils/ptr"

	"volcano.sh/apis/pkg/apis/batch/v1alpha1"
	schedulingapi "volcano.sh/apis/pkg/apis/scheduling/v1beta1"

	"volcano.sh/volcano/pkg/controllers/apis"
	"volcano.sh/volcano/pkg/controllers/job/state"
)

func TestKillJobFunc(t *testing.T) {
	namespace := "test"

	testcases := []struct {
		Name           string
		Job            *v1alpha1.Job
		PodGroup       *schedulingapi.PodGroup
		PodRetainPhase state.PhaseMap
		UpdateStatus   state.UpdateStatusFn
		JobInfo        *apis.JobInfo
		Services       []v1.Service
		ConfigMaps     []v1.ConfigMap
		Secrets        []v1.Secret
		Pods           map[string]*v1.Pod
		Plugins        []string
		ExpectVal      error
	}{
		{
			Name: "KillJob success Case",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job1",
					Namespace:       namespace,
					UID:             "e7f18111-1cec-11ea-b688-fa163ec79500",
					ResourceVersion: "100",
				},
			},
			PodGroup: &schedulingapi.PodGroup{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "job1-e7f18111-1cec-11ea-b688-fa163ec79500",
					Namespace: namespace,
				},
			},
			PodRetainPhase: state.PodRetainPhaseNone,
			UpdateStatus:   nil,
			JobInfo: &apis.JobInfo{
				Namespace: namespace,
				Name:      "jobinfo1",
				Pods: map[string]map[string]*v1.Pod{
					"task1": {
						"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
						"pod2": buildPod(namespace, "pod2", v1.PodRunning, nil),
					},
				},
			},
			Services: []v1.Service{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "job1",
						Namespace: namespace,
					},
				},
			},
			Secrets: []v1.Secret{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "job1-e7f18111-1cec-11ea-b688-fa163ec79500-ssh",
						Namespace: namespace,
					},
				},
			},
			Pods: map[string]*v1.Pod{
				"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
				"pod2": buildPod(namespace, "pod2", v1.PodRunning, nil),
			},
			Plugins:   []string{"svc", "ssh", "env"},
			ExpectVal: nil,
		},
		{
			Name: "KillJob compatibility with version lt 1.5 success case",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job2",
					Namespace:       namespace,
					UID:             "e7f18111-1cec-11ea-b688-fa163ec79500",
					ResourceVersion: "100",
				},
			},
			PodGroup: &schedulingapi.PodGroup{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "job2",
					Namespace: namespace,
				},
			},
			PodRetainPhase: state.PodRetainPhaseNone,
			UpdateStatus:   nil,
			JobInfo: &apis.JobInfo{
				Namespace: namespace,
				Name:      "jobinfo2",
				Pods: map[string]map[string]*v1.Pod{
					"task1": {
						"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
						"pod2": buildPod(namespace, "pod2", v1.PodRunning, nil),
					},
				},
			},
			Services: []v1.Service{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "job2",
						Namespace: namespace,
					},
				},
			},
			Secrets: []v1.Secret{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "job2-e7f18111-1cec-11ea-b688-fa163ec79500-ssh",
						Namespace: namespace,
					},
				},
			},
			Pods: map[string]*v1.Pod{
				"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
				"pod2": buildPod(namespace, "pod2", v1.PodRunning, nil),
			},
			Plugins:   []string{"svc", "ssh", "env"},
			ExpectVal: nil,
		},
	}

	for i, testcase := range testcases {

		t.Run(testcase.Name, func(t *testing.T) {
			fakeController := newFakeController()
			jobPlugins := make(map[string][]string)

			for _, service := range testcase.Services {
				_, err := fakeController.kubeClient.CoreV1().Services(namespace).Create(context.TODO(), &service, metav1.CreateOptions{})
				if err != nil {
					t.Error("Error While Creating Service")
				}
			}

			for _, secret := range testcase.Secrets {
				_, err := fakeController.kubeClient.CoreV1().Secrets(namespace).Create(context.TODO(), &secret, metav1.CreateOptions{})
				if err != nil {
					t.Error("Error While Creating Secret.")
				}
			}

			for _, pod := range testcase.Pods {
				_, err := fakeController.kubeClient.CoreV1().Pods(namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
				if err != nil {
					t.Error("Error While Creating ConfigMaps")
				}
			}

			_, err := fakeController.vcClient.BatchV1alpha1().Jobs(namespace).Create(context.TODO(), testcase.Job, metav1.CreateOptions{})
			if err != nil {
				t.Error("Error While Creating Jobs")
			}
			err = fakeController.cache.Add(testcase.Job)
			if err != nil {
				t.Error("Error While Adding Job in cache")
			}

			for _, plugin := range testcase.Plugins {
				jobPlugins[plugin] = make([]string, 0)
			}

			testcase.JobInfo.Job = testcase.Job
			testcase.JobInfo.Job.Spec.Plugins = jobPlugins

			testcase.JobInfo.Job.Status.ControlledResources = map[string]string{}
			for _, name := range testcase.Plugins {
				testcase.JobInfo.Job.Status.ControlledResources["plugin-"+name] = name
			}

			err = fakeController.killJob(testcase.JobInfo, testcase.PodRetainPhase, testcase.UpdateStatus)
			if err != nil {
				t.Errorf("Case %d (%s): expected: No Error, but got error %v.", i, testcase.Name, err)
			}

			for _, plugin := range testcase.Plugins {

				if plugin == "svc" {
					_, err = fakeController.kubeClient.CoreV1().Services(namespace).Get(context.TODO(), testcase.Job.Name, metav1.GetOptions{})
					if err == nil {
						t.Errorf("Case %d (%s): expected: Service to be deleted, but not deleted.", i, testcase.Name)
					}
				}

				if plugin == "ssh" {
					_, err := fakeController.kubeClient.CoreV1().ConfigMaps(namespace).Get(context.TODO(),
						fmt.Sprintf("%s-%s-%s", testcase.Job.Name, testcase.Job.UID, "ssh"), metav1.GetOptions{})
					if err == nil {
						t.Errorf("Case %d (%s): expected: Secret to be deleted, but not deleted.", i, testcase.Name)
					}
				}
			}
		})
	}
}

func TestSyncJobFunc(t *testing.T) {
	namespace := "test"

	testcases := []struct {
		Name           string
		Job            *v1alpha1.Job
		PodGroup       *schedulingapi.PodGroup
		PodRetainPhase state.PhaseMap
		UpdateStatus   state.UpdateStatusFn
		JobInfo        *apis.JobInfo
		Pods           map[string]*v1.Pod
		Plugins        []string
		TotalNumPods   int
		ExpectVal      error
	}{
		{
			Name: "SyncJob success Case",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job1",
					Namespace:       namespace,
					ResourceVersion: "100",
					UID:             "e7f18111-1cec-11ea-b688-fa163ec79500",
				},
				Spec: v1alpha1.JobSpec{
					Tasks: []v1alpha1.TaskSpec{
						{
							Name:     "task1",
							Replicas: 6,
							Template: v1.PodTemplateSpec{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "pods",
									Namespace: namespace,
								},
								Spec: v1.PodSpec{
									Containers: []v1.Container{
										{
											Name: "Containers",
										},
									},
								},
							},
						},
					},
				},
				Status: v1alpha1.JobStatus{
					State: v1alpha1.JobState{
						Phase: v1alpha1.Pending,
					},
				},
			},
			PodGroup: &schedulingapi.PodGroup{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "job1-e7f18111-1cec-11ea-b688-fa163ec79500",
					Namespace: namespace,
				},
				Spec: schedulingapi.PodGroupSpec{
					MinResources:  &v1.ResourceList{},
					MinTaskMember: map[string]int32{},
				},
				Status: schedulingapi.PodGroupStatus{
					Phase: schedulingapi.PodGroupInqueue,
				},
			},
			PodRetainPhase: state.PodRetainPhaseNone,
			UpdateStatus:   nil,
			JobInfo: &apis.JobInfo{
				Namespace: namespace,
				Name:      "jobinfo1",
				Pods: map[string]map[string]*v1.Pod{
					"task1": {
						"job1-task1-0": buildPod(namespace, "job1-task1-0", v1.PodRunning, nil),
						"job1-task1-1": buildPod(namespace, "job1-task1-1", v1.PodRunning, nil),
					},
				},
			},
			Pods: map[string]*v1.Pod{
				"job1-task1-0": buildPod(namespace, "job1-task1-0", v1.PodRunning, nil),
				"job1-task1-1": buildPod(namespace, "job1-task1-1", v1.PodRunning, nil),
			},
			TotalNumPods: 6,
			Plugins:      []string{"svc", "ssh", "env"},
			ExpectVal:    nil,
		},
		{
			Name: "SyncJob with dependsOn job can't find the dependent task",
			/*
				Work dependsOn Master task, preempt actions causes controller deadlock
				   controller                master,work             scheduler
				       |                           |  <---preempt-----  |
				       |                           |  <---kill work---  |
				       | ----watch work kill---->  |                    |
					   |                           |  <---kill master-- |
				       | ----create work pods--->  |                    |
				       | --wait master running-->  |                    |
				       |                           |                    |
					   | ---watch master kill--->  |                    |
				       | --push master to queue->  |                    |
				       | -wait process to create-> |                    |
			*/
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job1",
					Namespace:       namespace,
					ResourceVersion: "100",
					UID:             "e7f18111-1cec-11ea-b688-fa163ec79500",
				},
				Spec: v1alpha1.JobSpec{
					Tasks: []v1alpha1.TaskSpec{
						{
							Name:     "master",
							Replicas: 1,
							Template: v1.PodTemplateSpec{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "pods",
									Namespace: namespace,
								},
								Spec: v1.PodSpec{
									Containers: []v1.Container{
										{
											Name: "Containers",
										},
									},
								},
							},
						},
						{
							Name:     "work",
							Replicas: 3,
							Template: v1.PodTemplateSpec{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "pods",
									Namespace: namespace,
								},
								Spec: v1.PodSpec{
									Containers: []v1.Container{
										{
											Name: "Containers",
										},
									},
								},
							},
							DependsOn: &v1alpha1.DependsOn{
								Name: []string{"master"},
							},
						},
					},
				},
				Status: v1alpha1.JobStatus{
					State: v1alpha1.JobState{
						Phase: v1alpha1.Pending,
					},
				},
			},
			PodGroup: &schedulingapi.PodGroup{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "job1-e7f18111-1cec-11ea-b688-fa163ec79500",
					Namespace: namespace,
				},
				Spec: schedulingapi.PodGroupSpec{
					MinResources:  &v1.ResourceList{},
					MinTaskMember: map[string]int32{},
				},
				Status: schedulingapi.PodGroupStatus{
					Phase: schedulingapi.PodGroupInqueue,
				},
			},
			PodRetainPhase: state.PodRetainPhaseNone,
			UpdateStatus:   nil,
			JobInfo: &apis.JobInfo{
				Namespace: namespace,
				Name:      "jobinfo1",
				Pods: map[string]map[string]*v1.Pod{
					"work": {
						"job1-work-0": buildPod(namespace, "job1-work-0", v1.PodRunning, nil),
						"job1-work-1": buildPod(namespace, "job1-work-1", v1.PodRunning, nil),
					},
				},
			},
			Pods: map[string]*v1.Pod{
				"job1-work-0": buildPod(namespace, "job1-work-0", v1.PodRunning, nil),
				"job1-work-1": buildPod(namespace, "job1-work-1", v1.PodRunning, nil),
			},
			TotalNumPods: 4,
			Plugins:      []string{"svc", "ssh", "env"},
			ExpectVal:    nil,
		},
	}
	for i, testcase := range testcases {

		t.Run(testcase.Name, func(t *testing.T) {
			fakeController := newFakeController()

			patches := gomonkey.ApplyMethod(reflect.TypeOf(fakeController), "GetQueueInfo", func(_ *jobcontroller, _ string) (*schedulingapi.Queue, error) {
				return &schedulingapi.Queue{}, nil
			})

			defer patches.Reset()

			jobPlugins := make(map[string][]string)

			for _, plugin := range testcase.Plugins {
				jobPlugins[plugin] = make([]string, 0)
			}
			testcase.JobInfo.Job = testcase.Job
			testcase.JobInfo.Job.Spec.Plugins = jobPlugins

			fakeController.pgInformer.Informer().GetIndexer().Add(testcase.PodGroup)
			fakeController.vcClient.SchedulingV1beta1().PodGroups(testcase.PodGroup.Namespace).Create(context.TODO(), testcase.PodGroup, metav1.CreateOptions{})

			for _, pod := range testcase.Pods {
				_, err := fakeController.kubeClient.CoreV1().Pods(namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
				if err != nil {
					t.Error("Error While Creating pods")
				}
			}

			_, err := fakeController.vcClient.BatchV1alpha1().Jobs(namespace).Create(context.TODO(), testcase.Job, metav1.CreateOptions{})
			if err != nil {
				t.Errorf("Expected no Error while creating job, but got error: %s", err)
			}

			err = fakeController.cache.Add(testcase.Job)
			if err != nil {
				t.Error("Error While Adding Job in cache")
			}

			err = fakeController.syncJob(testcase.JobInfo, nil)
			if err != testcase.ExpectVal {
				t.Errorf("Expected no error while syncing job, but got error: %s", err)
			}

			podList, err := fakeController.kubeClient.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{})
			if err != nil {
				t.Errorf("Expected no error while listing pods, but got error %s in case %d", err, i)
			}
			if testcase.TotalNumPods != len(podList.Items) {
				t.Errorf("Expected Total number of pods to be same as podlist count: Expected: %d, Got: %d in case: %d", testcase.TotalNumPods, len(podList.Items), i)
			}
		})
	}
}

func TestCreateJobIOIfNotExistFunc(t *testing.T) {
	namespace := "test"

	testcases := []struct {
		Name      string
		Job       *v1alpha1.Job
		ExpextVal error
	}{
		{
			Name: "Create Job IO case",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job1",
					Namespace:       namespace,
					ResourceVersion: "100",
				},
				Spec: v1alpha1.JobSpec{
					Volumes: []v1alpha1.VolumeSpec{
						{
							VolumeClaimName: "pvc1",
						},
					},
				},
			},
			ExpextVal: errors.New("pvc pvc1 is not found, the job will be in the Pending state until the PVC is created"),
		},
	}

	for i, testcase := range testcases {

		t.Run(testcase.Name, func(t *testing.T) {
			fakeController := newFakeController()

			job, err := fakeController.createJobIOIfNotExist(testcase.Job)
			if testcase.ExpextVal == nil {
				if err != nil {
					t.Errorf("Expected Return value to be : %v, but got: %v in testcase %d", testcase.ExpextVal, err, i)
				}
			} else {
				if err == nil || err.Error() != testcase.ExpextVal.Error() {
					t.Errorf("Expected Return value to be : %v, but got: %v in testcase %d", testcase.ExpextVal.Error(), err.Error(), i)
				}
			}

			if len(job.Spec.Volumes) == 0 {
				t.Errorf("Expected number of volumes to be greater than 0 but got: %d in case: %d", len(job.Spec.Volumes), i)
			}
		})
	}
}

func TestCreatePVCFunc(t *testing.T) {
	namespace := "test"

	testcases := []struct {
		Name        string
		Job         *v1alpha1.Job
		VolumeClaim *v1.PersistentVolumeClaimSpec
		ExpextVal   error
	}{
		{
			Name: "CreatePVC success Case",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job1",
					Namespace:       namespace,
					ResourceVersion: "100",
				},
			},
			VolumeClaim: &v1.PersistentVolumeClaimSpec{
				VolumeName: "vol1",
			},
			ExpextVal: nil,
		},
	}

	for _, testcase := range testcases {
		t.Run(testcase.Name, func(t *testing.T) {
			fakeController := newFakeController()

			err := fakeController.createPVC(testcase.Job, "pvc1", testcase.VolumeClaim)
			if err != testcase.ExpextVal {
				t.Errorf("Expected return value to be equal to expected: %s, but got: %s", testcase.ExpextVal, err)
			}
			_, err = fakeController.kubeClient.CoreV1().PersistentVolumeClaims(namespace).Get(context.TODO(), "pvc1", metav1.GetOptions{})
			if err != nil {
				t.Error("Expected PVC to get created, but not created")
			}
		})
	}
}

func TestCreatePodGroupIfNotExistFunc(t *testing.T) {
	namespace := "test"
	highestTierAllowed := 1

	testcases := []struct {
		Name                string
		Job                 *v1alpha1.Job
		ExpectMinTaskMember map[string]int32
		ExpextVal           error
	}{
		{
			Name: "CreatePodGroup success Case",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: namespace,
					Name:      "job1",
					UID:       "e7f18111-1cec-11ea-b688-fa163ec79500",
				},
				Spec: v1alpha1.JobSpec{
					SchedulerName: "volcano",
					Volumes: []v1alpha1.VolumeSpec{
						{
							VolumeClaimName: "vc1",
							VolumeClaim: &v1.PersistentVolumeClaimSpec{
								VolumeName: "v1",
							},
						},
						{
							VolumeClaimName: "vc2",
						},
					},
					Tasks: []v1alpha1.TaskSpec{
						{
							Name:     "task1",
							Replicas: 6,
							PartitionPolicy: &v1alpha1.PartitionPolicySpec{
								TotalPartitions: 2,
								PartitionSize:   3,
								NetworkTopology: &v1alpha1.NetworkTopologySpec{
									Mode:               "hard",
									HighestTierAllowed: &highestTierAllowed,
								},
							},
							Template: v1.PodTemplateSpec{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "pods",
									Namespace: namespace,
								},
								Spec: v1.PodSpec{
									Containers: []v1.Container{
										{
											Name: "Containers",
										},
									},
								},
							},
						},
					},
				},
			},
			ExpectMinTaskMember: map[string]int32{"task1": 6},
			ExpextVal:           nil,
		},
		{
			Name: "CreatePodGroup success Case with MinPartitions",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: namespace,
					Name:      "job3",
					UID:       "e7f18111-1cec-11ea-b688-fa163ec79500",
				},
				Spec: v1alpha1.JobSpec{
					SchedulerName: "volcano",
					Volumes: []v1alpha1.VolumeSpec{
						{
							VolumeClaimName: "vc1",
							VolumeClaim: &v1.PersistentVolumeClaimSpec{
								VolumeName: "v1",
							},
						},
						{
							VolumeClaimName: "vc2",
						},
					},
					Tasks: []v1alpha1.TaskSpec{
						{
							Name:     "task1",
							Replicas: 6,
							PartitionPolicy: &v1alpha1.PartitionPolicySpec{
								TotalPartitions: 2,
								PartitionSize:   3,
								MinPartitions:   1,
								NetworkTopology: &v1alpha1.NetworkTopologySpec{
									Mode:               "hard",
									HighestTierAllowed: &highestTierAllowed,
								},
							},
							Template: v1.PodTemplateSpec{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "pods",
									Namespace: namespace,
								},
								Spec: v1.PodSpec{
									Containers: []v1.Container{
										{
											Name: "Containers",
										},
									},
								},
							},
						},
					},
				},
			},
			ExpectMinTaskMember: map[string]int32{"task1": 3},
			ExpextVal:           nil,
		},
		{
			Name: "CreatePodGroup success Case with highestTierName",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: namespace,
					Name:      "job2",
					UID:       "e7f18111-1cec-11ea-b688-fa163ec79500",
				},
				Spec: v1alpha1.JobSpec{
					SchedulerName: "volcano",
					Volumes: []v1alpha1.VolumeSpec{
						{
							VolumeClaimName: "vc1",
							VolumeClaim: &v1.PersistentVolumeClaimSpec{
								VolumeName: "v1",
							},
						},
						{
							VolumeClaimName: "vc2",
						},
					},
					Tasks: []v1alpha1.TaskSpec{
						{
							Name:     "task1",
							Replicas: 6,
							PartitionPolicy: &v1alpha1.PartitionPolicySpec{
								TotalPartitions: 2,
								PartitionSize:   3,
								NetworkTopology: &v1alpha1.NetworkTopologySpec{
									Mode:            "hard",
									HighestTierName: "volcano.sh/hypernode",
								},
							},
							Template: v1.PodTemplateSpec{
								ObjectMeta: metav1.ObjectMeta{
									Name:      "pods",
									Namespace: namespace,
								},
								Spec: v1.PodSpec{
									Containers: []v1.Container{
										{
											Name: "Containers",
										},
									},
								},
							},
						},
					},
				},
			},
			ExpectMinTaskMember: map[string]int32{"task1": 6},
			ExpextVal:           nil,
		},
	}

	for _, testcase := range testcases {
		t.Run(testcase.Name, func(t *testing.T) {
			fakeController := newFakeController()

			err := fakeController.createOrUpdatePodGroup(testcase.Job)
			if err != testcase.ExpextVal {
				t.Errorf("Expected return value to be equal to expected: %s, but got: %s", testcase.ExpextVal, err)
			}

			pgName := testcase.Job.Name + "-" + string(testcase.Job.UID)
			pg, err := fakeController.vcClient.SchedulingV1beta1().PodGroups(namespace).Get(context.TODO(), pgName, metav1.GetOptions{})
			if err != nil {
				t.Error("Expected PodGroup to get created, but not created")
			}
			if !reflect.DeepEqual(pg.Spec.MinTaskMember, testcase.ExpectMinTaskMember) {
				t.Errorf("Expected PodGroup.Spec.MinTaskMember to be created to: %v, but got: %v", testcase.ExpectMinTaskMember, pg.Spec.MinTaskMember)
			}
		})

	}
}

func TestUpdatePodGroupIfJobUpdateFunc(t *testing.T) {
	namespace := "test"
	highestTierAllowed := 1

	testcases := []struct {
		Name                string
		PodGroup            *schedulingapi.PodGroup
		Job                 *v1alpha1.Job
		ExpectMinTaskMember map[string]int32
		ExpectVal           error
	}{
		{
			Name: "UpdatePodGroup success Case",
			PodGroup: &schedulingapi.PodGroup{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: namespace,
					Name:      "job1-e7f18111-1cec-11ea-b688-fa163ec79500",
				},
				Spec: schedulingapi.PodGroupSpec{
					MinResources: &v1.ResourceList{},
				},
			},
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Namespace:       namespace,
					Name:            "job1",
					ResourceVersion: "100",
					UID:             "e7f18111-1cec-11ea-b688-fa163ec79500",
				},
				Spec: v1alpha1.JobSpec{
					PriorityClassName: "new",
				},
			},
			ExpectMinTaskMember: map[string]int32{},
			ExpectVal:           nil,
		},
		{
			Name: "UpdatePodGroup compatibility with version lt 1.5 success Case",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Namespace:       namespace,
					Name:            "job2",
					ResourceVersion: "100",
					UID:             "e7f18111-1cec-11ea-b688-fa163ec79500",
				},
				Spec: v1alpha1.JobSpec{
					PriorityClassName: "new",
				},
			},
			PodGroup: &schedulingapi.PodGroup{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: namespace,
					Name:      "job2",
				},
				Spec: schedulingapi.PodGroupSpec{
					MinResources: &v1.ResourceList{},
				},
			},
			ExpectMinTaskMember: map[string]int32{},
			ExpectVal:           nil,
		},
		{
			Name: "UpdatePodGroup compatibility with PartitionPolicy",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Namespace:       namespace,
					Name:            "job3",
					ResourceVersion: "100",
					UID:             "e7f18111-1cec-11ea-b688-fa163ec79500",
				},
				Spec: v1alpha1.JobSpec{
					PriorityClassName: "new",
					Tasks: []v1alpha1.TaskSpec{
						{
							Name:     "task1",
							Replicas: 6,
							PartitionPolicy: &v1alpha1.PartitionPolicySpec{
								TotalPartitions: 2,
								PartitionSize:   3,
								NetworkTopology: &v1alpha1.NetworkTopologySpec{
									Mode:               "hard",
									HighestTierAllowed: &highestTierAllowed,
								},
							},
						},
					},
				},
			},
			PodGroup: &schedulingapi.PodGroup{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: namespace,
					Name:      "job3",
				},
				Spec: schedulingapi.PodGroupSpec{
					MinResources: &v1.ResourceList{},
				},
			},
			ExpectMinTaskMember: map[string]int32{"task1": 6},
			ExpectVal:           nil,
		},
		{
			Name: "UpdatePodGroup compatibility with PartitionPolicy having MinPartitions",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Namespace:       namespace,
					Name:            "job4",
					ResourceVersion: "100",
					UID:             "e7f18111-1cec-11ea-b688-fa163ec79500",
				},
				Spec: v1alpha1.JobSpec{
					PriorityClassName: "new",
					Tasks: []v1alpha1.TaskSpec{
						{
							Name:     "task1",
							Replicas: 6,
							PartitionPolicy: &v1alpha1.PartitionPolicySpec{
								TotalPartitions: 2,
								PartitionSize:   3,
								MinPartitions:   1,
								NetworkTopology: &v1alpha1.NetworkTopologySpec{
									Mode:               "hard",
									HighestTierAllowed: &highestTierAllowed,
								},
							},
						},
					},
				},
			},
			PodGroup: &schedulingapi.PodGroup{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: namespace,
					Name:      "job4",
				},
				Spec: schedulingapi.PodGroupSpec{
					MinResources: &v1.ResourceList{},
				},
			},
			ExpectMinTaskMember: map[string]int32{"task1": 3},
			ExpectVal:           nil,
		},
		{
			Name: "UpdatePodGroup compatibility with highestTierName",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Namespace:       namespace,
					Name:            "job4",
					ResourceVersion: "100",
					UID:             "e7f18111-1cec-11ea-b688-fa163ec79500",
				},
				Spec: v1alpha1.JobSpec{
					PriorityClassName: "new",
					Tasks: []v1alpha1.TaskSpec{
						{
							Name:     "task1",
							Replicas: 6,
							PartitionPolicy: &v1alpha1.PartitionPolicySpec{
								TotalPartitions: 2,
								PartitionSize:   3,
								NetworkTopology: &v1alpha1.NetworkTopologySpec{
									Mode:            "hard",
									HighestTierName: "volcano.sh/hypernode",
								},
							},
						},
					},
				},
			},
			PodGroup: &schedulingapi.PodGroup{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: namespace,
					Name:      "job4",
				},
				Spec: schedulingapi.PodGroupSpec{
					MinResources: &v1.ResourceList{},
				},
			},
			ExpectMinTaskMember: map[string]int32{"task1": 6},
			ExpectVal:           nil,
		},
	}

	for _, testcase := range testcases {
		t.Run(testcase.Name, func(t *testing.T) {
			fakeController := newFakeController()
			fakeController.pgInformer.Informer().GetIndexer().Add(testcase.PodGroup)
			fakeController.vcClient.SchedulingV1beta1().PodGroups(testcase.PodGroup.Namespace).Create(context.TODO(), testcase.PodGroup, metav1.CreateOptions{})

			err := fakeController.createOrUpdatePodGroup(testcase.Job)
			if err != testcase.ExpectVal {
				t.Errorf("Expected return value to be equal to expected: %s, but got: %s", testcase.ExpectVal, err)
			}

			pgName := testcase.PodGroup.Name
			pg, err := fakeController.vcClient.SchedulingV1beta1().PodGroups(namespace).
				Get(context.TODO(), pgName, metav1.GetOptions{})
			if err != nil {
				t.Error("Expected PodGroup to be created, but not created")
			}
			if pg.Spec.PriorityClassName != testcase.Job.Spec.PriorityClassName {
				t.Errorf("Expected PodGroup.Spec.PriorityClassName to be updated to: %s, but got: %s", testcase.Job.Spec.PriorityClassName, pg.Spec.PriorityClassName)
			}
			if !reflect.DeepEqual(pg.Spec.MinTaskMember, testcase.ExpectMinTaskMember) {
				t.Errorf("Expected PodGroup.Spec.MinTaskMember to be updated: %v, but got: %v", testcase.ExpectMinTaskMember, pg.Spec.MinTaskMember)
			}
		})

	}

}

func TestDeleteJobPod(t *testing.T) {
	namespace := "test"

	testcases := []struct {
		Name      string
		Job       *v1alpha1.Job
		Pods      map[string]*v1.Pod
		DeletePod *v1.Pod
		ExpextVal error
	}{
		{
			Name: "DeleteJobPod success case",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job1",
					Namespace:       namespace,
					ResourceVersion: "100",
				},
			},
			Pods: map[string]*v1.Pod{
				"job1-task1-0": buildPod(namespace, "job1-task1-0", v1.PodRunning, nil),
				"job1-task1-1": buildPod(namespace, "job1-task1-1", v1.PodRunning, nil),
			},
			DeletePod: buildPod(namespace, "job1-task1-0", v1.PodRunning, nil),
			ExpextVal: nil,
		},
	}

	for _, testcase := range testcases {
		t.Run(testcase.Name, func(t *testing.T) {
			fakeController := newFakeController()

			for _, pod := range testcase.Pods {
				_, err := fakeController.kubeClient.CoreV1().Pods(namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
				if err != nil {
					t.Error("Expected error not to occur")
				}
			}

			err := fakeController.deleteJobPod(testcase.Job.Name, testcase.DeletePod)
			if err != testcase.ExpextVal {
				t.Errorf("Expected return value to be equal to expected: %s, but got: %s", testcase.ExpextVal, err)
			}

			_, err = fakeController.kubeClient.CoreV1().Pods(namespace).Get(context.TODO(), "job1-task1-0", metav1.GetOptions{})
			if err == nil {
				t.Error("Expected Pod to be deleted but not deleted")
			}
		})
	}
}

func TestRecordPodGroupEvent(t *testing.T) {
	job1 := &v1alpha1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:            "job1",
			Namespace:       "default",
			ResourceVersion: "100",
			UID:             "e7f18111-1cec-11ea-b688-fa163ec79500",
		},
		Spec: v1alpha1.JobSpec{
			Tasks: []v1alpha1.TaskSpec{},
		},
		Status: v1alpha1.JobStatus{
			State: v1alpha1.JobState{
				Phase: v1alpha1.Running,
			},
		},
	}

	pg1 := &schedulingapi.PodGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "job1-e7f18111-1cec-11ea-b688-fa163ec79500",
			Namespace: "default",
		},
		Spec: schedulingapi.PodGroupSpec{
			MinResources:  &v1.ResourceList{},
			MinTaskMember: map[string]int32{},
		},
		Status: schedulingapi.PodGroupStatus{
			Phase:      schedulingapi.PodGroupRunning,
			Conditions: []schedulingapi.PodGroupCondition{},
		},
	}
	testcases := []struct {
		Name         string
		pgConditions []schedulingapi.PodGroupCondition
		ExpectEvent  string
	}{
		{
			Name:         "pg with no conditions",
			pgConditions: []schedulingapi.PodGroupCondition{},
			ExpectEvent:  "",
		},
		{
			Name: "pg with Unschedulable condition",
			pgConditions: []schedulingapi.PodGroupCondition{
				{
					Type:               schedulingapi.PodGroupUnschedulableType,
					Status:             v1.ConditionTrue,
					Reason:             "test reason",
					Message:            "test message",
					LastTransitionTime: metav1.NewTime(time.Date(2024, time.July, 1, 15, 4, 5, 0, time.UTC)),
				},
			},
			ExpectEvent: "Warning PodGroupPending PodGroup default:job1 unschedulable, reason: test message",
		},
		{
			Name: "pg with Unschedulable later than Scheduled",
			pgConditions: []schedulingapi.PodGroupCondition{
				{
					Type:               schedulingapi.PodGroupUnschedulableType,
					Status:             v1.ConditionTrue,
					Reason:             "test reason",
					Message:            "test message",
					LastTransitionTime: metav1.NewTime(time.Date(2024, time.July, 1, 15, 15, 5, 0, time.UTC)),
				},
				{
					Type:               schedulingapi.PodGroupScheduled,
					Status:             v1.ConditionTrue,
					Reason:             "test reason",
					Message:            "test message",
					LastTransitionTime: metav1.NewTime(time.Date(2024, time.July, 1, 15, 4, 5, 0, time.UTC)),
				},
			},
			ExpectEvent: "Warning PodGroupPending PodGroup default:job1 unschedulable, reason: test message",
		},
		{
			Name: "pg with Scheduled later than Unschedulable ",
			pgConditions: []schedulingapi.PodGroupCondition{
				{
					Type:               schedulingapi.PodGroupUnschedulableType,
					Status:             v1.ConditionTrue,
					Reason:             "test reason",
					Message:            "test message",
					LastTransitionTime: metav1.NewTime(time.Date(2024, time.July, 1, 15, 15, 5, 0, time.UTC)),
				},
				{
					Type:               schedulingapi.PodGroupScheduled,
					Status:             v1.ConditionTrue,
					Reason:             "test reason",
					Message:            "test message",
					LastTransitionTime: metav1.NewTime(time.Date(2024, time.July, 1, 15, 20, 5, 0, time.UTC)),
				},
			},
			ExpectEvent: "",
		},
	}
	for _, tt := range testcases {
		t.Run(tt.Name, func(t *testing.T) {
			fakeController := newFakeController()
			fakeController.recorder = record.NewFakeRecorder(100)
			pg1.Status.Conditions = tt.pgConditions
			fakeController.recordPodGroupEvent(job1, pg1)
			r := fakeController.recorder.(*record.FakeRecorder)
			close(r.Events)
			event := <-r.Events

			if len(tt.ExpectEvent) == 0 {
				if len(event) != 0 {
					t.Errorf("Testcase %s failed, expect no event, but got %s", tt.Name, event)
				}
			} else {
				if tt.ExpectEvent != event {
					t.Errorf("Testcase %s failed, expect event: %s, but got %s", tt.Name, tt.ExpectEvent, event)
				}
			}
		})
	}

}

func TestPodsToKill(t *testing.T) {
	namespace := "test"

	testcases := []struct {
		Name           string
		Job            *v1alpha1.Job
		JobInfo        *apis.JobInfo
		Target         *state.Target
		PodRetainPhase state.PhaseMap
		UpdateStatus   state.UpdateStatusFn
		Pods           map[string]*v1.Pod
		ExpectVal      error
		ExpectedPods   map[string]bool // pod name -> should exist
	}{
		{
			Name: "KillPods with TargetTypeTask",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job1",
					Namespace:       namespace,
					ResourceVersion: "100",
				},
			},
			JobInfo: &apis.JobInfo{
				Job: &v1alpha1.Job{
					ObjectMeta: metav1.ObjectMeta{
						Name:            "job1",
						Namespace:       namespace,
						ResourceVersion: "100",
					},
				},
				Pods: map[string]map[string]*v1.Pod{
					"task1": {
						"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
						"pod2": buildPod(namespace, "pod2", v1.PodRunning, nil),
					},
					"task2": {
						"pod3": buildPod(namespace, "pod3", v1.PodRunning, nil),
					},
				},
			},
			Target: &state.Target{
				Type:     state.TargetTypeTask,
				TaskName: "task1",
			},
			PodRetainPhase: state.PodRetainPhaseNone,
			Pods: map[string]*v1.Pod{
				"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
				"pod2": buildPod(namespace, "pod2", v1.PodRunning, nil),
				"pod3": buildPod(namespace, "pod3", v1.PodRunning, nil),
			},
			ExpectVal: nil,
			ExpectedPods: map[string]bool{
				"pod1": false, // Should be deleted
				"pod2": false, // Should be deleted
				"pod3": true,  // Should exist
			},
		},
		{
			Name: "KillPods with TargetTypePod",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job1",
					Namespace:       namespace,
					ResourceVersion: "100",
				},
			},
			JobInfo: &apis.JobInfo{
				Job: &v1alpha1.Job{
					ObjectMeta: metav1.ObjectMeta{
						Name:            "job1",
						Namespace:       namespace,
						ResourceVersion: "100",
					},
				},
				Pods: map[string]map[string]*v1.Pod{
					"task1": {
						"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
						"pod2": buildPod(namespace, "pod2", v1.PodRunning, nil),
					},
				},
			},
			Target: &state.Target{
				Type:     state.TargetTypePod,
				TaskName: "task1",
				PodName:  "pod1",
			},
			PodRetainPhase: state.PodRetainPhaseNone,
			Pods: map[string]*v1.Pod{
				"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
				"pod2": buildPod(namespace, "pod2", v1.PodRunning, nil),
			},
			ExpectVal: nil,
			ExpectedPods: map[string]bool{
				"pod1": false, // Should be deleted
				"pod2": true,  // Should exist
			},
		},
		{
			Name: "KillPods with non-existent task",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job1",
					Namespace:       namespace,
					ResourceVersion: "100",
				},
			},
			JobInfo: &apis.JobInfo{
				Job: &v1alpha1.Job{
					ObjectMeta: metav1.ObjectMeta{
						Name:            "job1",
						Namespace:       namespace,
						ResourceVersion: "100",
					},
				},
				Pods: map[string]map[string]*v1.Pod{
					"task1": {
						"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
					},
				},
			},
			Target: &state.Target{
				Type:     state.TargetTypeTask,
				TaskName: "nonexistent",
			},
			PodRetainPhase: state.PodRetainPhaseNone,
			Pods: map[string]*v1.Pod{
				"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
			},
			ExpectVal: nil,
			ExpectedPods: map[string]bool{
				"pod1": true, // Should exist (not targeted)
			},
		},
		{
			Name: "KillPods with non-existent pod",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job1",
					Namespace:       namespace,
					ResourceVersion: "100",
				},
			},
			JobInfo: &apis.JobInfo{
				Job: &v1alpha1.Job{
					ObjectMeta: metav1.ObjectMeta{
						Name:            "job1",
						Namespace:       namespace,
						ResourceVersion: "100",
					},
				},
				Pods: map[string]map[string]*v1.Pod{
					"task1": {
						"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
					},
				},
			},
			Target: &state.Target{
				Type:     state.TargetTypePod,
				TaskName: "task1",
				PodName:  "nonexistent",
			},
			PodRetainPhase: state.PodRetainPhaseNone,
			Pods: map[string]*v1.Pod{
				"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
			},
			ExpectVal: nil,
			ExpectedPods: map[string]bool{
				"pod1": true, // Should exist (not targeted)
			},
		},
	}

	for i, testcase := range testcases {
		t.Run(testcase.Name, func(t *testing.T) {
			fakeController := newFakeController()

			// Create pods
			for _, pod := range testcase.Pods {
				_, err := fakeController.kubeClient.CoreV1().Pods(namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
				if err != nil {
					t.Fatalf("Error while creating pod: %v", err)
				}
			}

			// Create job
			_, err := fakeController.vcClient.BatchV1alpha1().Jobs(namespace).Create(context.TODO(), testcase.Job, metav1.CreateOptions{})
			if err != nil {
				t.Fatalf("Error while creating job: %v", err)
			}

			err = fakeController.cache.Add(testcase.Job)
			if err != nil {
				t.Fatalf("Error while adding job to cache: %v", err)
			}

			// Execute killPods
			err = fakeController.killPods(testcase.JobInfo, testcase.PodRetainPhase, testcase.Target, testcase.UpdateStatus)
			if !errors.Is(err, testcase.ExpectVal) {
				if testcase.ExpectVal == nil {
					t.Errorf("Test case %d (%s): expected no error, but got error %v", i, testcase.Name, err)
				} else if err == nil || err.Error() != testcase.ExpectVal.Error() {
					t.Errorf("Test case %d (%s): expected error %v, but got %v", i, testcase.Name, testcase.ExpectVal, err)
				}
			}

			// Check if pods exist as expected
			for podName, shouldExist := range testcase.ExpectedPods {
				_, err := fakeController.kubeClient.CoreV1().Pods(namespace).Get(context.TODO(), podName, metav1.GetOptions{})
				if shouldExist && err != nil {
					t.Errorf("Test case %d (%s): expected pod %s to exist, but got error %v", i, testcase.Name, podName, err)
				} else if !shouldExist && err == nil {
					t.Errorf("Test case %d (%s): expected pod %s to be deleted, but still exists", i, testcase.Name, podName)
				}
			}
		})
	}
}

func TestKillPodsPodGroupDeletion(t *testing.T) {
	namespace := "test"
	jobUID := "e7f18111-1cec-11ea-b688-fa163ec79500"

	testcases := []struct {
		Name               string
		Target             *state.Target
		ExpectPodGroupGone bool
	}{
		{
			Name:               "full job kill (target=nil) deletes PodGroup",
			Target:             nil,
			ExpectPodGroupGone: true,
		},
		{
			Name: "targeted pod restart (target!=nil) keeps PodGroup",
			Target: &state.Target{
				Type:     state.TargetTypePod,
				TaskName: "task1",
				PodName:  "pod1",
			},
			ExpectPodGroupGone: false,
		},
	}

	for _, testcase := range testcases {
		t.Run(testcase.Name, func(t *testing.T) {
			job := &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:            "job1",
					Namespace:       namespace,
					UID:             types.UID(jobUID),
					ResourceVersion: "100",
				},
			}
			pgName := fmt.Sprintf("%s-%s", job.Name, jobUID)
			pg := &schedulingapi.PodGroup{
				ObjectMeta: metav1.ObjectMeta{
					Name:      pgName,
					Namespace: namespace,
				},
			}
			jobInfo := &apis.JobInfo{
				Namespace: namespace,
				Name:      job.Name,
				Job:       job,
				Pods: map[string]map[string]*v1.Pod{
					"task1": {
						"pod1": buildPod(namespace, "pod1", v1.PodRunning, nil),
					},
				},
			}

			fakeController := newFakeController()

			fakeController.pgInformer.Informer().GetIndexer().Add(pg)
			_, err := fakeController.vcClient.SchedulingV1beta1().PodGroups(namespace).Create(context.TODO(), pg, metav1.CreateOptions{})
			if err != nil {
				t.Fatalf("Error creating PodGroup: %v", err)
			}

			_, err = fakeController.kubeClient.CoreV1().Pods(namespace).Create(context.TODO(), jobInfo.Pods["task1"]["pod1"], metav1.CreateOptions{})
			if err != nil {
				t.Fatalf("Error creating pod: %v", err)
			}

			_, err = fakeController.vcClient.BatchV1alpha1().Jobs(namespace).Create(context.TODO(), job, metav1.CreateOptions{})
			if err != nil {
				t.Fatalf("Error creating job: %v", err)
			}
			if err = fakeController.cache.Add(job); err != nil {
				t.Fatalf("Error adding job to cache: %v", err)
			}

			err = fakeController.killPods(jobInfo, state.PodRetainPhaseNone, testcase.Target, nil)
			if err != nil {
				t.Fatalf("killPods returned unexpected error: %v", err)
			}

			_, pgErr := fakeController.vcClient.SchedulingV1beta1().PodGroups(namespace).Get(context.TODO(), pgName, metav1.GetOptions{})
			pgGone := pgErr != nil && apierrors.IsNotFound(pgErr)
			if pgGone != testcase.ExpectPodGroupGone {
				if testcase.ExpectPodGroupGone {
					t.Errorf("expected PodGroup to be deleted, but it still exists")
				} else {
					t.Errorf("expected PodGroup to be kept, but it was deleted")
				}
			}
		})
	}
}

func TestGetSubGroupPolicy(t *testing.T) {
	highestTierAllowed := 1

	testCases := []struct {
		Name                 string
		TaskSpec             v1alpha1.TaskSpec
		ExpectedMinSubGroups *int32
		ExpectedSubGroupSize *int32
		Description          string
	}{
		{
			Name: "SubGroupPolicy with MinPartitions set",
			TaskSpec: v1alpha1.TaskSpec{
				Name:     "task1",
				Replicas: 6,
				PartitionPolicy: &v1alpha1.PartitionPolicySpec{
					TotalPartitions: 2,
					MinPartitions:   1,
					PartitionSize:   3,
					NetworkTopology: &v1alpha1.NetworkTopologySpec{
						Mode:               v1alpha1.HardNetworkTopologyMode,
						HighestTierAllowed: &highestTierAllowed,
					},
				},
			},
			ExpectedMinSubGroups: ptr.To(int32(1)),
			ExpectedSubGroupSize: ptr.To(int32(3)),
			Description:          "MinSubGroups should be set to MinPartitions when MinPartitions > 0",
		},
		{
			Name: "SubGroupPolicy with MinPartitions equal to TotalPartitions",
			TaskSpec: v1alpha1.TaskSpec{
				Name:     "task1",
				Replicas: 9,
				PartitionPolicy: &v1alpha1.PartitionPolicySpec{
					TotalPartitions: 3,
					MinPartitions:   3,
					PartitionSize:   3,
				},
			},
			ExpectedMinSubGroups: ptr.To(int32(3)),
			ExpectedSubGroupSize: ptr.To(int32(3)),
			Description:          "MinSubGroups should equal TotalPartitions when MinPartitions equals TotalPartitions",
		},
		{
			Name: "SubGroupPolicy with MinPartitions not set",
			TaskSpec: v1alpha1.TaskSpec{
				Name:     "task1",
				Replicas: 6,
				PartitionPolicy: &v1alpha1.PartitionPolicySpec{
					TotalPartitions: 2,
					MinPartitions:   0,
					PartitionSize:   3,
				},
			},
			ExpectedMinSubGroups: ptr.To(int32(0)),
			ExpectedSubGroupSize: ptr.To(int32(3)),
			Description:          "MinSubGroups should be nil when MinPartitions is 0",
		},
	}

	for _, tc := range testCases {
		t.Run(tc.Name, func(t *testing.T) {
			result := getSubGroupPolicy(tc.TaskSpec)

			// Check SubGroupSize
			if tc.ExpectedSubGroupSize != nil {
				if result.SubGroupSize == nil {
					t.Errorf("%s: Expected SubGroupSize=%d, got nil",
						tc.Description, *tc.ExpectedSubGroupSize)
				} else if *result.SubGroupSize != *tc.ExpectedSubGroupSize {
					t.Errorf("%s: Expected SubGroupSize=%d, got %d",
						tc.Description,
						*tc.ExpectedSubGroupSize,
						*result.SubGroupSize)
				}
			}

			// Check MinSubGroups
			if tc.ExpectedMinSubGroups != nil {
				if result.MinSubGroups == nil {
					t.Errorf("%s: Expected MinSubGroups=%d, got nil",
						tc.Description, *tc.ExpectedMinSubGroups)
				} else if *result.MinSubGroups != *tc.ExpectedMinSubGroups {
					t.Errorf("%s: Expected MinSubGroups=%d, got %d",
						tc.Description,
						*tc.ExpectedMinSubGroups,
						*result.MinSubGroups)
				}
			} else {
				if result.MinSubGroups != nil {
					t.Errorf("%s: Expected MinSubGroups to be nil, got %d",
						tc.Description, *result.MinSubGroups)
				}
			}

			// Check LabelSelector
			if result.LabelSelector == nil {
				t.Errorf("%s: Expected LabelSelector to be set", tc.Description)
			} else {
				if result.LabelSelector.MatchLabels == nil {
					t.Errorf("%s: Expected LabelSelector.MatchLabels to be set", tc.Description)
				} else if result.LabelSelector.MatchLabels["volcano.sh/task-spec"] != tc.TaskSpec.Name {
					t.Errorf("%s: Expected LabelSelector.MatchLabels['volcano.sh/task-spec']=%s, got %s",
						tc.Description,
						tc.TaskSpec.Name,
						result.LabelSelector.MatchLabels["volcano.sh/task-spec"])
				}
			}

			// Check NetworkTopology
			if tc.TaskSpec.PartitionPolicy.NetworkTopology != nil {
				if result.NetworkTopology == nil {
					t.Errorf("%s: Expected NetworkTopology to be set", tc.Description)
				}
			}
		})
	}
}

func TestAppendJobCondition(t *testing.T) {
	testcases := []struct {
		Name          string
		Conditions    []v1alpha1.JobCondition
		NewCondition  v1alpha1.JobCondition
		ExpectedLen   int
		ExpectedPhase v1alpha1.JobPhase
	}{
		{
			Name:       "Append to empty conditions",
			Conditions: []v1alpha1.JobCondition{},
			NewCondition: v1alpha1.JobCondition{
				Status: v1alpha1.Pending,
			},
			ExpectedLen:   1,
			ExpectedPhase: v1alpha1.Pending,
		},
		{
			Name: "Replace existing condition with same phase",
			Conditions: []v1alpha1.JobCondition{
				{Status: v1alpha1.Pending},
				{Status: v1alpha1.Running},
			},
			NewCondition: v1alpha1.JobCondition{
				Status: v1alpha1.Pending,
			},
			ExpectedLen:   2,
			ExpectedPhase: v1alpha1.Pending,
		},
		{
			Name: "Append if phase is different",
			Conditions: []v1alpha1.JobCondition{
				{Status: v1alpha1.Pending},
				{Status: v1alpha1.Running},
			},
			NewCondition: v1alpha1.JobCondition{
				Status: v1alpha1.Completed,
			},
			ExpectedLen:   3,
			ExpectedPhase: v1alpha1.Completed,
		},
	}

	for _, tc := range testcases {
		t.Run(tc.Name, func(t *testing.T) {
			result := appendJobCondition(tc.Conditions, tc.NewCondition)

			if len(result) != tc.ExpectedLen {
				t.Errorf("Expected length %d, got %d", tc.ExpectedLen, len(result))
			}

			if result[len(result)-1].Status != tc.ExpectedPhase {
				t.Errorf("Expected last phase %v, got %v", tc.ExpectedPhase, result[len(result)-1].Status)
			}
		})
	}
}

func TestIsDependsOnPodsReady(t *testing.T) {
	namespace := "test"
	jobName := "job1"

	testcases := []struct {
		Name          string
		Task          string
		Job           *v1alpha1.Job
		Pods          []*v1.Pod
		ExpectedReady bool
	}{
		{
			Name: "DependsOn task has all pods succeeded",
			Task: "task1",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:      jobName,
					Namespace: namespace,
				},
				Spec: v1alpha1.JobSpec{
					Tasks: []v1alpha1.TaskSpec{
						{
							Name:         "task1",
							Replicas:     2,
							MinAvailable: ptr.To[int32](2),
							Template: v1.PodTemplateSpec{
								ObjectMeta: metav1.ObjectMeta{
									Name: "task1",
								},
							},
						},
						{
							Name:     "task2",
							Replicas: 1,
							DependsOn: &v1alpha1.DependsOn{
								Name: []string{"task1"},
							},
						},
					},
				},
			},
			Pods: []*v1.Pod{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "job1-task1-0",
						Namespace: namespace,
					},
					Status: v1.PodStatus{
						Phase: v1.PodSucceeded,
					},
				},
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "job1-task1-1",
						Namespace: namespace,
					},
					Status: v1.PodStatus{
						Phase: v1.PodSucceeded,
					},
				},
			},
			ExpectedReady: true,
		},
		{
			Name: "DependsOn task has pods running and ready",
			Task: "task1",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:      jobName,
					Namespace: namespace,
				},
				Spec: v1alpha1.JobSpec{
					Tasks: []v1alpha1.TaskSpec{
						{
							Name:         "task1",
							Replicas:     1,
							MinAvailable: ptr.To[int32](1),
							Template: v1.PodTemplateSpec{
								ObjectMeta: metav1.ObjectMeta{
									Name: "task1",
								},
							},
						},
						{
							Name:     "task2",
							Replicas: 1,
							DependsOn: &v1alpha1.DependsOn{
								Name: []string{"task1"},
							},
						},
					},
				},
			},
			Pods: []*v1.Pod{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "job1-task1-0",
						Namespace: namespace,
					},
					Status: v1.PodStatus{
						Phase: v1.PodRunning,
						ContainerStatuses: []v1.ContainerStatus{
							{Ready: true},
						},
					},
				},
			},
			ExpectedReady: true,
		},
		{
			Name: "DependsOn task has pods running but not ready",
			Task: "task1",
			Job: &v1alpha1.Job{
				ObjectMeta: metav1.ObjectMeta{
					Name:      jobName,
					Namespace: namespace,
				},
				Spec: v1alpha1.JobSpec{
					Tasks: []v1alpha1.TaskSpec{
						{
							Name:         "task1",
							Replicas:     1,
							MinAvailable: ptr.To[int32](1),
							Template: v1.PodTemplateSpec{
								ObjectMeta: metav1.ObjectMeta{
									Name: "task1",
								},
							},
						},
						{
							Name:     "task2",
							Replicas: 1,
							DependsOn: &v1alpha1.DependsOn{
								Name: []string{"task1"},
							},
						},
					},
				},
			},
			Pods: []*v1.Pod{
				{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "job1-task1-0",
						Namespace: namespace,
					},
					Status: v1.PodStatus{
						Phase: v1.PodRunning,
						ContainerStatuses: []v1.ContainerStatus{
							{Ready: false},
						},
					},
				},
			},
			ExpectedReady: false,
		},
	}

	for i, tc := range testcases {
		t.Run(tc.Name, func(t *testing.T) {
			fakeController := newFakeController()

			_, err := fakeController.vcClient.BatchV1alpha1().Jobs(namespace).Create(context.TODO(), tc.Job, metav1.CreateOptions{})
			if err != nil {
				t.Fatalf("Failed to create job: %v", err)
			}
			err = fakeController.cache.Add(tc.Job)
			if err != nil {
				t.Fatalf("Failed to add job to cache: %v", err)
			}
			fakeController.jobInformer.Informer().GetIndexer().Add(tc.Job)

			for _, pod := range tc.Pods {
				_, err := fakeController.kubeClient.CoreV1().Pods(namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
				if err != nil {
					t.Fatalf("Failed to create pod: %v", err)
				}
				fakeController.podInformer.Informer().GetIndexer().Add(pod)
			}

			ready := fakeController.isDependsOnPodsReady(tc.Task, tc.Job)
			if ready != tc.ExpectedReady {
				t.Errorf("Case %d (%s): expected ready: %v, got: %v", i, tc.Name, tc.ExpectedReady, ready)
			}
		})
	}
}
