def _get_criterion(resource_id, member_id=None, is_owner=True):
    """Generates criterion for querying resource_member_v2 table."""

    # Resource owner query resource membership with member_id.
    if is_owner and member_id:
        return sa.and_(
            models.ResourceMember.project_id == security.get_project_id(),
            models.ResourceMember.resource_id == resource_id,
            models.ResourceMember.member_id == member_id
        )
    # Resource owner query resource memberships.
    elif is_owner and not member_id:
        return sa.and_(
            models.ResourceMember.project_id == security.get_project_id(),
            models.ResourceMember.resource_id == resource_id,
        )

    # Other members query other resource membership.
    elif not is_owner and member_id and member_id != security.get_project_id():
        return None

    # Resource member query resource memberships.
    return sa.and_(
        models.ResourceMember.member_id == security.get_project_id(),
        models.ResourceMember.resource_id == resource_id
    )

    def test_get_workflow_definitions(self):
        created0 = db_api.create_workflow_definition(WF_DEFINITIONS[0])
        created1 = db_api.create_workflow_definition(WF_DEFINITIONS[1])

        fetched0 = db_api.load_workflow_definition(created0.name)
        fetched1 = db_api.load_workflow_definition(created1.name)

        self.assertEqual(security.get_project_id(), fetched0.project_id)
        self.assertEqual(security.get_project_id(), fetched1.project_id)

        fetched = db_api.get_workflow_definitions()

        self.assertEqual(2, len(fetched))
        self.assertEqual(created0, fetched[0])
        self.assertEqual(created1, fetched[1])

    def _create_action_execution(self, input_dict, runtime_ctx,
                                 desc='', action_ex_id=None):
        action_ex_id = action_ex_id or utils.generate_unicode_uuid()

        values = {
            'id': action_ex_id,
            'name': self.action_def.name,
            'spec': self.action_def.spec,
            'state': states.RUNNING,
            'input': input_dict,
            'runtime_context': runtime_ctx,
            'description': desc
        }

        if self.task_ex:
            values.update({
                'task_execution_id': self.task_ex.id,
                'workflow_name': self.task_ex.workflow_name,
                'workflow_id': self.task_ex.workflow_id,
                'project_id': self.task_ex.project_id,
            })
        else:
            values.update({
                'project_id': security.get_project_id(),
            })

        self.action_ex = db_api.create_action_execution(values)

        if self.task_ex:
            # Add to collection explicitly so that it's in a proper
            # state within the current session.
            self.task_ex.executions.append(self.action_ex)

def delete_workflow_definition(identifier, session=None):
    wf_def = get_workflow_definition(identifier)

    if wf_def.project_id != security.get_project_id():
        raise exc.NotAllowedException(
            "Can not delete workflow of other users. [workflow_identifier=%s]"
            % identifier
        )

    if wf_def.is_system:
        msg = "Attempt to delete a system workflow: %s" % identifier
        raise exc.DataAccessException(msg)

    cron_triggers = get_cron_triggers(insecure=True, workflow_id=wf_def.id)
    if cron_triggers:
        raise exc.DBError(
            "Can't delete workflow that has cron triggers associated. "
            "[workflow_identifier=%s], [cron_trigger_id(s)=%s]" %
            (identifier, ', '.join([t.id for t in cron_triggers]))
        )

    event_triggers = get_event_triggers(insecure=True, workflow_id=wf_def.id)

    if event_triggers:
        raise exc.DBError(
            "Can't delete workflow that has event triggers associated. "
            "[workflow_identifier=%s], [event_trigger_id(s)=%s]" %
            (identifier, ', '.join([t.id for t in event_triggers]))
        )

    # Delete workflow members first.
    delete_resource_members(resource_type='workflow', resource_id=wf_def.id)

    session.delete(wf_def)

def update_resource_member(resource_id, res_type, member_id, values,
                           session=None):
    # Only member who is not the owner of the resource can update the
    # membership status.
    if member_id != security.get_project_id():
        raise exc.DBEntityNotFoundError(
            "Resource member not found [resource_id=%s, member_id=%s]" %
            (resource_id, member_id)
        )

    query = _secure_query(models.ResourceMember).filter_by(
        resource_type=res_type
    )

    res_member = query.filter(
        _get_criterion(resource_id, member_id, is_owner=False)
    ).first()

    if not res_member:
        raise exc.DBEntityNotFoundError(
            "Resource member not found [resource_id=%s, member_id=%s]" %
            (resource_id, member_id)
        )

    res_member.update(values.copy())

    return res_member

def update_workflow_definition(identifier, values, session=None):
    wf_def = get_workflow_definition(identifier)

    if wf_def.project_id != security.get_project_id():
        raise exc.NotAllowedException(
            "Can not update workflow of other tenants. "
            "[workflow_identifier=%s]" % identifier
        )

    if wf_def.is_system:
        raise exc.InvalidActionException(
            "Attempt to modify a system workflow: %s" % identifier
        )

    if wf_def.scope == 'public' and values['scope'] == 'private':
        cron_triggers = _get_associated_cron_triggers(identifier)

        try:
            [get_cron_trigger(name) for name in cron_triggers]
        except exc.DBEntityNotFoundError:
            raise exc.NotAllowedException(
                "Can not update scope of workflow that has triggers "
                "associated in other tenants."
                "[workflow_identifier=%s]" % identifier
            )

    wf_def.update(values.copy())

    return wf_def

def _secure_query(model, *columns):
    query = b.model_query(model, columns)

    if not issubclass(model, mb.MistralSecureModelBase):
        return query

    shared_res_ids = []
    res_type = RESOURCE_MAPPING.get(model, '')

    if res_type:
        shared_res = _get_accepted_resources(res_type)
        shared_res_ids = [res.resource_id for res in shared_res]

    query_criterion = sa.or_(
        model.project_id == security.get_project_id(),
        model.scope == 'public'
    )

    # NOTE(kong): Include IN_ predicate in query filter only if shared_res_ids
    # is not empty to avoid sqlalchemy SAWarning and wasting a db call.
    if shared_res_ids:
        query_criterion = sa.or_(
            query_criterion,
            model.id.in_(shared_res_ids)
        )

    query = query.filter(query_criterion)

    return query

def _secure_query(model, *columns):
    query = b.model_query(model, columns)

    if issubclass(model, mb.MistralSecureModelBase):
        query = query.filter(sa.or_(model.project_id == security.get_project_id(), model.scope == "public"))

    return query

def _get_accepted_resources(res_type):
    resources = _secure_query(models.ResourceMember).filter(
        sa.and_(
            models.ResourceMember.resource_type == res_type,
            models.ResourceMember.status == 'accepted',
            models.ResourceMember.member_id == security.get_project_id()
        )
    ).all()

    return resources

def create_action_execution(action_def, action_input, task_ex=None,
                            index=0, description=''):
    # TODO(rakhmerov): We can avoid hitting DB at all when calling something
    # create_action_execution(), these operations can be just done using
    # SQLAlchemy session (1-level cache) and session flush (on TX commit) would
    # send necessary SQL queries to DB. Currently, session flush happens
    # on every operation which may not be optimal. The problem with using just
    # session level cache is in generating ids. Ids are generated only on
    # session flush. And now we have a lot places where we need to have ids
    # before TX completion.

    # Assign the action execution ID here to minimize database calls.
    # Otherwise, the input property of the action execution DB object needs
    # to be updated with the action execution ID after the action execution
    # DB object is created.
    action_ex_id = utils.generate_unicode_uuid()

    if a_m.has_action_context(
            action_def.action_class, action_def.attributes or {}) and task_ex:
        action_input.update(a_m.get_action_context(task_ex, action_ex_id))

    values = {
        'id': action_ex_id,
        'name': action_def.name,
        'spec': action_def.spec,
        'state': states.RUNNING,
        'input': action_input,
        'runtime_context': {'with_items_index': index},
        'description': description
    }

    if task_ex:
        values.update({
            'task_execution_id': task_ex.id,
            'workflow_name': task_ex.workflow_name,
            'workflow_id': task_ex.workflow_id,
            'project_id': task_ex.project_id,
        })
    else:
        values.update({
            'project_id': security.get_project_id(),
        })

    action_ex = db_api.create_action_execution(values)

    if task_ex:
        # Add to collection explicitly so that it's in a proper
        # state within the current session.
        task_ex.executions.append(action_ex)

    return action_ex

def check_db_obj_access(db_obj):
    """Check accessibility to db object."""
    ctx = context.ctx()
    is_admin = ctx.is_admin

    if not is_admin and db_obj.project_id != security.get_project_id():
        raise exc.NotAllowedException(
            "Can not access %s resource of other projects, ID: %s" %
            (db_obj.__class__.__name__, db_obj.id)
        )

    if not is_admin and hasattr(db_obj, 'is_system') and db_obj.is_system:
        raise exc.InvalidActionException(
            "Can not modify a system %s resource, ID: %s" %
            (db_obj.__class__.__name__, db_obj.id)
        )

    def _create_action_execution(self, input_dict, runtime_ctx, desc=''):
        # Assign the action execution ID here to minimize database calls.
        # Otherwise, the input property of the action execution DB object needs
        # to be updated with the action execution ID after the action execution
        # DB object is created.
        action_ex_id = utils.generate_unicode_uuid()

        # TODO(rakhmerov): Bad place, we probably need to push action context
        # to all actions. It's related to
        # https://blueprints.launchpad.net/mistral/+spec/mistral-custom-actions-api
        if a_m.has_action_context(
                self.action_def.action_class,
                self.action_def.attributes or {}) and self.task_ex:
            input_dict.update(
                a_m.get_action_context(self.task_ex, action_ex_id)
            )

        values = {
            'id': action_ex_id,
            'name': self.action_def.name,
            'spec': self.action_def.spec,
            'state': states.RUNNING,
            'input': input_dict,
            'runtime_context': runtime_ctx,
            'description': desc
        }

        if self.task_ex:
            values.update({
                'task_execution_id': self.task_ex.id,
                'workflow_name': self.task_ex.workflow_name,
                'workflow_id': self.task_ex.workflow_id,
                'project_id': self.task_ex.project_id,
            })
        else:
            values.update({
                'project_id': security.get_project_id(),
            })

        self.action_ex = db_api.create_action_execution(values)

        if self.task_ex:
            # Add to collection explicitly so that it's in a proper
            # state within the current session.
            self.task_ex.executions.append(self.action_ex)

def update_workflow_definition(identifier, values, session=None):
    wf_def = get_workflow_definition(identifier)

    if wf_def.project_id != security.get_project_id():
        raise exc.NotAllowedException(
            "Can not update workflow of other tenants. "
            "[workflow_identifier=%s]" % identifier
        )

    if wf_def.is_system:
        raise exc.InvalidActionException(
            "Attempt to modify a system workflow: %s" % identifier
        )

    if wf_def.scope == 'public' and values['scope'] == 'private':
        # Check cron triggers.
        cron_triggers = get_cron_triggers(insecure=True, workflow_id=wf_def.id)

        for c_t in cron_triggers:
            if c_t.project_id != wf_def.project_id:
                raise exc.NotAllowedException(
                    "Can not update scope of workflow that has cron triggers "
                    "associated in other tenants. [workflow_identifier=%s]" %
                    identifier
                )

        # Check event triggers.
        event_triggers = get_event_triggers(
            insecure=True,
            workflow_id=wf_def.id
        )
        for e_t in event_triggers:
            if e_t.project_id != wf_def.project_id:
                raise exc.NotAllowedException(
                    "Can not update scope of workflow that has event triggers "
                    "associated in other tenants. [workflow_identifier=%s]" %
                    identifier
                )

    wf_def.update(values.copy())

    return wf_def

def _secure_query(model, *columns):
    query = b.model_query(model, columns)

    shared_res_ids = []
    res_type = RESOURCE_MAPPING.get(model, '')

    if res_type:
        shared_res = _get_accepted_resources(res_type)
        shared_res_ids = [res.resource_id for res in shared_res]

    if issubclass(model, mb.MistralSecureModelBase):
        query = query.filter(
            sa.or_(
                model.project_id == security.get_project_id(),
                model.scope == 'public',
                model.id.in_(shared_res_ids)
            )
        )

    return query

def delete_workflow_definition(identifier, session=None):
    wf_def = get_workflow_definition(identifier)

    if wf_def.project_id != security.get_project_id():
        raise exc.NotAllowedException(
            "Can not delete workflow of other users. [workflow_identifier=%s]"
            % identifier
        )

    if wf_def.is_system:
        msg = "Attempt to delete a system workflow: %s" % identifier
        raise exc.DataAccessException(msg)

    cron_triggers = _get_associated_cron_triggers(identifier)

    if cron_triggers:
        raise exc.DBException(
            "Can't delete workflow that has triggers associated. "
            "[workflow_identifier=%s], [cron_trigger_name(s)=%s]" %
            (identifier, ', '.join(cron_triggers))
        )

    session.delete(wf_def)

from oslo_utils import uuidutils
import sqlalchemy as sa

from mistral.db.v2 import api as db_api
from mistral.services import security
from mistral.tests.unit.api import base

GET_PROJECT_PATH = 'mistral.services.security.get_project_id'

WF_DEFINITION = {
    'name': 'test_wf',
    'definition': 'empty',
    'spec': {},
    'tags': ['mc'],
    'scope': 'private',
    'project_id': security.get_project_id(),
    'trust_id': '1234'
}

WORKFLOW_MEMBER_PENDING = {
    'member_id': '11-22-33',
    'project_id': '<default-project>',
    'resource_type': 'workflow',
    'status': 'pending'
}

WORKFLOW_MEMBER_ACCEPTED = {}

MEMBER_URL = None

def _set_project_id(target, value, oldvalue, initiator):
    return security.get_project_id()