test: exercise controller with envtest

internal/controller/job_controller_integration_test.go

@@ -0,0 +1,202 @@
+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
+}

internal/controller/job_controller_test.go

@@ -18,7 +18,13 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
-const defaultClassName = "default"
+const (
+	defaultClassName = "default"
+	testJobName      = "hello"
+	testSAName       = "runner"
+	testLabelKey     = "execution"
+	testLabelEnabled = "enabled"
+)
 
 //nolint:modernize // controller-runtime and Kubernetes API structs expose promoted embedded fields.
 func TestJobReconcilerKubernetesLifecycle(t *testing.T) {
@@ -26,7 +32,7 @@ func TestJobReconcilerKubernetesLifecycle(t *testing.T) {
 	now := time.Unix(1_700_000_000, 0)
 	reconciler, kubeClient := testReconciler(t, now, validObjects()...)
 	request := ctrl.Request{}
-	request.NamespacedName = types.NamespacedName{Namespace: "ci", Name: "hello"}
+	request.NamespacedName = types.NamespacedName{Namespace: "ci", Name: testJobName}
 
 	if _, err := reconciler.Reconcile(ctx, request); err != nil {
 		t.Fatalf("add finalizer: %v", err)
@@ -180,18 +186,18 @@ func TestJobReconcilerKeepsConfirmedSuccessDuringCancellation(t *testing.T) {
 //nolint:modernize // Kubernetes API structs expose ObjectMeta through embedded TypeMeta fields.
 func validObjects() []client.Object {
 	return []client.Object{
-		&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ci", Labels: map[string]string{"execution": "enabled"}}},
-		&corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: "runner", Namespace: "ci"}},
+		&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ci", Labels: map[string]string{testLabelKey: testLabelEnabled}}},
+		&corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: testSAName, Namespace: "ci"}},
 		&executionv1alpha1.JobClass{
 			ObjectMeta: metav1.ObjectMeta{Name: defaultClassName, UID: types.UID("class-uid")},
 			Spec: executionv1alpha1.JobClassSpec{
 				ControllerName:    kubernetesadapter.ControllerName,
 				ParametersRef:     executionv1alpha1.ParametersReference{Group: executionv1alpha1.GroupVersion.Group, Kind: "KubernetesExecutionParameters", Name: defaultClassName},
-				AllowedNamespaces: &metav1.LabelSelector{MatchLabels: map[string]string{"execution": "enabled"}},
+				AllowedNamespaces: &metav1.LabelSelector{MatchLabels: map[string]string{testLabelKey: testLabelEnabled}},
 			},
 		},
 		&executionv1alpha1.Job{
-			ObjectMeta: metav1.ObjectMeta{Name: "hello", Namespace: "ci", UID: types.UID("ayatori-job-uid")},
+			ObjectMeta: metav1.ObjectMeta{Name: testJobName, Namespace: "ci", UID: types.UID("ayatori-job-uid")},
 			Spec: executionv1alpha1.JobSpec{
 				JobClassName: defaultClassName, DesiredState: executionv1alpha1.JobDesiredStateRunning,
 				Task: executionv1alpha1.TaskSpec{Image: "alpine:3.22", Command: []string{"true"}},
@@ -199,7 +205,7 @@ func validObjects() []client.Object {
 		},
 		&executionv1alpha1.KubernetesExecutionParameters{
 			ObjectMeta: metav1.ObjectMeta{Name: defaultClassName, UID: types.UID("parameters-uid")},
-			Spec:       executionv1alpha1.KubernetesExecutionParametersSpec{ServiceAccountName: "runner", ImagePullPolicy: corev1.PullIfNotPresent},
+			Spec:       executionv1alpha1.KubernetesExecutionParametersSpec{ServiceAccountName: testSAName, ImagePullPolicy: corev1.PullIfNotPresent},
 		},
 	}
 }