[task-controller] reconcile happy path tests from Task creation until working transition

Author: justonedev1

control-operator/internal/controller/occ_server_test.go

@@ -0,0 +1,137 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2026 CERN and copyright holders of ALICE O².
+ * Author: Michal Tichak <michal.tichak@cern.ch>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * In applying this license CERN does not waive the privileges and
+ * immunities granted to it by virtue of its status as an
+ * Intergovernmental Organization or submit itself to any jurisdiction.
+ */
+
+package controller
+
+import (
+	"context"
+	"errors"
+	"net"
+	"strings"
+	"sync"
+
+	"google.golang.org/grpc"
+
+	pb "github.com/AliceO2Group/Control/control-operator/internal/controller/protos/generated"
+)
+
+// occServerStub is an in-process implementation of the OCC gRPC service defined
+// in occ/protos/occ.proto. Embedding pb.UnimplementedOccServer satisfies the
+// generated interface and keeps this type valid if the service gains methods.
+// The state is guarded because the gRPC server answers on its own goroutines while
+// the test reads and writes it.
+type occServerStub struct {
+	pb.UnimplementedOccServer
+
+	returnError bool
+	mu          sync.RWMutex
+	state       string
+}
+
+func (s *occServerStub) setState(state string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.state = state
+}
+
+func (s *occServerStub) getState() string {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.state
+}
+
+func (s *occServerStub) GetState(ctx context.Context, req *pb.GetStateRequest) (*pb.GetStateReply, error) {
+	if s.returnError == true {
+		return nil, errors.New("returning error as requested by user")
+	}
+	return &pb.GetStateReply{State: s.getState()}, nil
+}
+
+// Transition answers with the state the requested event leads to, looked up in the
+// same FSM table the controller used to pick the event, and moves the stub to that
+// state so a following GetState reports it.
+func (s *occServerStub) Transition(ctx context.Context, req *pb.TransitionRequest) (*pb.TransitionReply, error) {
+	if s.returnError == true {
+		return nil, errors.New("returning error as requested by user")
+	}
+
+	src, err := StateFromString(strings.ToLower(req.GetSrcState()))
+	if err != nil {
+		return nil, err
+	}
+	event, err := TransitionFromString(strings.ToLower(req.GetTransitionEvent()))
+	if err != nil {
+		return nil, err
+	}
+
+	for fromTo, transition := range fromStatesToTransition {
+		if fromTo.from == src && transition == event {
+			newState := fromTo.to.String()
+			s.setState(newState)
+			return &pb.TransitionReply{
+				Trigger:         pb.StateChangeTrigger_EXECUTOR,
+				State:           newState,
+				TransitionEvent: req.GetTransitionEvent(),
+				Ok:              true,
+			}, nil
+		}
+	}
+
+	// No rule for this event in this state, so the device stays where it is.
+	return &pb.TransitionReply{
+		Trigger:         pb.StateChangeTrigger_EXECUTOR,
+		State:           src.String(),
+		TransitionEvent: req.GetTransitionEvent(),
+		Ok:              false,
+	}, nil
+}
+
+func (s *occServerStub) StateStream(req *pb.StateStreamRequest, srv grpc.ServerStreamingServer[pb.StateStreamReply]) error {
+	<-srv.Context().Done()
+	return nil
+}
+
+func (s *occServerStub) EventStream(req *pb.EventStreamRequest, srv grpc.ServerStreamingServer[pb.EventStreamReply]) error {
+	<-srv.Context().Done()
+	return nil
+}
+
+// startOccServerStub serves a stub on a kernel-assigned loopback port and
+// returns it together with that port and a stop function.
+func startOccServerStub() (*occServerStub, int, func(), error) {
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		return nil, 0, nil, err
+	}
+
+	stub := &occServerStub{}
+	srv := grpc.NewServer()
+	pb.RegisterOccServer(srv, stub)
+
+	go func() {
+		_ = srv.Serve(lis)
+	}()
+
+	return stub, lis.Addr().(*net.TCPAddr).Port, srv.Stop, nil
+}

control-operator/internal/controller/task_controller_test.go

@@ -26,71 +26,219 @@ package controller
 
 import (
 	"context"
+	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/record"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
 	aliecsv1alpha1 "github.com/AliceO2Group/Control/control-operator/api/v1alpha1"
 )
 
 var _ = Describe("Task Controller", func() {
-	Context("When reconciling a resource", func() {
-		const resourceName = "test-resource"
+	// Reconcile takes the Task one step further on every pass and then returns to wait
+	// for the next event: first the finalizer, then the Pod, then the gRPC connection,
+	// then the state of the container, and finally the transition. These specs run in
+	// order and each one drives the pass responsible for the next step.
+	Describe("When reconciling a Task step by step", Ordered, func() {
+		const (
+			resourceName  = "test-resource"
+			namespace     = "default"
+			containerName = "placeholder-name"
+			containerImg  = "placeholder-image:latest"
+			// The OCC stub listens on loopback, so the Pod IP the controller dials has
+			// to point back at the test process.
+			podIP = "127.0.0.1"
+		)
 
-		ctx := context.Background()
+		var (
+			ctx        context.Context
+			taskName   types.NamespacedName
+			podName    types.NamespacedName
+			occServer  *occServerStub
+			reconciler *TaskReconciler
+		)
 
-		typeNamespacedName := types.NamespacedName{
-			Name:      resourceName,
-			Namespace: "default", // TODO(user):Modify as needed
+		reconcileOnce := func() reconcile.Result {
+			GinkgoHelper()
+			res, err := reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: taskName})
+			Expect(err).NotTo(HaveOccurred())
+			return res
 		}
-		task := &aliecsv1alpha1.Task{}
 
-		BeforeEach(func() {
-			By("creating the custom resource for the Kind Task")
-			err := k8sClient.Get(ctx, typeNamespacedName, task)
-			if err != nil && errors.IsNotFound(err) {
-				resource := &aliecsv1alpha1.Task{
-					ObjectMeta: metav1.ObjectMeta{
-						Name:      resourceName,
-						Namespace: "default",
-					},
-					Spec: aliecsv1alpha1.TaskSpec{
-						State: "standby",
-						Pod:   v1.PodSpec{Containers: []v1.Container{}},
-					},
-				}
-				Expect(k8sClient.Create(ctx, resource)).To(Succeed())
-			}
-		})
-
-		AfterEach(func() {
-			// TODO(user): Cleanup logic after each test, like removing the task instance.
+		getTask := func() *aliecsv1alpha1.Task {
+			GinkgoHelper()
 			task := &aliecsv1alpha1.Task{}
-			err := k8sClient.Get(ctx, typeNamespacedName, task)
+			Expect(k8sClient.Get(ctx, taskName, task)).To(Succeed())
+			return task
+		}
+
+		BeforeAll(func() {
+			ctx = context.Background()
+			taskName = types.NamespacedName{Name: resourceName, Namespace: namespace}
+			podName = types.NamespacedName{Name: podNameFromTask(resourceName), Namespace: namespace}
+
+			By("starting the OCC server stub that stands in for the task container")
+			stub, port, stopOccServer, err := startOccServerStub()
 			Expect(err).NotTo(HaveOccurred())
+			DeferCleanup(stopOccServer)
 
-			By("Cleanup the specific resource instance Task")
-			Expect(k8sClient.Delete(ctx, task)).To(Succeed())
-		})
-		It("should successfully reconcile the resource", func() {
-			By("Reconciling the created resource")
-			controllerReconciler := &TaskReconciler{
+			occServer = stub
+			occServer.setState("standby")
+
+			reconciler = &TaskReconciler{
 				Client: k8sClient,
 				Scheme: k8sClient.Scheme(),
+				// A zero-value FakeRecorder discards events; NewFakeRecorder would block
+				// once its buffer filled up.
+				Recorder: &record.FakeRecorder{},
 			}
 
-			_, err := controllerReconciler.Reconcile(ctx, reconcile.Request{
-				NamespacedName: typeNamespacedName,
+			By("creating the custom resource for the Kind Task")
+			task := &aliecsv1alpha1.Task{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      resourceName,
+					Namespace: namespace,
+				},
+				Spec: aliecsv1alpha1.TaskSpec{
+					Control: aliecsv1alpha1.TaskSpecControl{Mode: "direct", Port: port},
+					State:   "standby",
+					Pod: v1.PodSpec{
+						Containers: []v1.Container{{Name: containerName, Image: containerImg}},
+					},
+				},
+			}
+			Expect(k8sClient.Create(ctx, task)).To(Succeed())
+
+			// The last spec deletes the Task and its Pod through the controller, so this
+			// only clears what an earlier failure left behind.
+			DeferCleanup(func() {
+				if c, loaded := clientsForContainers.LoadAndDelete(resourceName); loaded {
+					Expect(c.(*OccClient).Close()).To(Succeed())
+				}
+
+				pod := &v1.Pod{}
+				if err := k8sClient.Get(ctx, podName, pod); err == nil {
+					Expect(client.IgnoreNotFound(k8sClient.Delete(ctx, pod, client.GracePeriodSeconds(0)))).To(Succeed())
+				}
+
+				leftover := &aliecsv1alpha1.Task{}
+				if err := k8sClient.Get(ctx, taskName, leftover); err == nil {
+					controllerutil.RemoveFinalizer(leftover, taskFinalizer)
+					Expect(k8sClient.Update(ctx, leftover)).To(Succeed())
+					Expect(client.IgnoreNotFound(k8sClient.Delete(ctx, leftover))).To(Succeed())
+				}
 			})
-			Expect(err).NotTo(HaveOccurred())
-			// TODO(user): Add more specific assertions depending on your controller's reconciliation logic.
-			// Example: If you expect a certain status condition after reconciliation, verify it here.
+		})
+
+		It("sets the finalizer before touching anything else", func() {
+			reconcileOnce()
+
+			Expect(controllerutil.ContainsFinalizer(getTask(), taskFinalizer)).To(BeTrue())
+			Expect(k8sClient.Get(ctx, podName, &v1.Pod{})).To(Satisfy(errors.IsNotFound),
+				"the Pod should wait until the finalizer is persisted")
+		})
+
+		It("creates the Pod for the Task", func() {
+			reconcileOnce()
+
+			pod := &v1.Pod{}
+			Expect(k8sClient.Get(ctx, podName, pod)).To(Succeed())
+
+			Expect(pod.Spec.Containers).To(HaveLen(1))
+			Expect(pod.Spec.Containers[0].Image).To(Equal(containerImg))
+			Expect(pod.Spec.RestartPolicy).To(Equal(v1.RestartPolicyNever))
+			Expect(pod.Labels).To(HaveKeyWithValue("task_name", resourceName))
+
+			Expect(pod.OwnerReferences).To(HaveLen(1))
+			Expect(pod.OwnerReferences[0].Name).To(Equal(resourceName))
+			Expect(pod.OwnerReferences[0].Controller).To(HaveValue(BeTrue()))
+		})
+
+		It("waits for the Pod to report an IP before connecting", func() {
+			reconcileOnce()
+
+			_, connected := clientsForContainers.Load(resourceName)
+			Expect(connected).To(BeFalse())
+			Expect(meta.FindStatusCondition(getTask().Status.Conditions, aliecsv1alpha1.ConditionPodReady)).To(BeNil())
+		})
+
+		It("connects to the container once the Pod is running", func() {
+			By("reporting the Pod status that a kubelet would report")
+			// envtest runs no kubelet, so nothing else moves the Pod to Running.
+			pod := &v1.Pod{}
+			Expect(k8sClient.Get(ctx, podName, pod)).To(Succeed())
+			pod.Status.Phase = v1.PodRunning
+			pod.Status.PodIP = podIP
+			Expect(k8sClient.Status().Update(ctx, pod)).To(Succeed())
+
+			reconcileOnce()
+
+			occClient, connected := clientsForContainers.Load(resourceName)
+			Expect(connected).To(BeTrue())
+
+			task := getTask()
+			Expect(meta.IsStatusConditionTrue(task.Status.Conditions, aliecsv1alpha1.ConditionPodReady)).To(BeTrue())
+			Expect(meta.IsStatusConditionTrue(task.Status.Conditions, aliecsv1alpha1.ConditionGRPCConnected)).To(BeTrue())
+
+			// This pass asked to be requeued until the connection is Ready, which is what
+			// the next step needs.
+			connectCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+			defer cancel()
+			Expect(occClient.(*OccClient).WaitUntilConnected(connectCtx)).To(Succeed())
+		})
+
+		It("reads the state from the container", func() {
+			reconcileOnce()
+
+			task := getTask()
+			Expect(task.Status.State).To(Equal(occServer.getState()))
+			Expect(task.Status.Error).To(BeEmpty())
+			Expect(meta.IsStatusConditionTrue(task.Status.Conditions, aliecsv1alpha1.ConditionStateAccessible)).To(BeTrue())
+		})
+
+		It("transitions the container to a newly requested state", func() {
+			By("asking the Task for a state the container is not in")
+			task := getTask()
+			task.Spec.State = "configured"
+			Expect(k8sClient.Update(ctx, task)).To(Succeed())
+
+			reconcileOnce()
+
+			task = getTask()
+			Expect(task.Status.State).To(Equal("configured"))
+			Expect(task.Status.Error).To(BeEmpty())
+			Expect(meta.IsStatusConditionTrue(task.Status.Conditions, aliecsv1alpha1.ConditionStateTransitionResult)).To(BeTrue())
+
+			By("leaving the container in the requested state")
+			Expect(occServer.getState()).To(Equal("configured"))
+		})
+
+		It("clears the connection and the Pod when the Task is deleted", func() {
+			Expect(k8sClient.Delete(ctx, getTask())).To(Succeed())
+
+			reconcileOnce()
+
+			_, connected := clientsForContainers.Load(resourceName)
+			Expect(connected).To(BeFalse())
+
+			// Nothing schedules the Pod in envtest, so the API server drops it right away
+			// instead of waiting for a kubelet to confirm termination.
+			Expect(k8sClient.Get(ctx, podName, &v1.Pod{})).To(Satisfy(errors.IsNotFound))
+			Expect(controllerutil.ContainsFinalizer(getTask(), taskFinalizer)).To(BeTrue(),
+				"this pass only deletes the Pod, the finalizer goes on the next one")
+
+			reconcileOnce()
+
+			Expect(k8sClient.Get(ctx, taskName, &aliecsv1alpha1.Task{})).To(Satisfy(errors.IsNotFound))
 		})
 	})
 })