rest: validate REANA specification

- before, "create_workflow" endpoint was not validating REANA specification; this commit adds validation.
- it also fixes some tests as they incorrectly put "workflow_name" in JSON body instead of query string.

closes #468

CHANGES.rst

@@ -4,6 +4,7 @@ Changes
 Version 0.9.0 (UNRELEASED)
 --------------------------
 
+- Adds validation of REANA specification to ``create_workflow`` and ``start_workflow`` endpoints.
 - Adds interactive sessions out-of-sync check to ``reana-admin check-workflows`` command.
 - Adds workspace retention rules validation to ``get_workspace_retention_rules`` function.
 - Adds ``queue-consume`` command that can be used by REANA administrators to remove specific messages from the queue.

reana_server/rest/workflows.py

@@ -16,6 +16,8 @@
 from bravado.exception import HTTPError
 from flask import Blueprint, Response
 from flask import jsonify, request, stream_with_context
+from jsonschema.exceptions import ValidationError
+
 from reana_commons.config import REANA_WORKFLOW_ENGINES
 from reana_commons.errors import REANAQuotaExceededError, REANAValidationError
 from reana_commons.validation.operational_options import validate_operational_options
@@ -30,7 +32,7 @@
 from reana_server.api_client import current_rwc_api_client
 from reana_server.config import REANA_HOSTNAME
 from reana_server.decorators import check_quota, signin_required
-from reana_server.validation import validate_workspace_path
+from reana_server.validation import validate_workspace_path, validate_workflow
 from reana_server.utils import (
     _fail_gitlab_commit_build_status,
     RequestStreamWithLen,
@@ -523,6 +525,9 @@ def create_workflow(user):  # noqa
         if is_uuid_v4(workflow_name):
             return jsonify({"message": "Workflow name cannot be a valid UUIDv4."}), 400
 
+        validate_workflow(reana_spec_file, input_parameters={})
+        workspace_root_path = reana_spec_file.get("workspace", {}).get("root_path")
+
         workflow_engine = reana_spec_file["workflow"]["type"]
         if workflow_engine not in REANA_WORKFLOW_ENGINES:
             raise Exception("Unknown workflow type.")
@@ -531,9 +536,6 @@ def create_workflow(user):  # noqa
             workflow_engine, reana_spec_file.get("inputs", {}).get("options", {})
         )
 
-        workspace_root_path = reana_spec_file.get("workspace", {}).get("root_path")
-        validate_workspace_path(reana_spec_file)
-
         retention_days = reana_spec_file.get("workspace", {}).get("retention_days")
         retention_rules = get_workspace_retention_rules(retention_days)
 
@@ -1196,7 +1198,7 @@ def start_workflow(workflow_id_or_name, user):  # noqa
     except HTTPError as e:
         logging.error(traceback.format_exc())
         return jsonify(e.response.json()), e.response.status_code
-    except REANAValidationError as e:
+    except (REANAValidationError, ValidationError) as e:
         logging.error(traceback.format_exc())
         return jsonify({"message": str(e)}), 400
     except ValueError as e:

reana_server/utils.py

@@ -72,7 +72,7 @@
     WORKSPACE_RETENTION_PERIOD,
     DEFAULT_WORKSPACE_RETENTION_RULE,
 )
-from reana_server.validation import validate_retention_rule
+from reana_server.validation import validate_retention_rule, validate_workflow
 
 
 def is_uuid_v4(uuid_or_name):
@@ -594,6 +594,8 @@ def get_workspace_retention_rules(
 def clone_workflow(workflow, reana_spec, restart_type):
     """Create a copy of workflow in DB for restarting."""
     reana_specification = reana_spec or workflow.reana_specification
+    validate_workflow(reana_specification, input_parameters={})
+
     retention_days = reana_specification.get("workspace", {}).get("retention_days")
     retention_rules = get_workspace_retention_rules(retention_days)
     try:

tests/test_views.py

@@ -8,6 +8,7 @@
 
 """Test server views."""
 
+import copy
 import json
 from io import BytesIO
 from uuid import uuid4
@@ -16,7 +17,7 @@
 from flask import url_for
 from mock import Mock, patch
 from pytest_reana.test_utils import make_mock_api_client
-from reana_db.models import User, InteractiveSessionType
+from reana_db.models import User, InteractiveSessionType, RunStatus
 
 from reana_server.utils import (
     _create_and_associate_local_user,
@@ -53,7 +54,9 @@ def test_get_workflows(app, default_user, _get_user_mock):
             assert res.status_code == 200
 
 
-def test_create_workflow(app, default_user, _get_user_mock):
+def test_create_workflow(
+    app, session, default_user, _get_user_mock, sample_serial_workflow_in_db
+):
     """Test create_workflow view."""
     with app.test_client() as client:
         with patch(
@@ -89,55 +92,105 @@ def test_create_workflow(app, default_user, _get_user_mock):
             assert res.status_code == 500
 
             # unknown workflow engine
-            workflow_data = {
-                "workflow": {"specification": {}, "type": "unknown"},
-                "workflow_name": "test",
-            }
+            workflow_specification = copy.deepcopy(
+                sample_serial_workflow_in_db.reana_specification
+            )
+            workflow_specification["workflow"]["type"] = "unknown"
             res = client.post(
                 url_for("workflows.create_workflow"),
                 headers={"Content-Type": "application/json"},
-                query_string={"access_token": default_user.access_token},
-                data=json.dumps(workflow_data),
+                query_string={
+                    "access_token": default_user.access_token,
+                    "workflow_name": "test",
+                },
+                data=json.dumps(workflow_specification),
             )
             assert res.status_code == 500
 
             # name cannot be valid uuid4
-            workflow_data["workflow"]["type"] = "serial"
             res = client.post(
                 url_for("workflows.create_workflow"),
                 headers={"Content-Type": "application/json"},
                 query_string={
                     "access_token": default_user.access_token,
                     "workflow_name": str(uuid4()),
                 },
-                data=json.dumps(workflow_data),
+                data=json.dumps(sample_serial_workflow_in_db.reana_specification),
             )
             assert res.status_code == 400
 
             # wrong specification json
-            workflow_data = {"nonsense": {"specification": {}, "type": "unknown"}}
+            workflow_specification = {
+                "nonsense": {"specification": {}, "type": "unknown"}
+            }
             res = client.post(
                 url_for("workflows.create_workflow"),
                 headers={"Content-Type": "application/json"},
-                query_string={"access_token": default_user.access_token},
-                data=json.dumps(workflow_data),
+                query_string={
+                    "access_token": default_user.access_token,
+                    "workflow_name": "test",
+                },
+                data=json.dumps(workflow_specification),
             )
             assert res.status_code == 400
 
-            # correct case
-            workflow_data = {
+            # not valid specification
+            workflow_specification = {
                 "workflow": {"specification": {}, "type": "serial"},
-                "workflow_name": "test",
             }
             res = client.post(
                 url_for("workflows.create_workflow"),
                 headers={"Content-Type": "application/json"},
-                query_string={"access_token": default_user.access_token},
-                data=json.dumps(workflow_data),
+                query_string={
+                    "access_token": default_user.access_token,
+                    "workflow_name": "test",
+                },
+                data=json.dumps(workflow_specification),
+            )
+            assert res.status_code == 400
+
+            # correct case
+            workflow_specification = sample_serial_workflow_in_db.reana_specification
+            res = client.post(
+                url_for("workflows.create_workflow"),
+                headers={"Content-Type": "application/json"},
+                query_string={
+                    "access_token": default_user.access_token,
+                    "workflow_name": "test",
+                },
+                data=json.dumps(workflow_specification),
             )
             assert res.status_code == 200
 
 
+def test_restart_workflow_validates_specification(
+    app, session, default_user, sample_serial_workflow_in_db
+):
+    with app.test_client() as client:
+        sample_serial_workflow_in_db.status = RunStatus.finished
+        sample_serial_workflow_in_db.name = "test"
+        session.add(sample_serial_workflow_in_db)
+        session.commit()
+
+        workflow_specification = copy.deepcopy(
+            sample_serial_workflow_in_db.reana_specification
+        )
+        workflow_specification["workflow"]["type"] = "unknown"
+        body = {
+            "reana_specification": workflow_specification,
+            "restart": "can be anything here doesnt matter",
+        }
+        res = client.post(
+            url_for("workflows.start_workflow", workflow_id_or_name="test"),
+            headers={"Content-Type": "application/json"},
+            query_string={
+                "access_token": default_user.access_token,
+            },
+            data=json.dumps(body),
+        )
+        assert res.status_code == 400
+
+
 def test_get_workflow_specification(
     app, default_user, _get_user_mock, sample_yadage_workflow_in_db
 ):