Get "cron trigger" list using model query.

If cron-trigger is created in different tenant
than workflow's tenant, associated cron-trigger
can not be got. So "model_query" has been used
instead of "secure_query".

Change-Id: Ief78f2cad76e6ca16893bff80c5e75c9015e9c74
Closes-bug: #1522286

mistral/db/v2/sqlalchemy/api.py

@@ -364,9 +364,9 @@ def delete_workflow_definition(name, session=None):
 
 
 def _get_associated_cron_triggers(wf_name):
-    cron_triggers = _secure_query(
+    cron_triggers = b.model_query(
         models.CronTrigger,
-        models.CronTrigger.name
+        [models.CronTrigger.name]
     ).filter_by(workflow_name=wf_name).all()
 
     return [t[0] for t in cron_triggers]

mistral/tests/functional/api/v2/test_mistral_basic_v2.py

@@ -338,12 +338,45 @@ class WorkflowTestsV2(base.TestCase):
         resp, body = self.client.create_cron_trigger(
             tr_name, name, None, '5 * * * *')
 
-        self.assertRaises(exceptions.BadRequest,
-                          self.client.delete_obj,
-                          'workflows', name)
+        try:
+            self.assertRaises(
+                exceptions.BadRequest,
+                self.client.delete_obj,
+                'workflows',
+                name
+            )
+        finally:
+            self.client.delete_obj('cron_triggers', tr_name)
+            self.client.triggers.remove(tr_name)
 
-        self.client.delete_obj('cron_triggers', tr_name)
-        self.client.triggers.remove(tr_name)
+    @test.attr(type='negative')
+    def test_delete_wf_with_trigger_associate_in_other_tenant(self):
+        tr_name = 'trigger'
+        _, body = self.client.create_workflow('wf_v2.yaml', scope='public')
+        name = body['workflows'][0]['name']
+        resp, body = self.alt_client.create_cron_trigger(
+            tr_name,
+            name,
+            None,
+            '5 * * * *'
+        )
+
+        try:
+            exception = self.assertRaises(
+                exceptions.BadRequest,
+                self.client.delete_obj,
+                'workflows',
+                name
+            )
+
+            self.assertEqual(
+                "Can't delete workflow that has triggers " +
+                "[workflow_name=wf2],[cron_trigger_name(s)=trigger]",
+                exception.resp_body['faultstring']
+            )
+        finally:
+            self.alt_client.delete_obj('cron_triggers', tr_name)
+            self.alt_client.triggers.remove(tr_name)
 
     @test.attr(type='negative')
     def test_delete_nonexistent_wf(self):

mistral/tests/functional/base.py

@@ -154,8 +154,11 @@ class MistralClientV2(MistralClientBase):
 
         return resp, json.loads(body)
 
-    def create_workflow(self, yaml_file):
-        resp, body = self.post_request('workflows', yaml_file)
+    def create_workflow(self, yaml_file, scope=None):
+        if scope:
+            resp, body = self.post_request('workflows?scope=public', yaml_file)
+        else:
+            resp, body = self.post_request('workflows', yaml_file)
 
         for wf in json.loads(body)['workflows']:
             self.workflows.append(wf['name'])
@@ -259,9 +262,16 @@ class TestCase(test.BaseTestCase):
             )
             cls.mgr = clients.Manager(cls.creds)
 
+            cls.alt_creds = creds.get_configured_credentials(
+                credential_type='alt_user'
+            )
+            cls.alt_mgr = clients.Manager(cls.alt_creds)
+
         if cls._service == 'workflowv2':
             cls.client = MistralClientV2(
                 cls.mgr.auth_provider, cls._service)
+            cls.alt_client = MistralClientV2(
+                cls.alt_mgr.auth_provider, cls._service)
 
     def setUp(self):
         super(TestCase, self).setUp()