From 2e4054513a543e221118f4dde1d1670278807285 Mon Sep 17 00:00:00 2001
From: hardik <hardik.parekh@nectechnologies.in>
Date: Sun, 20 Sep 2015 12:23:50 +0530
Subject: [PATCH] Check for trigger before delete wf.

When wf which has trigger is deleted
DBRefrenceException was thrown.

Change-Id: Icb8d73fafeede057ac3e698c91ec0bd7e63c2ba0
Closes-bug: #1383170
---
 mistral/db/v2/sqlalchemy/api.py                 | 17 +++++++++++++++++
 .../functional/api/v2/test_mistral_basic_v2.py  | 15 +++++++++++++++
 2 files changed, 32 insertions(+)

diff --git a/mistral/db/v2/sqlalchemy/api.py b/mistral/db/v2/sqlalchemy/api.py
index 5894d6d0..d19c37c2 100644
--- a/mistral/db/v2/sqlalchemy/api.py
+++ b/mistral/db/v2/sqlalchemy/api.py
@@ -343,9 +343,26 @@ def delete_workflow_definition(name, session=None):
             "Workflow not found [workflow_name=%s]" % name
         )
 
+    cron_triggers = _get_associated_cron_triggers(name)
+
+    if cron_triggers:
+        raise exc.DBException(
+            "Can't delete workflow that has triggers [workflow_name=%s],"
+            "[cron_trigger_name(s)=%s]" % (name, ', '.join(cron_triggers))
+        )
+
     session.delete(wf_def)
 
 
+def _get_associated_cron_triggers(wf_name):
+    cron_triggers = _secure_query(
+        models.CronTrigger,
+        models.CronTrigger.name
+    ).filter_by(workflow_name=wf_name).all()
+
+    return [t[0] for t in cron_triggers]
+
+
 @b.session_aware()
 def delete_workflow_definitions(**kwargs):
     return _delete_all(models.WorkflowDefinition, **kwargs)
diff --git a/mistral/tests/functional/api/v2/test_mistral_basic_v2.py b/mistral/tests/functional/api/v2/test_mistral_basic_v2.py
index 1a21c4bf..5093a14d 100644
--- a/mistral/tests/functional/api/v2/test_mistral_basic_v2.py
+++ b/mistral/tests/functional/api/v2/test_mistral_basic_v2.py
@@ -330,6 +330,21 @@ class WorkflowTestsV2(base.TestCase):
                           self.client.update_request,
                           'workflows', 'wb_v1.yaml')
 
+    @test.attr(type='negative')
+    def test_delete_wf_with_trigger_associate(self):
+        tr_name = 'trigger'
+        resp, body = self.client.create_workflow('wf_v2.yaml')
+        name = body['workflows'][0]['name']
+        resp, body = self.client.create_cron_trigger(
+            tr_name, name, None, '5 * * * *')
+
+        self.assertRaises(exceptions.BadRequest,
+                          self.client.delete_obj,
+                          'workflows', name)
+
+        self.client.delete_obj('cron_triggers', tr_name)
+        self.client.triggers.remove(tr_name)
+
     @test.attr(type='negative')
     def test_delete_nonexistent_wf(self):
         self.assertRaises(exceptions.NotFound,