ayatori/internal/controller/job_controller_integration_test.go

package controller

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

	executionv1alpha1 "git.ddupan.top/panxiao81/ayatori/api/execution/v1alpha1"
	kubernetesadapter "git.ddupan.top/panxiao81/ayatori/internal/adapter/kubernetes"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/envtest"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
)

const (
	integrationNamespace = "controller-integration"
	integrationClass     = "integration"
)

//nolint:modernize // Kubernetes API structs expose ObjectMeta through embedded TypeMeta fields.
func TestJobControllerIntegration(t *testing.T) {
	if os.Getenv("KUBEBUILDER_ASSETS") == "" {
		t.Skip("KUBEBUILDER_ASSETS is unset; run make test to execute controller integration tests")
	}

	scheme := runtime.NewScheme()
	for _, addToScheme := range []func(*runtime.Scheme) error{
		corev1.AddToScheme,
		batchv1.AddToScheme,
		executionv1alpha1.AddToScheme,
	} {
		if err := addToScheme(scheme); err != nil {
			t.Fatal(err)
		}
	}

	crdPath, err := filepath.Abs("../../config/crd/bases")
	if err != nil {
		t.Fatal(err)
	}
	environment := &envtest.Environment{CRDDirectoryPaths: []string{crdPath}}
	config, err := environment.Start()
	if err != nil {
		t.Fatalf("start envtest: %v", err)
	}
	t.Cleanup(func() {
		if err := environment.Stop(); err != nil {
			t.Errorf("stop envtest: %v", err)
		}
	})

	manager, err := ctrl.NewManager(config, ctrl.Options{
		Scheme:  scheme,
		Metrics: metricsserver.Options{BindAddress: "0"},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := (&JobReconciler{Client: manager.GetClient()}).SetupWithManager(manager); err != nil {
		t.Fatal(err)
	}

	managerContext, cancelManager := context.WithCancel(context.Background())
	t.Cleanup(cancelManager)
	managerErrors := make(chan error, 1)
	go func() {
		managerErrors <- manager.Start(managerContext)
	}()
	if !manager.GetCache().WaitForCacheSync(managerContext) {
		t.Fatal("manager cache did not synchronize")
	}

	directClient, err := client.New(config, client.Options{Scheme: scheme})
	if err != nil {
		t.Fatal(err)
	}
	ctx := context.Background()
	objects := []client.Object{
		&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: integrationNamespace, Labels: map[string]string{testLabelKey: testLabelEnabled}}},
		&corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: testSAName, Namespace: integrationNamespace}},
		&executionv1alpha1.KubernetesExecutionParameters{
			ObjectMeta: metav1.ObjectMeta{Name: integrationClass},
			Spec: executionv1alpha1.KubernetesExecutionParametersSpec{
				ServiceAccountName: testSAName,
				ImagePullPolicy:    corev1.PullIfNotPresent,
			},
		},
		&executionv1alpha1.JobClass{
			ObjectMeta: metav1.ObjectMeta{Name: integrationClass},
			Spec: executionv1alpha1.JobClassSpec{
				ControllerName: kubernetesadapter.ControllerName,
				ParametersRef: executionv1alpha1.ParametersReference{
					Group: executionv1alpha1.GroupVersion.Group,
					Kind:  "KubernetesExecutionParameters",
					Name:  integrationClass,
				},
				AllowedNamespaces: &metav1.LabelSelector{MatchLabels: map[string]string{testLabelKey: testLabelEnabled}},
			},
		},
	}
	for _, object := range objects {
		if err := directClient.Create(ctx, object); err != nil {
			t.Fatalf("create %T: %v", object, err)
		}
	}

	job := &executionv1alpha1.Job{
		ObjectMeta: metav1.ObjectMeta{Name: testJobName, Namespace: integrationNamespace},
		Spec: executionv1alpha1.JobSpec{
			JobClassName: integrationClass,
			DesiredState: executionv1alpha1.JobDesiredStateRunning,
			Task:         executionv1alpha1.TaskSpec{Image: "alpine:3.22", Command: []string{"true"}},
		},
	}
	if err := directClient.Create(ctx, job); err != nil {
		t.Fatal(err)
	}

	backend := &batchv1.Job{}
	eventually(t, 10*time.Second, func() (bool, error) {
		err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, backend)
		return err == nil, client.IgnoreNotFound(err)
	})
	if backend.Labels[kubernetesadapter.JobUIDLabel] != string(job.UID) {
		t.Fatalf("backend identity label = %q, want %q", backend.Labels[kubernetesadapter.JobUIDLabel], job.UID)
	}

	eventually(t, 10*time.Second, func() (bool, error) {
		if err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, job); err != nil {
			return false, err
		}
		return conditionStatus(job, executionv1alpha1.JobConditionScheduled) == metav1.ConditionTrue, nil
	})

	completed := metav1.Now()
	if err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, backend); err != nil {
		t.Fatal(err)
	}
	backend.Status.StartTime = &completed
	backend.Status.CompletionTime = &completed
	backend.Status.Conditions = []batchv1.JobCondition{
		{Type: batchv1.JobSuccessCriteriaMet, Status: corev1.ConditionTrue, Reason: "CompletionsReached"},
		{Type: batchv1.JobComplete, Status: corev1.ConditionTrue, Reason: "Completed"},
	}
	if err := directClient.Status().Update(ctx, backend); err != nil {
		t.Fatal(err)
	}

	eventually(t, 10*time.Second, func() (bool, error) {
		if err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, job); err != nil {
			return false, err
		}
		return conditionStatus(job, executionv1alpha1.JobConditionSucceeded) == metav1.ConditionTrue, nil
	})
	if job.Status.StartTime == nil || job.Status.CompletionTime == nil || job.Status.Execution == nil {
		t.Fatalf("controller did not persist execution status: %#v", job.Status)
	}

	cancelManager()
	select {
	case err := <-managerErrors:
		if err != nil {
			t.Fatalf("manager stopped with error: %v", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("manager did not stop")
	}
}

func eventually(t *testing.T, timeout time.Duration, check func() (bool, error)) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		ready, err := check()
		if err != nil {
			t.Fatal(err)
		}
		if ready {
			return
		}
		time.Sleep(100 * time.Millisecond)
	}
	t.Fatal(fmt.Errorf("condition was not met within %s", timeout))
}

func conditionStatus(job *executionv1alpha1.Job, conditionType string) metav1.ConditionStatus {
	condition := meta.FindStatusCondition(job.Status.Conditions, conditionType)
	if condition == nil {
		return metav1.ConditionUnknown
	}
	return condition.Status
}