Fix update workflow by admin

Change-Id: I38cada2e467d2b2b2ca0f332e9331caa2fc64f47
Closes-Bug: #1667174

mistral/db/v2/sqlalchemy/api.py

@@ -259,8 +259,8 @@ def _get_db_object_by_id(model, id):
     return _secure_query(model).filter_by(id=id).first()
 
 
-def _get_db_object_by_name_or_id(model, identifier):
-    query = _secure_query(model)
+def _get_db_object_by_name_or_id(model, identifier, insecure=False):
+    query = b.model_query(model) if insecure else _secure_query(model)
     query = query.filter(
         sa.or_(
             model.id == identifier,
@@ -417,16 +417,19 @@ def delete_workbooks(session=None, **kwargs):
 # Workflow definitions.
 
 @b.session_aware()
-def get_workflow_definition(identifier, session=None):
+def get_workflow_definition(identifier, session=None, insecure=False):
     """Gets workflow definition by name or uuid.
 
     :param identifier: Identifier could be in the format of plain string or
                        uuid.
+    :param insecure: If True, will search all records, otherwise will narrow
+        scope to project id.
     :return: Workflow definition.
     """
     wf_def = _get_db_object_by_name_or_id(
         models.WorkflowDefinition,
-        identifier
+        identifier,
+        insecure=insecure
     )
 
     if not wf_def:
@@ -487,7 +490,7 @@ def create_workflow_definition(values, session=None):
 
 @b.session_aware()
 def update_workflow_definition(identifier, values, session=None):
-    wf_def = get_workflow_definition(identifier)
+    wf_def = get_workflow_definition(identifier, insecure=True)
     ctx = auth_ctx.ctx()
 
     if not ctx.is_admin and wf_def.project_id != security.get_project_id():

mistral/tests/resources/single_wf.yaml

@@ -0,0 +1,11 @@
+---
+version: '2.0'
+
+single_wf:
+  type: direct
+
+  tasks:
+    hello:
+      action: std.echo output="Hello"
+      publish:
+        result: <% task(hello).result %>

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

@@ -616,7 +616,7 @@ class WorkflowDefinitionTest(SQLAlchemyTest):
         )
 
     def test_update_other_project_workflow_by_admin(self):
-        created = db_api.create_workflow_definition(WF_DEFINITIONS[0])
+        created = db_api.create_workflow_definition(WF_DEFINITIONS[1])
 
         # Switch to admin.
         auth_context.set_ctx(test_base.get_context(default=False, admin=True))

mistral_tempest_tests/tests/api/v2/test_workflows.py

@@ -235,10 +235,10 @@ class WorkflowTestsV2(base.TestCase):
     @decorators.idempotent_id('f516aad0-9a50-4ace-a217-fa1931fd9335')
     def test_update_workflow(self):
         self.useFixture(lockutils.LockFixture('mistral-workflow'))
-        _, body = self.client.create_workflow('wf_v2.yaml')
+        _, body = self.client.create_workflow('single_wf.yaml')
         name = body['workflows'][0]['name']
 
-        resp, body = self.client.update_request('workflows', 'wf_v2.yaml')
+        resp, body = self.client.update_request('workflows', 'single_wf.yaml')
 
         self.assertEqual(200, resp.status)
         self.assertEqual(name, body['workflows'][0]['name'])