Disallow user to change workflow scope

If workflow has any cron-trigger assoaciated
outside of the workflow's tenant, user is
not allowed to change its scope from 'public'
to 'private'.

Change-Id: I90bd903d46140af26a729d62176e079e02406eae
Closes-bug: #1518655

mistral/db/v2/sqlalchemy/api.py

@@ -346,6 +346,18 @@ def update_workflow_definition(identifier, values, session=None):
             "Attempt to modify a system workflow: %s" % identifier
         )
 
+    if wf_def.scope == 'public' and values['scope'] == 'private':
+        cron_triggers = _get_associated_cron_triggers(identifier)
+
+        try:
+            [get_cron_trigger(name) for name in cron_triggers]
+        except exc.NotFoundException:
+            raise exc.NotAllowedException(
+                "Can not update scope of workflow that has triggers "
+                "associated in other tenants."
+                "[workflow_identifier=%s]" % identifier
+            )
+
     wf_def.update(values.copy())
 
     return wf_def

mistral/tests/unit/db/v2/test_sqlalchemy_db_api.py

@@ -243,6 +243,20 @@ WF_DEFINITIONS = [
 ]
 
 
+CRON_TRIGGER = {
+    'name': 'trigger1',
+    'pattern': '* * * * *',
+    'workflow_name': 'my_wf1',
+    'workflow_id': None,
+    'workflow_input': {},
+    'next_execution_time':
+    datetime.datetime.now() + datetime.timedelta(days=1),
+    'remaining_executions': 42,
+    'scope': 'private',
+    'project_id': '<default-project>'
+}
+
+
 class WorkflowDefinitionTest(SQLAlchemyTest):
     def test_create_and_get_and_load_workflow_definition(self):
         created = db_api.create_workflow_definition(WF_DEFINITIONS[0])
@@ -281,7 +295,7 @@ class WorkflowDefinitionTest(SQLAlchemyTest):
         # Update workflow using workflow name as identifier.
         updated = db_api.update_workflow_definition(
             created['name'],
-            {'definition': 'my new definition'}
+            {'definition': 'my new definition', 'scope': 'private'}
         )
 
         self.assertEqual('my new definition', updated.definition)
@@ -294,7 +308,11 @@ class WorkflowDefinitionTest(SQLAlchemyTest):
         # Update workflow using workflow uuid as identifier.
         updated = db_api.update_workflow_definition(
             created['id'],
-            {'name': 'updated_name', 'definition': 'my new definition'}
+            {
+                'name': 'updated_name',
+                'definition': 'my new definition',
+                'scope': 'private'
+            }
         )
 
         self.assertEqual('updated_name', updated.name)
@@ -315,7 +333,7 @@ class WorkflowDefinitionTest(SQLAlchemyTest):
             exc.NotAllowedException,
             db_api.update_workflow_definition,
             created.name,
-            {'definition': 'my new definition'}
+            {'definition': 'my new definition', 'scope': 'private'}
         )
 
     def test_create_or_update_workflow_definition(self):
@@ -333,7 +351,7 @@ class WorkflowDefinitionTest(SQLAlchemyTest):
 
         updated = db_api.create_or_update_workflow_definition(
             created.name,
-            {'definition': 'my new definition'}
+            {'definition': 'my new definition', 'scope': 'private'}
         )
 
         self.assertEqual('my new definition', updated.definition)
@@ -346,6 +364,34 @@ class WorkflowDefinitionTest(SQLAlchemyTest):
 
         self.assertEqual(updated, fetched)
 
+    def test_update_wf_scope_cron_trigger_associated_in_diff_tenant(self):
+        created = db_api.create_workflow_definition(WF_DEFINITIONS[0])
+
+        # Create a new user.
+        auth_context.set_ctx(test_base.get_context(default=False))
+
+        db_api.create_cron_trigger(CRON_TRIGGER)
+
+        auth_context.set_ctx(test_base.get_context(default=True))
+
+        self.assertRaises(
+            exc.NotAllowedException,
+            db_api.update_workflow_definition,
+            created['name'],
+            {'scope': 'private'}
+        )
+
+    def test_update_wf_scope_cron_trigger_associated_in_same_tenant(self):
+        created = db_api.create_workflow_definition(WF_DEFINITIONS[0])
+
+        db_api.create_cron_trigger(CRON_TRIGGER)
+        updated = db_api.update_workflow_definition(
+            created['name'],
+            {'scope': 'private'}
+        )
+
+        self.assertEqual('private', updated.scope)
+
     def test_get_workflow_definitions(self):
         created0 = db_api.create_workflow_definition(WF_DEFINITIONS[0])
         created1 = db_api.create_workflow_definition(WF_DEFINITIONS[1])