openstack/mistral
Code Issues Proposed changes

Merge "Add unit test for deleting workflows by admin"

Browse Source
This commit is contained in:
Jenkins 2017-03-16 08:53:55 +00:00 committed by Gerrit Code Review
commit b950d52919
2 changed files with 20 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
mistral/db/v2/sqlalchemy/api.py
View File

@ -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:

17
mistral/tests/unit/db/v2/test_sqlalchemy_db_api.py
View File

@ -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.