def test_get_roles_context_is_admin_rule_missing(self):
     rules = dict((k, common_policy.parse_rule(v)) for k, v in {
         "some_other_rule": "role:admin",
     }.items())
     common_policy.set_rules(common_policy.Rules(rules))
     # 'admin' role is expected for bw compatibility
     self.assertEqual(['admin'], policy.get_admin_roles())

 def test_get_roles_with_rule_check(self):
     rules = dict((k, common_policy.parse_rule(v)) for k, v in {
         policy.ADMIN_CTX_POLICY: "rule:some_other_rule",
         "some_other_rule": "role:admin",
     }.items())
     common_policy.set_rules(common_policy.Rules(rules))
     self.assertEqual(['admin'], policy.get_admin_roles())

    def __init__(self, user_id, tenant_id, is_admin=None, read_deleted="no",
                 roles=None, timestamp=None, load_admin_roles=True, **kwargs):
        """Object initialization.

        :param read_deleted: 'no' indicates deleted records are hidden, 'yes'
            indicates deleted records are visible, 'only' indicates that
            *only* deleted records are visible.
        """
        if kwargs:
            LOG.warn(_('Arguments dropped when creating '
                       'context: %s'), kwargs)
        super(ContextBase, self).__init__(user=user_id, tenant=tenant_id,
                                          is_admin=is_admin)
        self.read_deleted = read_deleted
        if not timestamp:
            timestamp = datetime.utcnow()
        self.timestamp = timestamp
        self._session = None
        self.roles = roles or []
        if self.is_admin is None:
            self.is_admin = policy.check_is_admin(self)
        elif self.is_admin and load_admin_roles:
            # Ensure context is populated with admin roles
            admin_roles = policy.get_admin_roles()
            if admin_roles:
                self.roles = list(set(self.roles) | set(admin_roles))

    def __init__(self, user_id, tenant_id, is_admin=None, read_deleted="no",
                 roles=None, timestamp=None, **kwargs):
        """Object initialization.

        :param read_deleted: 'no' indicates deleted records are hidden, 'yes'
            indicates deleted records are visible, 'only' indicates that
            *only* deleted records are visible.
        """
        if kwargs:
            LOG.warn(_('Arguments dropped when creating '
                       'context: %s'), kwargs)
        super(ContextBase, self).__init__(user=user_id, tenant=tenant_id,
                                          is_admin=is_admin)
        self.read_deleted = read_deleted
        if not timestamp:
            timestamp = datetime.utcnow()
        self.timestamp = timestamp
        self._session = None
        self.roles = roles or []
        if self.is_admin is None:
            self.is_admin = policy.check_is_admin(self)
        elif self.is_admin:
            # Ensure context is populated with admin roles
            # TODO(salvatore-orlando): It should not be necessary
            # to populate roles in artificially-generated contexts
            # address in bp/make-authz-orthogonal
            admin_roles = policy.get_admin_roles()
            if admin_roles:
                self.roles = list(set(self.roles) | set(admin_roles))

 def test_get_roles_with_or_check(self):
     self.rules = dict((k, common_policy.parse_rule(v)) for k, v in {
         policy.ADMIN_CTX_POLICY: "rule:rule1 or rule:rule2",
         "rule1": "role:admin_1",
         "rule2": "role:admin_2"
     }.items())
     self.assertEqual(['admin_1', 'admin_2'],
                      policy.get_admin_roles())

    def __init__(
        self,
        user_id,
        tenant_id,
        is_admin=None,
        read_deleted="no",
        roles=None,
        timestamp=None,
        load_admin_roles=True,
        request_id=None,
        tenant_name=None,
        user_name=None,
        overwrite=True,
        auth_token=None,
        **kwargs
    ):
        """Object initialization.

        :param read_deleted: 'no' indicates deleted records are hidden, 'yes'
            indicates deleted records are visible, 'only' indicates that
            *only* deleted records are visible.

        :param overwrite: Set to False to ensure that the greenthread local
            copy of the index is not overwritten.

        :param kwargs: Extra arguments that might be present, but we ignore
            because they possibly came in from older rpc messages.
        """
        super(ContextBase, self).__init__(
            auth_token=auth_token, user=user_id, tenant=tenant_id, is_admin=is_admin, request_id=request_id
        )
        self.user_name = user_name
        self.tenant_name = tenant_name

        self.read_deleted = read_deleted
        if not timestamp:
            timestamp = datetime.datetime.utcnow()
        self.timestamp = timestamp
        self._session = None
        self.roles = roles or []
        self.is_advsvc = policy.check_is_advsvc(self)
        if self.is_admin is None:
            self.is_admin = policy.check_is_admin(self)
        elif self.is_admin and load_admin_roles:
            # Ensure context is populated with admin roles
            admin_roles = policy.get_admin_roles()
            if admin_roles:
                self.roles = list(set(self.roles) | set(admin_roles))
        # Allow openstack.common.log to access the context
        if overwrite or not hasattr(local.store, "context"):
            local.store.context = self

        # Log only once the context has been configured to prevent
        # format errors.
        if kwargs:
            LOG.debug(_("Arguments dropped when creating " "context: %s"), kwargs)

    def __init__(self, user_id, tenant_id, is_admin=None, read_deleted="no",
                 roles=None, timestamp=None, load_admin_roles=True,
                 request_id=None, tenant_name=None, user_name=None,
                 overwrite=True, auth_token=None, **kwargs):
        """Object initialization.

        :param read_deleted: 'no' indicates deleted records are hidden, 'yes'
            indicates deleted records are visible, 'only' indicates that
            *only* deleted records are visible.

        :param overwrite: Set to False to ensure that the greenthread local
            copy of the index is not overwritten.

        :param kwargs: Extra arguments that might be present, but we ignore
            because they possibly came in from older rpc messages.
        """
        super(ContextBase, self).__init__(auth_token=auth_token,
                                          user=user_id, tenant=tenant_id,
                                          is_admin=is_admin,
                                          request_id=request_id,
                                          overwrite=overwrite)
        self.user_name = user_name
        self.tenant_name = tenant_name

        self.read_deleted = read_deleted
        if not timestamp:
            timestamp = datetime.datetime.utcnow()
        self.timestamp = timestamp
        self._session = None
        self.roles = roles or []
        self.is_advsvc = policy.check_is_advsvc(self)
        if self.is_admin is None:
            self.is_admin = policy.check_is_admin(self)
        elif self.is_admin and load_admin_roles:
            # Ensure context is populated with admin roles
            admin_roles = policy.get_admin_roles()
            if admin_roles:
                self.roles = list(set(self.roles) | set(admin_roles))

 def test_get_roles_with_other_rules(self):
     self.rules = dict((k, common_policy.parse_rule(v)) for k, v in {
         policy.ADMIN_CTX_POLICY: "role:xxx or other:value",
     }.items())
     self.assertEqual(['xxx'], policy.get_admin_roles())

 def test_get_roles_with_role_check(self):
     rules = dict((k, common_policy.parse_rule(v)) for k, v in {policy.ADMIN_CTX_POLICY: "role:admin"}.items())
     policy.set_rules(common_policy.Rules(rules))
     self.assertEqual(["admin"], policy.get_admin_roles())