Adding CheckAndFetchInputsForExecution on single task executions (fly…

…teorg#495) * adding CheckAndFetchInputsForExecution on single task executions Signed-off-by: Dan Rammer <[email protected]> * fixed unit test Signed-off-by: Dan Rammer <[email protected]> * fixed typo Signed-off-by: Dan Rammer <[email protected]> Signed-off-by: Dan Rammer <[email protected]>
eapolinario · Dec 13, 2022 · 905e7ee · 905e7ee
1 parent a1804d5
commit 905e7ee
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 37 deletions.
diff --git a/pkg/manager/impl/execution_manager.go b/pkg/manager/impl/execution_manager.go
@@ -575,6 +575,18 @@ func (m *ExecutionManager) launchSingleTaskExecution(
 		return nil, nil, err
 	}
 
+	executionInputs, err := validation.CheckAndFetchInputsForExecution(
+		request.Inputs,
+		launchPlan.Spec.FixedInputs,
+		launchPlan.Closure.ExpectedInputs,
+	)
+	if err != nil {
+		logger.Debugf(ctx, "Failed to CheckAndFetchInputsForExecution with request.Inputs: %+v"+
+			"fixed inputs: %+v and expected inputs: %+v with err %v",
+			request.Inputs, launchPlan.Spec.FixedInputs, launchPlan.Closure.ExpectedInputs, err)
+		return nil, nil, err
+	}
+
 	name := util.GetExecutionName(request)
 	workflowExecutionID := core.WorkflowExecutionIdentifier{
 		Project: request.Project,
@@ -647,7 +659,7 @@ func (m *ExecutionManager) launchSingleTaskExecution(
 	}
 
 	executionParameters := workflowengineInterfaces.ExecutionParameters{
-		Inputs:              request.Inputs,
+		Inputs:              executionInputs,
 		AcceptedAt:          requestedAt,
 		Labels:              labels,
 		Annotations:         annotations,

diff --git a/pkg/manager/impl/execution_manager_test.go b/pkg/manager/impl/execution_manager_test.go
@@ -4217,6 +4217,18 @@ func TestCreateSingleTaskExecution(t *testing.T) {
 			return nil
 		})
 
+	var launchplan *models.LaunchPlan
+	repository.LaunchPlanRepo().(*repositoryMocks.MockLaunchPlanRepo).SetCreateCallback(func(input models.LaunchPlan) error {
+		launchplan = &input
+		return nil
+	})
+	repository.LaunchPlanRepo().(*repositoryMocks.MockLaunchPlanRepo).SetGetCallback(func(input interfaces.Identifier) (models.LaunchPlan, error) {
+		if launchplan == nil {
+			return models.LaunchPlan{}, flyteAdminErrors.NewFlyteAdminError(codes.NotFound, "launchplan not found")
+		}
+		return *launchplan, nil
+	})
+
 	mockStorage := getMockStorageForExecTest(context.Background())
 	workflowManager := NewWorkflowManager(
 		repository,
@@ -4255,45 +4267,17 @@ func TestCreateSingleTaskExecution(t *testing.T) {
 			},
 		},
 	}
+
 	marshaller := jsonpb.Marshaler{}
-	stringReq, ferr := marshaller.MarshalToString(&request)
+	_, ferr := marshaller.MarshalToString(&request)
 	assert.NoError(t, ferr)
-	println(fmt.Sprintf("req: %+v", stringReq))
-	_, err := execManager.CreateExecution(context.TODO(), admin.ExecutionCreateRequest{
-		Project: "flytekit",
-		Domain:  "production",
-		Name:    "singletaskexec",
-		Spec: &admin.ExecutionSpec{
-			LaunchPlan: &core.Identifier{
-				Project:      "flytekit",
-				Domain:       "production",
-				Name:         "simple_task",
-				Version:      "12345",
-				ResourceType: core.ResourceType_TASK,
-			},
-			AuthRole: &admin.AuthRole{
-				KubernetesServiceAccount: "foo",
-			},
-		},
-		Inputs: &core.LiteralMap{
-			Literals: map[string]*core.Literal{
-				"a": {
-					Value: &core.Literal_Scalar{
-						Scalar: &core.Scalar{
-							Value: &core.Scalar_Primitive{
-								Primitive: &core.Primitive{
-									Value: &core.Primitive_Integer{
-										Integer: 999,
-									},
-								},
-							},
-						},
-					},
-				},
-			},
-		},
-	}, time.Now())
 
+	// test once to create an initial launchplan
+	_, err := execManager.CreateExecution(context.TODO(), request, time.Now())
+	assert.NoError(t, err)
+
+	// test again to ensure existing launchplan retrieval works
+	_, err = execManager.CreateExecution(context.TODO(), request, time.Now())
 	assert.NoError(t, err)
 }