From 8c11e2d33d221243e1b1e06e486d8c4b72f4ab9b Mon Sep 17 00:00:00 2001
From: Marco Donadoni <marco.donadoni@cern.ch>
Date: Tue, 25 Jun 2024 17:02:19 +0200
Subject: [PATCH] fix(start): validate endpoint parameters (#689)

Closes reanahub/reana-client#718
---
 reana_server/rest/workflows.py | 36 +++++++++++++++++++++-------------
 1 file changed, 22 insertions(+), 14 deletions(-)

diff --git a/reana_server/rest/workflows.py b/reana_server/rest/workflows.py
index de6f7861..e8ffaf22 100644
--- a/reana_server/rest/workflows.py
+++ b/reana_server/rest/workflows.py
@@ -1132,8 +1132,16 @@ def get_workflow_status(workflow_id_or_name, user):  # noqa
 
 @blueprint.route("/workflows/<workflow_id_or_name>/start", methods=["POST"])
 @signin_required()
+@use_kwargs(
+    {
+        "operational_options": fields.Dict(missing={}, location="json"),
+        "input_parameters": fields.Dict(missing={}, location="json"),
+        "restart": fields.Boolean(location="json"),
+        "reana_specification": fields.Raw(location="json"),
+    }
+)
 @check_quota
-def start_workflow(workflow_id_or_name, user):  # noqa
+def start_workflow(workflow_id_or_name, user, **kwargs):  # noqa
     r"""Start workflow.
     ---
     post:
@@ -1285,17 +1293,23 @@ def start_workflow(workflow_id_or_name, user):  # noqa
                 "message": "Status resume is not supported yet."
               }
     """
+
+    operational_options = kwargs["operational_options"]
+    input_parameters = kwargs["input_parameters"]
+    restart = kwargs.get("restart", False)
+    reana_specification = kwargs.get("reana_specification")
+
     try:
         if not workflow_id_or_name:
             raise ValueError("workflow_id_or_name is not supplied")
-        parameters = request.json if request.is_json else {}
+
         workflow = _get_workflow_with_uuid_or_name(workflow_id_or_name, str(user.id_))
-        operational_options = parameters.get("operational_options", {})
         operational_options = validate_operational_options(
             workflow.type_, operational_options
         )
+
         restart_type = None
-        if "restart" in parameters:
+        if restart:
             if workflow.status not in [RunStatus.finished, RunStatus.failed]:
                 raise ValueError("Only finished or failed workflows can be restarted.")
             if workflow.workspace_has_pending_retention_rules():
@@ -1303,14 +1317,9 @@ def start_workflow(workflow_id_or_name, user):  # noqa
                     "The workflow cannot be restarted because some retention rules are "
                     "currently being applied to the workspace. Please retry later."
                 )
-            restart_type = (
-                parameters.get("reana_specification", {})
-                .get("workflow", {})
-                .get("type", None)
-            )
-            workflow = clone_workflow(
-                workflow, parameters.get("reana_specification", None), restart_type
-            )
+            if reana_specification:
+                restart_type = reana_specification.get("workflow", {}).get("type", None)
+            workflow = clone_workflow(workflow, reana_specification, restart_type)
         elif workflow.status != RunStatus.created:
             raise ValueError(
                 "Workflow {} is already {} and cannot be started "
@@ -1319,12 +1328,11 @@ def start_workflow(workflow_id_or_name, user):  # noqa
         if "yadage" in (workflow.type_, restart_type):
             _load_and_save_yadage_spec(workflow, operational_options)
 
-        input_parameters = parameters.get("input_parameters", {})
         validate_workflow(
             workflow.reana_specification, input_parameters=input_parameters
         )
 
-        publish_workflow_submission(workflow, user.id_, parameters)
+        publish_workflow_submission(workflow, user.id_, kwargs)
         response = {
             "message": "Workflow submitted.",
             "workflow_id": workflow.id_,