openstack
/
keystone
Code Issues Proposed changes

Fix D202: No blank lines after function docstring (PEP257) 

No other PEP257 violations were addressed in this patch.

Change-Id: Ic2e2305d3f8e2c3c4b70c61c0868d310f14355be
This commit is contained in:
Dolph Mathews 2015-10-01 13:42:47 +00:00 committed by Brant Knudson
parent c00797d972
commit 618cb4a894
67 changed files with 3 additions and 231 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
keystone/assignment/controllers.py
View File

@ -505,7 +505,6 @@ class RoleAssignmentV3(controller.V3Controller):
}
"""
formatted_entity = {'links': {}}
inherited_assignment = entity.get('inherited_to_projects')

9
keystone/assignment/core.py
View File

@ -147,7 +147,6 @@ class Manager(manager.Manager):
keystone.exception.DomainNotFound
"""
def _get_group_domain_roles(user_id, domain_id):
role_list = []
group_ids = self._get_group_ids_for_user_id(user_id)
@ -185,7 +184,6 @@ class Manager(manager.Manager):
def get_roles_for_groups(self, group_ids, project_id=None, domain_id=None):
"""Get a list of roles for this group on domain and/or project."""
if project_id is not None:
project = self.resource_api.get_project(project_id)
role_ids = self.list_role_ids_for_groups_on_project(
@ -509,10 +507,8 @@ class Manager(manager.Manager):
filter the result on those values.
"""
def create_group_assignment(base_ref, user_id):
"""Creates a group assignment from the provided ref."""
ref = copy.deepcopy(base_ref)
ref['user_id'] = user_id
@ -682,7 +678,6 @@ class Manager(manager.Manager):
specified, hence avoiding retrieving a huge list.
"""
def list_role_assignments_for_actor(
role_id, inherited, user_id=None,
group_ids=None, project_id=None, domain_id=None):
@ -711,7 +706,6 @@ class Manager(manager.Manager):
response are included.
"""
# List direct project role assignments
project_ids = [project_id] if project_id else None
@ -827,7 +821,6 @@ class Manager(manager.Manager):
inherited roles retrieval, inherited role assignments will be ignored.
"""
if not CONF.os_inherit.enabled:
if inherited:
return []
@ -981,7 +974,6 @@ class AssignmentDriverV8(object):
domain_id=None, project_id=None,
inherited_to_projects=False):
"""Lists role ids for assignments/grants."""
raise exception.NotImplemented() # pragma: no cover
@abc.abstractmethod
@ -1139,7 +1131,6 @@ class AssignmentDriverV8(object):
@abc.abstractmethod
def delete_role_assignments(self, role_id):
"""Deletes all assignments for a role."""
raise exception.NotImplemented() # pragma: no cover
@abc.abstractmethod

2
keystone/auth/controllers.py
View File

@ -93,7 +93,6 @@ class AuthContext(dict):
earliest value.
"""
# identity attributes need to be reconciled among the auth plugins
IDENTITY_ATTRIBUTES = frozenset(['user_id', 'project_id',
'access_token_id', 'domain_id',
@ -479,7 +478,6 @@ class Auth(controller.V3Controller):
def authenticate(self, context, auth_info, auth_context):
"""Authenticate user."""
# The 'external' method allows any 'REMOTE_USER' based authentication
# In some cases the server can set REMOTE_USER as '' instead of
# dropping it, so this must be filtered out

1
keystone/auth/plugins/external.py
View File

@ -78,7 +78,6 @@ class Domain(Base):
The domain will be extracted from the REMOTE_DOMAIN environment
variable if present. If not, the default domain will be used.
"""
username = remote_user
try:
domain_name = context['environment']['REMOTE_DOMAIN']

1
keystone/auth/plugins/mapped.py
View File

@ -56,7 +56,6 @@ class Mapped(auth.AuthMethodHandler):
``OS-FEDERATION:protocol``
"""
if 'id' in auth_payload:
token_ref = self._get_token_ref(auth_payload)
handle_scoped_token(context, auth_payload, auth_context, token_ref,

1
keystone/auth/plugins/oauth1.py
View File

@ -31,7 +31,6 @@ LOG = log.getLogger(__name__)
class OAuth(auth.AuthMethodHandler):
def authenticate(self, context, auth_info, auth_context):
"""Turn a signed request with an access key into a keystone token."""
headers = context['headers']
oauth_headers = oauth.get_oauth_headers(headers)
access_token_id = oauth_headers.get('oauth_token')

1
keystone/catalog/core.py
View File

@ -52,7 +52,6 @@ def format_url(url, substitutions, silent_keyerror_failures=None):
:returns: a formatted URL
"""
substitutions = utils.WhiteListedItemFilter(
WHITELISTED_PROPERTIES,
substitutions)

5
keystone/common/controller.py
View File

@ -167,7 +167,6 @@ def protected(callback=None):
def filterprotected(*filters):
"""Wraps filtered API calls with role based access controls (RBAC)."""
def _filterprotected(f):
@functools.wraps(f)
def wrapper(self, context, **kwargs):
@ -300,7 +299,6 @@ class V2Controller(wsgi.Application):
If ref is a list type, we will iterate through each element and do the
conversion.
"""
def _format_default_project_id(ref):
"""Convert default_project_id to tenantId for v2 calls."""
default_project_id = ref.pop('default_project_id', None)
@ -342,7 +340,6 @@ class V2Controller(wsgi.Application):
If ref is a list type, we will iterate through each element and do the
conversion.
"""
def _filter_project_properties(ref):
"""Run through the various filter methods."""
V2Controller.filter_domain_id(ref)
@ -450,7 +447,6 @@ class V3Controller(wsgi.Application):
True, including the absence of a value
"""
if (isinstance(filter_value, six.string_types) and
filter_value == '0'):
val = False
@ -545,7 +541,6 @@ class V3Controller(wsgi.Application):
@classmethod
def filter_by_attributes(cls, refs, hints):
"""Filters a list of references by filter values."""
def _attr_match(ref_attr, val_attr):
"""Matches attributes allowing for booleans as strings.

1
keystone/common/dependency.py
View File

@ -225,6 +225,5 @@ def reset():
This is useful for unit testing to ensure that tests don't use providers
from previous tests.
"""
_REGISTRY.clear()
_future_dependencies.clear()

2
keystone/common/environment/eventlet_server.py
View File

@ -93,7 +93,6 @@ class Server(service.ServiceBase):
Raises Exception if this has already been called.
"""
# TODO(dims): eventlet's green dns/socket module does not actually
# support IPv6 in getaddrinfo(). We need to get around this in the
# future or monitor upstream for a fix.
@ -121,7 +120,6 @@ class Server(service.ServiceBase):
def start(self, key=None, backlog=128):
"""Run a WSGI server with the given application."""
if self.socket is None:
self.listen(key=key, backlog=backlog)

1
keystone/common/extension.py
View File

@ -41,5 +41,4 @@ def register_admin_extension(url_prefix, extension_data):
def register_public_extension(url_prefix, extension_data):
"""Same as register_admin_extension but for public extensions."""
PUBLIC_EXTENSIONS[url_prefix] = extension_data

1
keystone/common/json_home.py
View File

@ -79,7 +79,6 @@ class Status(object):
def translate_urls(json_home, new_prefix):
"""Given a JSON Home document, sticks new_prefix on each of the urls."""
for dummy_rel, resource in json_home['resources'].items():
if 'href' in resource:
resource['href'] = new_prefix + resource['href']

7
keystone/common/ldap/core.py
View File

@ -110,7 +110,6 @@ def py2ldap(val):
def enabled2py(val):
"""Similar to ldap2py, only useful for the enabled attribute."""
try:
return LDAP_VALUES[val]
except KeyError:
@ -239,7 +238,6 @@ def is_ava_value_equal(attribute_type, val1, val2):
that function apply here.
"""
return prep_case_insensitive(val1) == prep_case_insensitive(val2)
@ -259,7 +257,6 @@ def is_rdn_equal(rdn1, rdn2):
limitations of that function apply here.
"""
if len(rdn1) != len(rdn2):
return False
@ -292,7 +289,6 @@ def is_dn_equal(dn1, dn2):
:param dn2: Either a string DN or a DN parsed by ldap.dn.str2dn.
"""
if not isinstance(dn1, list):
dn1 = ldap.dn.str2dn(utf8_encode(dn1))
if not isinstance(dn2, list):
@ -314,7 +310,6 @@ def dn_startswith(descendant_dn, dn):
:param dn: Either a string DN or a DN parsed by ldap.dn.str2dn.
"""
if not isinstance(descendant_dn, list):
descendant_dn = ldap.dn.str2dn(utf8_encode(descendant_dn))
if not isinstance(dn, list):
@ -815,7 +810,6 @@ class PooledLDAPHandler(LDAPHandler):
which requested msgId and used it in result3 exits.
"""
conn, msg_id = msgid
return conn.result3(msg_id, all, timeout)
@ -858,7 +852,6 @@ class KeystoneLDAPHandler(LDAPHandler):
OpenStack.
"""
def __init__(self, conn=None):
super(KeystoneLDAPHandler, self).__init__(conn=conn)
self.page_size = 0

1
keystone/common/manager.py
View File

@ -121,7 +121,6 @@ def create_legacy_driver(driver_class):
Driver = create_legacy_driver(CatalogDriverV8)
"""
module_name = driver_class.__module__
class_name = driver_class.__name__

1
keystone/common/sql/core.py
View File

@ -68,7 +68,6 @@ flag_modified = flag_modified
def initialize():
"""Initialize the module."""
db_options.set_defaults(
CONF,
connection="sqlite:///keystone.db")

1
keystone/common/sql/migrate_repo/versions/073_insert_assignment_inherited_pk.py
View File

@ -26,7 +26,6 @@ def upgrade(migrate_engine):
assignment table with the new PK constraint and migrates the existing data.
"""
ASSIGNMENT_TABLE_NAME = 'assignment'
metadata = sql.MetaData()

1
keystone/common/sql/migration_helpers.py
View File

@ -117,7 +117,6 @@ def rename_tables_with_constraints(renames, constraints, engine):
`renames` is a dict, mapping {'to_table_name': from_table, ...}
"""
if engine.name != 'sqlite':
# Sqlite doesn't support constraints, so nothing to remove.
remove_constraints(constraints)

1
keystone/common/tokenless_auth.py
View File

@ -45,7 +45,6 @@ class TokenlessAuthHelper(object):
SSL_CLIENT_S_DN_CN, SSL_CLIENT_S_DN_O
:type env: dict
"""
self.env = env
def _build_scope_info(self):

5
keystone/common/utils.py
View File

@ -274,7 +274,6 @@ def get_unix_user(user=None):
:return: tuple of (uid, name)
"""
if isinstance(user, six.string_types):
try:
user_info = pwd.getpwnam(user)
@ -331,7 +330,6 @@ def get_unix_group(group=None):
:return: tuple of (gid, name)
"""
if isinstance(group, six.string_types):
try:
group_info = grp.getgrnam(group)
@ -380,7 +378,6 @@ def set_permissions(path, mode=None, user=None, group=None, log=None):
if None no logging is performed.
"""
if user is None:
user_uid, user_name = None, None
else:
@ -447,7 +444,6 @@ def make_dirs(path, mode=None, user=None, group=None, log=None):
if None no logging is performed.
"""
if log:
if mode is None:
mode_string = str(mode)
@ -483,7 +479,6 @@ _ISO8601_TIME_FORMAT = '%Y-%m-%dT%H:%M:%S'
def isotime(at=None, subsecond=False):
"""Stringify time in ISO 8601 format."""
# Python provides a similar instance method for datetime.datetime objects
# called isoformat(). The format of the strings generated by isoformat()
# have a couple of problems:

6
keystone/common/wsgi.py
View File

@ -115,7 +115,6 @@ def best_match_language(req):
"""Determines the best available locale from the Accept-Language
HTTP header passed in the request.
"""
if not req.accept_language:
return None
return req.accept_language.best_match(
@ -124,7 +123,6 @@ def best_match_language(req):
class BaseApplication(object):
"""Base WSGI application wrapper. Subclasses need to implement __call__."""
@classmethod
def factory(cls, global_config, **local_config):
"""Used for paste app factories in paste.deploy config files.
@ -305,7 +303,6 @@ class Application(BaseApplication):
does not have the admin role
"""
if not context['is_admin']:
user_token_ref = utils.get_token_ref(context)
@ -398,7 +395,6 @@ class Middleware(Application):
behavior.
"""
@classmethod
def factory(cls, global_config, **local_config):
"""Used for paste app factories in paste.deploy config files.
@ -477,7 +473,6 @@ class Debug(Middleware):
about the request and response.
"""
@webob.dec.wsgify()
def __call__(self, req):
if not hasattr(LOG, 'isEnabledFor') or LOG.isEnabledFor(LOG.debug):
@ -789,7 +784,6 @@ def render_response(body=None, status=None, headers=None, method=None):
def render_exception(error, context=None, request=None, user_locale=None):
"""Forms a WSGI response based on the current error."""
error_message = error.args[0]
message = oslo_i18n.translate(error_message, desired_locale=user_locale)
if message is error_message:

1
keystone/config.py
View File

@ -39,7 +39,6 @@ def set_default_for_default_log_levels():
This function needs to be called before CONF().
"""
extra_log_level_defaults = [
'dogpile=INFO',
'routes=INFO',

5
keystone/contrib/ec2/controllers.py
View File

@ -111,7 +111,6 @@ class Ec2ControllerCommon(object):
:returns: user_ref, tenant_ref, metadata_ref, roles_ref, catalog_ref
"""
# FIXME(ja): validate that a service token was used!
# NOTE(termie): backwards compat hack
@ -171,7 +170,6 @@ class Ec2ControllerCommon(object):
:param tenant_id: id of tenant
:returns: credential: dict of ec2 credential
"""
self.identity_api.get_user(user_id)
self.resource_api.get_project(tenant_id)
trust_id = self._get_trust_id_for_request(context)
@ -193,7 +191,6 @@ class Ec2ControllerCommon(object):
:param user_id: id of user
:returns: credentials: list of ec2 credential dicts
"""
self.identity_api.get_user(user_id)
credential_refs = self.credential_api.list_credentials_for_user(
user_id)
@ -210,7 +207,6 @@ class Ec2ControllerCommon(object):
:param credential_id: access key for credentials
:returns: credential: dict of ec2 credential
"""
self.identity_api.get_user(user_id)
return {'credential': self._get_credentials(credential_id)}
@ -223,7 +219,6 @@ class Ec2ControllerCommon(object):
:param credential_id: access key for credentials
:returns: bool: success
"""
self.identity_api.get_user(user_id)
self._get_credentials(credential_id)
ec2_credential_id = utils.hash_access_key(credential_id)

1
keystone/contrib/endpoint_filter/controllers.py
View File

@ -153,7 +153,6 @@ class EndpointGroupV3Controller(_ControllerBase):
@classmethod
def base_url(cls, context, path=None):
"""Construct a path and pass it to V3Controller.base_url method."""
path = '/OS-EP-FILTER/' + cls.collection_name
return super(EndpointGroupV3Controller, cls).base_url(context,
path=path)

4
keystone/contrib/federation/controllers.py
View File

@ -44,7 +44,6 @@ class _ControllerBase(controller.V3Controller):
@classmethod
def base_url(cls, context, path=None):
"""Construct a path and pass it to V3Controller.base_url method."""
path = '/OS-FEDERATION/' + cls.collection_name
return super(_ControllerBase, cls).base_url(context, path=path)
@ -327,7 +326,6 @@ class Auth(auth_controllers.Auth):
def render_html_response(self, host, token_id):
"""Forms an HTML Form from a template with autosubmit."""
headers = [('Content-Type', 'text/html')]
with open(CONF.federation.sso_callback_template) as template:
@ -380,7 +378,6 @@ class Auth(auth_controllers.Auth):
:param auth: Dictionary that contains a token and service provider ID
:returns: SAML Assertion based on properties from the token
"""
t = self._create_base_saml_assertion(context, auth)
(response, service_provider) = t
@ -396,7 +393,6 @@ class Auth(auth_controllers.Auth):
:param auth: Dictionary that contains a token and service provider ID
:returns: ECP Assertion based on properties from the token
"""
t = self._create_base_saml_assertion(context, auth)
(saml_assertion, service_provider) = t
relay_state_prefix = service_provider.get('relay_state_prefix')

4
keystone/contrib/federation/idp.py
View File

@ -195,7 +195,6 @@ class SAMLGenerator(object):
:return: XML <AttributeStatement> object
"""
def _build_attribute(attribute_name, attribute_values):
attribute = saml.Attribute()
attribute.name = attribute_name
@ -550,7 +549,6 @@ class MetadataGenerator(object):
def _ensure_required_values_present(self):
"""Ensure idp_sso_endpoint and idp_entity_id have values."""
if CONF.saml.idp_entity_id is None:
msg = _('Ensure configuration option idp_entity_id is set.')
raise exception.ValidationError(msg)
@ -560,7 +558,6 @@ class MetadataGenerator(object):
def _check_contact_person_values(self):
"""Determine if contact information is included in metadata."""
# Check if we should include contact information
params = [CONF.saml.idp_contact_company,
CONF.saml.idp_contact_name,
@ -582,7 +579,6 @@ class MetadataGenerator(object):
def _check_organization_values(self):
"""Determine if organization information is included in metadata."""
params = [CONF.saml.idp_organization_name,
CONF.saml.idp_organization_display_name,
CONF.saml.idp_organization_url]

8
keystone/contrib/federation/utils.py
View File

@ -216,7 +216,6 @@ def get_remote_id_parameter(protocol):
def validate_idp(idp, protocol, assertion):
"""The IdP providing the assertion should be registered for the mapping."""
remote_id_parameter = get_remote_id_parameter(protocol)
if not remote_id_parameter or not idp['remote_ids']:
LOG.debug('Impossible to identify the IdP %s ', idp['id'])
@ -324,7 +323,6 @@ def transform_to_group_ids(group_names, mapping_id,
exist in the backend.
"""
def resolve_domain(domain):
"""Return domain id.
@ -385,7 +383,6 @@ class RuleProcessor(object):
:type rules: dict
"""
self.rules = rules
def process(self, assertion_data):
@ -438,7 +435,6 @@ class RuleProcessor(object):
}
"""
# Assertions will come in as string key-value pairs, and will use a
# semi-colon to indicate multiple values, i.e. groups.
# This will create a new dictionary where the values are arrays, and
@ -505,7 +501,6 @@ class RuleProcessor(object):
:rtype: dict
"""
def extract_groups(groups_by_domain):
for groups in list(groups_by_domain.values()):
for group in list({g['name']: g for g in groups}.values()):
@ -513,7 +508,6 @@ class RuleProcessor(object):
def normalize_user(user):
"""Parse and validate user mapping."""
user_type = user.get('type')
if user_type and user_type not in (UserType.EPHEMERAL,
@ -601,7 +595,6 @@ class RuleProcessor(object):
{'user': {'name': 'Bob Thompson', 'email': 'bob@example.org'}}
"""
LOG.debug('direct_maps: %s', direct_maps)
LOG.debug('local: %s', local)
new = {}
@ -666,7 +659,6 @@ class RuleProcessor(object):
:rtype: keystone.contrib.federation.utils.DirectMaps or None
"""
direct_maps = DirectMaps()
for requirement in requirements:

1
keystone/contrib/oauth1/controllers.py
View File

@ -50,7 +50,6 @@ class ConsumerCrudV3(controller.V3Controller):
@classmethod
def base_url(cls, context, path=None):
"""Construct a path and pass it to V3Controller.base_url method."""
# NOTE(stevemar): Overriding path to /OS-OAUTH1/consumers so that
# V3Controller.base_url handles setting the self link correctly.
path = '/OS-OAUTH1/' + cls.collection_name

1
keystone/endpoint_policy/core.py
View File

@ -127,7 +127,6 @@ class Manager(manager.Manager):
:returns: list of endpoints that match
"""
if region_id in regions_examined:
msg = _LE('Circular reference or a repeated entry found '
'in region tree - %(region_id)s.')

3
keystone/identity/backends/ldap.py
View File

@ -272,7 +272,6 @@ class UserApi(common_ldap.EnabledEmuMixIn, common_ldap.BaseLdap):
def is_user(self, dn):
"""Returns True if the entry is a user."""
# NOTE(blk-u): It's easy to check if the DN is under the User tree,
# but may not be accurate. A more accurate test would be to fetch the
# entry to see if it's got the user objectclass, but this could be
@ -349,7 +348,6 @@ class GroupApi(common_ldap.BaseLdap):
def list_user_groups(self, user_dn):
"""Return a list of groups for which the user is a member."""
user_dn_esc = ldap.filter.escape_filter_chars(user_dn)
query = '(%s=%s)%s' % (self.member_attribute,
user_dn_esc,
@ -358,7 +356,6 @@ class GroupApi(common_ldap.BaseLdap):
def list_user_groups_filtered(self, user_dn, hints):
"""Return a filtered list of groups for which the user is a member."""
user_dn_esc = ldap.filter.escape_filter_chars(user_dn)
query = '(%s=%s)%s' % (self.member_attribute,
user_dn_esc,

2
keystone/identity/core.py
View File

@ -424,7 +424,6 @@ def domains_configured(f):
def exception_translated(exception_type):
"""Wraps API calls to map to correct exception."""
def _exception_translated(f):
@functools.wraps(f)
def wrapper(self, *args, **kwargs):
@ -485,7 +484,6 @@ class Manager(manager.Manager):
mapping by default is a more prudent way to introduce this functionality.
"""
driver_namespace = 'keystone.identity'
_USER = 'user'

1
keystone/middleware/core.py
View File

@ -259,7 +259,6 @@ class AuthContextMiddleware(wsgi.Middleware):
:type env: dict
:returns: True if client_issuer is trusted; otherwise False
"""
if not CONF.tokenless_auth.trusted_issuer:
return False

10
keystone/notifications.py
View File

@ -326,7 +326,6 @@ def listener(cls):
}
"""
def init_wrapper(init):
@functools.wraps(init)
def __new_init__(self, *args, **kwargs):
@ -424,7 +423,6 @@ def _create_cadf_payload(operation, resource_type, resource_id,
:param outcome: outcomes of the operation (SUCCESS, FAILURE, etc)
:param initiator: CADF representation of the user that created the request
"""
if resource_type not in CADF_TYPE_MAP:
target_uri = taxonomy.UNKNOWN
else:
@ -484,7 +482,6 @@ def _get_request_audit_info(context, user_id=None):
:returns: Auditing data about the request
:rtype: :class:`pycadf.Resource`
"""
remote_addr = None
http_user_agent = None
project_id = None
@ -536,7 +533,6 @@ class CadfNotificationWrapper(object):
def __call__(self, f):
def wrapper(wrapped_self, context, user_id, *args, **kwargs):
"""Always send a notification."""
initiator = _get_request_audit_info(context, user_id)
target = resource.Resource(typeURI=taxonomy.ACCOUNT_USER)
try:
@ -583,9 +579,9 @@ class CadfRoleAssignmentNotificationWrapper(object):
def __call__(self, f):
def wrapper(wrapped_self, role_id, *args, **kwargs):
"""Send a notification if the wrapped callable is successful."""
"""Send a notification if the wrapped callable is successful.
""" NOTE(stevemar): The reason we go through checking kwargs
NOTE(stevemar): The reason we go through checking kwargs
and args for possible target and actor values is because the
create_grant() (and delete_grant()) method are called
differently in various tests.
@ -686,7 +682,6 @@ def send_saml_audit_notification(action, context, user_id, group_ids,
:param outcome: One of :class:`pycadf.cadftaxonomy`
:type outcome: str
"""
initiator = _get_request_audit_info(context)
target = resource.Resource(typeURI=taxonomy.ACCOUNT_USER)
audit_type = SAML_AUDIT_TYPE
@ -718,7 +713,6 @@ def _send_audit_notification(action, initiator, outcome, target,
key-value pairs to the CADF event.
"""
event = eventfactory.EventFactory().new_event(
eventType=cadftype.EVENTTYPE_ACTIVITY,
outcome=outcome,

1
keystone/resource/config_backends/sql.py
View File

@ -143,7 +143,6 @@ class DomainConfig(resource.DomainConfigDriverV8):
def release_registration(self, domain_id, type=None):
"""Silently delete anything registered for the domain specified."""
with sql.transaction() as session:
query = session.query(ConfigRegister)
if type:

4
keystone/resource/core.py
View File

@ -483,7 +483,6 @@ class Manager(manager.Manager):
associated with them as well as revoking any relevant tokens.
"""
def _delete_projects(project, projects, examined):
if project['id'] in examined:
msg = _LE('Circular reference or a repeated entry found '
@ -925,7 +924,6 @@ class DomainConfigManager(manager.Manager):
def _config_to_list(self, config):
"""Build whitelisted and sensitive lists for use by backend drivers."""
whitelisted = []
sensitive = []
for group in config:
@ -1083,7 +1081,6 @@ class DomainConfigManager(manager.Manager):
"""
def _assert_valid_update(domain_id, config, group=None, option=None):
"""Ensure the combination of config, group and option is valid."""
self._assert_valid_config(config)
self._assert_valid_group_and_option(group, option)
@ -1142,7 +1139,6 @@ class DomainConfigManager(manager.Manager):
def _update_or_create(domain_id, option, sensitive):
"""Update the option, if it doesn't exist then create it."""
try:
self.create_config_option(
domain_id, option['group'], option['option'],

1
keystone/tests/hacking/checks.py
View File

@ -283,7 +283,6 @@ class CheckForLoggingIssues(BaseASTChecker):
def visit_Call(self, node):
"""Look for the 'LOG.*' calls."""
# obj.method
if isinstance(node.func, ast.Attribute):
obj_name = self._find_name(node.func.value)

6
keystone/tests/unit/backend/domain_config/core.py
View File

@ -71,7 +71,6 @@ class DomainConfigTests(object):
def _list_domain_config(self, sensitive):
"""Test listing by combination of domain, group & option."""
config1 = {'group': uuid.uuid4().hex, 'option': uuid.uuid4().hex,
'value': uuid.uuid4().hex}
# Put config2 in the same group as config1
@ -113,7 +112,6 @@ class DomainConfigTests(object):
def _delete_domain_configs(self, sensitive):
"""Test deleting by combination of domain, group & option."""
config1 = {'group': uuid.uuid4().hex, 'option': uuid.uuid4().hex,
'value': uuid.uuid4().hex}
# Put config2 and config3 in the same group as config1
@ -162,7 +160,6 @@ class DomainConfigTests(object):
def _create_domain_config_twice(self, sensitive):
"""Test conflict error thrown if create the same option twice."""
config = {'group': uuid.uuid4().hex, 'option': uuid.uuid4().hex,
'value': uuid.uuid4().hex}
@ -182,7 +179,6 @@ class DomainConfigTests(object):
def test_delete_domain_deletes_configs(self):
"""Test domain deletion clears the domain configs."""
domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
self.resource_api.create_domain(domain['id'], domain)
config1 = {'group': uuid.uuid4().hex, 'option': uuid.uuid4().hex,
@ -479,7 +475,6 @@ class DomainConfigTests(object):
def test_invalid_sensitive_substitution_in_domain_config(self):
"""Check that invalid substitutions raise warnings."""
mock_log = mock.Mock()
invalid_option_config = {
@ -504,7 +499,6 @@ class DomainConfigTests(object):
def test_escaped_sequence_in_domain_config(self):
"""Check that escaped '%(' doesn't get interpreted."""
mock_log = mock.Mock()
escaped_option_config = {

1
keystone/tests/unit/common/test_notifications.py
View File

@ -1004,7 +1004,6 @@ class CadfNotificationsWrapperTestCase(test_v3.RestfulTestCase):
'id': 'openstack:782689dd-f428-4f13-99c7-5c70f94a5ac1'
}
"""
note = self._notifications[-1]
event = note['event']
if project:

12
keystone/tests/unit/contrib/federation/test_utils.py
View File

@ -50,7 +50,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
a direct mapping for the users name.
"""
mapping = mapping_fixtures.MAPPING_LARGE
assertion = mapping_fixtures.ADMIN_ASSERTION
rp = mapping_utils.RuleProcessor(mapping['rules'])
@ -74,7 +73,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
RuleProcessor should return list of empty group_ids.
"""
mapping = mapping_fixtures.MAPPING_LARGE
assertion = mapping_fixtures.BAD_TESTER_ASSERTION
rp = mapping_utils.RuleProcessor(mapping['rules'])
@ -93,7 +91,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
a match.
"""
mapping = mapping_fixtures.MAPPING_TESTER_REGEX
assertion = mapping_fixtures.TESTER_ASSERTION
rp = mapping_utils.RuleProcessor(mapping['rules'])
@ -116,7 +113,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
mapping.
"""
mapping = mapping_fixtures.MAPPING_SMALL
assertion = mapping_fixtures.CONTRACTOR_ASSERTION
rp = mapping_utils.RuleProcessor(mapping['rules'])
@ -138,7 +134,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
has `not_any_of`, and direct mapping to a username, no group.
"""
mapping = mapping_fixtures.MAPPING_LARGE
assertion = mapping_fixtures.CUSTOMER_ASSERTION
rp = mapping_utils.RuleProcessor(mapping['rules'])
@ -160,7 +155,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
rules must be matched, including a `not_any_of`.
"""
mapping = mapping_fixtures.MAPPING_SMALL
assertion = mapping_fixtures.EMPLOYEE_ASSERTION
rp = mapping_utils.RuleProcessor(mapping['rules'])
@ -183,7 +177,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
regex set to True.
"""
mapping = mapping_fixtures.MAPPING_DEVELOPER_REGEX
assertion = mapping_fixtures.DEVELOPER_ASSERTION
rp = mapping_utils.RuleProcessor(mapping['rules'])
@ -206,7 +199,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
RuleProcessor should return list of empty group_ids.
"""
mapping = mapping_fixtures.MAPPING_DEVELOPER_REGEX
assertion = mapping_fixtures.BAD_DEVELOPER_ASSERTION
rp = mapping_utils.RuleProcessor(mapping['rules'])
@ -223,7 +215,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
Expect DEVELOPER_GROUP_ID and TESTER_GROUP_ID in the results.
"""
mapping = mapping_fixtures.MAPPING_LARGE
rp = mapping_utils.RuleProcessor(mapping['rules'])
values = rp.process(assertion)
@ -317,7 +308,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
correctly filters out Manager and only allows Developer and Contractor.
"""
mapping = mapping_fixtures.MAPPING_GROUPS_WHITELIST
assertion = mapping_fixtures.EMPLOYEE_ASSERTION_MULTIPLE_GROUPS
rp = mapping_utils.RuleProcessor(mapping['rules'])
@ -354,7 +344,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
correctly filters out Manager and Developer and only allows Contractor.
"""
mapping = mapping_fixtures.MAPPING_GROUPS_BLACKLIST
assertion = mapping_fixtures.EMPLOYEE_ASSERTION_MULTIPLE_GROUPS
rp = mapping_utils.RuleProcessor(mapping['rules'])
@ -383,7 +372,6 @@ class MappingRuleEngineTests(unit.BaseTestCase):
entry in the remote rules.
"""
mapping = mapping_fixtures.MAPPING_GROUPS_BLACKLIST_MULTIPLES
assertion = mapping_fixtures.EMPLOYEE_ASSERTION_MULTIPLE_GROUPS
rp = mapping_utils.RuleProcessor(mapping['rules'])

1
keystone/tests/unit/core.py
View File

@ -515,7 +515,6 @@ class TestCase(BaseTestCase):
def load_backends(self):
"""Initializes each manager and assigns them to an attribute."""
# TODO(blk-u): Shouldn't need to clear the registry here, but some
# tests call load_backends multiple times. These should be fixed to
# only call load_backends once.

1
keystone/tests/unit/fakeldap.py
View File

@ -132,7 +132,6 @@ def _paren_groups(source):
def _match(key, value, attrs):
"""Match a given key and value against an attribute list."""
def match_with_wildcards(norm_val, val_list):
# Case insensitive checking with wildcards
if norm_val.startswith('*'):

3
keystone/tests/unit/filtering.py
View File

@ -49,7 +49,6 @@ class FilterTests(object):
one.
"""
f = getattr(self.identity_api, 'create_%s' % entity_type, None)
if f is None:
f = getattr(self.resource_api, 'create_%s' % entity_type, None)
@ -65,7 +64,6 @@ class FilterTests(object):
one.
"""
f = getattr(self.identity_api, 'delete_%s' % entity_type, None)
if f is None:
f = getattr(self.resource_api, 'delete_%s' % entity_type, None)
@ -81,7 +79,6 @@ class FilterTests(object):
one.
"""
f = getattr(self.identity_api, 'list_%ss' % entity_type, None)
if f is None:
f = getattr(self.resource_api, 'list_%ss' % entity_type, None)

1
keystone/tests/unit/identity_mapping.py
View File

@ -17,7 +17,6 @@ from keystone.identity.mapping_backends import sql as mapping_sql
def list_id_mappings():
"""List all id_mappings for testing purposes."""
a_session = sql.get_session()
refs = a_session.query(mapping_sql.IDMapping).all()
return [x.to_dict() for x in refs]

15
keystone/tests/unit/test_backend.py
View File

@ -197,7 +197,6 @@ class AssignmentTestHelperMixin(object):
def _create_entity_in_domain(entity_type, domain_id):
"""Create a user or group entity in the domain."""
new_entity = {'name': uuid.uuid4().hex, 'domain_id': domain_id}
if entity_type == 'users':
new_entity = self.identity_api.create_user(new_entity)
@ -288,7 +287,6 @@ class AssignmentTestHelperMixin(object):
def create_group_memberships(self, group_pattern, test_data):
"""Create the group memberships specified in the test plan."""
for group_spec in group_pattern:
# Each membership specification is a dict of the form:
#
@ -304,7 +302,6 @@ class AssignmentTestHelperMixin(object):
def create_assignments(self, assignment_pattern, test_data):
"""Create the assignments specified in the test plan."""
# First store how many assignments are already in the system,
# so during the tests we can check the number of new assignments
# created.
@ -337,7 +334,6 @@ class AssignmentTestHelperMixin(object):
def execute_assignment_tests(self, test_plan, test_data):
"""Execute the test plan, based on the created test_data."""
def check_results(expected, actual, param_arg_count):
if param_arg_count == 0:
# It was an unfiltered call, so default fixture assignments
@ -885,7 +881,6 @@ class IdentityTests(AssignmentTestHelperMixin):
def test_list_role_assignments_unfiltered(self):
"""Test unfiltered listing of role assignments."""
test_plan = {
# Create a domain, with a user, group & project
'entities': {'domains': {'users': 1, 'groups': 1, 'projects': 1},
@ -908,7 +903,6 @@ class IdentityTests(AssignmentTestHelperMixin):
def test_list_role_assignments_filtered_by_role(self):
"""Test listing of role assignments filtered by role ID."""
test_plan = {
# Create a user, group & project in the default domain
'entities': {'domains': {'id': DEFAULT_DOMAIN_ID,
@ -1647,7 +1641,6 @@ class IdentityTests(AssignmentTestHelperMixin):
def test_grant_crud_throws_exception_if_invalid_role(self):
"""Ensure RoleNotFound thrown if role does not exist."""
def assert_role_not_found_exception(f, **kwargs):
self.assertRaises(exception.RoleNotFound, f,
role_id=uuid.uuid4().hex, **kwargs)
@ -1990,7 +1983,6 @@ class IdentityTests(AssignmentTestHelperMixin):
def test_list_role_assignment_by_domain(self):
"""Test listing of role assignment filtered by domain."""
test_plan = {
# A domain with 3 users, 1 group, a spoiler domain and 2 roles.
'entities': {'domains': [{'users': 3, 'groups': 1}, 1],
@ -2020,7 +2012,6 @@ class IdentityTests(AssignmentTestHelperMixin):
def test_list_role_assignment_by_user_with_domain_group_roles(self):
"""Test listing assignments by user, with group roles on a domain."""
test_plan = {
# A domain with 3 users, 3 groups, a spoiler domain
# plus 3 roles.
@ -5706,7 +5697,6 @@ class CatalogTests(object):
def test_get_catalog_endpoint_disabled(self):
"""Get back only enabled endpoints when get the v2 catalog."""
service_ref, enabled_endpoint_ref, dummy_disabled_endpoint_ref = (
self._create_endpoints())
@ -5725,7 +5715,6 @@ class CatalogTests(object):
def test_get_v3_catalog_endpoint_disabled(self):
"""Get back only enabled endpoints when get the v3 catalog."""
enabled_endpoint_ref = self._create_endpoints()[1]
user_id = uuid.uuid4().hex
@ -5927,7 +5916,6 @@ class InheritanceTests(AssignmentTestHelperMixin):
('project_id' or 'domain_id'), respectively.
"""
# Create a new role to avoid assignments loaded from default fixtures
role = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
role = self.role_api.create_role(role['id'], role)
@ -6700,7 +6688,6 @@ class FilterTests(filtering.FilterTests):
name, both restrictions have been enforced on what is returned.
"""
number_of_groups = 10
group_name_data = {
# entity index: name for entity
@ -6800,7 +6787,6 @@ class LimitTests(filtering.FilterTests):
def setUp(self):
"""Setup for Limit Test Cases."""
self.domain1 = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
self.resource_api.create_domain(self.domain1['id'], self.domain1)
self.addCleanup(self.clean_up_domain)
@ -6817,7 +6803,6 @@ class LimitTests(filtering.FilterTests):
def clean_up_domain(self):
"""Clean up domain test data from Limit Test Cases."""
self.domain1['enabled'] = False
self.resource_api.update_domain(self.domain1['id'], self.domain1)
self.resource_api.delete_domain(self.domain1['id'])

1
keystone/tests/unit/test_backend_endpoint_policy.py
View File

@ -51,7 +51,6 @@ class PolicyAssociationTests(object):
5 - region 2, Service 0
"""
def new_endpoint(region_id, service_id):
endpoint = {'id': uuid.uuid4().hex, 'interface': 'test',
'region_id': region_id, 'service_id': service_id,

9
keystone/tests/unit/test_backend_ldap.py
View File

@ -582,7 +582,6 @@ class BaseLDAPIdentity(test_backend.IdentityTests):
are returned.
"""
# Create a group
group = dict(name=uuid.uuid4().hex,
domain_id=CONF.identity.default_domain_id)
@ -818,7 +817,6 @@ class BaseLDAPIdentity(test_backend.IdentityTests):
def test_user_id_comma(self):
"""Even if the user has a , in their ID, groups can be listed."""
# Create a user with a , in their ID
# NOTE(blk-u): the DN for this user is hard-coded in fakeldap!
@ -871,7 +869,6 @@ class BaseLDAPIdentity(test_backend.IdentityTests):
def test_user_id_comma_grants(self):
"""List user and group grants, even with a comma in the user's ID."""
# Create a user with a , in their ID
# NOTE(blk-u): the DN for this user is hard-coded in fakeldap!
@ -2483,7 +2480,6 @@ class BaseMultiLDAPandSQLIdentity(object):
def create_users_across_domains(self):
"""Create a set of users, each with a role on their own domain."""
# We also will check that the right number of id mappings get created
initial_mappings = len(mapping_sql.list_id_mappings())
@ -2952,7 +2948,6 @@ class MultiLDAPandSQLIdentityDomainConfigsInSQL(MultiLDAPandSQLIdentity):
def test_reloading_domain_config(self):
"""Ensure domain drivers are reloaded on a config modification."""
domain_cfgs = self.identity_api.domain_configs
# Create a new config for the default domain, hence overwriting the
@ -2984,7 +2979,6 @@ class MultiLDAPandSQLIdentityDomainConfigsInSQL(MultiLDAPandSQLIdentity):
def test_setting_multiple_sql_driver_raises_exception(self):
"""Ensure setting multiple domain specific sql drivers is prevented."""
new_config = {'identity': {'driver': 'sql'}}
self.domain_config_api.create_config(
CONF.identity.default_domain_id, new_config)
@ -2998,7 +2992,6 @@ class MultiLDAPandSQLIdentityDomainConfigsInSQL(MultiLDAPandSQLIdentity):
def test_same_domain_gets_sql_driver(self):
"""Ensure we can set an SQL driver if we have had it before."""
new_config = {'identity': {'driver': 'sql'}}
self.domain_config_api.create_config(
CONF.identity.default_domain_id, new_config)
@ -3016,7 +3009,6 @@ class MultiLDAPandSQLIdentityDomainConfigsInSQL(MultiLDAPandSQLIdentity):
def test_delete_domain_clears_sql_registration(self):
"""Ensure registration is deleted when a domain is deleted."""
domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
domain = self.resource_api.create_domain(domain['id'], domain)
new_config = {'identity': {'driver': 'sql'}}
@ -3044,7 +3036,6 @@ class MultiLDAPandSQLIdentityDomainConfigsInSQL(MultiLDAPandSQLIdentity):
def test_orphaned_registration_does_not_prevent_getting_sql_driver(self):
"""Ensure we self heal an orphaned sql registration."""
domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
domain = self.resource_api.create_domain(domain['id'], domain)
new_config = {'identity': {'driver': 'sql'}}

1
keystone/tests/unit/test_backend_sql.py
View File

@ -748,7 +748,6 @@ class SqlFilterTests(SqlTests, test_backend.FilterTests):
def clean_up_entities(self):
"""Clean up entity test data from Filter Test Cases."""
for entity in ['user', 'group', 'project']:
self._delete_test_data(entity, self.entity_list[entity])
self._delete_test_data(entity, self.domain1_entity_list[entity])

1
keystone/tests/unit/test_revoke.py
View File

@ -67,7 +67,6 @@ def _matches(event, token_values):
:returns if the token matches the revocation event, indicating the
token has been revoked
"""
# The token has three attributes that can match the user_id
if event.user_id is not None:
for attribute_name in ['user_id', 'trustor_id', 'trustee_id']:

3
keystone/tests/unit/test_sql_upgrade.py
View File

@ -502,7 +502,6 @@ class SqlUpgradeTests(SqlMigrateBase):
def does_pk_exist(self, table, pk_column):
"""Checks whether a column is primary key on a table."""
inspector = reflection.Inspector.from_engine(self.engine)
pk_columns = inspector.get_pk_constraint(table)['constrained_columns']
@ -856,7 +855,6 @@ class VersionTests(SqlMigrateBase):
def test_unexpected_extension(self):
"""The version for a non-existent extension raises ImportError."""
extension_name = uuid.uuid4().hex
self.assertRaises(ImportError,
migration_helpers.get_db_version,
@ -864,7 +862,6 @@ class VersionTests(SqlMigrateBase):
def test_unversioned_extension(self):
"""The version for extensions without migrations raise an exception."""
self.assertRaises(exception.MigrationNotProvided,
migration_helpers.get_db_version,
extension='admin_crud')

2
keystone/tests/unit/test_v2.py
View File

@ -1245,7 +1245,6 @@ class V2TestCase(RestfulTestCase, CoreApiTests, LegacyV2UsernameTests):
"""If the server is configured for md5, then the revocation list has
tokens hashed with MD5.
"""
# The default hash algorithm is md5.
hash_algorithm = 'md5'
@ -1257,7 +1256,6 @@ class V2TestCase(RestfulTestCase, CoreApiTests, LegacyV2UsernameTests):
"""If the server is configured for sha256, then the revocation list has
tokens hashed with SHA256
"""
hash_algorithm = 'sha256'
self.config_fixture.config(group='token',
hash_algorithm=hash_algorithm)

1
keystone/tests/unit/test_v2_controller.py
View File

@ -74,7 +74,6 @@ class TenantTestCase(unit.TestCase):
def test_list_projects_default_domain(self):
"""Test that list projects only returns those in the default domain."""
domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
'enabled': True}
self.resource_api.create_domain(domain['id'], domain)

4
keystone/tests/unit/test_v3.py
View File

@ -406,7 +406,6 @@ class RestfulTestCase(unit.SQLDriverOverrides, rest.RestfulTestCase,
def get_requested_token(self, auth):
"""Request the specific token we want."""
r = self.v3_create_token(auth)
return r.headers.get('X-Subject-Token')
@ -1307,7 +1306,6 @@ class AssignmentTestMixin(object):
Available filters are: domain_id, project_id, user_id, group_id,
role_id and inherited_to_projects.
"""
query_params = '?effective' if effective else ''
for k, v in filters.items():
@ -1332,7 +1330,6 @@ class AssignmentTestMixin(object):
Provided attributes are expected to contain: domain_id or project_id,
user_id or group_id, role_id and, optionally, inherited_to_projects.
"""
if attribs.get('domain_id'):
link = '/domains/' + attribs['domain_id']
else:
@ -1356,7 +1353,6 @@ class AssignmentTestMixin(object):
Provided attributes are expected to contain: domain_id or project_id,
user_id or group_id, role_id and, optionally, inherited_to_projects.
"""
entity = {'links': {'assignment': (
link or self.build_role_assignment_link(**attribs))}}

18
keystone/tests/unit/test_v3_assignment.py
View File

@ -184,7 +184,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
def test_delete_enabled_domain_fails(self):
"""Call ``DELETE /domains/{domain_id}`` (when domain enabled)."""
# Try deleting an enabled domain, which should fail
self.delete('/domains/%(domain_id)s' % {
'domain_id': self.domain['id']},
@ -210,7 +209,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
- Check entities in self.domain are unaffected
"""
# Create a 2nd set of entities in a 2nd domain
self.domain2 = self.new_domain_ref()
self.resource_api.create_domain(self.domain2['id'], self.domain2)
@ -341,7 +339,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
becomes invalid once that domain is disabled.
"""
self.domain = self.new_domain_ref()
self.resource_api.create_domain(self.domain['id'], self.domain)
@ -457,7 +454,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
This includes operations like create, update, delete.
"""
non_default_name = 'beta_federated_domain'
self.config_fixture.config(group='federation',
federated_domain_name=non_default_name)
@ -516,7 +512,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
@utils.wip('waiting for projects acting as domains implementation')
def test_create_project_without_parent_id_and_without_domain_id(self):
"""Call ``POST /projects``."""
# Grant a domain role for the user
collection_url = (
'/domains/%(domain_id)s/users/%(user_id)s/roles' % {
@ -716,7 +711,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
'project' and 'parent'.
"""
# Create the project hierarchy
parent, project, subproject = self._create_projects_hierarchy(2)
@ -747,7 +741,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
'parent'.
"""
# Create the project hierarchy
parent, project, subproject = self._create_projects_hierarchy(2)
@ -883,7 +876,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
and 'subproject'.
"""
# Create the project hierarchy
parent, project, subproject = self._create_projects_hierarchy(2)
@ -913,7 +905,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
- Check that calling subtree_as_list on 'parent' returns 'subproject'.
"""
# Create the project hierarchy
parent, project, subproject = self._create_projects_hierarchy(2)
@ -1153,7 +1144,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
returns Not Found for the user.
"""
user_id = uuid.uuid4().hex
collection_url = (
@ -1192,7 +1182,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
returns 404 Not Found for the user.
"""
user_id = uuid.uuid4().hex
collection_url = (
@ -1231,7 +1220,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
server returns 404 Not Found for the group.
"""
group_id = uuid.uuid4().hex
collection_url = (
@ -1271,7 +1259,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
returns 404 Not Found for the group.
"""
group_id = uuid.uuid4().hex
collection_url = (
@ -1392,7 +1379,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
been removed
"""
# Since the default fixtures already assign some roles to the
# user it creates, we also need a new user that will not have any
# existing assignments
@ -1641,7 +1627,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase,
token (all effective roles for a user on a project)
"""
# Since the default fixtures already assign some roles to the
# user it creates, we also need a new user that will not have any
# existing assignments
@ -1888,7 +1873,6 @@ class RoleAssignmentBaseTestCase(test_v3.RestfulTestCase,
queried URL.
"""
query_url = self._get_role_assignments_query_url(**filters)
response = self.get(query_url, expected_status=expected_status)
@ -1967,7 +1951,6 @@ class RoleAssignmentDirectTestCase(RoleAssignmentBaseTestCase):
group_id, user_id and inherited_to_projects.
"""
# Fills default assignment with provided filters
test_assignment = self._set_default_assignment_attributes(**filters)
@ -2511,7 +2494,6 @@ class AssignmentInheritanceTestCase(test_v3.RestfulTestCase,
shows up.
"""
role_list = []
for _ in range(4):
role = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}

6
keystone/tests/unit/test_v3_auth.py
View File

@ -981,7 +981,6 @@ class TestTokenRevokeById(test_v3.RestfulTestCase):
the one for Project C
"""
self.role_data_fixtures()
# Now we are ready to start issuing requests
@ -1281,7 +1280,6 @@ class TestTokenRevokeById(test_v3.RestfulTestCase):
def test_removing_role_assignment_does_not_affect_other_users(self):
"""Revoking a role from one user should not affect other users."""
# This group grant is not needed for the test
self.delete(
'/projects/%(project_id)s/groups/%(group_id)s/roles/%(role_id)s' %
@ -1990,7 +1988,6 @@ class TestAuth(test_v3.RestfulTestCase):
def test_auth_catalog_disabled_endpoint(self):
"""On authenticate, get a catalog that excludes disabled endpoints."""
# Create a disabled endpoint that's like the enabled one.
disabled_endpoint_ref = copy.copy(self.endpoint)
disabled_endpoint_id = uuid.uuid4().hex
@ -2044,7 +2041,6 @@ class TestAuth(test_v3.RestfulTestCase):
tokens
"""
domainA = self.new_domain_ref()
self.resource_api.create_domain(domainA['id'], domainA)
projectA = self.new_project_ref(domain_id=domainA['id'])
@ -4507,7 +4503,6 @@ class TestFernetTokenProvider(test_v3.RestfulTestCase):
Test that validating an domain scoped token in v2.0
returns unauthorized.
"""
# Grant user access to domain
self.assignment_api.create_grant(self.role['id'],
user_id=self.user['id'],
@ -4524,7 +4519,6 @@ class TestFernetTokenProvider(test_v3.RestfulTestCase):
Test that validating an trust scoped token in v2.0 returns
unauthorized.
"""
trustee_user, trust = self._create_trust()
trust_scoped_token = self._get_trust_scoped_token(trustee_user, trust)
self.assertRaises(exception.Unauthorized,

1
keystone/tests/unit/test_v3_catalog.py
View File

@ -238,7 +238,6 @@ class CatalogTestCase(test_v3.RestfulTestCase):
def test_delete_region(self):
"""Call ``DELETE /regions/{region_id}``."""
ref = self.new_region_ref()
r = self.post(
'/regions',

5
keystone/tests/unit/test_v3_endpoint_policy.py
View File

@ -69,7 +69,6 @@ class EndpointPolicyTestCase(test_v3.RestfulTestCase):
def test_crud_for_policy_for_explicit_endpoint(self):
"""PUT, HEAD and DELETE for explicit endpoint policy."""
url = ('/policies/%(policy_id)s/OS-ENDPOINT-POLICY'
'/endpoints/%(endpoint_id)s') % {
'policy_id': self.policy['id'],
@ -78,7 +77,6 @@ class EndpointPolicyTestCase(test_v3.RestfulTestCase):
def test_crud_for_policy_for_service(self):
"""PUT, HEAD and DELETE for service endpoint policy."""
url = ('/policies/%(policy_id)s/OS-ENDPOINT-POLICY'
'/services/%(service_id)s') % {
'policy_id': self.policy['id'],
@ -87,7 +85,6 @@ class EndpointPolicyTestCase(test_v3.RestfulTestCase):
def test_crud_for_policy_for_region_and_service(self):
"""PUT, HEAD and DELETE for region and service endpoint policy."""
url = ('/policies/%(policy_id)s/OS-ENDPOINT-POLICY'
'/services/%(service_id)s/regions/%(region_id)s') % {
'policy_id': self.policy['id'],
@ -97,7 +94,6 @@ class EndpointPolicyTestCase(test_v3.RestfulTestCase):
def test_get_policy_for_endpoint(self):
"""GET /endpoints/{endpoint_id}/policy."""
self.put('/policies/%(policy_id)s/OS-ENDPOINT-POLICY'
'/endpoints/%(endpoint_id)s' % {
'policy_id': self.policy['id'],
@ -115,7 +111,6 @@ class EndpointPolicyTestCase(test_v3.RestfulTestCase):
def test_list_endpoints_for_policy(self):
"""GET /policies/%(policy_id}/endpoints."""
self.put('/policies/%(policy_id)s/OS-ENDPOINT-POLICY'
'/endpoints/%(endpoint_id)s' % {
'policy_id': self.policy['id'],

23
keystone/tests/unit/test_v3_federation.py
View File

@ -137,7 +137,6 @@ class FederatedSetupMixin(object):
def assertValidMappedUser(self, token):
"""Check if user object meets all the criteria."""
user = token['user']
self.assertIn('id', user)
self.assertIn('name', user)
@ -209,7 +208,6 @@ class FederatedSetupMixin(object):
def load_federation_sample_data(self):
"""Inject additional data."""
# Create and add domains
self.domainA = self.new_domain_ref()
self.resource_api.create_domain(self.domainA['id'],
@ -856,7 +854,6 @@ class FederatedIdentityProviderTests(FederationTests):
def test_create_idp(self):
"""Creates the IdentityProvider entity associated to remote_ids."""
keys_to_check = list(self.idp_keys)
body = self.default_body.copy()
body['description'] = uuid.uuid4().hex
@ -867,7 +864,6 @@ class FederatedIdentityProviderTests(FederationTests):
def test_create_idp_remote(self):
"""Creates the IdentityProvider entity associated to remote_ids."""
keys_to_check = list(self.idp_keys)
keys_to_check.append('remote_ids')
body = self.default_body.copy()
@ -889,7 +885,6 @@ class FederatedIdentityProviderTests(FederationTests):
Expect HTTP 409 Conflict code for the latter call.
"""
body = self.default_body.copy()
repeated_remote_id = uuid.uuid4().hex
body['remote_ids'] = [uuid.uuid4().hex,
@ -906,7 +901,6 @@ class FederatedIdentityProviderTests(FederationTests):
def test_create_idp_remote_empty(self):
"""Creates an IdP with empty remote_ids."""
keys_to_check = list(self.idp_keys)
keys_to_check.append('remote_ids')
body = self.default_body.copy()
@ -919,7 +913,6 @@ class FederatedIdentityProviderTests(FederationTests):
def test_create_idp_remote_none(self):
"""Creates an IdP with a None remote_ids."""
keys_to_check = list(self.idp_keys)
keys_to_check.append('remote_ids')
body = self.default_body.copy()
@ -1072,7 +1065,6 @@ class FederatedIdentityProviderTests(FederationTests):
def test_delete_idp_also_deletes_assigned_protocols(self):
"""Deleting an IdP will delete its assigned protocol."""
# create default IdP
default_resp = self._create_default_idp()
default_idp = self._fetch_attribute_from_response(default_resp,
@ -1178,7 +1170,6 @@ class FederatedIdentityProviderTests(FederationTests):
def test_assign_protocol_to_idp(self):
"""Assign a protocol to existing IdP."""
self._assign_protocol_to_idp(expected_status=http_client.CREATED)
def test_protocol_composite_pk(self):
@ -1224,7 +1215,6 @@ class FederatedIdentityProviderTests(FederationTests):
Expect HTTP 404 Not Found code.
"""
idp_id = uuid.uuid4().hex
kwargs = {'expected_status': http_client.NOT_FOUND}
self._assign_protocol_to_idp(proto='saml2',
@ -1234,7 +1224,6 @@ class FederatedIdentityProviderTests(FederationTests):
def test_get_protocol(self):
"""Create and later fetch protocol tied to IdP."""
resp, idp_id, proto = self._assign_protocol_to_idp(
expected_status=http_client.CREATED)
proto_id = self._fetch_attribute_from_response(resp, 'protocol')['id']
@ -1280,7 +1269,6 @@ class FederatedIdentityProviderTests(FederationTests):
def test_update_protocols_attribute(self):
"""Update protocol's attribute."""
resp, idp_id, proto = self._assign_protocol_to_idp(
expected_status=http_client.CREATED)
new_mapping_id = uuid.uuid4().hex
@ -1697,7 +1685,6 @@ class FederatedTokenTests(FederationTests, FederatedSetupMixin):
def test_scope_to_bad_project(self):
"""Scope unscoped token with a project we don't have access to."""
self.v3_create_token(
self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_CUSTOMER,
expected_status=http_client.UNAUTHORIZED)
@ -1711,7 +1698,6 @@ class FederatedTokenTests(FederationTests, FederatedSetupMixin):
* Employees' project
"""
bodies = (self.TOKEN_SCOPE_PROJECT_EMPLOYEE_FROM_ADMIN,
self.TOKEN_SCOPE_PROJECT_CUSTOMER_FROM_ADMIN)
project_ids = (self.proj_employees['id'],
@ -1892,7 +1878,6 @@ class FederatedTokenTests(FederationTests, FederatedSetupMixin):
* Scope token to one of available projects
"""
r = self._issue_unscoped_token()
token_resp = r.json_body['token']
self.assertValidMappedUser(token_resp)
@ -2054,7 +2039,6 @@ class FederatedTokenTests(FederationTests, FederatedSetupMixin):
assigned
"""
domain_id = self.domainA['id']
domain_name = self.domainA['name']
@ -2129,7 +2113,6 @@ class FederatedTokenTests(FederationTests, FederatedSetupMixin):
assigned
"""
domain_id = self.domainA['id']
domain_name = self.domainA['name']
@ -2799,7 +2782,6 @@ class SAMLGenerationTests(FederationTests):
Raises exception.SchemaValidationError() - error 400 Bad Request
"""
token_id = uuid.uuid4().hex
body = self._create_generate_saml_request(token_id,
self.SERVICE_PROVDIER_ID)
@ -2814,7 +2796,6 @@ class SAMLGenerationTests(FederationTests):
Raises exception.SchemaValidationError() - error 400 Bad Request
"""
token_id = uuid.uuid4().hex
body = self._create_generate_saml_request(token_id,
self.SERVICE_PROVDIER_ID)
@ -2837,7 +2818,6 @@ class SAMLGenerationTests(FederationTests):
def test_sp_disabled(self):
"""Try generating assertion for disabled Service Provider."""
# Disable Service Provider
sp_ref = {'enabled': False}
self.federation_api.update_sp(self.SERVICE_PROVDIER_ID, sp_ref)
@ -2854,7 +2834,6 @@ class SAMLGenerationTests(FederationTests):
Raises exception.TokenNotFound() - error Not Found 404
"""
token_id = uuid.uuid4().hex
body = self._create_generate_saml_request(token_id,
self.SERVICE_PROVDIER_ID)
@ -2870,7 +2849,6 @@ class SAMLGenerationTests(FederationTests):
The controller should return a SAML assertion that is wrapped in a
SOAP envelope.
"""
self.config_fixture.config(group='saml', idp_entity_id=self.ISSUER)
token_id = self._fetch_valid_token()
body = self._create_generate_saml_request(token_id,
@ -3439,7 +3417,6 @@ class K2KServiceCatalogTests(FederationTests):
def test_service_providers_in_token(self):
"""Check if service providers are listed in service catalog."""
token = self.token_v3_helper.get_token_data(self.user_id, ['password'])
ref = {}
for r in (self.sp_alpha, self.sp_beta, self.sp_gamma):

7
keystone/tests/unit/test_v3_filters.py
View File

@ -33,7 +33,6 @@ class IdentityTestFilteredCase(filtering.FilterTests,
def setUp(self):
"""Setup for Identity Filter Test Cases."""
super(IdentityTestFilteredCase, self).setUp()
self.tempfile = self.useFixture(temporaryfile.SecureTempFile())
self.tmpfilename = self.tempfile.file_name
@ -329,7 +328,6 @@ class IdentityTestListLimitCase(IdentityTestFilteredCase):
def setUp(self):
"""Setup for Identity Limit Test Cases."""
super(IdentityTestListLimitCase, self).setUp()
# Create 10 entries for each of the entities we are going to test
@ -359,18 +357,15 @@ class IdentityTestListLimitCase(IdentityTestFilteredCase):
def clean_up_entity(self, entity):
"""Clean up entity test data from Identity Limit Test Cases."""
self._delete_test_data(entity, self.entity_lists[entity])
def clean_up_service(self):
"""Clean up service test data from Identity Limit Test Cases."""
for service in self.service_list:
self.catalog_api.delete_service(service['id'])
def clean_up_policy(self):
"""Clean up policy test data from Identity Limit Test Cases."""
for policy in self.policy_list:
self.policy_api.delete_policy(policy['id'])
@ -430,7 +425,6 @@ class IdentityTestListLimitCase(IdentityTestFilteredCase):
def test_no_limit(self):
"""Check truncated attribute not set when list not limited."""
self._set_policy({"identity:list_services": []})
r = self.get('/services', auth=self.auth)
self.assertEqual(10, len(r.result.get('services')))
@ -438,7 +432,6 @@ class IdentityTestListLimitCase(IdentityTestFilteredCase):
def test_at_limit(self):
"""Check truncated attribute not set when list at max size."""
# Test this by overriding the general limit with a higher
# driver-specific limit (allowing all entities to be returned
# in the collection), which should result in a non truncated list

1
keystone/tests/unit/test_v3_identity.py
View File

@ -209,7 +209,6 @@ class IdentityTestCase(test_v3.RestfulTestCase):
def test_list_groups_for_user(self):
"""Call ``GET /users/{user_id}/groups``."""
self.user1 = self.new_user_ref(
domain_id=self.domain['id'])
password = self.user1['password']

1
keystone/tests/unit/test_v3_oauth1.py
View File

@ -801,7 +801,6 @@ class OAuthNotificationTests(OAuth1Tests,
notifications for request token creation, and access token
creation/deletion are emitted.
"""
consumer = self._create_single_consumer()
consumer_id = consumer['id']
consumer_secret = consumer['secret']

1
keystone/tests/unit/utils.py
View File

@ -72,7 +72,6 @@ def wip(message):
>>> pass
"""
def _wip(f):
@six.wraps(f)
def run_test(*args, **kwargs):

2
keystone/token/controllers.py
View File

@ -82,7 +82,6 @@ class Auth(controller.V2Controller):
Alternatively, this call accepts auth with only a token and tenant
that will return a token that is scoped to that tenant.
"""
if auth is None:
raise exception.ValidationError(attribute='auth',
target='request body')
@ -379,7 +378,6 @@ class Auth(controller.V2Controller):
def _get_project_roles_and_ref(self, user_id, tenant_id):
"""Returns the project roles for this user, and the project ref."""
tenant_ref = None
role_list = []
if tenant_id:

3
keystone/token/persistence/backends/sql.py
View File

@ -53,7 +53,6 @@ def _expiry_range_batched(session, upper_bound_func, batch_size):
Return the timestamp of the next token that is `batch_size` rows from
being the oldest expired token.
"""
# This expiry strategy splits the tokens into roughly equal sized batches
# to be deleted. It does this by finding the timestamp of a token
# `batch_size` rows from the oldest token and yielding that to the caller.
@ -79,7 +78,6 @@ def _expiry_range_batched(session, upper_bound_func, batch_size):
def _expiry_range_all(session, upper_bound_func):
"""Expires all tokens in one pass."""
yield upper_bound_func()
@ -245,7 +243,6 @@ class Token(token.persistence.TokenDriverV8):
Based on the DB dialect, select an expiry range callable that is
appropriate.
"""
# DB2 and MySQL can both benefit from a batched strategy. On DB2 the
# transaction log can fill up and on MySQL w/Galera, large
# transactions can exceed the maximum write set size.

1
keystone/token/persistence/core.py
View File

@ -212,7 +212,6 @@ class Manager(object):
def __getattr__(self, item):
"""Forward calls to the `token_provider_api` persistence manager."""
# NOTE(morganfainberg): Prevent infinite recursion, raise an
# AttributeError for 'token_provider_api' ensuring that the dep
# injection doesn't infinitely try and lookup self.token_provider_api

1
keystone/token/provider.py
View File

@ -266,7 +266,6 @@ class Manager(manager.Manager):
def _is_valid_token(self, token):
"""Verify the token is valid format and has not expired."""
current_time = timeutils.normalize_time(timeutils.utcnow())
try:

2
keystone/token/providers/fernet/token_formatters.py
View File

@ -596,7 +596,6 @@ class FederatedUnscopedPayload(BasePayload):
:returns: the payload of a federated token
"""
b_user_id = cls.attempt_convert_uuid_hex_to_bytes(user_id)
methods = auth_plugins.convert_method_list_to_integer(methods)
b_group_ids = list(map(cls.pack_group_id,
@ -622,7 +621,6 @@ class FederatedUnscopedPayload(BasePayload):
federated domain ID
"""
(is_stored_as_bytes, user_id) = payload[0]
if is_stored_as_bytes:
user_id = cls.attempt_convert_uuid_bytes_to_hex(user_id)

2
keystone/trust/controllers.py
View File

@ -56,7 +56,6 @@ class TrustV3(controller.V3Controller):
@classmethod
def base_url(cls, context, path=None):
"""Construct a path and pass it to V3Controller.base_url method."""
# NOTE(stevemar): Overriding path to /OS-TRUST/trusts so that
# V3Controller.base_url handles setting the self link correctly.
path = '/OS-TRUST/' + cls.collection_name
@ -128,7 +127,6 @@ class TrustV3(controller.V3Controller):
The user creating the trust must be the trustor.
"""
auth_context = context.get('environment',
{}).get('KEYSTONE_AUTH_CONTEXT', {})

2
tox.ini
View File

@ -115,7 +115,7 @@ show-source = true
# D205: Blank line required between one-line summary and description.
# D400: First line should end with a period.
# D401: First line should be in imperative mood.
ignore = H405,D100,D101,D102,D103,D104,D105,D202,D203,D204,D205,D400,D401
ignore = H405,D100,D101,D102,D103,D104,D105,D203,D204,D205,D400,D401
exclude=.venv,.git,.tox,build,dist,doc,*openstack/common*,*lib/python*,*egg,tools,vendor,.update-venv,*.ini,*.po,*.pot
max-complexity=24