openstack
/
neutron
Code Issues Proposed changes

Start enforcing E125 flake8 directive 

Removed E125 (continuation line does not distinguish itself
from next logical line) from the ignore list and fixed all
the indentation issues.  Didn't think it was going to be
close to 100 files when I started.

Change-Id: I0a6f5efec4b7d8d3632dd9dbb43e0ab58af9dff3
This commit is contained in:
Brian Haley 2019-07-18 10:30:42 -04:00
parent b3d9d9742a
commit b79842f289
100 changed files with 385 additions and 424 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
neutron/agent/l3/dvr_local_router.py
View File

@ -756,8 +756,8 @@ class DvrLocalRouter(dvr_router_base.DvrRouterBase):
operation, route, fip_ns_name, tbl_index)
super(DvrLocalRouter, self).update_routing_table(operation, route)
def _update_fip_route_table_with_next_hop_routes(
self, operation, route, fip_ns_name, tbl_index):
def _update_fip_route_table_with_next_hop_routes(self, operation, route,
fip_ns_name, tbl_index):
cmd = ['ip', 'route', operation, 'to', route['destination'],
'via', route['nexthop'], 'table', tbl_index]
ip_wrapper = ip_lib.IPWrapper(namespace=fip_ns_name)
@ -768,8 +768,8 @@ class DvrLocalRouter(dvr_router_base.DvrRouterBase):
"router %(id)s",
{'ns': fip_ns_name, 'id': self.router_id})
def _check_if_route_applicable_to_fip_namespace(
self, route, agent_gateway_port):
def _check_if_route_applicable_to_fip_namespace(self, route,
agent_gateway_port):
ip_cidrs = common_utils.fixed_ip_cidrs(agent_gateway_port['fixed_ips'])
nexthop_cidr = netaddr.IPAddress(route['nexthop'])
for gw_cidr in ip_cidrs:

4
neutron/agent/linux/utils.py
View File

@ -56,8 +56,8 @@ class RootwrapDaemonHelper(object):
def get_client(cls):
with cls.__lock:
if cls.__client is None:
if xenapi_root_helper.ROOT_HELPER_DAEMON_TOKEN == \
cfg.CONF.AGENT.root_helper_daemon:
if (xenapi_root_helper.ROOT_HELPER_DAEMON_TOKEN ==
cfg.CONF.AGENT.root_helper_daemon):
cls.__client = xenapi_root_helper.XenAPIClient()
else:
cls.__client = client.Client(

3
neutron/db/l3_db.py
View File

@ -2091,8 +2091,7 @@ class L3_NAT_db_mixin(L3_NAT_dbonly_mixin, L3RpcNotifierMixin):
super(L3_NAT_db_mixin, self).notify_routers_updated(
context, list(router_ids), 'disassociate_floatingips', {})
def _migrate_router_ports(
self, context, router_db, old_owner, new_owner):
def _migrate_router_ports(self, context, router_db, old_owner, new_owner):
"""Update the model to support the dvr case of a router."""
for rp in router_db.attached_ports:
if rp.port_type == old_owner:

49
neutron/db/l3_dvr_db.py
View File

@ -186,8 +186,8 @@ class DVRResourceOperationHandler(object):
not old_router.get(l3_apidef.EXTERNAL_GW_INFO))
if not do_create:
return
if not self._create_snat_intf_ports_if_not_exists(
context.elevated(), router_db):
if not self._create_snat_intf_ports_if_not_exists(context.elevated(),
router_db):
LOG.debug("SNAT interface ports not created: %s",
router_db['id'])
return router_db
@ -220,8 +220,9 @@ class DVRResourceOperationHandler(object):
msg = _("Unable to create the SNAT Interface Port")
raise n_exc.BadRequest(resource='router', msg=msg)
with plugin_utils.delete_port_on_error(
self.l3plugin._core_plugin, context.elevated(), snat_port['id']):
with plugin_utils.delete_port_on_error(self.l3plugin._core_plugin,
context.elevated(),
snat_port['id']):
l3_obj.RouterPort(
context,
port_id=snat_port['id'],
@ -313,8 +314,8 @@ class DVRResourceOperationHandler(object):
self.l3plugin.l3_rpc_notifier.delete_fipnamespace_for_ext_net(
context, network_id)
def delete_floatingip_agent_gateway_port(
self, context, host_id, ext_net_id):
def delete_floatingip_agent_gateway_port(self, context, host_id,
ext_net_id):
"""Function to delete FIP gateway port with given ext_net_id."""
# delete any fip agent gw port
device_filter = {'device_owner': [const.DEVICE_OWNER_AGENT_GW],
@ -327,8 +328,8 @@ class DVRResourceOperationHandler(object):
if host_id:
return
def _get_ports_for_allowed_address_pair_ip(
self, context, network_id, fixed_ip):
def _get_ports_for_allowed_address_pair_ip(self, context, network_id,
fixed_ip):
"""Return all active ports associated with the allowed_addr_pair ip."""
query = context.session.query(
models_v2.Port).filter(
@ -391,8 +392,7 @@ class DVRResourceOperationHandler(object):
self._inherit_service_port_and_arp_update(
context, addr_pair_active_service_port_list[0])
def _inherit_service_port_and_arp_update(
self, context, service_port):
def _inherit_service_port_and_arp_update(self, context, service_port):
"""Function inherits port host bindings for allowed_address_pair."""
service_port_dict = self.l3plugin._core_plugin._make_port_dict(
service_port)
@ -473,8 +473,8 @@ class DVRResourceOperationHandler(object):
port['id'])
LOG.debug("CSNAT port updated for IPv6 subnet: %s", updated_port)
def _find_v6_router_port_by_network_and_device_owner(
self, router, net_id, device_owner):
def _find_v6_router_port_by_network_and_device_owner(self, router, net_id,
device_owner):
for port in router.attached_ports:
p = port['port']
if (p['network_id'] == net_id and
@ -482,8 +482,8 @@ class DVRResourceOperationHandler(object):
self.l3plugin._port_has_ipv6_address(p)):
return self.l3plugin._core_plugin._make_port_dict(p)
def _check_for_multiprefix_csnat_port_and_update(
self, context, router, network_id, subnet_id):
def _check_for_multiprefix_csnat_port_and_update(self, context, router,
network_id, subnet_id):
"""Checks if the csnat port contains multiple ipv6 prefixes.
If the csnat port contains multiple ipv6 prefixes for the given
@ -889,7 +889,7 @@ class _DVRAgentInterfaceMixin(object):
self._get_dvr_migrating_service_port_hostid(
context, fip['port_id'], port=vm_port))
vm_port_agent_mode = vm_port.get('agent', None)
if vm_port_agent_mode != (
if (vm_port_agent_mode !=
const.L3_AGENT_MODE_DVR_NO_EXTERNAL):
# For floatingip configured on ports that do not
# reside on a 'dvr_no_external' agent, add the
@ -926,8 +926,8 @@ class _DVRAgentInterfaceMixin(object):
port_db = port or self._core_plugin.get_port(context, port_id)
return port_db[portbindings.HOST_ID] or None
def _get_dvr_migrating_service_port_hostid(
self, context, port_id, port=None):
def _get_dvr_migrating_service_port_hostid(self, context, port_id,
port=None):
"""Returns the migrating host_id from the migrating profile."""
port_db = port or self._core_plugin.get_port(context, port_id)
port_profile = port_db.get(portbindings.PROFILE)
@ -963,8 +963,8 @@ class _DVRAgentInterfaceMixin(object):
self.create_fip_agent_gw_port_if_not_exists(
context.elevated(), network_id, host)
def create_fip_agent_gw_port_if_not_exists(
self, context, network_id, host):
def create_fip_agent_gw_port_if_not_exists(self, context, network_id,
host):
"""Function to return the FIP Agent GW port.
This function will create a FIP Agent GW port
@ -1010,8 +1010,8 @@ class _DVRAgentInterfaceMixin(object):
self._populate_mtu_and_subnets_for_ports(context, [agent_port])
return agent_port
def _generate_arp_table_and_notify_agent(
self, context, fixed_ip, mac_address, notifier):
def _generate_arp_table_and_notify_agent(self, context, fixed_ip,
mac_address, notifier):
"""Generates the arp table entry and notifies the l3 agent."""
ip_address = fixed_ip['ip_address']
subnet = fixed_ip['subnet_id']
@ -1025,8 +1025,7 @@ class _DVRAgentInterfaceMixin(object):
for router_id in routers:
notifier(context, router_id, arp_table)
def _get_subnet_id_for_given_fixed_ip(
self, context, fixed_ip, port_dict):
def _get_subnet_id_for_given_fixed_ip(self, context, fixed_ip, port_dict):
"""Returns the subnet_id that matches the fixedip on a network."""
filters = {'network_id': [port_dict['network_id']]}
subnets = self._core_plugin.get_subnets(context, filters)
@ -1072,8 +1071,8 @@ class _DVRAgentInterfaceMixin(object):
context, fixed_ip, port_dict['mac_address'],
self.l3_rpc_notifier.add_arp_entry)
def delete_arp_entry_for_dvr_service_port(
self, context, port_dict, fixed_ips_to_delete=None):
def delete_arp_entry_for_dvr_service_port(self, context, port_dict,
fixed_ips_to_delete=None):
"""Notify L3 agents of ARP table entry for dvr service port.
When a dvr service port goes down, look for the DVR

22
neutron/db/portsecurity_db_common.py
View File

@ -30,8 +30,8 @@ class PortSecurityDbCommon(object):
response_data[psec.PORTSECURITY] = (
db_data['port_security'][psec.PORTSECURITY])
def _process_port_security_create(
self, context, obj_cls, res_name, req, res):
def _process_port_security_create(self, context, obj_cls, res_name, req,
res):
obj = obj_cls(
context,
id=res['id'],
@ -41,14 +41,13 @@ class PortSecurityDbCommon(object):
res[psec.PORTSECURITY] = req[psec.PORTSECURITY]
return self._make_port_security_dict(obj, res_name)
def _process_port_port_security_create(
self, context, port_req, port_res):
def _process_port_port_security_create(self, context, port_req, port_res):
self._process_port_security_create(
context, p_ps.PortSecurity, 'port',
port_req, port_res)
def _process_network_port_security_create(
self, context, network_req, network_res):
def _process_network_port_security_create(self, context, network_req,
network_res):
self._process_port_security_create(
context, network.NetworkPortSecurity, 'network',
network_req, network_res)
@ -66,19 +65,18 @@ class PortSecurityDbCommon(object):
def _get_port_security_binding(self, context, port_id):
return self._get_security_binding(context, p_ps.PortSecurity, port_id)
def _process_port_port_security_update(
self, context, port_req, port_res):
def _process_port_port_security_update(self, context, port_req, port_res):
self._process_port_security_update(
context, p_ps.PortSecurity, 'port', port_req, port_res)
def _process_network_port_security_update(
self, context, network_req, network_res):
def _process_network_port_security_update(self, context, network_req,
network_res):
self._process_port_security_update(
context, network.NetworkPortSecurity, 'network',
network_req, network_res)
def _process_port_security_update(
self, context, obj_cls, res_name, req, res):
def _process_port_security_update(self, context, obj_cls, res_name, req,
res):
if psec.PORTSECURITY not in req:
return
port_security_enabled = req[psec.PORTSECURITY]

3
neutron/db/quota/driver.py
View File

@ -118,8 +118,7 @@ class DbQuotaDriver(object):
Raise a "not found" error if the quota for the given tenant was
never defined.
"""
if quota_obj.Quota.delete_objects(
context, project_id=tenant_id) < 1:
if quota_obj.Quota.delete_objects(context, project_id=tenant_id) < 1:
# No record deleted means the quota was not found
raise exceptions.TenantQuotaNotFound(tenant_id=tenant_id)

5
neutron/plugins/ml2/drivers/l2pop/mech_driver.py
View File

@ -254,8 +254,9 @@ class L2populationMechanismDriver(api.MechanismDriver):
agent_host = context.host
l3plugin = directory.get_plugin(plugin_constants.L3)
# when agent transitions to backup, don't remove flood flows
if agent_host and l3plugin and getattr(
l3plugin, "list_router_ids_on_host", None):
if agent_host and l3plugin and getattr(l3plugin,
"list_router_ids_on_host",
None):
admin_context = n_context.get_admin_context()
port_context = context._plugin_context
fdb_entries = self._get_agent_fdb(

3
neutron/plugins/ml2/drivers/linuxbridge/agent/linuxbridge_neutron_agent.py
View File

@ -874,8 +874,7 @@ class LinuxBridgeManager(amb.CommonAgentManagerBase):
return lconst.EXTENSION_DRIVER_TYPE
class LinuxBridgeRpcCallbacks(
sg_rpc.SecurityGroupAgentRpcCallbackMixin,
class LinuxBridgeRpcCallbacks(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
l2pop_rpc.L2populationRpcCallBackMixin,
amb.CommonAgentManagerRpcCallBackBase):

3
neutron/plugins/ml2/drivers/openvswitch/agent/ovs_neutron_agent.py
View File

@ -2300,8 +2300,7 @@ class OVSNeutronAgent(l2population_rpc.L2populationRpcCallBackTunnelMixin,
# REVISIT (rossella_s) Define a method "reset" in
# BasePollingManager that will be implemented by AlwaysPoll as
# no action and by InterfacePollingMinimizer as start/stop
if isinstance(
polling_manager, polling.InterfacePollingMinimizer):
if isinstance(polling_manager, polling.InterfacePollingMinimizer):
polling_manager.stop()
polling_manager.start()

4
neutron/plugins/ml2/plugin.py
View File

@ -1933,8 +1933,8 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
self._post_delete_port(
context, port, router_ids, bound_mech_contexts)
def _post_delete_port(
self, context, port, router_ids, bound_mech_contexts):
def _post_delete_port(self, context, port, router_ids,
bound_mech_contexts):
kwargs = {
'context': context,
'port': port,

3
neutron/scheduler/dhcp_agent_scheduler.py
View File

@ -68,8 +68,7 @@ class AutoScheduler(object):
segments_on_host = {s.segment_id for s in segment_host_mapping}
for dhcp_agent in dhcp_agents:
if agent_utils.is_agent_down(
dhcp_agent.heartbeat_timestamp):
if agent_utils.is_agent_down(dhcp_agent.heartbeat_timestamp):
LOG.warning('DHCP agent %s is not active', dhcp_agent.id)
continue
for net_id, is_routed_network in net_ids.items():

5
neutron/services/auto_allocate/db.py
View File

@ -307,8 +307,9 @@ class AutoAllocatedTopologyMixin(object):
network_id = network['id'] if network else None
raise exceptions.UnknownProvisioningError(e, network_id=network_id)
def _provision_external_connectivity(
self, context, default_external_network, subnets, tenant_id):
def _provision_external_connectivity(self, context,
default_external_network, subnets,
tenant_id):
"""Uplink tenant subnet(s) to external network."""
router_args = {
'name': 'auto_allocated_router',

4
neutron/services/metering/drivers/iptables/iptables_driver.py
View File

@ -346,8 +346,8 @@ class IptablesMeteringDriver(abstract_driver.MeteringAbstractDriver):
rm.iptables_manager.ipv4['filter'].add_rule(
label_chain, '', wrap=False)
def _process_metering_rule_action_based_on_ns(
self, router, action, ext_dev, im):
def _process_metering_rule_action_based_on_ns(self, router, action,
ext_dev, im):
'''Process metering rule actions based specific namespaces.'''
rm = self.routers.get(router['id'])
with IptablesManagerTransaction(im):

3
neutron/services/segments/exceptions.py
View File

@ -23,8 +23,7 @@ class SegmentNotFound(exceptions.NotFound):
message = _("Segment %(segment_id)s could not be found.")
class NoUpdateSubnetWhenMultipleSegmentsOnNetwork(
exceptions.BadRequest):
class NoUpdateSubnetWhenMultipleSegmentsOnNetwork(exceptions.BadRequest):
message = _("The network '%(network_id)s' has multiple segments, it is "
"only possible to associate an existing subnet with a segment "
"on networks with a single segment.")

3
neutron/services/trunk/drivers/openvswitch/agent/driver.py
View File

@ -75,8 +75,7 @@ class OVSTrunkSkeleton(agent.TrunkSkeleton):
{'event': event_type, 'subports': subports, 'err': e})
@local_registry.receives(resources.TRUNK, [local_events.BEFORE_CREATE])
def check_trunk_dependencies(
self, resource, event, trigger, **kwargs):
def check_trunk_dependencies(self, resource, event, trigger, **kwargs):
# The OVS trunk driver does not work with iptables firewall and QoS.
# We should validate the environment configuration and signal that
# something might be wrong.

4
neutron/services/trunk/rules.py
View File

@ -256,8 +256,8 @@ class SubPortsValidator(object):
if subport['port_id'] == self.trunk_port_id:
raise trunk_exc.ParentPortInUse(port_id=subport['port_id'])
def _raise_subport_invalid_mtu(
self, context, subport, trunk_port_mtu, subport_mtus):
def _raise_subport_invalid_mtu(self, context, subport, trunk_port_mtu,
subport_mtus):
# Check MTU sanity - subport MTU must not exceed trunk MTU.
# If for whatever reason trunk_port_mtu is not available,
# the MTU sanity check cannot be enforced.

4
neutron/tests/common/agents/l2_extensions.py
View File

@ -88,8 +88,8 @@ def wait_until_dscp_marking_rule_applied_ovs(bridge, port_vif, rule):
common_utils.wait_until_true(_dscp_marking_rule_applied)
def wait_until_dscp_marking_rule_applied_linuxbridge(
namespace, port_vif, expected_rule):
def wait_until_dscp_marking_rule_applied_linuxbridge(namespace, port_vif,
expected_rule):
iptables = iptables_manager.IptablesManager(
namespace=namespace)

8
neutron/tests/functional/agent/l3/framework.py
View File

@ -680,8 +680,8 @@ class L3AgentTestFramework(base.BaseSudoTestCase):
self._assert_ip_address_on_interface(namespace, interface,
ip_address)
def _assert_ip_address_not_on_interface(
self, namespace, interface, ip_address):
def _assert_ip_address_not_on_interface(self,
namespace, interface, ip_address):
self.assertNotIn(
ip_address, self._get_addresses_on_device(namespace, interface))
@ -690,8 +690,8 @@ class L3AgentTestFramework(base.BaseSudoTestCase):
self.assertIn(
ip_address, self._get_addresses_on_device(namespace, interface))
def _assert_ping_reply_from_expected_address(
self, ping_result, expected_address):
def _assert_ping_reply_from_expected_address(self, ping_result,
expected_address):
ping_results = ping_result.split('\n')
self.assertGreater(
len(ping_results), 1,

47
neutron/tests/functional/agent/l3/test_dvr_router.py
View File

@ -232,8 +232,8 @@ class TestDvrRouter(framework.L3AgentTestFramework):
self._assert_fip_namespace_deleted(external_gw_port)
@mock.patch.object(dvr_fip_ns.FipNamespace, 'subscribe')
def test_dvr_process_fips_with_no_gw_port_in_namespace(
self, fip_subscribe):
def test_dvr_process_fips_with_no_gw_port_in_namespace(self,
fip_subscribe):
self.agent.conf.agent_mode = 'dvr'
# Create the router with external net
@ -584,8 +584,7 @@ class TestDvrRouter(framework.L3AgentTestFramework):
self.agent._register_router_cls(self.agent.router_factory)
return router
def _get_fip_agent_gw_port_for_router(
self, external_gw_port):
def _get_fip_agent_gw_port_for_router(self, external_gw_port):
# Add fip agent gateway port information to the router_info
if external_gw_port:
# Get values from external gateway port
@ -915,16 +914,16 @@ class TestDvrRouter(framework.L3AgentTestFramework):
self.assertFalse(self._fixed_ip_rule_exists(router_ns, fixed_ip))
self.assertTrue(self._fixed_ip_rule_exists(router_ns, new_fixed_ip))
def _assert_iptables_rules_exist(
self, router_iptables_manager, table_name, expected_rules):
def _assert_iptables_rules_exist(self, router_iptables_manager,
table_name, expected_rules):
rules = router_iptables_manager.get_rules_for_table(table_name)
for rule in expected_rules:
self.assertIn(
str(iptables_manager.IptablesRule(rule[0], rule[1])), rules)
return True
def _assert_iptables_rules_not_exist(
self, router_iptables_manager, table_name, expected_rules):
def _assert_iptables_rules_not_exist(self, router_iptables_manager,
table_name, expected_rules):
rules = router_iptables_manager.get_rules_for_table(table_name)
for rule in expected_rules:
self.assertNotIn(
@ -1092,8 +1091,8 @@ class TestDvrRouter(framework.L3AgentTestFramework):
router.router_id)
self.assertTrue(self._namespace_exists(namespace))
def _get_dvr_snat_namespace_device_status(
self, router, internal_dev_name=None):
def _get_dvr_snat_namespace_device_status(self, router,
internal_dev_name=None):
"""Function returns the internal and external device status."""
snat_ns = dvr_snat_ns.SnatNamespace.get_snat_ns_name(
router.router_id)
@ -1626,9 +1625,9 @@ class TestDvrRouter(framework.L3AgentTestFramework):
self.failover_agent._process_updated_router(router2.router)
self.assertFalse(r2_chsfr.called)
def _setup_dvr_router_static_routes(
self, router_namespace=True,
check_fpr_int_rule_delete=False, enable_ha=False):
def _setup_dvr_router_static_routes(self, router_namespace=True,
check_fpr_int_rule_delete=False,
enable_ha=False):
"""Test to validate the extra routes on dvr routers."""
self.agent.conf.agent_mode = 'dvr_snat'
router_info = self.generate_dvr_router_info(
@ -1745,14 +1744,14 @@ class TestDvrRouter(framework.L3AgentTestFramework):
def test_dvr_ha_rtr_static_routes_in_rtr_namespace(self):
self._setup_dvr_router_static_routes(enable_ha=True)
def test_dvr_router_rule_and_route_table_cleared_when_fip_removed(
self):
def test_dvr_router_rule_and_route_table_cleared_when_fip_removed(self):
self._setup_dvr_router_static_routes(
router_namespace=False, check_fpr_int_rule_delete=True)
def _assert_fip_namespace_interface_static_routes(
self, address_scopes, fpr_device,
router_info, rtr_2_fip, fpr_device_name):
def _assert_fip_namespace_interface_static_routes(self, address_scopes,
fpr_device, router_info,
rtr_2_fip,
fpr_device_name):
fixed_ips_1 = router_info[lib_constants.INTERFACE_KEY][0]['fixed_ips']
fixed_ips_2 = router_info[lib_constants.INTERFACE_KEY][1]['fixed_ips']
actual_routes = fpr_device.route.list_routes(
@ -1790,9 +1789,11 @@ class TestDvrRouter(framework.L3AgentTestFramework):
else:
self.assertEqual([], actual_routes)
def _assert_interface_rules_on_gateway_remove(
self, router, agent, address_scopes, agent_gw_port,
rfp_device, fpr_device, no_external=False):
def _assert_interface_rules_on_gateway_remove(self, router, agent,
address_scopes,
agent_gw_port, rfp_device,
fpr_device,
no_external=False):
router.router[lib_constants.SNAT_ROUTER_INTF_KEY] = []
router.router['gw_port'] = ""
@ -1978,8 +1979,8 @@ class TestDvrRouter(framework.L3AgentTestFramework):
self.assertIsNone(ex_gw_device.route.get_gateway())
self.assertIsNone(fg_device.route.get_gateway())
def _assert_fip_namespace_deleted(
self, ext_gateway_port, assert_ovs_interface=True):
def _assert_fip_namespace_deleted(self, ext_gateway_port,
assert_ovs_interface=True):
ext_net_id = ext_gateway_port['network_id']
fip_ns = self.agent.get_fip_ns(ext_net_id)
fip_ns.unsubscribe = mock.Mock()

4
neutron/tests/functional/services/l3_router/test_l3_dvr_router_plugin.py
View File

@ -369,7 +369,7 @@ class L3DvrTestCase(L3DvrTestCaseBase):
self.l3_plugin.create_floatingip(
self.context, {'floatingip': floating_ip})
if dvr:
if test_agent_mode == (
if (test_agent_mode ==
constants.L3_AGENT_MODE_DVR_NO_EXTERNAL):
if router['ha']:
expected_calls = [
@ -465,7 +465,7 @@ class L3DvrTestCase(L3DvrTestCaseBase):
self.context, floating_ip['id'],
{'floatingip': updated_floating_ip})
if dvr:
if test_agent_mode == (
if (test_agent_mode ==
constants.L3_AGENT_MODE_DVR_NO_EXTERNAL):
if router1['ha'] and router2['ha']:
self.assertEqual(

4
neutron/tests/unit/agent/dhcp/test_agent.py
View File

@ -793,8 +793,8 @@ class TestDhcpAgentEventHandler(base.BaseTestCase):
self.plugin.get_network_info.return_value = fake_network_ipv6_ipv4
self.call_driver.return_value = False
cfg.CONF.set_override('enable_isolated_metadata', True)
with mock.patch.object(
self.dhcp, 'enable_isolated_metadata_proxy') as enable_metadata:
with mock.patch.object(self.dhcp,
'enable_isolated_metadata_proxy') as enable_metadata:
self.dhcp.enable_dhcp_helper(fake_network_ipv6_ipv4.id)
self.plugin.assert_has_calls(
[mock.call.get_network_info(fake_network_ipv6_ipv4.id)])

8
neutron/tests/unit/agent/l2/extensions/test_qos.py
View File

@ -370,8 +370,8 @@ class QosExtensionRpcTestCase(QosExtensionBaseTestCase):
self.assertIsNone(self.qos_ext.policy_map.get_port_policy(port))
def test__handle_notification_ignores_all_event_types_except_updated(self):
with mock.patch.object(
self.qos_ext, '_process_update_policy') as update_mock:
with mock.patch.object(self.qos_ext,
'_process_update_policy') as update_mock:
for event_type in set(events.VALID) - {events.UPDATED}:
self.qos_ext._handle_notification(mock.Mock(), 'QOS',
@ -379,8 +379,8 @@ class QosExtensionRpcTestCase(QosExtensionBaseTestCase):
self.assertFalse(update_mock.called)
def test__handle_notification_passes_update_events(self):
with mock.patch.object(
self.qos_ext, '_process_update_policy') as update_mock:
with mock.patch.object(self.qos_ext,
'_process_update_policy') as update_mock:
policy_obj = mock.Mock()
self.qos_ext._handle_notification(mock.Mock(), 'QOS',

16
neutron/tests/unit/agent/l3/test_agent.py
View File

@ -958,8 +958,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
self.assertIsNotNone(ri.snat_namespace)
self.assertEqual(ri.snat_namespace.name, ri.get_gw_ns_name())
def test_ext_gw_updated_calling_snat_ns_delete_if_gw_port_host_none(
self):
def test_ext_gw_updated_calling_snat_ns_delete_if_gw_port_host_none(self):
"""Test to check the impact of snat_namespace object.
This function specifically checks the impact of the snat
@ -1409,8 +1408,8 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
self.assertEqual(1, ri.fip_ns.create_rtr_2_fip_link.call_count)
@mock.patch.object(lla.LinkLocalAllocator, '_write')
def test_floating_ip_not_configured_if_no_host_or_dest_host(
self, lla_write):
def test_floating_ip_not_configured_if_no_host_or_dest_host(self,
lla_write):
fake_network_id = _uuid()
subnet_id = _uuid()
fake_floatingips = {'floatingips': [
@ -2269,8 +2268,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
def test_process_router_floatingip_disabled(self):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
with mock.patch.object(
agent.plugin_rpc,
with mock.patch.object(agent.plugin_rpc,
'update_floatingip_statuses') as mock_update_fip_status:
fip_id = _uuid()
router = l3_test_common.prepare_router_data(num_internal_ports=1)
@ -2302,8 +2300,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
def test_process_router_floatingip_exception(self):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
with mock.patch.object(
agent.plugin_rpc,
with mock.patch.object(agent.plugin_rpc,
'update_floatingip_statuses') as mock_update_fip_status:
fip_id = _uuid()
router = l3_test_common.prepare_router_data(num_internal_ports=1)
@ -2327,8 +2324,7 @@ class TestBasicRouterOperations(BasicRouterOperationsFramework):
def test_process_external_iptables_exception(self):
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
with mock.patch.object(
agent.plugin_rpc,
with mock.patch.object(agent.plugin_rpc,
'update_floatingip_statuses') as mock_update_fip_status:
fip_id = _uuid()
router = l3_test_common.prepare_router_data(num_internal_ports=1)

8
neutron/tests/unit/agent/l3/test_dvr_fip_ns.py
View File

@ -119,8 +119,8 @@ class TestDvrFipNs(base.BaseTestCase):
@mock.patch.object(ip_lib, 'send_ip_addr_adv_notif')
@mock.patch.object(dvr_fip_ns.FipNamespace, 'subscribe')
@mock.patch.object(dvr_fip_ns.FipNamespace, '_add_default_gateway_for_fip')
def test_update_gateway_port(
self, def_gw, fip_sub, send_adv_notif, IPDevice):
def test_update_gateway_port(self, def_gw, fip_sub, send_adv_notif,
IPDevice):
fip_sub.return_value = False
self.fip_ns._check_for_gateway_ip_change = mock.Mock(return_value=True)
agent_gw_port = self._get_agent_gw_port()
@ -142,8 +142,8 @@ class TestDvrFipNs(base.BaseTestCase):
@mock.patch.object(dvr_fip_ns.FipNamespace, 'subscribe')
@mock.patch.object(dvr_fip_ns.FipNamespace, 'delete')
@mock.patch.object(dvr_fip_ns.FipNamespace, 'unsubscribe')
def test_update_gateway_port_raises_exception(
self, fip_unsub, fip_delete, fip_sub, exists):
def test_update_gateway_port_raises_exception(self, fip_unsub, fip_delete,
fip_sub, exists):
agent_gw_port = self._get_agent_gw_port()
self.fip_ns._create_gateway_port = mock.Mock()
self.fip_ns.create_or_update_gateway_port(agent_gw_port)

3
neutron/tests/unit/agent/l3/test_router_info.py
View File

@ -196,7 +196,8 @@ class TestRouterInfo(base.BaseTestCase):
ri = router_info.RouterInfo(mock.Mock(), _uuid(), {}, **self.ri_kwargs)
ri.router = {'id': _uuid()}
with mock.patch.object(ri, '_process_internal_ports') as p_i_p,\
mock.patch.object(ri, '_process_external_on_delete') as p_e_o_d:
mock.patch.object(ri,
'_process_external_on_delete') as p_e_o_d:
self.mock_ip.netns.exists.return_value = False
ri.process_delete()
self.assertFalse(p_i_p.called)

3
neutron/tests/unit/agent/linux/test_dhcp.py
View File

@ -2739,7 +2739,8 @@ class TestDnsmasq(TestBase):
self.conf, FakeDualNetworkDualDHCP()))
def _test__generate_opts_per_subnet_helper(
self, config_opts, expected_mdt_ip, network_class=FakeNetworkDhcpPort):
self, config_opts, expected_mdt_ip,
network_class=FakeNetworkDhcpPort):
for key, value in config_opts.items():
self.conf.set_override(key, value)
dm = self._get_dnsmasq(network_class())

3
neutron/tests/unit/agent/metadata/test_agent.py
View File

@ -149,8 +149,7 @@ class _TestMetadataProxyHandlerCacheMixin(object):
router_id = 'router-id'
ports = [{'network_id': 'network_id1', 'something': 42}]
expected_networks = ('network_id1',)
with mock.patch(
'oslo_utils.timeutils.utcnow_ts', return_value=0):
with mock.patch('oslo_utils.timeutils.utcnow_ts', return_value=0):
mock_get_ports = self.handler.plugin_rpc.get_ports
mock_get_ports.return_value = ports
networks = self.handler._get_router_networks(router_id)

4
neutron/tests/unit/agent/test_rpc.py
View File

@ -134,8 +134,8 @@ class AgentPluginReportState(base.BaseTestCase):
class AgentRPCMethods(base.BaseTestCase):
def _test_create_consumers(
self, endpoints, method, expected, topics, listen):
def _test_create_consumers(self, endpoints, method, expected, topics,
listen):
with mock.patch.object(n_rpc, 'Connection') as create_connection:
rpc.create_consumers(
endpoints, method, topics, start_listening=listen)

3
neutron/tests/unit/agent/test_securitygroups_rpc.py
View File

@ -966,8 +966,7 @@ class SecurityGroupAgentRpcTestCase(BaseSecurityGroupAgentRpcTestCase):
self.assertFalse(self.agent.firewall.defer_apply.called)
class SecurityGroupAgentEnhancedRpcTestCase(
BaseSecurityGroupAgentRpcTestCase):
class SecurityGroupAgentEnhancedRpcTestCase(BaseSecurityGroupAgentRpcTestCase):
def setUp(self, defer_refresh_firewall=False):
super(SecurityGroupAgentEnhancedRpcTestCase, self).setUp(

5
neutron/tests/unit/api/rpc/agentnotifiers/test_dhcp_rpc_agent_api.py
View File

@ -179,8 +179,9 @@ class TestDhcpAgentNotifyAPI(base.BaseTestCase):
{}, 'foo_network_id')
self.assertEqual(1, self.mock_fanout.call_count)
def _test__notify_agents_with_function(
self, function, expected_scheduling=0, expected_casts=0):
def _test__notify_agents_with_function(self, function,
expected_scheduling=0,
expected_casts=0):
with mock.patch.object(self.notifier, '_schedule_network') as f:
with mock.patch.object(self.notifier, '_get_enabled_agents') as g:
agent = agent_obj.Agent(mock.ANY, id=uuidutils.generate_uuid(),

3
neutron/tests/unit/api/rpc/callbacks/test_resource_manager.py
View File

@ -125,8 +125,7 @@ class ConsumerResourceCallbacksManagerTestCase(
self.mgr.register(callback2, 'TYPE')
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_get_callbacks_fails_on_unregistered_callback(
self, *mocks):
def test_get_callbacks_fails_on_unregistered_callback(self, *mocks):
self.assertRaises(
rpc_exc.CallbackNotFound,
self.mgr.get_callbacks, 'TYPE')

9
neutron/tests/unit/db/test_db_base_plugin_v2.py
View File

@ -1806,8 +1806,7 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
self.assertIn(ips[1]['ip_address'], network_ip_net)
def test_update_port_invalid_fixed_ip_address_v6_slaac(self):
with self.subnet(
cidr='2607:f0d0:1002:51::/64',
with self.subnet(cidr='2607:f0d0:1002:51::/64',
ip_version=constants.IP_VERSION_6,
ipv6_address_mode=constants.IPV6_SLAAC,
gateway_ip=constants.ATTR_NOT_SPECIFIED) as subnet:
@ -3603,8 +3602,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
cidr='10.0.1.0/24') as subnet2:
subnet1_id = subnet1['subnet']['id']
subnet2_id = subnet2['subnet']['id']
with self.port(
subnet=subnet1,
with self.port(subnet=subnet1,
fixed_ips=[{'subnet_id': subnet1_id}]):
req = self.new_delete_request('subnets', subnet2_id)
res = req.get_response(self.api)
@ -4700,7 +4698,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
with self.network() as network:
with self.subnet(network=network), (
mock.patch.object(self.plugin, 'update_port')) as update_port:
mock.patch.object(self.plugin,
'update_port')) as update_port:
# Create port on second network
network2 = self._make_network(self.fmt, 'net2', True)

7
neutron/tests/unit/db/test_l3_db.py
View File

@ -337,14 +337,15 @@ class L3_NAT_db_mixin(base.BaseTestCase):
with mock.patch.object(l3_db.L3_NAT_dbonly_mixin, '_create_router_db',
return_value=router_db) as crd,\
mock.patch.object(l3_db.L3_NAT_dbonly_mixin, '_make_router_dict',
mock.patch.object(l3_db.L3_NAT_dbonly_mixin,
'_make_router_dict',
return_value=router_dict),\
mock.patch.object(l3_db.L3_NAT_dbonly_mixin,
'_update_router_gw_info') as urgi,\
mock.patch.object(l3_db.L3_NAT_dbonly_mixin, '_get_router',
return_value=router_db),\
mock.patch.object(l3_db.L3_NAT_db_mixin, 'notify_router_updated')\
as nru:
mock.patch.object(l3_db.L3_NAT_db_mixin,
'notify_router_updated') as nru:
self.db.create_router(mock.Mock(), router_input)
self.assertTrue(crd.called)

11
neutron/tests/unit/db/test_l3_dvr_db.py
View File

@ -459,8 +459,7 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
self.assertIn(const.FLOATINGIP_KEY, router)
self.assertIn(fip, router[const.FLOATINGIP_KEY])
def _setup_test_create_floatingip(
self, fip, floatingip_db, router_db):
def _setup_test_create_floatingip(self, fip, floatingip_db, router_db):
port = {
'id': '1234',
portbindings.HOST_ID: 'myhost',
@ -975,8 +974,8 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
self.ctx, router['id'],
{'router': {'external_gateway_info':
{'network_id': ext_net_id}}})
with mock.patch.object(
self.mixin, '_add_csnat_router_interface_port') as f:
with mock.patch.object(self.mixin,
'_add_csnat_router_interface_port') as f:
f.side_effect = RuntimeError()
self.assertRaises(
l3_exc.RouterInterfaceAttachmentConflict,
@ -1006,8 +1005,8 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
{'router': {'external_gateway_info':
{'network_id': ext_net_id}}})
net_id = subnet['subnet']['network_id']
with mock.patch.object(
router_obj.RouterPort, 'create') as rtrport_update:
with mock.patch.object(router_obj.RouterPort,
'create') as rtrport_update:
rtrport_update.side_effect = Exception()
self.assertRaises(
l3_exc.RouterInterfaceAttachmentConflict,

3
neutron/tests/unit/db/test_securitygroups_db.py
View File

@ -86,8 +86,7 @@ class SecurityGroupDbMixinTestCase(testlib_api.SqlTestCase):
mock.patch.object(self.mixin, '_get_security_group'),\
mock.patch.object(registry, "notify") as mock_notify:
mock_notify.side_effect = exceptions.CallbackFailure(Exception())
with testtools.ExpectedException(
securitygroup.SecurityGroupInUse):
with testtools.ExpectedException(securitygroup.SecurityGroupInUse):
self.mixin.delete_security_group(self.ctx, mock.ANY)
def test_update_security_group_conflict(self):

6
neutron/tests/unit/extensions/test_dns.py
View File

@ -350,8 +350,7 @@ class DnsExtensionTestCase(test_plugin.Ml2PluginV2TestCase):
res = self._test_create_port_with_multiple_ipv4_and_ipv6_subnets()
self.assertEqual(201, res.status_code)
def test_create_port_multiple_v4_v6_subnets_bad_fqdn_and_dns_domain(
self):
def test_create_port_multiple_v4_v6_subnets_bad_fqdn_and_dns_domain(self):
cfg.CONF.set_override('dns_domain', 'example.com')
res = self._test_create_port_with_multiple_ipv4_and_ipv6_subnets(
dns_name='vm1.bad-domain.com.')
@ -361,8 +360,7 @@ class DnsExtensionTestCase(test_plugin.Ml2PluginV2TestCase):
'neutron.conf')
self.assertIn(expected_error, res.text)
def test_create_port_multiple_v4_v6_subnets_bad_pqdn_and_dns_domain(
self):
def test_create_port_multiple_v4_v6_subnets_bad_pqdn_and_dns_domain(self):
cfg.CONF.set_override('dns_domain', 'example.com')
num_labels = int(
math.floor(db_const.FQDN_FIELD_SIZE / constants.DNS_LABEL_MAX_LEN))

12
neutron/tests/unit/extensions/test_l3.py
View File

@ -567,9 +567,9 @@ class L3NatTestCaseMixin(object):
floatingip['floatingip']['id'])
@contextlib.contextmanager
def floatingip_no_assoc_with_public_sub(
self, private_sub, fmt=None, set_context=False,
public_sub=None, flavor_id=None, **kwargs):
def floatingip_no_assoc_with_public_sub(self, private_sub, fmt=None,
set_context=False, public_sub=None,
flavor_id=None, **kwargs):
self._set_net_external(public_sub['subnet']['network_id'])
args_list = {}
if flavor_id:
@ -920,7 +920,8 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
def test_router_update_gateway_add_multiple_prefixes_ipv6(self):
with self.network() as n:
with self.subnet(network=n) as s1, \
self.subnet(network=n, ip_version=lib_constants.IP_VERSION_6,
self.subnet(network=n,
ip_version=lib_constants.IP_VERSION_6,
cidr='2001:db8::/32') \
as s2, (self.router()) as r:
self._set_net_external(n['network']['id'])
@ -1421,8 +1422,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
p['port']['id'])
def test_update_router_interface_port_ip_not_allowed(self):
with self.router() as r, (
self.port()) as p:
with self.router() as r, self.port() as p:
body = self._router_interface_action('add',
r['router']['id'],
None,

4
neutron/tests/unit/extensions/test_l3_ext_gw_mode.py
View File

@ -464,8 +464,8 @@ class ExtGwModeIntTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
'external_fixed_ips': [{
'ip_address': mock.ANY,
'subnet_id': s['subnet']['id']}]})]
with self.router(
name=name, admin_state_up=True, tenant_id=tenant_id,
with self.router(name=name, admin_state_up=True,
tenant_id=tenant_id,
external_gateway_info=input_value) as router:
res = self._show('routers', router['router']['id'])
for k, v in expected_value:

3
neutron/tests/unit/extensions/test_portsecurity.py
View File

@ -35,8 +35,7 @@ DB_PLUGIN_KLASS = ('neutron.tests.unit.extensions.test_portsecurity.'
'PortSecurityTestPlugin')
class PortSecurityTestCase(
test_securitygroup.SecurityGroupsTestCase,
class PortSecurityTestCase(test_securitygroup.SecurityGroupsTestCase,
test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
def setUp(self, plugin=None):

9
neutron/tests/unit/extensions/test_subnet_onboard.py
View File

@ -112,12 +112,10 @@ class SubnetOnboardTestsBase(object):
with self.subnetpool(self.ip_version,
prefixes=[self.cidr_to_onboard],
address_scope_id=addr_scope['id']) as source:
with self.subnetpool(
self.ip_version,
with self.subnetpool(self.ip_version,
address_scope_id=addr_scope['id'],
prefixes=self.subnetpool_prefixes) as target:
with self.subnet(
cidr=self.cidr_to_onboard,
with self.subnet(cidr=self.cidr_to_onboard,
ip_version=self.ip_version) as subnet_to_onboard:
subnet_to_onboard = subnet_to_onboard['subnet']
@ -139,8 +137,7 @@ class SubnetOnboardTestsBase(object):
prefixes=self.subnetpool_prefixes) as source:
with self.subnetpool(self.ip_version,
prefixes=self.subnetpool_prefixes) as target:
with self.subnet(
cidr=self.cidr_to_onboard,
with self.subnet(cidr=self.cidr_to_onboard,
ip_version=self.ip_version) as subnet_to_onboard:
subnet_to_onboard = subnet_to_onboard['subnet']

8
neutron/tests/unit/extensions/test_subnetpool_prefix_ops.py
View File

@ -166,9 +166,7 @@ class SubnetpoolPrefixOpsTestBase(object):
default_prefixlen=self.default_prefixlen,
min_prefixlen=self.default_prefixlen,
prefixes=self.subnetpool_prefixes) as subnetpool:
with self.subnet(
cidr=None,
subnetpool_id=subnetpool['id'],
with self.subnet(cidr=None, subnetpool_id=subnetpool['id'],
ip_version=self.ip_version) as subnet:
subnet = subnet['subnet']
prefixes_to_remove = [subnet['cidr']]
@ -185,9 +183,7 @@ class SubnetpoolPrefixOpsTestBase(object):
default_prefixlen=self.default_prefixlen,
min_prefixlen=self.default_prefixlen,
prefixes=[self.subnetpool_prefixes[0]]) as subnetpool:
with self.subnet(
cidr=None,
subnetpool_id=subnetpool['id'],
with self.subnet(cidr=None, subnetpool_id=subnetpool['id'],
ip_version=self.ip_version) as subnet:
subnet = subnet['subnet']
prefixes_to_remove = [self.overlapping_cidr]

3
neutron/tests/unit/notifiers/test_nova.py
View File

@ -183,8 +183,7 @@ class TestNovaNotify(base.BaseTestCase):
def test_delete_floatingip_deleted_port_no_notify(self):
port_id = 'bee50827-bcee-4cc8-91c1-a27b0ce54222'
with mock.patch.object(
directory.get_plugin(), 'get_port',
with mock.patch.object(directory.get_plugin(), 'get_port',
side_effect=n_exc.PortNotFound(port_id=port_id)):
returned_obj = {'floatingip':
{'port_id': port_id}}

3
neutron/tests/unit/objects/port/extensions/test_uplink_status_propagation.py
View File

@ -22,8 +22,7 @@ class UplinkStatusPropagationIfaceObjectTestCase(
class UplinkStatusPropagationDbObjectTestCase(
obj_test_base.BaseDbObjectTestCase,
testlib_api.SqlTestCase):
obj_test_base.BaseDbObjectTestCase, testlib_api.SqlTestCase):
_test_class = uplink_status_propagation.PortUplinkStatusPropagation

3
neutron/tests/unit/objects/qos/test_policy.py
View File

@ -120,8 +120,7 @@ class QosPolicyObjectTestCase(test_base.BaseObjectIfaceTestCase):
[test_base.get_obj_persistent_fields(obj) for obj in objs])
def test_get_objects_valid_fields(self):
with mock.patch.object(
db_api, 'get_objects',
with mock.patch.object(db_api, 'get_objects',
return_value=[self.db_objs[0]]) as get_objects_mock:
objs = self._test_class.get_objects(
self.context,

3
neutron/tests/unit/objects/test_base.py
View File

@ -1251,8 +1251,7 @@ class BaseObjectIfaceTestCase(_BaseObjectTestCase, test_base.BaseTestCase):
with mock.patch.object(base.NeutronDbObject,
'_get_changed_persistent_fields',
return_value=fields_to_update):
with mock.patch.object(
obj_db_api, 'get_objects',
with mock.patch.object(obj_db_api, 'get_objects',
side_effect=self.fake_get_objects):
obj.update()
self._check_equal(self.objs[0], obj)

6
neutron/tests/unit/objects/test_quota.py
View File

@ -21,8 +21,7 @@ from neutron.tests.unit.objects import test_base as obj_test_base
from neutron.tests.unit import testlib_api
class ResourceDeltaObjectIfaceTestCase(
obj_test_base.BaseObjectIfaceTestCase):
class ResourceDeltaObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
_test_class = quota.ResourceDelta
@ -42,8 +41,7 @@ class ResourceDeltaDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
self._reservation.create()
class ReservationObjectIfaceTestCase(
obj_test_base.BaseObjectIfaceTestCase):
class ReservationObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
_test_class = quota.Reservation

3
neutron/tests/unit/objects/test_router.py
View File

@ -19,8 +19,7 @@ from neutron.tests.unit.objects import test_base as obj_test_base
from neutron.tests.unit import testlib_api
class RouterRouteIfaceObjectTestCase(
obj_test_base.BaseObjectIfaceTestCase):
class RouterRouteIfaceObjectTestCase(obj_test_base.BaseObjectIfaceTestCase):
_test_class = router.RouterRoute

3
neutron/tests/unit/objects/test_servicetype.py
View File

@ -25,7 +25,6 @@ class ProviderResourceAssociationIfaceObjectTestCase(
class ProviderResourceAssociationDbObjectTestCase(
obj_test_base.BaseDbObjectTestCase,
testlib_api.SqlTestCase):
obj_test_base.BaseDbObjectTestCase, testlib_api.SqlTestCase):
_test_class = servicetype.ProviderResourceAssociation

3
neutron/tests/unit/plugins/ml2/drivers/agent/test__common_agent.py
View File

@ -64,7 +64,8 @@ class TestCommonAgentLoop(base.BaseTestCase):
manager.get_agent_configurations.return_value = {}
manager.get_rpc_consumers.return_value = []
with mock.patch.object(ca.CommonAgentLoop, '_validate_manager_class'),\
mock.patch.object(ca.CommonAgentLoop, '_validate_rpc_endpoints'):
mock.patch.object(ca.CommonAgentLoop,
'_validate_rpc_endpoints'):
self.agent = ca.CommonAgentLoop(manager, 0, 10, 'fake_agent',
'foo-binary')
with mock.patch.object(self.agent, "daemon_loop"):

8
neutron/tests/unit/plugins/ml2/drivers/l2pop/rpc_manager/test_l2population_rpc.py
View File

@ -127,8 +127,8 @@ class TestL2populationRpcCallBackTunnelMixin(
sorted(mock_add_fdb_flow.call_args_list))
def test_fdb_remove_tun(self):
with mock.patch.object(
self.fakeagent, 'del_fdb_flow') as mock_del_fdb_flow:
with mock.patch.object(self.fakeagent,
'del_fdb_flow') as mock_del_fdb_flow:
self.fakeagent.fdb_remove_tun('context', self.fakebr, self.lvm0,
self.agent_ports,
self._tunnel_port_lookup)
@ -168,8 +168,8 @@ class TestL2populationRpcCallBackTunnelMixin(
def test_fdb_remove_tun_non_existence_key_in_ofports(self):
del self.ofports[self.type_gre][self.ports[1].ip]
with mock.patch.object(
self.fakeagent, 'del_fdb_flow') as mock_del_fdb_flow:
with mock.patch.object(self.fakeagent,
'del_fdb_flow') as mock_del_fdb_flow:
self.fakeagent.fdb_remove_tun('context', self.fakebr, self.lvm0,
self.agent_ports,
self._tunnel_port_lookup)

16
neutron/tests/unit/plugins/ml2/drivers/linuxbridge/agent/extension_drivers/test_qos_driver.py
View File

@ -200,8 +200,8 @@ class QosLinuxbridgeAgentDriverTestCase(base.BaseTestCase):
tag=self._dscp_rule_tag(self.port['device'])
)
]
with mock.patch.object(
self.qos_driver, "iptables_manager") as iptables_manager:
with mock.patch.object(self.qos_driver,
"iptables_manager") as iptables_manager:
iptables_manager.ip4['mangle'] = mock.Mock()
iptables_manager.ip6['mangle'] = mock.Mock()
@ -225,8 +225,8 @@ class QosLinuxbridgeAgentDriverTestCase(base.BaseTestCase):
tag=self._dscp_rule_tag(self.port['device'])
)
]
with mock.patch.object(
self.qos_driver, "iptables_manager") as iptables_manager:
with mock.patch.object(self.qos_driver,
"iptables_manager") as iptables_manager:
iptables_manager.ip4['mangle'] = mock.Mock()
iptables_manager.ip6['mangle'] = mock.Mock()
@ -243,8 +243,8 @@ class QosLinuxbridgeAgentDriverTestCase(base.BaseTestCase):
mock.call.remove_chain(
dscp_chain_name),
]
with mock.patch.object(
self.qos_driver, "iptables_manager") as iptables_manager:
with mock.patch.object(self.qos_driver,
"iptables_manager") as iptables_manager:
iptables_manager.ip4['mangle'] = mock.Mock()
iptables_manager.ip6['mangle'] = mock.Mock()
@ -265,8 +265,8 @@ class QosLinuxbridgeAgentDriverTestCase(base.BaseTestCase):
mock.call.clear_rules_by_tag(
self._dscp_rule_tag(self.port['device'])),
]
with mock.patch.object(
self.qos_driver, "iptables_manager") as iptables_manager:
with mock.patch.object(self.qos_driver,
"iptables_manager") as iptables_manager:
iptables_manager.ip4['mangle'] = mock.Mock()
iptables_manager.ip6['mangle'] = mock.Mock()

3
neutron/tests/unit/plugins/ml2/drivers/linuxbridge/agent/test_linuxbridge_neutron_agent.py
View File

@ -452,8 +452,7 @@ class TestLinuxBridgeManager(base.BaseTestCase):
with mock.patch.object(vxlan_dev, 'disable_ipv6') as dv6_fn,\
mock.patch.object(self.lbm.ip, 'add_vxlan',
return_value=vxlan_dev) as add_vxlan_fn,\
mock.patch.object(
vxlan_dev.link, 'set_mtu',
mock.patch.object(vxlan_dev.link, 'set_mtu',
side_effect=ip_lib.InvalidArgument(
parameter="MTU", value=mtu)),\
mock.patch.object(ip_lib, 'get_device_mtu',

4
neutron/tests/unit/plugins/ml2/drivers/macvtap/agent/test_macvtap_neutron_agent.py
View File

@ -116,8 +116,8 @@ class TestMacvtapManager(base.BaseTestCase):
listing = ['foo', 'macvtap0', 'macvtap1', 'bar']
# set some mac mappings to make sure they are cleaned up
self.mgr.mac_device_name_mappings = {'foo': 'bar'}
with mock.patch.object(os, 'listdir', return_value=listing)\
as mock_ld,\
with mock.patch.object(os, 'listdir',
return_value=listing) as mock_ld,\
mock.patch.object(ip_lib, 'get_device_mac') as mock_gdn:
mock_gdn.side_effect = ['mac0', 'mac1']

7
neutron/tests/unit/plugins/ml2/drivers/mech_sriov/agent/test_eswitch_manager.py
View File

@ -707,8 +707,7 @@ class TestPciOsWrapper(base.BaseTestCase):
@mock.patch("os.listdir", return_value=["eth0", "eth1"])
@mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
"PciDeviceIPWrapper.is_macvtap_assigned", return_value=True)
def test_is_assigned_vf_macvtap(
self, mock_is_macvtap_assigned, *args):
def test_is_assigned_vf_macvtap(self, mock_is_macvtap_assigned, *args):
esm.PciOsWrapper.is_assigned_vf(self.DEV_NAME, self.VF_INDEX, '')
mock_is_macvtap_assigned.called_with(self.VF_INDEX, "eth0")
@ -716,8 +715,8 @@ class TestPciOsWrapper(base.BaseTestCase):
@mock.patch("os.listdir", side_effect=OSError())
@mock.patch("neutron.plugins.ml2.drivers.mech_sriov.agent.pci_lib."
"PciDeviceIPWrapper.is_macvtap_assigned")
def test_is_assigned_vf_macvtap_failure(
self, mock_is_macvtap_assigned, *args):
def test_is_assigned_vf_macvtap_failure(self, mock_is_macvtap_assigned,
*args):
esm.PciOsWrapper.is_assigned_vf(self.DEV_NAME, self.VF_INDEX, '')
self.assertFalse(mock_is_macvtap_assigned.called)

4
neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/test_ovs_neutron_agent.py
View File

@ -2206,8 +2206,8 @@ class TestOvsNeutronAgent(object):
self.assertFalse(cleanup.called)
def test_set_rpc_timeout(self):
with mock.patch.object(
n_rpc.BackingOffClient, 'set_max_timeout') as smt:
with mock.patch.object(n_rpc.BackingOffClient,
'set_max_timeout') as smt:
self.agent._handle_sigterm(None, None)
for rpc_client in (self.agent.plugin_rpc.client,
self.agent.sg_plugin_rpc.client,

10
neutron/tests/unit/plugins/ml2/test_plugin.py
View File

@ -2414,8 +2414,8 @@ class TestMl2PortBinding(Ml2PluginV2TestCase,
'id': 'foo_port_id',
portbindings.HOST_ID: 'foo_host',
}
with mock.patch.object(
ml2_db, 'ensure_distributed_port_binding') as mock_dist:
with mock.patch.object(ml2_db,
'ensure_distributed_port_binding') as mock_dist:
plugin.update_distributed_port_binding(
self.context, 'foo_port_id', {'port': port})
self.assertFalse(mock_dist.called)
@ -3288,8 +3288,7 @@ class TestML2Segments(Ml2PluginV2TestCase):
def test_create_network_mtu_on_precommit(self):
with mock.patch.object(mech_test.TestMechanismDriver,
'create_network_precommit') as bmp:
with mock.patch.object(
self.driver, '_get_network_mtu') as mtu:
with mock.patch.object(self.driver, '_get_network_mtu') as mtu:
mtu.return_value = 1100
with self.network() as network:
self.assertIn('mtu', network['network'])
@ -3310,8 +3309,7 @@ class TestML2Segments(Ml2PluginV2TestCase):
def test_reserve_segment_update_network_mtu(self):
with self.network() as network:
network_id = network['network']['id']
with mock.patch.object(
self.driver, '_get_network_mtu') as mtu:
with mock.patch.object(self.driver, '_get_network_mtu') as mtu:
mtu.return_value = 100
self._reserve_segment(network)
updated_network = self.driver.get_network(self.context,

3
neutron/tests/unit/plugins/ml2/test_security_group.py
View File

@ -163,7 +163,6 @@ class TestMl2SecurityGroups(Ml2SecurityGroupsTestCase,
self.assertFalse(self.was_active)
class TestMl2SGServerRpcCallBack(
Ml2SecurityGroupsTestCase,
class TestMl2SGServerRpcCallBack(Ml2SecurityGroupsTestCase,
test_sg_rpc.SGServerRpcCallBackTestCase):
pass

6
neutron/tests/unit/scheduler/test_dhcp_agent_scheduler.py
View File

@ -525,8 +525,7 @@ class TestNetworksFailover(TestDhcpSchedulerBaseTestCase,
def test_reschedule_network_from_down_agent_failed_on_unexpected(self):
agents = self._create_and_set_agents_down(['host-a'], 1)
self._test_schedule_bind_network([agents[0]], self.network_id)
with mock.patch.object(
self, '_filter_bindings',
with mock.patch.object(self, '_filter_bindings',
side_effect=Exception()):
# just make sure that no exception is raised
self.remove_networks_from_down_agents()
@ -542,8 +541,7 @@ class TestNetworksFailover(TestDhcpSchedulerBaseTestCase,
def test_reschedule_doesnt_occur_if_no_agents(self):
agents = self._create_and_set_agents_down(['host-a', 'host-b'], 2)
self._test_schedule_bind_network([agents[0]], self.network_id)
with mock.patch.object(
self, 'remove_network_from_dhcp_agent') as rn:
with mock.patch.object(self, 'remove_network_from_dhcp_agent') as rn:
self.remove_networks_from_down_agents()
self.assertFalse(rn.called)

3
neutron/tests/unit/tests/common/test_net_helpers.py
View File

@ -56,8 +56,7 @@ class PortAllocationTestCase(base.DietTestCase):
for p in range(1024, 32767):
ss_output2 += ss_output_template % p
with mock.patch('neutron.agent.linux.ip_lib.IPWrapper') \
as ipwrapper, \
with mock.patch('neutron.agent.linux.ip_lib.IPWrapper') as ipwrapper, \
mock.patch('neutron.agent.linux.utils.execute') as ex:
m = mock.MagicMock()
m.netns.execute.return_value = ss_output2

3
tox.ini
View File

@ -156,7 +156,6 @@ deps = -r{toxinidir}/doc/requirements.txt
commands = sphinx-build -W -b linkcheck doc/source doc/build/linkcheck
[flake8]
# E125 continuation line does not distinguish itself from next logical line
# E126 continuation line over-indented for hanging indent
# E128 continuation line under-indented for visual indent
# H405 multi line docstring summary not separated with an empty line
@ -164,7 +163,7 @@ commands = sphinx-build -W -b linkcheck doc/source doc/build/linkcheck
# TODO(amotoki) check the following new rules should be fixed or ignored
# E731 do not assign a lambda expression, use a def
# W504 line break after binary operator
ignore = E125,E126,E128,E731,H405,N530,W504
ignore = E126,E128,E731,H405,N530,W504
# H106: Don't put vim configuration in source files
# H203: Use assertIs(Not)None to check for None
# H204: Use assert(Not)Equal to check for equality