Merge "Enables H404 check (multi line docstring) in flake8"

This commit is contained in:
Jenkins
2013-07-01 19:38:03 +00:00
committed by Gerrit Code Review
7 changed files with 115 additions and 292 deletions

View File

@@ -238,9 +238,10 @@ class HTTPClient(httplib2.Http):
'endpoint_url': self.endpoint_url} 'endpoint_url': self.endpoint_url}
def get_status_code(self, response): def get_status_code(self, response):
""" """Returns the integer status code from the response.
Returns the integer status code from the response, which
can be either a Webob.Response (used in testing) or httplib.Response Either a Webob.Response (used in testing) or httplib.Response
is returned.
""" """
if hasattr(response, 'status_int'): if hasattr(response, 'status_int'):
return response.status_int return response.status_int

View File

@@ -91,17 +91,10 @@ class AlreadyAttachedClient(QuantumClientException):
class Unauthorized(QuantumClientException): class Unauthorized(QuantumClientException):
"""
HTTP 401 - Unauthorized: bad credentials.
"""
message = _("Unauthorized: bad credentials.") message = _("Unauthorized: bad credentials.")
class Forbidden(QuantumClientException): class Forbidden(QuantumClientException):
"""
HTTP 403 - Forbidden: your credentials don't give you access to this
resource.
"""
message = _("Forbidden: your credentials don't give you access to this " message = _("Forbidden: your credentials don't give you access to this "
"resource.") "resource.")

View File

@@ -66,7 +66,8 @@ class JSONDictSerializer(DictSerializer):
class XMLDictSerializer(DictSerializer): class XMLDictSerializer(DictSerializer):
def __init__(self, metadata=None, xmlns=None): def __init__(self, metadata=None, xmlns=None):
""" """XMLDictSerializer constructor.
:param metadata: information needed to deserialize xml into :param metadata: information needed to deserialize xml into
a dictionary. a dictionary.
:param xmlns: XML namespace to include with serialized xml :param xmlns: XML namespace to include with serialized xml
@@ -80,7 +81,8 @@ class XMLDictSerializer(DictSerializer):
self.xmlns = xmlns self.xmlns = xmlns
def default(self, data): def default(self, data):
""" """Default serializer of XMLDictSerializer.
:param data: expect data to contain a single key as XML root, or :param data: expect data to contain a single key as XML root, or
contain another '*_links' key as atom links. Other contain another '*_links' key as atom links. Other
case will use 'VIRTUAL_ROOT_KEY' as XML root. case will use 'VIRTUAL_ROOT_KEY' as XML root.
@@ -232,7 +234,8 @@ class JSONDeserializer(TextDeserializer):
class XMLDeserializer(TextDeserializer): class XMLDeserializer(TextDeserializer):
def __init__(self, metadata=None): def __init__(self, metadata=None):
""" """XMLDeserializer constructor.
:param metadata: information needed to deserialize xml into :param metadata: information needed to deserialize xml into
a dictionary. a dictionary.
""" """

View File

@@ -31,9 +31,9 @@ from quantumclient.openstack.common import strutils
def env(*vars, **kwargs): def env(*vars, **kwargs):
""" """Returns the first environment variable set.
returns the first environment variable set
if none are non-empty, defaults to '' or keyword arg default if none are non-empty, defaults to '' or keyword arg default.
""" """
for v in vars: for v in vars:
value = os.environ.get(v) value = os.environ.get(v)
@@ -76,7 +76,7 @@ def loads(s):
def import_class(import_str): def import_class(import_str):
"""Returns a class from a string including module and class """Returns a class from a string including module and class.
:param import_str: a string representation of the class name :param import_str: a string representation of the class name
:rtype: the requested class :rtype: the requested class
@@ -141,7 +141,7 @@ def str2bool(strbool):
def str2dict(strdict): def str2dict(strdict):
''' '''Convert key1=value1,key2=value2,... string into dictionary.
:param strdict: key1=value1,key2=value2 :param strdict: key1=value1,key2=value2
''' '''

View File

@@ -433,10 +433,7 @@ class QuantumShell(app.App):
return parser return parser
def _bash_completion(self): def _bash_completion(self):
""" """Prints all of the commands and options for bash-completion."""
Prints all of the commands and options to stdout so that the
quantum's bash-completion script doesn't have to hard code them.
"""
commands = set() commands = set()
options = set() options = set()
for option, _action in self.parser._option_string_actions.items(): for option, _action in self.parser._option_string_actions.items():

View File

@@ -254,633 +254,465 @@ class Client(object):
@APIParamsCall @APIParamsCall
def list_ports(self, retrieve_all=True, **_params): def list_ports(self, retrieve_all=True, **_params):
""" """Fetches a list of all networks for a tenant."""
Fetches a list of all networks for a tenant
"""
# Pass filters in "params" argument to do_request # Pass filters in "params" argument to do_request
return self.list('ports', self.ports_path, retrieve_all, return self.list('ports', self.ports_path, retrieve_all,
**_params) **_params)
@APIParamsCall @APIParamsCall
def show_port(self, port, **_params): def show_port(self, port, **_params):
""" """Fetches information of a certain network."""
Fetches information of a certain network
"""
return self.get(self.port_path % (port), params=_params) return self.get(self.port_path % (port), params=_params)
@APIParamsCall @APIParamsCall
def create_port(self, body=None): def create_port(self, body=None):
""" """Creates a new port."""
Creates a new port
"""
return self.post(self.ports_path, body=body) return self.post(self.ports_path, body=body)
@APIParamsCall @APIParamsCall
def update_port(self, port, body=None): def update_port(self, port, body=None):
""" """Updates a port."""
Updates a port
"""
return self.put(self.port_path % (port), body=body) return self.put(self.port_path % (port), body=body)
@APIParamsCall @APIParamsCall
def delete_port(self, port): def delete_port(self, port):
""" """Deletes the specified port."""
Deletes the specified port
"""
return self.delete(self.port_path % (port)) return self.delete(self.port_path % (port))
@APIParamsCall @APIParamsCall
def list_networks(self, retrieve_all=True, **_params): def list_networks(self, retrieve_all=True, **_params):
""" """Fetches a list of all networks for a tenant."""
Fetches a list of all networks for a tenant
"""
# Pass filters in "params" argument to do_request # Pass filters in "params" argument to do_request
return self.list('networks', self.networks_path, retrieve_all, return self.list('networks', self.networks_path, retrieve_all,
**_params) **_params)
@APIParamsCall @APIParamsCall
def show_network(self, network, **_params): def show_network(self, network, **_params):
""" """Fetches information of a certain network."""
Fetches information of a certain network
"""
return self.get(self.network_path % (network), params=_params) return self.get(self.network_path % (network), params=_params)
@APIParamsCall @APIParamsCall
def create_network(self, body=None): def create_network(self, body=None):
""" """Creates a new network."""
Creates a new network
"""
return self.post(self.networks_path, body=body) return self.post(self.networks_path, body=body)
@APIParamsCall @APIParamsCall
def update_network(self, network, body=None): def update_network(self, network, body=None):
""" """Updates a network."""
Updates a network
"""
return self.put(self.network_path % (network), body=body) return self.put(self.network_path % (network), body=body)
@APIParamsCall @APIParamsCall
def delete_network(self, network): def delete_network(self, network):
""" """Deletes the specified network."""
Deletes the specified network
"""
return self.delete(self.network_path % (network)) return self.delete(self.network_path % (network))
@APIParamsCall @APIParamsCall
def list_subnets(self, retrieve_all=True, **_params): def list_subnets(self, retrieve_all=True, **_params):
""" """Fetches a list of all networks for a tenant."""
Fetches a list of all networks for a tenant
"""
return self.list('subnets', self.subnets_path, retrieve_all, return self.list('subnets', self.subnets_path, retrieve_all,
**_params) **_params)
@APIParamsCall @APIParamsCall
def show_subnet(self, subnet, **_params): def show_subnet(self, subnet, **_params):
""" """Fetches information of a certain subnet."""
Fetches information of a certain subnet
"""
return self.get(self.subnet_path % (subnet), params=_params) return self.get(self.subnet_path % (subnet), params=_params)
@APIParamsCall @APIParamsCall
def create_subnet(self, body=None): def create_subnet(self, body=None):
""" """Creates a new subnet."""
Creates a new subnet
"""
return self.post(self.subnets_path, body=body) return self.post(self.subnets_path, body=body)
@APIParamsCall @APIParamsCall
def update_subnet(self, subnet, body=None): def update_subnet(self, subnet, body=None):
""" """Updates a subnet."""
Updates a subnet
"""
return self.put(self.subnet_path % (subnet), body=body) return self.put(self.subnet_path % (subnet), body=body)
@APIParamsCall @APIParamsCall
def delete_subnet(self, subnet): def delete_subnet(self, subnet):
""" """Deletes the specified subnet."""
Deletes the specified subnet
"""
return self.delete(self.subnet_path % (subnet)) return self.delete(self.subnet_path % (subnet))
@APIParamsCall @APIParamsCall
def list_routers(self, retrieve_all=True, **_params): def list_routers(self, retrieve_all=True, **_params):
""" """Fetches a list of all routers for a tenant."""
Fetches a list of all routers for a tenant
"""
# Pass filters in "params" argument to do_request # Pass filters in "params" argument to do_request
return self.list('routers', self.routers_path, retrieve_all, return self.list('routers', self.routers_path, retrieve_all,
**_params) **_params)
@APIParamsCall @APIParamsCall
def show_router(self, router, **_params): def show_router(self, router, **_params):
""" """Fetches information of a certain router."""
Fetches information of a certain router
"""
return self.get(self.router_path % (router), params=_params) return self.get(self.router_path % (router), params=_params)
@APIParamsCall @APIParamsCall
def create_router(self, body=None): def create_router(self, body=None):
""" """Creates a new router."""
Creates a new router
"""
return self.post(self.routers_path, body=body) return self.post(self.routers_path, body=body)
@APIParamsCall @APIParamsCall
def update_router(self, router, body=None): def update_router(self, router, body=None):
""" """Updates a router."""
Updates a router
"""
return self.put(self.router_path % (router), body=body) return self.put(self.router_path % (router), body=body)
@APIParamsCall @APIParamsCall
def delete_router(self, router): def delete_router(self, router):
""" """Deletes the specified router."""
Deletes the specified router
"""
return self.delete(self.router_path % (router)) return self.delete(self.router_path % (router))
@APIParamsCall @APIParamsCall
def add_interface_router(self, router, body=None): def add_interface_router(self, router, body=None):
""" """Adds an internal network interface to the specified router."""
Adds an internal network interface to the specified router
"""
return self.put((self.router_path % router) + "/add_router_interface", return self.put((self.router_path % router) + "/add_router_interface",
body=body) body=body)
@APIParamsCall @APIParamsCall
def remove_interface_router(self, router, body=None): def remove_interface_router(self, router, body=None):
""" """Removes an internal network interface from the specified router."""
Removes an internal network interface from the specified router
"""
return self.put((self.router_path % router) + return self.put((self.router_path % router) +
"/remove_router_interface", body=body) "/remove_router_interface", body=body)
@APIParamsCall @APIParamsCall
def add_gateway_router(self, router, body=None): def add_gateway_router(self, router, body=None):
""" """Adds an external network gateway to the specified router."""
Adds an external network gateway to the specified router
"""
return self.put((self.router_path % router), return self.put((self.router_path % router),
body={'router': {'external_gateway_info': body}}) body={'router': {'external_gateway_info': body}})
@APIParamsCall @APIParamsCall
def remove_gateway_router(self, router): def remove_gateway_router(self, router):
""" """Removes an external network gateway from the specified router."""
Removes an external network gateway from the specified router
"""
return self.put((self.router_path % router), return self.put((self.router_path % router),
body={'router': {'external_gateway_info': {}}}) body={'router': {'external_gateway_info': {}}})
@APIParamsCall @APIParamsCall
def list_floatingips(self, retrieve_all=True, **_params): def list_floatingips(self, retrieve_all=True, **_params):
""" """Fetches a list of all floatingips for a tenant."""
Fetches a list of all floatingips for a tenant
"""
# Pass filters in "params" argument to do_request # Pass filters in "params" argument to do_request
return self.list('floatingips', self.floatingips_path, retrieve_all, return self.list('floatingips', self.floatingips_path, retrieve_all,
**_params) **_params)
@APIParamsCall @APIParamsCall
def show_floatingip(self, floatingip, **_params): def show_floatingip(self, floatingip, **_params):
""" """Fetches information of a certain floatingip."""
Fetches information of a certain floatingip
"""
return self.get(self.floatingip_path % (floatingip), params=_params) return self.get(self.floatingip_path % (floatingip), params=_params)
@APIParamsCall @APIParamsCall
def create_floatingip(self, body=None): def create_floatingip(self, body=None):
""" """Creates a new floatingip."""
Creates a new floatingip
"""
return self.post(self.floatingips_path, body=body) return self.post(self.floatingips_path, body=body)
@APIParamsCall @APIParamsCall
def update_floatingip(self, floatingip, body=None): def update_floatingip(self, floatingip, body=None):
""" """Updates a floatingip."""
Updates a floatingip
"""
return self.put(self.floatingip_path % (floatingip), body=body) return self.put(self.floatingip_path % (floatingip), body=body)
@APIParamsCall @APIParamsCall
def delete_floatingip(self, floatingip): def delete_floatingip(self, floatingip):
""" """Deletes the specified floatingip."""
Deletes the specified floatingip
"""
return self.delete(self.floatingip_path % (floatingip)) return self.delete(self.floatingip_path % (floatingip))
@APIParamsCall @APIParamsCall
def create_security_group(self, body=None): def create_security_group(self, body=None):
""" """Creates a new security group."""
Creates a new security group
"""
return self.post(self.security_groups_path, body=body) return self.post(self.security_groups_path, body=body)
@APIParamsCall @APIParamsCall
def update_security_group(self, security_group, body=None): def update_security_group(self, security_group, body=None):
""" """Updates a security group."""
Updates a security group
"""
return self.put(self.security_group_path % return self.put(self.security_group_path %
security_group, body=body) security_group, body=body)
@APIParamsCall @APIParamsCall
def list_security_groups(self, retrieve_all=True, **_params): def list_security_groups(self, retrieve_all=True, **_params):
""" """Fetches a list of all security groups for a tenant."""
Fetches a list of all security groups for a tenant
"""
return self.list('security_groups', self.security_groups_path, return self.list('security_groups', self.security_groups_path,
retrieve_all, **_params) retrieve_all, **_params)
@APIParamsCall @APIParamsCall
def show_security_group(self, security_group, **_params): def show_security_group(self, security_group, **_params):
""" """Fetches information of a certain security group."""
Fetches information of a certain security group
"""
return self.get(self.security_group_path % (security_group), return self.get(self.security_group_path % (security_group),
params=_params) params=_params)
@APIParamsCall @APIParamsCall
def delete_security_group(self, security_group): def delete_security_group(self, security_group):
""" """Deletes the specified security group."""
Deletes the specified security group
"""
return self.delete(self.security_group_path % (security_group)) return self.delete(self.security_group_path % (security_group))
@APIParamsCall @APIParamsCall
def create_security_group_rule(self, body=None): def create_security_group_rule(self, body=None):
""" """Creates a new security group rule."""
Creates a new security group rule
"""
return self.post(self.security_group_rules_path, body=body) return self.post(self.security_group_rules_path, body=body)
@APIParamsCall @APIParamsCall
def delete_security_group_rule(self, security_group_rule): def delete_security_group_rule(self, security_group_rule):
""" """Deletes the specified security group rule."""
Deletes the specified security group rule
"""
return self.delete(self.security_group_rule_path % return self.delete(self.security_group_rule_path %
(security_group_rule)) (security_group_rule))
@APIParamsCall @APIParamsCall
def list_security_group_rules(self, retrieve_all=True, **_params): def list_security_group_rules(self, retrieve_all=True, **_params):
""" """Fetches a list of all security group rules for a tenant."""
Fetches a list of all security group rules for a tenant
"""
return self.list('security_group_rules', return self.list('security_group_rules',
self.security_group_rules_path, self.security_group_rules_path,
retrieve_all, **_params) retrieve_all, **_params)
@APIParamsCall @APIParamsCall
def show_security_group_rule(self, security_group_rule, **_params): def show_security_group_rule(self, security_group_rule, **_params):
""" """Fetches information of a certain security group rule."""
Fetches information of a certain security group rule
"""
return self.get(self.security_group_rule_path % (security_group_rule), return self.get(self.security_group_rule_path % (security_group_rule),
params=_params) params=_params)
@APIParamsCall @APIParamsCall
def list_vips(self, retrieve_all=True, **_params): def list_vips(self, retrieve_all=True, **_params):
""" """Fetches a list of all load balancer vips for a tenant."""
Fetches a list of all load balancer vips for a tenant
"""
# Pass filters in "params" argument to do_request # Pass filters in "params" argument to do_request
return self.list('vips', self.vips_path, retrieve_all, return self.list('vips', self.vips_path, retrieve_all,
**_params) **_params)
@APIParamsCall @APIParamsCall
def show_vip(self, vip, **_params): def show_vip(self, vip, **_params):
""" """Fetches information of a certain load balancer vip."""
Fetches information of a certain load balancer vip
"""
return self.get(self.vip_path % (vip), params=_params) return self.get(self.vip_path % (vip), params=_params)
@APIParamsCall @APIParamsCall
def create_vip(self, body=None): def create_vip(self, body=None):
""" """Creates a new load balancer vip."""
Creates a new load balancer vip
"""
return self.post(self.vips_path, body=body) return self.post(self.vips_path, body=body)
@APIParamsCall @APIParamsCall
def update_vip(self, vip, body=None): def update_vip(self, vip, body=None):
""" """Updates a load balancer vip."""
Updates a load balancer vip
"""
return self.put(self.vip_path % (vip), body=body) return self.put(self.vip_path % (vip), body=body)
@APIParamsCall @APIParamsCall
def delete_vip(self, vip): def delete_vip(self, vip):
""" """Deletes the specified load balancer vip."""
Deletes the specified load balancer vip
"""
return self.delete(self.vip_path % (vip)) return self.delete(self.vip_path % (vip))
@APIParamsCall @APIParamsCall
def list_pools(self, retrieve_all=True, **_params): def list_pools(self, retrieve_all=True, **_params):
""" """Fetches a list of all load balancer pools for a tenant."""
Fetches a list of all load balancer pools for a tenant
"""
# Pass filters in "params" argument to do_request # Pass filters in "params" argument to do_request
return self.list('pools', self.pools_path, retrieve_all, return self.list('pools', self.pools_path, retrieve_all,
**_params) **_params)
@APIParamsCall @APIParamsCall
def show_pool(self, pool, **_params): def show_pool(self, pool, **_params):
""" """Fetches information of a certain load balancer pool."""
Fetches information of a certain load balancer pool
"""
return self.get(self.pool_path % (pool), params=_params) return self.get(self.pool_path % (pool), params=_params)
@APIParamsCall @APIParamsCall
def create_pool(self, body=None): def create_pool(self, body=None):
""" """Creates a new load balancer pool."""
Creates a new load balancer pool
"""
return self.post(self.pools_path, body=body) return self.post(self.pools_path, body=body)
@APIParamsCall @APIParamsCall
def update_pool(self, pool, body=None): def update_pool(self, pool, body=None):
""" """Updates a load balancer pool."""
Updates a load balancer pool
"""
return self.put(self.pool_path % (pool), body=body) return self.put(self.pool_path % (pool), body=body)
@APIParamsCall @APIParamsCall
def delete_pool(self, pool): def delete_pool(self, pool):
""" """Deletes the specified load balancer pool."""
Deletes the specified load balancer pool
"""
return self.delete(self.pool_path % (pool)) return self.delete(self.pool_path % (pool))
@APIParamsCall @APIParamsCall
def retrieve_pool_stats(self, pool, **_params): def retrieve_pool_stats(self, pool, **_params):
""" """Retrieves stats for a certain load balancer pool."""
Retrieves stats for a certain load balancer pool
"""
return self.get(self.pool_path_stats % (pool), params=_params) return self.get(self.pool_path_stats % (pool), params=_params)
@APIParamsCall @APIParamsCall
def list_members(self, retrieve_all=True, **_params): def list_members(self, retrieve_all=True, **_params):
""" """Fetches a list of all load balancer members for a tenant."""
Fetches a list of all load balancer members for a tenant
"""
# Pass filters in "params" argument to do_request # Pass filters in "params" argument to do_request
return self.list('members', self.members_path, retrieve_all, return self.list('members', self.members_path, retrieve_all,
**_params) **_params)
@APIParamsCall @APIParamsCall
def show_member(self, member, **_params): def show_member(self, member, **_params):
""" """Fetches information of a certain load balancer member."""
Fetches information of a certain load balancer member
"""
return self.get(self.member_path % (member), params=_params) return self.get(self.member_path % (member), params=_params)
@APIParamsCall @APIParamsCall
def create_member(self, body=None): def create_member(self, body=None):
""" """Creates a new load balancer member."""
Creates a new load balancer member
"""
return self.post(self.members_path, body=body) return self.post(self.members_path, body=body)
@APIParamsCall @APIParamsCall
def update_member(self, member, body=None): def update_member(self, member, body=None):
""" """Updates a load balancer member."""
Updates a load balancer member
"""
return self.put(self.member_path % (member), body=body) return self.put(self.member_path % (member), body=body)
@APIParamsCall @APIParamsCall
def delete_member(self, member): def delete_member(self, member):
""" """Deletes the specified load balancer member."""
Deletes the specified load balancer member
"""
return self.delete(self.member_path % (member)) return self.delete(self.member_path % (member))
@APIParamsCall @APIParamsCall
def list_health_monitors(self, retrieve_all=True, **_params): def list_health_monitors(self, retrieve_all=True, **_params):
""" """Fetches a list of all load balancer health monitors for a tenant."""
Fetches a list of all load balancer health monitors for a tenant
"""
# Pass filters in "params" argument to do_request # Pass filters in "params" argument to do_request
return self.list('health_monitors', self.health_monitors_path, return self.list('health_monitors', self.health_monitors_path,
retrieve_all, **_params) retrieve_all, **_params)
@APIParamsCall @APIParamsCall
def show_health_monitor(self, health_monitor, **_params): def show_health_monitor(self, health_monitor, **_params):
""" """Fetches information of a certain load balancer health monitor."""
Fetches information of a certain load balancer health monitor
"""
return self.get(self.health_monitor_path % (health_monitor), return self.get(self.health_monitor_path % (health_monitor),
params=_params) params=_params)
@APIParamsCall @APIParamsCall
def create_health_monitor(self, body=None): def create_health_monitor(self, body=None):
""" """Creates a new load balancer health monitor."""
Creates a new load balancer health monitor
"""
return self.post(self.health_monitors_path, body=body) return self.post(self.health_monitors_path, body=body)
@APIParamsCall @APIParamsCall
def update_health_monitor(self, health_monitor, body=None): def update_health_monitor(self, health_monitor, body=None):
""" """Updates a load balancer health monitor."""
Updates a load balancer health monitor
"""
return self.put(self.health_monitor_path % (health_monitor), body=body) return self.put(self.health_monitor_path % (health_monitor), body=body)
@APIParamsCall @APIParamsCall
def delete_health_monitor(self, health_monitor): def delete_health_monitor(self, health_monitor):
""" """Deletes the specified load balancer health monitor."""
Deletes the specified load balancer health monitor
"""
return self.delete(self.health_monitor_path % (health_monitor)) return self.delete(self.health_monitor_path % (health_monitor))
@APIParamsCall @APIParamsCall
def associate_health_monitor(self, pool, body): def associate_health_monitor(self, pool, body):
""" """Associate specified load balancer health monitor and pool."""
Associate specified load balancer health monitor and pool
"""
return self.post(self.associate_pool_health_monitors_path % (pool), return self.post(self.associate_pool_health_monitors_path % (pool),
body=body) body=body)
@APIParamsCall @APIParamsCall
def disassociate_health_monitor(self, pool, health_monitor): def disassociate_health_monitor(self, pool, health_monitor):
""" """Disassociate specified load balancer health monitor and pool."""
Disassociate specified load balancer health monitor and pool
"""
path = (self.disassociate_pool_health_monitors_path % path = (self.disassociate_pool_health_monitors_path %
{'pool': pool, 'health_monitor': health_monitor}) {'pool': pool, 'health_monitor': health_monitor})
return self.delete(path) return self.delete(path)
@APIParamsCall @APIParamsCall
def create_qos_queue(self, body=None): def create_qos_queue(self, body=None):
""" """Creates a new queue."""
Creates a new queue
"""
return self.post(self.qos_queues_path, body=body) return self.post(self.qos_queues_path, body=body)
@APIParamsCall @APIParamsCall
def list_qos_queues(self, **_params): def list_qos_queues(self, **_params):
""" """Fetches a list of all queues for a tenant."""
Fetches a list of all queues for a tenant
"""
return self.get(self.qos_queues_path, params=_params) return self.get(self.qos_queues_path, params=_params)
@APIParamsCall @APIParamsCall
def show_qos_queue(self, queue, **_params): def show_qos_queue(self, queue, **_params):
""" """Fetches information of a certain queue."""
Fetches information of a certain queue
"""
return self.get(self.qos_queue_path % (queue), return self.get(self.qos_queue_path % (queue),
params=_params) params=_params)
@APIParamsCall @APIParamsCall
def delete_qos_queue(self, queue): def delete_qos_queue(self, queue):
""" """Deletes the specified queue."""
Deletes the specified queue
"""
return self.delete(self.qos_queue_path % (queue)) return self.delete(self.qos_queue_path % (queue))
@APIParamsCall @APIParamsCall
def list_agents(self, **_params): def list_agents(self, **_params):
""" """Fetches agents."""
Fetches agents
"""
# Pass filters in "params" argument to do_request # Pass filters in "params" argument to do_request
return self.get(self.agents_path, params=_params) return self.get(self.agents_path, params=_params)
@APIParamsCall @APIParamsCall
def show_agent(self, agent, **_params): def show_agent(self, agent, **_params):
""" """Fetches information of a certain agent."""
Fetches information of a certain agent
"""
return self.get(self.agent_path % (agent), params=_params) return self.get(self.agent_path % (agent), params=_params)
@APIParamsCall @APIParamsCall
def update_agent(self, agent, body=None): def update_agent(self, agent, body=None):
""" """Updates an agent."""
Updates an agent
"""
return self.put(self.agent_path % (agent), body=body) return self.put(self.agent_path % (agent), body=body)
@APIParamsCall @APIParamsCall
def delete_agent(self, agent): def delete_agent(self, agent):
""" """Deletes the specified agent."""
Deletes the specified agent
"""
return self.delete(self.agent_path % (agent)) return self.delete(self.agent_path % (agent))
@APIParamsCall @APIParamsCall
def list_network_gateways(self, **_params): def list_network_gateways(self, **_params):
""" """Retrieve network gateways."""
Retrieve network gateways
"""
return self.get(self.network_gateways_path, params=_params) return self.get(self.network_gateways_path, params=_params)
@APIParamsCall @APIParamsCall
def show_network_gateway(self, gateway_id, **_params): def show_network_gateway(self, gateway_id, **_params):
""" """Fetch a network gateway."""
Fetch a network gateway
"""
return self.get(self.network_gateway_path % gateway_id, params=_params) return self.get(self.network_gateway_path % gateway_id, params=_params)
@APIParamsCall @APIParamsCall
def create_network_gateway(self, body=None): def create_network_gateway(self, body=None):
""" """Create a new network gateway."""
Create a new network gateway
"""
return self.post(self.network_gateways_path, body=body) return self.post(self.network_gateways_path, body=body)
@APIParamsCall @APIParamsCall
def update_network_gateway(self, gateway_id, body=None): def update_network_gateway(self, gateway_id, body=None):
""" """Update a network gateway."""
Update a network gateway
"""
return self.put(self.network_gateway_path % gateway_id, body=body) return self.put(self.network_gateway_path % gateway_id, body=body)
@APIParamsCall @APIParamsCall
def delete_network_gateway(self, gateway_id): def delete_network_gateway(self, gateway_id):
""" """Delete the specified network gateway."""
Delete the specified network gateway
"""
return self.delete(self.network_gateway_path % gateway_id) return self.delete(self.network_gateway_path % gateway_id)
@APIParamsCall @APIParamsCall
def connect_network_gateway(self, gateway_id, body=None): def connect_network_gateway(self, gateway_id, body=None):
""" """Connect a network gateway to the specified network."""
Connect a network gateway to the specified network
"""
base_uri = self.network_gateway_path % gateway_id base_uri = self.network_gateway_path % gateway_id
return self.put("%s/connect_network" % base_uri, body=body) return self.put("%s/connect_network" % base_uri, body=body)
@APIParamsCall @APIParamsCall
def disconnect_network_gateway(self, gateway_id, body=None): def disconnect_network_gateway(self, gateway_id, body=None):
""" """Disconnect a network from the specified gateway."""
Disconnect a network from the specified gateway
"""
base_uri = self.network_gateway_path % gateway_id base_uri = self.network_gateway_path % gateway_id
return self.put("%s/disconnect_network" % base_uri, body=body) return self.put("%s/disconnect_network" % base_uri, body=body)
@APIParamsCall @APIParamsCall
def list_dhcp_agent_hosting_networks(self, network, **_params): def list_dhcp_agent_hosting_networks(self, network, **_params):
""" """Fetches a list of dhcp agents hosting a network."""
Fetches a list of dhcp agents hosting a network.
"""
return self.get((self.network_path + self.DHCP_AGENTS) % network, return self.get((self.network_path + self.DHCP_AGENTS) % network,
params=_params) params=_params)
@APIParamsCall @APIParamsCall
def list_networks_on_dhcp_agent(self, dhcp_agent, **_params): def list_networks_on_dhcp_agent(self, dhcp_agent, **_params):
""" """Fetches a list of dhcp agents hosting a network."""
Fetches a list of dhcp agents hosting a network.
"""
return self.get((self.agent_path + self.DHCP_NETS) % dhcp_agent, return self.get((self.agent_path + self.DHCP_NETS) % dhcp_agent,
params=_params) params=_params)
@APIParamsCall @APIParamsCall
def add_network_to_dhcp_agent(self, dhcp_agent, body=None): def add_network_to_dhcp_agent(self, dhcp_agent, body=None):
""" """Adds a network to dhcp agent."""
Adds a network to dhcp agent.
"""
return self.post((self.agent_path + self.DHCP_NETS) % dhcp_agent, return self.post((self.agent_path + self.DHCP_NETS) % dhcp_agent,
body=body) body=body)
@APIParamsCall @APIParamsCall
def remove_network_from_dhcp_agent(self, dhcp_agent, network_id): def remove_network_from_dhcp_agent(self, dhcp_agent, network_id):
""" """Remove a network from dhcp agent."""
Remove a network from dhcp agent.
"""
return self.delete((self.agent_path + self.DHCP_NETS + "/%s") % ( return self.delete((self.agent_path + self.DHCP_NETS + "/%s") % (
dhcp_agent, network_id)) dhcp_agent, network_id))
@APIParamsCall @APIParamsCall
def list_l3_agent_hosting_routers(self, router, **_params): def list_l3_agent_hosting_routers(self, router, **_params):
""" """Fetches a list of L3 agents hosting a router."""
Fetches a list of L3 agents hosting a router.
"""
return self.get((self.router_path + self.L3_AGENTS) % router, return self.get((self.router_path + self.L3_AGENTS) % router,
params=_params) params=_params)
@APIParamsCall @APIParamsCall
def list_routers_on_l3_agent(self, l3_agent, **_params): def list_routers_on_l3_agent(self, l3_agent, **_params):
""" """Fetches a list of L3 agents hosting a router."""
Fetches a list of L3 agents hosting a router.
"""
return self.get((self.agent_path + self.L3_ROUTERS) % l3_agent, return self.get((self.agent_path + self.L3_ROUTERS) % l3_agent,
params=_params) params=_params)
@APIParamsCall @APIParamsCall
def add_router_to_l3_agent(self, l3_agent, body): def add_router_to_l3_agent(self, l3_agent, body):
""" """Adds a router to L3 agent."""
Adds a router to L3 agent.
"""
return self.post((self.agent_path + self.L3_ROUTERS) % l3_agent, return self.post((self.agent_path + self.L3_ROUTERS) % l3_agent,
body=body) body=body)
@APIParamsCall @APIParamsCall
def remove_router_from_l3_agent(self, l3_agent, router_id): def remove_router_from_l3_agent(self, l3_agent, router_id):
""" """Remove a router from l3 agent."""
Remove a router from l3 agent.
"""
return self.delete((self.agent_path + self.L3_ROUTERS + "/%s") % ( return self.delete((self.agent_path + self.L3_ROUTERS + "/%s") % (
l3_agent, router_id)) l3_agent, router_id))
@@ -941,9 +773,10 @@ class Client(object):
return self.httpclient.get_auth_info() return self.httpclient.get_auth_info()
def get_status_code(self, response): def get_status_code(self, response):
""" """Returns the integer status code from the response.
Returns the integer status code from the response, which
can be either a Webob.Response (used in testing) or httplib.Response Either a Webob.Response (used in testing) or httplib.Response
is returned.
""" """
if hasattr(response, 'status_int'): if hasattr(response, 'status_int'):
return response.status_int return response.status_int
@@ -951,9 +784,10 @@ class Client(object):
return response.status return response.status
def serialize(self, data): def serialize(self, data):
""" """Serializes a dictionary into either xml or json.
Serializes a dictionary with a single key (which can contain any
structure) into either xml or json A dictionary with a single key can be passed and
it can contain any structure.
""" """
if data is None: if data is None:
return None return None
@@ -965,28 +799,25 @@ class Client(object):
type(data)) type(data))
def deserialize(self, data, status_code): def deserialize(self, data, status_code):
""" """Deserializes an xml or json string into a dictionary."""
Deserializes an xml or json string into a dictionary
"""
if status_code == 204: if status_code == 204:
return data return data
return serializer.Serializer(self.get_attr_metadata()).deserialize( return serializer.Serializer(self.get_attr_metadata()).deserialize(
data, self.content_type())['body'] data, self.content_type())['body']
def content_type(self, _format=None): def content_type(self, _format=None):
""" """Returns the mime-type for either 'xml' or 'json'.
Returns the mime-type for either 'xml' or 'json'. Defaults to the
currently set format Defaults to the currently set format.
""" """
_format = _format or self.format _format = _format or self.format
return "application/%s" % (_format) return "application/%s" % (_format)
def retry_request(self, method, action, body=None, def retry_request(self, method, action, body=None,
headers=None, params=None): headers=None, params=None):
""" """Call do_request with the default retry configuration.
Call do_request with the default retry configuration. Only
idempotent requests should retry failed connection attempts.
Only idempotent requests should retry failed connection attempts.
:raises: ConnectionFailed if the maximum # of retries is exceeded :raises: ConnectionFailed if the maximum # of retries is exceeded
""" """
max_attempts = self.retries + 1 max_attempts = self.retries + 1

View File

@@ -25,10 +25,8 @@ downloadcache = ~/cache/pip
[flake8] [flake8]
# E125 continuation line does not distinguish itself from next logical line # E125 continuation line does not distinguish itself from next logical line
# H301 one import per line
# H302 import only modules # H302 import only modules
# TODO(marun) H404 multi line docstring should start with a summary ignore = E125,H302
ignore = E125,H301,H302,H404
show-source = true show-source = true
builtins = _ builtins = _
exclude=.venv,.git,.tox,dist,doc,*openstack/common*,*lib/python*,*egg,tools exclude=.venv,.git,.tox,dist,doc,*openstack/common*,*lib/python*,*egg,tools