Add unit test for deleting workflows by admin

Partially implements: blueprint mistral-rbac

Change-Id: I8daf439fb4f05105bce1870901b77b502763802c
This commit is contained in:
Lingxian Kong 2017-03-16 00:33:06 +13:00
parent e938d47a25
commit 6229282dd5
2 changed files with 20 additions and 3 deletions

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 base as b
from mistral.db.sqlalchemy import model_base as mb from mistral.db.sqlalchemy import model_base as mb
from mistral.db.sqlalchemy import sqlite_lock 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 filters as db_filters
from mistral.db.v2.sqlalchemy import models from mistral.db.v2.sqlalchemy import models
from mistral import exceptions as exc 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): def update_workflow_definition(identifier, values, session=None):
wf_def = get_workflow_definition(identifier, insecure=True) 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': if wf_def.scope == 'public' and values['scope'] == 'private':
# Check cron triggers. # Check cron triggers.
@ -536,7 +536,7 @@ def create_or_update_workflow_definition(name, values, session=None):
def delete_workflow_definition(identifier, session=None): def delete_workflow_definition(identifier, session=None):
wf_def = get_workflow_definition(identifier) 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) cron_triggers = get_cron_triggers(insecure=True, workflow_id=wf_def.id)
if cron_triggers: if cron_triggers:

View File

@ -818,6 +818,23 @@ class WorkflowDefinitionTest(SQLAlchemyTest):
created.name 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): def test_workflow_definition_private(self):
# Create a workflow(scope=private) as under one project # Create a workflow(scope=private) as under one project
# then make sure it's NOT visible for other projects. # then make sure it's NOT visible for other projects.