diff --git a/mistral/db/v2/sqlalchemy/api.py b/mistral/db/v2/sqlalchemy/api.py index b8a07556b..a752e7a67 100644 --- a/mistral/db/v2/sqlalchemy/api.py +++ b/mistral/db/v2/sqlalchemy/api.py @@ -31,7 +31,7 @@ from sqlalchemy.sql.expression import Insert from mistral.db.sqlalchemy import base as b from mistral.db.sqlalchemy import model_base as mb from mistral.db.sqlalchemy import sqlite_lock -from mistral.db import utils as mdb_utils +from mistral.db import utils as m_dbutils from mistral.db.v2.sqlalchemy import filters as db_filters from mistral.db.v2.sqlalchemy import models from mistral import exceptions as exc @@ -492,7 +492,7 @@ def create_workflow_definition(values, session=None): def update_workflow_definition(identifier, values, session=None): wf_def = get_workflow_definition(identifier, insecure=True) - mdb_utils.check_db_obj_access(wf_def) + m_dbutils.check_db_obj_access(wf_def) if wf_def.scope == 'public' and values['scope'] == 'private': # Check cron triggers. @@ -536,7 +536,7 @@ def create_or_update_workflow_definition(name, values, session=None): def delete_workflow_definition(identifier, session=None): wf_def = get_workflow_definition(identifier) - mdb_utils.check_db_obj_access(wf_def) + m_dbutils.check_db_obj_access(wf_def) cron_triggers = get_cron_triggers(insecure=True, workflow_id=wf_def.id) if cron_triggers: diff --git a/mistral/tests/unit/db/v2/test_sqlalchemy_db_api.py b/mistral/tests/unit/db/v2/test_sqlalchemy_db_api.py index e845de5f2..b76ba9cfb 100644 --- a/mistral/tests/unit/db/v2/test_sqlalchemy_db_api.py +++ b/mistral/tests/unit/db/v2/test_sqlalchemy_db_api.py @@ -818,6 +818,23 @@ class WorkflowDefinitionTest(SQLAlchemyTest): created.name ) + def test_delete_other_project_workflow_definition_by_admin(self): + created = db_api.create_workflow_definition(WF_DEFINITIONS[0]) + + # Switch to admin. + auth_context.set_ctx(test_base.get_context(default=False, admin=True)) + + db_api.delete_workflow_definition(created['id']) + + # Switch back. + auth_context.set_ctx(test_base.get_context()) + + self.assertRaises( + exc.DBEntityNotFoundError, + db_api.get_workflow_definition, + created['id'] + ) + def test_workflow_definition_private(self): # Create a workflow(scope=private) as under one project # then make sure it's NOT visible for other projects.