e2e and unit test

Signed-off-by: Julie Vogelman <[email protected]>
numaproj · Sep 19, 2024 · d693eee · d693eee
1 parent 8368cc1
commit d693eee
Show file tree

Hide file tree

Showing 3 changed files with 35 additions and 6 deletions.
diff --git a/internal/controller/pause.go b/internal/controller/pause.go
@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/numaproj/numaplane/internal/common"
 	"github.com/numaproj/numaplane/internal/util/kubernetes"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/client-go/rest"
@@ -115,12 +114,28 @@ func (pm *PauseModule) updatePipelineLifecycle(ctx context.Context, restConfig *
 		return err
 	}
 
+	// TODO: I noticed if any of these fields are nil, this function errors out - but can't remember why they'd be nil
 	err = unstructured.SetNestedField(unstruc.Object, status, "spec", "lifecycle", "desiredPhase")
 	if err != nil {
 		return err
 	}
 
-	return kubernetes.UpdateUnstructuredCR(ctx, restConfig, unstruc, common.PipelineGVR, pipeline.Namespace, pipeline.Name)
+	resultObj, err := kubernetes.UnstructuredToObject(unstruc)
+	if err != nil {
+		return err
+	}
+	if resultObj == nil {
+		return fmt.Errorf("error converting unstructured %+v to object, result is nil?", unstruc.Object)
+	}
+
+	err = kubernetes.UpdateCR(ctx, restConfig, resultObj, "pipelines")
+	if err != nil {
+		return err
+	}
+	*pipeline = *resultObj
+	return nil
+	//return kubernetes.UpdateUnstructuredCR(ctx, restConfig, unstruc, common.PipelineGVR, pipeline.Namespace, pipeline.Name)*/
+
 }
 
 func (pm *PauseModule) getNumaflowControllerKey(namespace string) string {

diff --git a/internal/controller/pipelinerollout_controller.go b/internal/controller/pipelinerollout_controller.go
@@ -30,7 +30,6 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	k8stypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
@@ -571,19 +570,20 @@ func (r *PipelineRolloutReconciler) processExistingPipelineWithPPND(ctx context.
 			numaLogger.Infof("it's safe to update Pipeline so updating now")
 			r.recorder.Eventf(pipelineRollout, "Normal", "PipelineUpdate", "it's safe to update Pipeline so updating now")
 			// make sure lifecycle is left set to "Paused" in the new spec
-			unstruc, err := kubernetes.ObjectToUnstructured(newPipelineDef)
+			/*unstruc, err := kubernetes.ObjectToUnstructured(newPipelineDef)
 			if err != nil {
 				return false, err
 			}
-			// TODO: I noticed if any of these fields are nil, this function errors out - but can't remember why they'd be nil
 			err = unstructured.SetNestedField(unstruc.Object, "Paused", "spec", "lifecycle", "desiredPhase")
 			if err != nil {
 				return false, err
 			}
 			err = kubernetes.UpdateUnstructuredCR(ctx, r.restConfig, unstruc, common.PipelineGVR, pipelineRollout.Namespace, pipelineRollout.Name)
 			if err != nil {
 				return false, err
-			}
+			}*/
+			GetPauseModule().updatePipelineLifecycle(ctx, r.restConfig, newPipelineDef, "Paused")
+
 			pipelineRollout.Status.MarkDeployed(pipelineRollout.Generation)
 		}
 	}

diff --git a/internal/controller/pipelinerollout_controller_test.go b/internal/controller/pipelinerollout_controller_test.go
@@ -942,6 +942,20 @@ func Test_processExistingPipeline_PPND(t *testing.T) {
 				return reflect.DeepEqual(withDesiredPhase(pipelineSpecWithTopologyChange, numaflowv1.PipelinePhasePaused), spec)
 			},
 		},
+		{
+			name:                           "PPND in progress, spec applied, done being reconciled",
+			newPipelineSpec:                pipelineSpecWithTopologyChange,
+			existingPipelineDef:            *createPipelineOfSpec(withDesiredPhase(pipelineSpecWithTopologyChange, numaflowv1.PipelinePhasePaused), numaflowv1.PipelinePhasePaused, true),
+			initialRolloutPhase:            apiv1.PhaseDeployed,
+			initialInProgressStrategy:      &ppndUpgradeStrategy,
+			numaflowControllerPauseRequest: &falseValue,
+			isbServicePauseRequest:         &falseValue,
+			expectedInProgressStrategy:     apiv1.UpgradeStrategyNoOp,
+			expectedRolloutPhase:           apiv1.PhaseDeployed,
+			expectedPipelineSpecResult: func(spec numaflowv1.PipelineSpec) bool {
+				return reflect.DeepEqual(withDesiredPhase(pipelineSpecWithTopologyChange, numaflowv1.PipelinePhaseRunning), spec)
+			},
+		},
 	}
 
 	for _, tc := range testCases {