test failure in executor

Signed-off-by: Babis Kiosidis <[email protected]>

pkg/controller/handler.go

@@ -245,7 +245,7 @@ func (p *Propeller) Handle(ctx context.Context, namespace, name string) error {
 		t := p.metrics.RoundTime.Start(ctx)
 		mutatedWf, err := p.TryMutateWorkflow(ctx, w)
 
-		if wfClosureCrdFields != nil  {
+		if wfClosureCrdFields != nil {
 			// strip data populated from WorkflowClosureReference
 			w.SubWorkflows, w.Tasks, w.WorkflowSpec = nil, nil, nil
 			if mutatedWf != nil {

pkg/controller/handler_test.go

@@ -894,4 +894,24 @@ func TestPropellerHandler_OffloadedWorkflowClosure(t *testing.T) {
 		err := p.Handle(ctx, namespace, name)
 		assert.Error(t, err)
 	})
+
+	t.Run("TryMutate failure is handled", func(t *testing.T) {
+		scope := promutils.NewTestScope()
+
+		protoStore := &storagemocks.ComposedProtobufStore{}
+		protoStore.OnReadProtobufMatch(mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("foo"))
+		exec.HandleCb = func(ctx context.Context, w *v1alpha1.FlyteWorkflow) error {
+			return fmt.Errorf("foo")
+		}
+		dataStore := storage.NewCompositeDataStore(storage.URLPathConstructor{}, protoStore)
+		p := NewPropellerHandler(ctx, cfg, dataStore, s, exec, scope)
+
+		err := p.Handle(ctx, namespace, name)
+		assert.Error(t, err, "foo")
+
+		r, err := s.Get(ctx, namespace, name)
+		assert.Nil(t, r.WorkflowSpec)
+		assert.Nil(t, r.SubWorkflows)
+		assert.Nil(t, r.Tasks)
+	})
 }