From 422588e13338dab4a5ba1973c96256690ba4adf5 Mon Sep 17 00:00:00 2001 From: =?utf8?q?Adrien=20Verg=C3=A9?= Date: Fri, 29 May 2015 22:54:33 +0200 Subject: [PATCH] Get completely rid of contextlib.nested `contextlib.nested` is deprecated since Python 2.7 and incompatible with Python 3. This patch removes all its occurences by using the helper script at [1]. This is a necessary step to allow us running all unit tests with Python 3 (not just a small subset as it is done now). [1]: https://github.com/adrienverge/context_unnester Change-Id: I8d1de09ff38ed0af9fb56f423a2c43476408e0fb Blueprint: neutron-python3 Closes-Bug: #1428424 --- neutron/hacking/checks.py | 13 - .../unit/db/metering/test_metering_db.py | 60 +- .../tests/unit/db/test_db_base_plugin_v2.py | 370 ++++----- neutron/tests/unit/db/test_l3_dvr_db.py | 123 ++- .../tests/unit/db/test_securitygroups_db.py | 10 +- .../unit/extensions/test_external_net.py | 4 +- .../tests/unit/extensions/test_extraroute.py | 34 +- neutron/tests/unit/extensions/test_l3.py | 150 ++-- .../unit/extensions/test_securitygroup.py | 141 ++-- .../unit/plugins/ibm/test_sdnve_agent.py | 15 +- .../unit/plugins/ibm/test_sdnve_plugin.py | 23 +- .../agent/test_linuxbridge_neutron_agent.py | 261 +++--- .../plugins/ml2/drivers/base_type_tunnel.py | 17 +- .../ml2/drivers/l2pop/test_mech_driver.py | 22 +- .../plugins/ml2/test_extension_driver_api.py | 37 +- neutron/tests/unit/plugins/ml2/test_plugin.py | 115 ++- neutron/tests/unit/plugins/ml2/test_rpc.py | 9 +- .../unit/plugins/ml2/test_security_group.py | 18 +- .../plugins/oneconvergence/test_nvsd_agent.py | 43 +- .../oneconvergence/test_nvsd_plugin.py | 3 +- .../agent/test_ovs_neutron_agent.py | 753 +++++++++--------- .../openvswitch/test_agent_scheduler.py | 114 ++- .../plugins/openvswitch/test_ovs_tunnel.py | 26 +- .../sriovnicagent/test_eswitch_manager.py | 170 ++-- .../scheduler/test_dhcp_agent_scheduler.py | 36 +- .../unit/scheduler/test_l3_agent_scheduler.py | 277 +++---- 26 files changed, 1276 insertions(+), 1568 deletions(-) diff --git a/neutron/hacking/checks.py b/neutron/hacking/checks.py index 3a7d8df48..c6160072d 100644 --- a/neutron/hacking/checks.py +++ b/neutron/hacking/checks.py @@ -149,19 +149,6 @@ def check_no_contextlib_nested(logical_line, filename): "docs.python.org/2/library/contextlib.html#contextlib.nested for " "more information.") - # TODO(ankit): The following check is temporary. - # A series of patches will be submitted to address - # these issues. It should be removed completely - # when bug 1428424 is closed. - ignore_dirs = [ - "neutron/tests/unit/db", - "neutron/tests/unit/extensions", - "neutron/tests/unit/plugins", - "neutron/tests/unit/scheduler"] - for directory in ignore_dirs: - if directory in filename: - return - if contextlib_nested.match(logical_line): yield(0, msg) diff --git a/neutron/tests/unit/db/metering/test_metering_db.py b/neutron/tests/unit/db/metering/test_metering_db.py index bd513696b..c9e185163 100644 --- a/neutron/tests/unit/db/metering/test_metering_db.py +++ b/neutron/tests/unit/db/metering/test_metering_db.py @@ -167,9 +167,9 @@ class TestMetering(MeteringPluginDbTestCase): name = 'my label' description = 'my metering label' - with contextlib.nested( - self.metering_label(name, description), - self.metering_label(name, description)) as metering_label: + with self.metering_label(name, description) as v1,\ + self.metering_label(name, description) as v2: + metering_label = (v1, v2) self._test_list_resources('metering-label', metering_label) @@ -224,15 +224,15 @@ class TestMetering(MeteringPluginDbTestCase): remote_ip_prefix = '192.168.0.0/24' excluded = True - with contextlib.nested( - self.metering_label_rule(metering_label_id, - direction, - remote_ip_prefix, - excluded), - self.metering_label_rule(metering_label_id, - 'ingress', - remote_ip_prefix, - excluded)) as metering_label_rule: + with self.metering_label_rule(metering_label_id, + direction, + remote_ip_prefix, + excluded) as v1,\ + self.metering_label_rule(metering_label_id, + 'ingress', + remote_ip_prefix, + excluded) as v2: + metering_label_rule = (v1, v2) self._test_list_resources('metering-label-rule', metering_label_rule) @@ -248,15 +248,15 @@ class TestMetering(MeteringPluginDbTestCase): remote_ip_prefix = '192.168.0.0/24' excluded = True - with contextlib.nested( - self.metering_label_rule(metering_label_id, - direction, - remote_ip_prefix, - excluded), - self.metering_label_rule(metering_label_id, - direction, - n_consts.IPv4_ANY, - False)) as metering_label_rule: + with self.metering_label_rule(metering_label_id, + direction, + remote_ip_prefix, + excluded) as v1,\ + self.metering_label_rule(metering_label_id, + direction, + n_consts.IPv4_ANY, + False) as v2: + metering_label_rule = (v1, v2) self._test_list_resources('metering-label-rule', metering_label_rule) @@ -299,15 +299,15 @@ class TestMetering(MeteringPluginDbTestCase): remote_ip_prefix = '192.168.0.0/24' excluded = True - with contextlib.nested( - self.metering_label_rule(metering_label_id1, - direction, - remote_ip_prefix, - excluded), - self.metering_label_rule(metering_label_id2, - direction, - remote_ip_prefix, - excluded)) as metering_label_rule: + with self.metering_label_rule(metering_label_id1, + direction, + remote_ip_prefix, + excluded) as v1,\ + self.metering_label_rule(metering_label_id2, + direction, + remote_ip_prefix, + excluded) as v2: + metering_label_rule = (v1, v2) self._test_list_resources('metering-label-rule', metering_label_rule) diff --git a/neutron/tests/unit/db/test_db_base_plugin_v2.py b/neutron/tests/unit/db/test_db_base_plugin_v2.py index f34beb85a..21989c0bf 100644 --- a/neutron/tests/unit/db/test_db_base_plugin_v2.py +++ b/neutron/tests/unit/db/test_db_base_plugin_v2.py @@ -1040,15 +1040,14 @@ class TestPortsV2(NeutronDbPluginV2TestCase): def test_list_ports(self): # for this test we need to enable overlapping ips cfg.CONF.set_default('allow_overlapping_ips', True) - with contextlib.nested(self.port(), - self.port(), - self.port()) as ports: + with self.port() as v1, self.port() as v2, self.port() as v3: + ports = (v1, v2, v3) self._test_list_resources('port', ports) def test_list_ports_filtered_by_fixed_ip(self): # for this test we need to enable overlapping ips cfg.CONF.set_default('allow_overlapping_ips', True) - with contextlib.nested(self.port(), self.port()) as (port1, port2): + with self.port() as port1, self.port(): fixed_ips = port1['port']['fixed_ips'][0] query_params = """ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s @@ -1061,9 +1060,8 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s def test_list_ports_public_network(self): with self.network(shared=True) as network: with self.subnet(network) as subnet: - with contextlib.nested(self.port(subnet, tenant_id='tenant_1'), - self.port(subnet, tenant_id='tenant_2') - ) as (port1, port2): + with self.port(subnet, tenant_id='tenant_1') as port1,\ + self.port(subnet, tenant_id='tenant_2') as port2: # Admin request - must return both ports self._test_list_resources('port', [port1, port2]) # Tenant_1 request - must return single port @@ -1079,13 +1077,12 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s if self._skip_native_sorting: self.skipTest("Skip test for not implemented sorting feature") cfg.CONF.set_default('allow_overlapping_ips', True) - with contextlib.nested(self.port(admin_state_up='True', - mac_address='00:00:00:00:00:01'), - self.port(admin_state_up='False', - mac_address='00:00:00:00:00:02'), - self.port(admin_state_up='False', - mac_address='00:00:00:00:00:03') - ) as (port1, port2, port3): + with self.port(admin_state_up='True', + mac_address='00:00:00:00:00:01') as port1,\ + self.port(admin_state_up='False', + mac_address='00:00:00:00:00:02') as port2,\ + self.port(admin_state_up='False', + mac_address='00:00:00:00:00:03') as port3: self._test_list_with_sort('port', (port3, port2, port1), [('admin_state_up', 'asc'), ('mac_address', 'desc')]) @@ -1096,13 +1093,12 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s new=_fake_get_sorting_helper) helper_patcher.start() cfg.CONF.set_default('allow_overlapping_ips', True) - with contextlib.nested(self.port(admin_state_up='True', - mac_address='00:00:00:00:00:01'), - self.port(admin_state_up='False', - mac_address='00:00:00:00:00:02'), - self.port(admin_state_up='False', - mac_address='00:00:00:00:00:03') - ) as (port1, port2, port3): + with self.port(admin_state_up='True', + mac_address='00:00:00:00:00:01') as port1,\ + self.port(admin_state_up='False', + mac_address='00:00:00:00:00:02') as port2,\ + self.port(admin_state_up='False', + mac_address='00:00:00:00:00:03') as port3: self._test_list_with_sort('port', (port3, port2, port1), [('admin_state_up', 'asc'), ('mac_address', 'desc')]) @@ -1111,10 +1107,9 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s if self._skip_native_pagination: self.skipTest("Skip test for not implemented pagination feature") cfg.CONF.set_default('allow_overlapping_ips', True) - with contextlib.nested(self.port(mac_address='00:00:00:00:00:01'), - self.port(mac_address='00:00:00:00:00:02'), - self.port(mac_address='00:00:00:00:00:03') - ) as (port1, port2, port3): + with self.port(mac_address='00:00:00:00:00:01') as port1,\ + self.port(mac_address='00:00:00:00:00:02') as port2,\ + self.port(mac_address='00:00:00:00:00:03') as port3: self._test_list_with_pagination('port', (port1, port2, port3), ('mac_address', 'asc'), 2, 2) @@ -1125,10 +1120,9 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s new=_fake_get_pagination_helper) helper_patcher.start() cfg.CONF.set_default('allow_overlapping_ips', True) - with contextlib.nested(self.port(mac_address='00:00:00:00:00:01'), - self.port(mac_address='00:00:00:00:00:02'), - self.port(mac_address='00:00:00:00:00:03') - ) as (port1, port2, port3): + with self.port(mac_address='00:00:00:00:00:01') as port1,\ + self.port(mac_address='00:00:00:00:00:02') as port2,\ + self.port(mac_address='00:00:00:00:00:03') as port3: self._test_list_with_pagination('port', (port1, port2, port3), ('mac_address', 'asc'), 2, 2) @@ -1137,10 +1131,9 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s if self._skip_native_pagination: self.skipTest("Skip test for not implemented pagination feature") cfg.CONF.set_default('allow_overlapping_ips', True) - with contextlib.nested(self.port(mac_address='00:00:00:00:00:01'), - self.port(mac_address='00:00:00:00:00:02'), - self.port(mac_address='00:00:00:00:00:03') - ) as (port1, port2, port3): + with self.port(mac_address='00:00:00:00:00:01') as port1,\ + self.port(mac_address='00:00:00:00:00:02') as port2,\ + self.port(mac_address='00:00:00:00:00:03') as port3: self._test_list_with_pagination_reverse('port', (port1, port2, port3), ('mac_address', 'asc'), @@ -1152,10 +1145,9 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s new=_fake_get_pagination_helper) helper_patcher.start() cfg.CONF.set_default('allow_overlapping_ips', True) - with contextlib.nested(self.port(mac_address='00:00:00:00:00:01'), - self.port(mac_address='00:00:00:00:00:02'), - self.port(mac_address='00:00:00:00:00:03') - ) as (port1, port2, port3): + with self.port(mac_address='00:00:00:00:00:01') as port1,\ + self.port(mac_address='00:00:00:00:00:02') as port2,\ + self.port(mac_address='00:00:00:00:00:03') as port3: self._test_list_with_pagination_reverse('port', (port1, port2, port3), ('mac_address', 'asc'), @@ -1662,14 +1654,13 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s def test_requested_subnet_id_v4_and_v6_slaac(self): with self.network() as network: - with contextlib.nested( - self.subnet(network), - self.subnet(network, - cidr='2607:f0d0:1002:51::/64', - ip_version=6, - gateway_ip='fe80::1', - ipv6_address_mode=constants.IPV6_SLAAC) - ) as (subnet, subnet2): + with self.subnet(network) as subnet,\ + self.subnet( + network, + cidr='2607:f0d0:1002:51::/64', + ip_version=6, + gateway_ip='fe80::1', + ipv6_address_mode=constants.IPV6_SLAAC) as subnet2: with self.port( subnet, fixed_ips=[{'subnet_id': subnet['subnet']['id']}, @@ -1689,13 +1680,12 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s def test_create_router_port_ipv4_and_ipv6_slaac_no_fixed_ips(self): with self.network() as network: # Create an IPv4 and an IPv6 SLAAC subnet on the network - with contextlib.nested( - self.subnet(network), - self.subnet(network, - cidr='2607:f0d0:1002:51::/64', - ip_version=6, - gateway_ip='fe80::1', - ipv6_address_mode=constants.IPV6_SLAAC)): + with self.subnet(network),\ + self.subnet(network, + cidr='2607:f0d0:1002:51::/64', + ip_version=6, + gateway_ip='fe80::1', + ipv6_address_mode=constants.IPV6_SLAAC): # Create a router port without specifying fixed_ips port = self._make_port( self.fmt, network['network']['id'], @@ -2144,11 +2134,9 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s plugin = manager.NeutronManager.get_plugin() ctx = context.get_admin_context() with self.subnet() as subnet: - with contextlib.nested( - self.port(subnet=subnet, device_id='owner1'), - self.port(subnet=subnet, device_id='owner1'), - self.port(subnet=subnet, device_id='owner2'), - ) as (p1, p2, p3): + with self.port(subnet=subnet, device_id='owner1') as p1,\ + self.port(subnet=subnet, device_id='owner1') as p2,\ + self.port(subnet=subnet, device_id='owner2') as p3: network_id = subnet['subnet']['network_id'] plugin.delete_ports_by_device_id(ctx, 'owner1', network_id) @@ -2162,11 +2150,9 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s def _test_delete_ports_by_device_id_second_call_failure(self, plugin): ctx = context.get_admin_context() with self.subnet() as subnet: - with contextlib.nested( - self.port(subnet=subnet, device_id='owner1'), - self.port(subnet=subnet, device_id='owner1'), - self.port(subnet=subnet, device_id='owner2'), - ) as (p1, p2, p3): + with self.port(subnet=subnet, device_id='owner1') as p1,\ + self.port(subnet=subnet, device_id='owner1') as p2,\ + self.port(subnet=subnet, device_id='owner2') as p3: orig = plugin.delete_port with mock.patch.object(plugin, 'delete_port') as del_port: @@ -2194,10 +2180,8 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s def _test_delete_ports_ignores_port_not_found(self, plugin): ctx = context.get_admin_context() with self.subnet() as subnet: - with contextlib.nested( - self.port(subnet=subnet, device_id='owner1'), - mock.patch.object(plugin, 'delete_port') - ) as (p, del_port): + with self.port(subnet=subnet, device_id='owner1') as p,\ + mock.patch.object(plugin, 'delete_port') as del_port: del_port.side_effect = n_exc.PortNotFound( port_id=p['port']['id'] ) @@ -2507,21 +2491,16 @@ class TestNetworksV2(NeutronDbPluginV2TestCase): ) def test_list_networks(self): - with contextlib.nested(self.network(), - self.network(), - self.network()) as networks: + with self.network() as v1, self.network() as v2, self.network() as v3: + networks = (v1, v2, v3) self._test_list_resources('network', networks) def test_list_networks_with_sort_native(self): if self._skip_native_sorting: self.skipTest("Skip test for not implemented sorting feature") - with contextlib.nested(self.network(admin_status_up=True, - name='net1'), - self.network(admin_status_up=False, - name='net2'), - self.network(admin_status_up=False, - name='net3') - ) as (net1, net2, net3): + with self.network(admin_status_up=True, name='net1') as net1,\ + self.network(admin_status_up=False, name='net2') as net2,\ + self.network(admin_status_up=False, name='net3') as net3: self._test_list_with_sort('network', (net3, net2, net1), [('admin_state_up', 'asc'), ('name', 'desc')]) @@ -2529,13 +2508,9 @@ class TestNetworksV2(NeutronDbPluginV2TestCase): def test_list_networks_with_sort_extended_attr_native_returns_400(self): if self._skip_native_sorting: self.skipTest("Skip test for not implemented sorting feature") - with contextlib.nested(self.network(admin_status_up=True, - name='net1'), - self.network(admin_status_up=False, - name='net2'), - self.network(admin_status_up=False, - name='net3') - ): + with self.network(admin_status_up=True, name='net1'),\ + self.network(admin_status_up=False, name='net2'),\ + self.network(admin_status_up=False, name='net3'): req = self.new_list_request( 'networks', params='sort_key=provider:segmentation_id&sort_dir=asc') @@ -2545,13 +2520,9 @@ class TestNetworksV2(NeutronDbPluginV2TestCase): def test_list_networks_with_sort_remote_key_native_returns_400(self): if self._skip_native_sorting: self.skipTest("Skip test for not implemented sorting feature") - with contextlib.nested(self.network(admin_status_up=True, - name='net1'), - self.network(admin_status_up=False, - name='net2'), - self.network(admin_status_up=False, - name='net3') - ): + with self.network(admin_status_up=True, name='net1'),\ + self.network(admin_status_up=False, name='net2'),\ + self.network(admin_status_up=False, name='net3'): req = self.new_list_request( 'networks', params='sort_key=subnets&sort_dir=asc') res = req.get_response(self.api) @@ -2562,13 +2533,9 @@ class TestNetworksV2(NeutronDbPluginV2TestCase): 'neutron.api.v2.base.Controller._get_sorting_helper', new=_fake_get_sorting_helper) helper_patcher.start() - with contextlib.nested(self.network(admin_status_up=True, - name='net1'), - self.network(admin_status_up=False, - name='net2'), - self.network(admin_status_up=False, - name='net3') - ) as (net1, net2, net3): + with self.network(admin_status_up=True, name='net1') as net1,\ + self.network(admin_status_up=False, name='net2') as net2,\ + self.network(admin_status_up=False, name='net3') as net3: self._test_list_with_sort('network', (net3, net2, net1), [('admin_state_up', 'asc'), ('name', 'desc')]) @@ -2576,10 +2543,9 @@ class TestNetworksV2(NeutronDbPluginV2TestCase): def test_list_networks_with_pagination_native(self): if self._skip_native_pagination: self.skipTest("Skip test for not implemented pagination feature") - with contextlib.nested(self.network(name='net1'), - self.network(name='net2'), - self.network(name='net3') - ) as (net1, net2, net3): + with self.network(name='net1') as net1,\ + self.network(name='net2') as net2,\ + self.network(name='net3') as net3: self._test_list_with_pagination('network', (net1, net2, net3), ('name', 'asc'), 2, 2) @@ -2589,10 +2555,9 @@ class TestNetworksV2(NeutronDbPluginV2TestCase): 'neutron.api.v2.base.Controller._get_pagination_helper', new=_fake_get_pagination_helper) helper_patcher.start() - with contextlib.nested(self.network(name='net1'), - self.network(name='net2'), - self.network(name='net3') - ) as (net1, net2, net3): + with self.network(name='net1') as net1,\ + self.network(name='net2') as net2,\ + self.network(name='net3') as net3: self._test_list_with_pagination('network', (net1, net2, net3), ('name', 'asc'), 2, 2) @@ -2602,13 +2567,9 @@ class TestNetworksV2(NeutronDbPluginV2TestCase): 'neutron.api.v2.base.Controller._get_pagination_helper', new=_fake_get_pagination_helper) helper_patcher.start() - with contextlib.nested(self.network(name='net1', - shared=True), - self.network(name='net2', - shared=False), - self.network(name='net3', - shared=True) - ) as (net1, net2, net3): + with self.network(name='net1', shared=True) as net1,\ + self.network(name='net2', shared=False) as net2,\ + self.network(name='net3', shared=True) as net3: self._test_list_with_pagination('network', (net1, net2, net3), ('name', 'asc'), 2, 2, @@ -2618,10 +2579,9 @@ class TestNetworksV2(NeutronDbPluginV2TestCase): def test_list_networks_without_pk_in_fields_pagination_native(self): if self._skip_native_pagination: self.skipTest("Skip test for not implemented pagination feature") - with contextlib.nested(self.network(name='net1'), - self.network(name='net2'), - self.network(name='net3') - ) as (net1, net2, net3): + with self.network(name='net1') as net1,\ + self.network(name='net2') as net2,\ + self.network(name='net3') as net3: self._test_list_with_pagination('network', (net1, net2, net3), ('name', 'asc'), 2, 2, @@ -2631,10 +2591,9 @@ class TestNetworksV2(NeutronDbPluginV2TestCase): def test_list_networks_with_pagination_reverse_native(self): if self._skip_native_pagination: self.skipTest("Skip test for not implemented pagination feature") - with contextlib.nested(self.network(name='net1'), - self.network(name='net2'), - self.network(name='net3') - ) as (net1, net2, net3): + with self.network(name='net1') as net1,\ + self.network(name='net2') as net2,\ + self.network(name='net3') as net3: self._test_list_with_pagination_reverse('network', (net1, net2, net3), ('name', 'asc'), 2, 2) @@ -2644,18 +2603,16 @@ class TestNetworksV2(NeutronDbPluginV2TestCase): 'neutron.api.v2.base.Controller._get_pagination_helper', new=_fake_get_pagination_helper) helper_patcher.start() - with contextlib.nested(self.network(name='net1'), - self.network(name='net2'), - self.network(name='net3') - ) as (net1, net2, net3): + with self.network(name='net1') as net1,\ + self.network(name='net2') as net2,\ + self.network(name='net3') as net3: self._test_list_with_pagination_reverse('network', (net1, net2, net3), ('name', 'asc'), 2, 2) def test_list_networks_with_parameters(self): - with contextlib.nested(self.network(name='net1', - admin_state_up=False), - self.network(name='net2')) as (net1, net2): + with self.network(name='net1', admin_state_up=False) as net1,\ + self.network(name='net2') as net2: query_params = 'admin_state_up=False' self._test_list_resources('network', [net1], query_params=query_params) @@ -2674,25 +2631,23 @@ class TestNetworksV2(NeutronDbPluginV2TestCase): self.assertIsNone(res['networks'][0].get('id')) def test_list_networks_with_parameters_invalid_values(self): - with contextlib.nested(self.network(name='net1', - admin_state_up=False), - self.network(name='net2')) as (net1, net2): + with self.network(name='net1', admin_state_up=False),\ + self.network(name='net2'): req = self.new_list_request('networks', params='admin_state_up=fake') res = req.get_response(self.api) self.assertEqual(webob.exc.HTTPClientError.code, res.status_int) def test_list_shared_networks_with_non_admin_user(self): - with contextlib.nested(self.network(shared=False, - name='net1', - tenant_id='tenant1'), - self.network(shared=True, - name='net2', - tenant_id='another_tenant'), - self.network(shared=False, - name='net3', - tenant_id='another_tenant') - ) as (net1, net2, net3): + with self.network(shared=False, + name='net1', + tenant_id='tenant1') as net1,\ + self.network(shared=True, + name='net2', + tenant_id='another_tenant') as net2,\ + self.network(shared=False, + name='net3', + tenant_id='another_tenant'): ctx = context.Context(user_id='non_admin', tenant_id='tenant1', is_admin=False) @@ -2931,8 +2886,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): cidr_2 = '10.0.0.0/24' cfg.CONF.set_override('allow_overlapping_ips', True) - with contextlib.nested(self.subnet(cidr=cidr_1), - self.subnet(cidr=cidr_2)): + with self.subnet(cidr=cidr_1), self.subnet(cidr=cidr_2): pass def test_create_2_subnets_overlapping_cidr_not_allowed_returns_400(self): @@ -2941,8 +2895,7 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): cfg.CONF.set_override('allow_overlapping_ips', False) with testlib_api.ExpectedException( webob.exc.HTTPClientError) as ctx_manager: - with contextlib.nested(self.subnet(cidr=cidr_1), - self.subnet(cidr=cidr_2)): + with self.subnet(cidr=cidr_1), self.subnet(cidr=cidr_2): pass self.assertEqual(ctx_manager.exception.code, webob.exc.HTTPClientError.code) @@ -3100,10 +3053,9 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): def test_delete_subnet_with_other_subnet_on_network_still_in_use(self): with self.network() as network: - with contextlib.nested( - self.subnet(network=network), - self.subnet(network=network, cidr='10.0.1.0/24'), - ) as (subnet1, subnet2): + with self.subnet(network=network) as subnet1,\ + self.subnet(network=network, + cidr='10.0.1.0/24') as subnet2: subnet1_id = subnet1['subnet']['id'] subnet2_id = subnet2['subnet']['id'] with self.port( @@ -3219,16 +3171,15 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): set_context=False) def test_create_subnet_nonzero_cidr(self): - with contextlib.nested( - self.subnet(cidr='10.129.122.5/8'), - self.subnet(cidr='11.129.122.5/15'), - self.subnet(cidr='12.129.122.5/16'), - self.subnet(cidr='13.129.122.5/18'), - self.subnet(cidr='14.129.122.5/22'), - self.subnet(cidr='15.129.122.5/24'), - self.subnet(cidr='16.129.122.5/28'), - self.subnet(cidr='17.129.122.5/32', enable_dhcp=False) - ) as subs: + with self.subnet(cidr='10.129.122.5/8') as v1,\ + self.subnet(cidr='11.129.122.5/15') as v2,\ + self.subnet(cidr='12.129.122.5/16') as v3,\ + self.subnet(cidr='13.129.122.5/18') as v4,\ + self.subnet(cidr='14.129.122.5/22') as v5,\ + self.subnet(cidr='15.129.122.5/24') as v6,\ + self.subnet(cidr='16.129.122.5/28') as v7,\ + self.subnet(cidr='17.129.122.5/32', enable_dhcp=False) as v8: + subs = (v1, v2, v3, v4, v5, v6, v7, v8) # the API should accept and correct these for users self.assertEqual(subs[0]['subnet']['cidr'], '10.0.0.0/8') self.assertEqual(subs[1]['subnet']['cidr'], '11.128.0.0/15') @@ -4218,15 +4169,16 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): def test_list_subnets(self): with self.network() as network: - with contextlib.nested(self.subnet(network=network, - gateway_ip='10.0.0.1', - cidr='10.0.0.0/24'), - self.subnet(network=network, - gateway_ip='10.0.1.1', - cidr='10.0.1.0/24'), - self.subnet(network=network, - gateway_ip='10.0.2.1', - cidr='10.0.2.0/24')) as subnets: + with self.subnet(network=network, + gateway_ip='10.0.0.1', + cidr='10.0.0.0/24') as v1,\ + self.subnet(network=network, + gateway_ip='10.0.1.1', + cidr='10.0.1.0/24') as v2,\ + self.subnet(network=network, + gateway_ip='10.0.2.1', + cidr='10.0.2.0/24') as v3: + subnets = (v1, v2, v3) self._test_list_resources('subnet', subnets) def test_list_subnets_shared(self): @@ -4253,13 +4205,13 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): def test_list_subnets_with_parameter(self): with self.network() as network: - with contextlib.nested(self.subnet(network=network, - gateway_ip='10.0.0.1', - cidr='10.0.0.0/24'), - self.subnet(network=network, - gateway_ip='10.0.1.1', - cidr='10.0.1.0/24') - ) as subnets: + with self.subnet(network=network, + gateway_ip='10.0.0.1', + cidr='10.0.0.0/24') as v1,\ + self.subnet(network=network, + gateway_ip='10.0.1.1', + cidr='10.0.1.0/24') as v2: + subnets = (v1, v2) query_params = 'ip_version=4&ip_version=6' self._test_list_resources('subnet', subnets, query_params=query_params) @@ -4270,13 +4222,9 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): def test_list_subnets_with_sort_native(self): if self._skip_native_sorting: self.skipTest("Skip test for not implemented sorting feature") - with contextlib.nested(self.subnet(enable_dhcp=True, - cidr='10.0.0.0/24'), - self.subnet(enable_dhcp=False, - cidr='11.0.0.0/24'), - self.subnet(enable_dhcp=False, - cidr='12.0.0.0/24') - ) as (subnet1, subnet2, subnet3): + with self.subnet(enable_dhcp=True, cidr='10.0.0.0/24') as subnet1,\ + self.subnet(enable_dhcp=False, cidr='11.0.0.0/24') as subnet2,\ + self.subnet(enable_dhcp=False, cidr='12.0.0.0/24') as subnet3: self._test_list_with_sort('subnet', (subnet3, subnet2, subnet1), [('enable_dhcp', 'asc'), ('cidr', 'desc')]) @@ -4286,13 +4234,9 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): 'neutron.api.v2.base.Controller._get_sorting_helper', new=_fake_get_sorting_helper) helper_patcher.start() - with contextlib.nested(self.subnet(enable_dhcp=True, - cidr='10.0.0.0/24'), - self.subnet(enable_dhcp=False, - cidr='11.0.0.0/24'), - self.subnet(enable_dhcp=False, - cidr='12.0.0.0/24') - ) as (subnet1, subnet2, subnet3): + with self.subnet(enable_dhcp=True, cidr='10.0.0.0/24') as subnet1,\ + self.subnet(enable_dhcp=False, cidr='11.0.0.0/24') as subnet2,\ + self.subnet(enable_dhcp=False, cidr='12.0.0.0/24') as subnet3: self._test_list_with_sort('subnet', (subnet3, subnet2, subnet1), @@ -4302,10 +4246,9 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): def test_list_subnets_with_pagination_native(self): if self._skip_native_pagination: self.skipTest("Skip test for not implemented sorting feature") - with contextlib.nested(self.subnet(cidr='10.0.0.0/24'), - self.subnet(cidr='11.0.0.0/24'), - self.subnet(cidr='12.0.0.0/24') - ) as (subnet1, subnet2, subnet3): + with self.subnet(cidr='10.0.0.0/24') as subnet1,\ + self.subnet(cidr='11.0.0.0/24') as subnet2,\ + self.subnet(cidr='12.0.0.0/24') as subnet3: self._test_list_with_pagination('subnet', (subnet1, subnet2, subnet3), ('cidr', 'asc'), 2, 2) @@ -4315,10 +4258,9 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): 'neutron.api.v2.base.Controller._get_pagination_helper', new=_fake_get_pagination_helper) helper_patcher.start() - with contextlib.nested(self.subnet(cidr='10.0.0.0/24'), - self.subnet(cidr='11.0.0.0/24'), - self.subnet(cidr='12.0.0.0/24') - ) as (subnet1, subnet2, subnet3): + with self.subnet(cidr='10.0.0.0/24') as subnet1,\ + self.subnet(cidr='11.0.0.0/24') as subnet2,\ + self.subnet(cidr='12.0.0.0/24') as subnet3: self._test_list_with_pagination('subnet', (subnet1, subnet2, subnet3), ('cidr', 'asc'), 2, 2) @@ -4326,10 +4268,9 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): def test_list_subnets_with_pagination_reverse_native(self): if self._skip_native_sorting: self.skipTest("Skip test for not implemented sorting feature") - with contextlib.nested(self.subnet(cidr='10.0.0.0/24'), - self.subnet(cidr='11.0.0.0/24'), - self.subnet(cidr='12.0.0.0/24') - ) as (subnet1, subnet2, subnet3): + with self.subnet(cidr='10.0.0.0/24') as subnet1,\ + self.subnet(cidr='11.0.0.0/24') as subnet2,\ + self.subnet(cidr='12.0.0.0/24') as subnet3: self._test_list_with_pagination_reverse('subnet', (subnet1, subnet2, subnet3), @@ -4340,10 +4281,9 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): 'neutron.api.v2.base.Controller._get_pagination_helper', new=_fake_get_pagination_helper) helper_patcher.start() - with contextlib.nested(self.subnet(cidr='10.0.0.0/24'), - self.subnet(cidr='11.0.0.0/24'), - self.subnet(cidr='12.0.0.0/24') - ) as (subnet1, subnet2, subnet3): + with self.subnet(cidr='10.0.0.0/24') as subnet1,\ + self.subnet(cidr='11.0.0.0/24') as subnet2,\ + self.subnet(cidr='12.0.0.0/24') as subnet3: self._test_list_with_pagination_reverse('subnet', (subnet1, subnet2, subnet3), @@ -4599,9 +4539,8 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase): self.assertEqual(res.status_int, webob.exc.HTTPNoContent.code) def test_delete_subnet_with_callback(self): - with contextlib.nested( - self.subnet(), - mock.patch.object(registry, 'notify')) as (subnet, notify): + with self.subnet() as subnet,\ + mock.patch.object(registry, 'notify') as notify: errors = [ exceptions.NotificationError( @@ -5557,18 +5496,11 @@ class NeutronDbPluginV2AsMixinTestCase(NeutronDbPluginV2TestCase, def test_get_user_allocation_for_dhcp_port_returns_none(self): plugin = manager.NeutronManager.get_plugin() - with contextlib.nested( - self.network(), - self.network() - ) as (net, net1): - with contextlib.nested( - self.subnet(network=net, cidr='10.0.0.0/24'), - self.subnet(network=net1, cidr='10.0.1.0/24') - ) as (subnet, subnet1): - with contextlib.nested( - self.port(subnet=subnet, device_owner='network:dhcp'), - self.port(subnet=subnet1) - ) as (p, p2): + with self.network() as net, self.network() as net1: + with self.subnet(network=net, cidr='10.0.0.0/24') as subnet,\ + self.subnet(network=net1, cidr='10.0.1.0/24') as subnet1: + with self.port(subnet=subnet, device_owner='network:dhcp'),\ + self.port(subnet=subnet1): # check that user allocations on another network don't # affect _subnet_get_user_allocation method res = plugin._subnet_get_user_allocation( diff --git a/neutron/tests/unit/db/test_l3_dvr_db.py b/neutron/tests/unit/db/test_l3_dvr_db.py index cf76942ac..a37c52057 100644 --- a/neutron/tests/unit/db/test_l3_dvr_db.py +++ b/neutron/tests/unit/db/test_l3_dvr_db.py @@ -13,7 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -import contextlib import mock from neutron.common import constants as l3_const @@ -205,12 +204,11 @@ class L3DvrTestCase(testlib_api.SqlTestCase): router_db = self._create_router(router) router_id = router_db['id'] self.assertTrue(router_db.extra_attributes.distributed) - with contextlib.nested( - mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, - '_create_gw_port'), - mock.patch.object(self.mixin, - '_create_snat_intf_ports_if_not_exists') - ) as (cw, cs): + with mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, + '_create_gw_port'),\ + mock.patch.object( + self.mixin, + '_create_snat_intf_ports_if_not_exists') as cs: self.mixin._create_gw_port( self.ctx, router_id, router_db, mock.ANY, mock.ANY) @@ -228,16 +226,15 @@ class L3DvrTestCase(testlib_api.SqlTestCase): 'fixed_port_id': _uuid(), 'floating_network_id': _uuid() } - with contextlib.nested( - mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, - '_get_floatingip'), - mock.patch.object(self.mixin, - '_get_vm_port_hostid'), - mock.patch.object(self.mixin, - '_check_fips_availability_on_host_ext_net'), - mock.patch.object(self.mixin, - '_delete_floatingip_agent_gateway_port') - ) as (gfips, gvm, cfips, dfips): + with mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, + '_get_floatingip') as gfips,\ + mock.patch.object(self.mixin, '_get_vm_port_hostid') as gvm,\ + mock.patch.object( + self.mixin, + '_check_fips_availability_on_host_ext_net') as cfips,\ + mock.patch.object( + self.mixin, + '_delete_floatingip_agent_gateway_port') as dfips: gfips.return_value = floatingip gvm.return_value = 'my-host' cfips.return_value = True @@ -254,10 +251,9 @@ class L3DvrTestCase(testlib_api.SqlTestCase): 'network_id': 'ext_network_id', 'device_owner': l3_const.DEVICE_OWNER_AGENT_GW } - with contextlib.nested( - mock.patch.object(manager.NeutronManager, 'get_plugin'), - mock.patch.object(self.mixin, - '_get_vm_port_hostid')) as (gp, vm_host): + with mock.patch.object(manager.NeutronManager, 'get_plugin') as gp,\ + mock.patch.object(self.mixin, + '_get_vm_port_hostid') as vm_host: plugin = mock.Mock() gp.return_value = plugin plugin.get_ports.return_value = [port] @@ -271,23 +267,20 @@ class L3DvrTestCase(testlib_api.SqlTestCase): def _delete_floatingip_test_setup(self, floatingip): fip_id = floatingip['id'] - with contextlib.nested( - mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, - '_get_floatingip'), - mock.patch.object(self.mixin, - '_clear_unused_fip_agent_gw_port'), - mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, - 'delete_floatingip')) as (gf, vf, df): + with mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, + '_get_floatingip') as gf,\ + mock.patch.object(self.mixin, + '_clear_unused_fip_agent_gw_port') as vf,\ + mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, + 'delete_floatingip'): gf.return_value = floatingip self.mixin.delete_floatingip(self.ctx, fip_id) return vf def _disassociate_floatingip_setup(self, port_id=None, floatingip=None): - with contextlib.nested( - mock.patch.object(self.mixin, '_get_floatingip_on_port'), - mock.patch.object(self.mixin, - '_clear_unused_fip_agent_gw_port'), - ) as (gf, vf): + with mock.patch.object(self.mixin, '_get_floatingip_on_port') as gf,\ + mock.patch.object(self.mixin, + '_clear_unused_fip_agent_gw_port') as vf: gf.return_value = floatingip self.mixin.disassociate_floatingips( self.ctx, port_id, do_notify=False) @@ -386,14 +379,11 @@ class L3DvrTestCase(testlib_api.SqlTestCase): 'router_id': 'foo_router_id' } router = {'id': 'foo_router_id', 'distributed': True} - with contextlib.nested( - mock.patch.object(self.mixin, - 'get_router'), - mock.patch.object(self.mixin, - '_clear_unused_fip_agent_gw_port'), - mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, - '_update_fip_assoc'), - ) as (grtr, vf, cf): + with mock.patch.object(self.mixin, 'get_router') as grtr,\ + mock.patch.object(self.mixin, + '_clear_unused_fip_agent_gw_port') as vf,\ + mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, + '_update_fip_assoc'): grtr.return_value = router self.mixin._update_fip_assoc( self.ctx, fip, floatingip, mock.ANY) @@ -407,18 +397,15 @@ class L3DvrTestCase(testlib_api.SqlTestCase): 'network_id': 'external_net' } - with contextlib.nested( - mock.patch.object(self.mixin, - 'get_router'), - mock.patch.object(self.mixin, - '_get_vm_port_hostid'), - mock.patch.object(self.mixin, - '_clear_unused_fip_agent_gw_port'), - mock.patch.object(self.mixin, - 'create_fip_agent_gw_port_if_not_exists'), - mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, - '_update_fip_assoc'), - ) as (grtr, vmp, d_fip, c_fip, up_fip): + with mock.patch.object(self.mixin, 'get_router') as grtr,\ + mock.patch.object(self.mixin, '_get_vm_port_hostid') as vmp,\ + mock.patch.object(self.mixin, + '_clear_unused_fip_agent_gw_port') as d_fip,\ + mock.patch.object( + self.mixin, + 'create_fip_agent_gw_port_if_not_exists') as c_fip,\ + mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin, + '_update_fip_assoc'): grtr.return_value = router_db vmp.return_value = 'my-host' self.mixin._update_fip_assoc( @@ -502,23 +489,19 @@ class L3DvrTestCase(testlib_api.SqlTestCase): return_value=False) plugin.remove_router_from_l3_agent = mock.Mock( return_value=None) - with contextlib.nested( - mock.patch.object(self.mixin, - '_get_router'), - mock.patch.object(self.mixin, - '_get_device_owner'), - mock.patch.object(self.mixin, - '_remove_interface_by_subnet'), - mock.patch.object(self.mixin, - 'delete_csnat_router_interface_ports'), - mock.patch.object(manager.NeutronManager, - 'get_service_plugins'), - mock.patch.object(self.mixin, - '_make_router_interface_info'), - mock.patch.object(self.mixin, - 'notify_router_interface_action'), - ) as (grtr, gdev, rmintf, delintf, gplugin, - mkintf, notify): + with mock.patch.object(self.mixin, '_get_router') as grtr,\ + mock.patch.object(self.mixin, '_get_device_owner') as gdev,\ + mock.patch.object(self.mixin, + '_remove_interface_by_subnet') as rmintf,\ + mock.patch.object( + self.mixin, + 'delete_csnat_router_interface_ports') as delintf,\ + mock.patch.object(manager.NeutronManager, + 'get_service_plugins') as gplugin,\ + mock.patch.object(self.mixin, + '_make_router_interface_info') as mkintf,\ + mock.patch.object(self.mixin, + 'notify_router_interface_action') as notify: grtr.return_value = router gdev.return_value = mock.Mock() rmintf.return_value = (mock.MagicMock(), mock.MagicMock()) diff --git a/neutron/tests/unit/db/test_securitygroups_db.py b/neutron/tests/unit/db/test_securitygroups_db.py index 7f87802ac..0626f9ca6 100644 --- a/neutron/tests/unit/db/test_securitygroups_db.py +++ b/neutron/tests/unit/db/test_securitygroups_db.py @@ -11,7 +11,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -import contextlib import mock import testtools @@ -45,11 +44,10 @@ class SecurityGroupDbMixinTestCase(testlib_api.SqlTestCase): self.mixin.create_security_group(self.ctx, secgroup) def test_delete_security_group_in_use(self): - with contextlib.nested( - mock.patch.object(self.mixin, '_get_port_security_group_bindings'), - mock.patch.object(self.mixin, '_get_security_group'), - mock.patch.object(registry, "notify"), - ) as (_, _, mock_notify): + with mock.patch.object(self.mixin, + '_get_port_security_group_bindings'),\ + 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): diff --git a/neutron/tests/unit/extensions/test_external_net.py b/neutron/tests/unit/extensions/test_external_net.py index b2ccce818..0f68cae38 100644 --- a/neutron/tests/unit/extensions/test_external_net.py +++ b/neutron/tests/unit/extensions/test_external_net.py @@ -13,7 +13,6 @@ # License for the specific language governing permissions and limitations # under the License. -import contextlib import itertools import mock @@ -90,8 +89,7 @@ class ExtNetDBTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase): def test_list_nets_external_pagination(self): if self._skip_native_pagination: self.skipTest("Skip test for not implemented pagination feature") - with contextlib.nested(self.network(name='net1'), - self.network(name='net3')) as (n1, n3): + with self.network(name='net1') as n1, self.network(name='net3') as n3: self._set_net_external(n1['network']['id']) self._set_net_external(n3['network']['id']) with self.network(name='net2') as n2: diff --git a/neutron/tests/unit/extensions/test_extraroute.py b/neutron/tests/unit/extensions/test_extraroute.py index 78e555f0b..278567f4c 100644 --- a/neutron/tests/unit/extensions/test_extraroute.py +++ b/neutron/tests/unit/extensions/test_extraroute.py @@ -13,8 +13,6 @@ # License for the specific language governing permissions and limitations # under the License. -import contextlib - from oslo_config import cfg from oslo_log import log as logging from webob import exc @@ -146,13 +144,10 @@ class ExtraRouteDBTestCaseBase(object): 'nexthop': '10.0.0.3'}] routes2 = [{'destination': '12.0.0.0/8', 'nexthop': '10.0.0.4'}] - with contextlib.nested( - self.router(), - self.router(), - self.subnet(cidr='10.0.0.0/24')) as (r1, r2, s): - with contextlib.nested( - self.port(subnet=s), - self.port(subnet=s)) as (p1, p2): + with self.router() as r1,\ + self.router() as r2,\ + self.subnet(cidr='10.0.0.0/24') as s: + with self.port(subnet=s) as p1, self.port(subnet=s) as p2: body = self._routes_update_prepare(r1['router']['id'], None, p1['port']['id'], routes1) @@ -427,27 +422,24 @@ class ExtraRouteDBTestCaseBase(object): self.assertIsNone(gw_info) def test_router_list_with_sort(self): - with contextlib.nested(self.router(name='router1'), - self.router(name='router2'), - self.router(name='router3') - ) as (router1, router2, router3): + with self.router(name='router1') as router1,\ + self.router(name='router2') as router2,\ + self.router(name='router3') as router3: self._test_list_with_sort('router', (router3, router2, router1), [('name', 'desc')]) def test_router_list_with_pagination(self): - with contextlib.nested(self.router(name='router1'), - self.router(name='router2'), - self.router(name='router3') - ) as (router1, router2, router3): + with self.router(name='router1') as router1,\ + self.router(name='router2') as router2,\ + self.router(name='router3') as router3: self._test_list_with_pagination('router', (router1, router2, router3), ('name', 'asc'), 2, 2) def test_router_list_with_pagination_reverse(self): - with contextlib.nested(self.router(name='router1'), - self.router(name='router2'), - self.router(name='router3') - ) as (router1, router2, router3): + with self.router(name='router1') as router1,\ + self.router(name='router2') as router2,\ + self.router(name='router3') as router3: self._test_list_with_pagination_reverse('router', (router1, router2, router3), diff --git a/neutron/tests/unit/extensions/test_l3.py b/neutron/tests/unit/extensions/test_l3.py index 2e67d40f0..2392adc03 100644 --- a/neutron/tests/unit/extensions/test_l3.py +++ b/neutron/tests/unit/extensions/test_l3.py @@ -620,11 +620,10 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): def test_router_create_with_gwinfo_ext_ip_subnet(self): with self.network() as n: - with contextlib.nested( - self.subnet(network=n), - self.subnet(network=n, cidr='1.0.0.0/24'), - self.subnet(network=n, cidr='2.0.0.0/24'), - ) as subnets: + with self.subnet(network=n) as v1,\ + self.subnet(network=n, cidr='1.0.0.0/24') as v2,\ + self.subnet(network=n, cidr='2.0.0.0/24') as v3: + subnets = (v1, v2, v3) self._set_net_external(n['network']['id']) for s in subnets: ext_info = { @@ -658,16 +657,13 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): self.assertEqual(res.status_int, exc.HTTPForbidden.code) def test_router_list(self): - with contextlib.nested(self.router(), - self.router(), - self.router() - ) as routers: + with self.router() as v1, self.router() as v2, self.router() as v3: + routers = (v1, v2, v3) self._test_list_resources('router', routers) def test_router_list_with_parameters(self): - with contextlib.nested(self.router(name='router1'), - self.router(name='router2'), - ) as (router1, router2): + with self.router(name='router1') as router1,\ + self.router(name='router2') as router2: query_params = 'name=router1' self._test_list_resources('router', [router1], query_params=query_params) @@ -679,27 +675,24 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): query_params=query_params) def test_router_list_with_sort(self): - with contextlib.nested(self.router(name='router1'), - self.router(name='router2'), - self.router(name='router3') - ) as (router1, router2, router3): + with self.router(name='router1') as router1,\ + self.router(name='router2') as router2,\ + self.router(name='router3') as router3: self._test_list_with_sort('router', (router3, router2, router1), [('name', 'desc')]) def test_router_list_with_pagination(self): - with contextlib.nested(self.router(name='router1'), - self.router(name='router2'), - self.router(name='router3') - ) as (router1, router2, router3): + with self.router(name='router1') as router1,\ + self.router(name='router2') as router2,\ + self.router(name='router3') as router3: self._test_list_with_pagination('router', (router1, router2, router3), ('name', 'asc'), 2, 2) def test_router_list_with_pagination_reverse(self): - with contextlib.nested(self.router(name='router1'), - self.router(name='router2'), - self.router(name='router3') - ) as (router1, router2, router3): + with self.router(name='router1') as router1,\ + self.router(name='router2') as router2,\ + self.router(name='router3') as router3: self._test_list_with_pagination_reverse('router', (router1, router2, router3), @@ -767,11 +760,9 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): expected_code=exc.HTTPBadRequest.code) def test_router_update_gateway_with_invalid_external_subnet(self): - with contextlib.nested( - self.subnet(), - self.subnet(cidr='1.0.0.0/24'), - self.router() - ) as (s1, s2, r): + with self.subnet() as s1,\ + self.subnet(cidr='1.0.0.0/24') as s2,\ + self.router() as r: self._set_net_external(s1['subnet']['network_id']) self._add_external_gateway_to_router( r['router']['id'], @@ -782,11 +773,9 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): def test_router_update_gateway_with_different_external_subnet(self): with self.network() as n: - with contextlib.nested( - self.subnet(network=n), - self.subnet(network=n, cidr='1.0.0.0/24'), - self.router() - ) as (s1, s2, r): + with self.subnet(network=n) as s1,\ + self.subnet(network=n, cidr='1.0.0.0/24') as s2,\ + self.router() as r: self._set_net_external(n['network']['id']) res1 = self._add_external_gateway_to_router( r['router']['id'], @@ -998,7 +987,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): 'address_mode': stateless}] for uc in use_cases: fake_notifier.reset() - with contextlib.nested(self.router(), self.network()) as (r, n): + with self.router() as r, self.network() as n: with self.subnet(network=n, cidr='fd00::1/64', gateway_ip='fd00::1', ip_version=6, ipv6_ra_mode=uc['ra_mode'], @@ -1106,7 +1095,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): 'ra_mode': None, 'address_mode': l3_constants.DHCPV6_STATELESS}] for uc in use_cases: - with contextlib.nested(self.router(), self.network()) as (r, n): + with self.router() as r, self.network() as n: with self.subnet(network=n, cidr='fd00::1/64', gateway_ip='fd00::1', ip_version=6, ipv6_ra_mode=uc['ra_mode'], @@ -1163,13 +1152,11 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): def test_router_add_interface_subnet_with_port_from_other_tenant(self): tenant_id = _uuid() other_tenant_id = _uuid() - with contextlib.nested( - self.router(tenant_id=tenant_id), - self.network(tenant_id=tenant_id), - self.network(tenant_id=other_tenant_id)) as (r, n1, n2): - with contextlib.nested( - self.subnet(network=n1, cidr='10.0.0.0/24'), - self.subnet(network=n2, cidr='10.1.0.0/24')) as (s1, s2): + with self.router(tenant_id=tenant_id) as r,\ + self.network(tenant_id=tenant_id) as n1,\ + self.network(tenant_id=other_tenant_id) as n2: + with self.subnet(network=n1, cidr='10.0.0.0/24') as s1,\ + self.subnet(network=n2, cidr='10.1.0.0/24') as s2: body = self._router_interface_action( 'add', r['router']['id'], @@ -1594,10 +1581,9 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): expected_code=exc.HTTPConflict.code) def test_router_remove_interface_callback_failure_returns_409(self): - with contextlib.nested( - self.router(), - self.subnet(), - mock.patch.object(registry, 'notify')) as (r, s, notify): + with self.router() as r,\ + self.subnet() as s,\ + mock.patch.object(registry, 'notify') as notify: errors = [ exceptions.NotificationError( 'foo_callback_id', n_exc.InUse()), @@ -1619,10 +1605,9 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): exc.HTTPConflict.code) def test_router_clear_gateway_callback_failure_returns_409(self): - with contextlib.nested( - self.router(), - self.subnet(), - mock.patch.object(registry, 'notify')) as (r, s, notify): + with self.router() as r,\ + self.subnet() as s,\ + mock.patch.object(registry, 'notify') as notify: errors = [ exceptions.NotificationError( 'foo_callback_id', n_exc.InUse()), @@ -1980,11 +1965,9 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): def test_floatingip_update_different_router(self): # Create subnet with different CIDRs to account for plugins which # do not support overlapping IPs - with contextlib.nested(self.subnet(cidr='10.0.0.0/24'), - self.subnet(cidr='10.0.1.0/24')) as ( - s1, s2): - with contextlib.nested(self.port(subnet=s1), - self.port(subnet=s2)) as (p1, p2): + with self.subnet(cidr='10.0.0.0/24') as s1,\ + self.subnet(cidr='10.0.1.0/24') as s2: + with self.port(subnet=s1) as p1, self.port(subnet=s2) as p2: private_sub1 = {'subnet': {'id': p1['port']['fixed_ips'][0]['subnet_id']}} @@ -1992,12 +1975,12 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): {'id': p2['port']['fixed_ips'][0]['subnet_id']}} with self.subnet(cidr='12.0.0.0/24') as public_sub: - with contextlib.nested( + with self.floatingip_no_assoc_with_public_sub( + private_sub1, + public_sub=public_sub) as (fip1, r1),\ self.floatingip_no_assoc_with_public_sub( - private_sub1, public_sub=public_sub), - self.floatingip_no_assoc_with_public_sub( - private_sub2, public_sub=public_sub)) as ( - (fip1, r1), (fip2, r2)): + private_sub2, + public_sub=public_sub) as (fip2, r2): def assert_no_assoc(fip): body = self._show('floatingips', @@ -2160,10 +2143,9 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): self.assertEqual(res.status_int, 400) def test_floatingip_list_with_sort(self): - with contextlib.nested(self.subnet(cidr="10.0.0.0/24"), - self.subnet(cidr="11.0.0.0/24"), - self.subnet(cidr="12.0.0.0/24") - ) as (s1, s2, s3): + with self.subnet(cidr="10.0.0.0/24") as s1,\ + self.subnet(cidr="11.0.0.0/24") as s2,\ + self.subnet(cidr="12.0.0.0/24") as s3: network_id1 = s1['subnet']['network_id'] network_id2 = s2['subnet']['network_id'] network_id3 = s3['subnet']['network_id'] @@ -2186,10 +2168,9 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): self.assertEqual(len(res['floatingips']), 0) def test_floatingip_list_with_pagination(self): - with contextlib.nested(self.subnet(cidr="10.0.0.0/24"), - self.subnet(cidr="11.0.0.0/24"), - self.subnet(cidr="12.0.0.0/24") - ) as (s1, s2, s3): + with self.subnet(cidr="10.0.0.0/24") as s1,\ + self.subnet(cidr="11.0.0.0/24") as s2,\ + self.subnet(cidr="12.0.0.0/24") as s3: network_id1 = s1['subnet']['network_id'] network_id2 = s2['subnet']['network_id'] network_id3 = s3['subnet']['network_id'] @@ -2204,10 +2185,9 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): ('floating_ip_address', 'asc'), 2, 2) def test_floatingip_list_with_pagination_reverse(self): - with contextlib.nested(self.subnet(cidr="10.0.0.0/24"), - self.subnet(cidr="11.0.0.0/24"), - self.subnet(cidr="12.0.0.0/24") - ) as (s1, s2, s3): + with self.subnet(cidr="10.0.0.0/24") as s1,\ + self.subnet(cidr="11.0.0.0/24") as s2,\ + self.subnet(cidr="12.0.0.0/24") as s3: network_id1 = s1['subnet']['network_id'] network_id2 = s2['subnet']['network_id'] network_id3 = s3['subnet']['network_id'] @@ -2222,21 +2202,19 @@ class L3NatTestCaseBase(L3NatTestCaseMixin): ('floating_ip_address', 'asc'), 2, 2) def test_floatingip_multi_external_one_internal(self): - with contextlib.nested(self.subnet(cidr="10.0.0.0/24"), - self.subnet(cidr="11.0.0.0/24"), - self.subnet(cidr="12.0.0.0/24") - ) as (exs1, exs2, ins1): + with self.subnet(cidr="10.0.0.0/24") as exs1,\ + self.subnet(cidr="11.0.0.0/24") as exs2,\ + self.subnet(cidr="12.0.0.0/24") as ins1: network_ex_id1 = exs1['subnet']['network_id'] network_ex_id2 = exs2['subnet']['network_id'] self._set_net_external(network_ex_id1) self._set_net_external(network_ex_id2) r2i_fixed_ips = [{'ip_address': '12.0.0.2'}] - with contextlib.nested(self.router(no_delete=True), - self.router(no_delete=True), - self.port(subnet=ins1, - fixed_ips=r2i_fixed_ips) - ) as (r1, r2, r2i_port): + with self.router(no_delete=True) as r1,\ + self.router(no_delete=True) as r2,\ + self.port(subnet=ins1, + fixed_ips=r2i_fixed_ips) as r2i_port: self._add_external_gateway_to_router( r1['router']['id'], network_ex_id1) @@ -2589,9 +2567,7 @@ class L3NatDBIntAgentSchedulingTestCase(L3BaseForIntTests, self.assertEqual(agents[0]['host'], agent_host) def test_update_gateway_agent_exists_supporting_network(self): - with contextlib.nested(self.router(), - self.subnet(), - self.subnet()) as (r, s1, s2): + with self.router() as r, self.subnet() as s1, self.subnet() as s2: self._set_net_external(s1['subnet']['network_id']) l3_rpc_cb = l3_rpc.L3RpcCallback() helpers.register_l3_agent( @@ -2616,9 +2592,7 @@ class L3NatDBIntAgentSchedulingTestCase(L3BaseForIntTests, self._assert_router_on_agent(r['router']['id'], 'host2') def test_update_gateway_agent_exists_supporting_multiple_network(self): - with contextlib.nested(self.router(), - self.subnet(), - self.subnet()) as (r, s1, s2): + with self.router() as r, self.subnet() as s1, self.subnet() as s2: self._set_net_external(s1['subnet']['network_id']) l3_rpc_cb = l3_rpc.L3RpcCallback() helpers.register_l3_agent( diff --git a/neutron/tests/unit/extensions/test_securitygroup.py b/neutron/tests/unit/extensions/test_securitygroup.py index e21813b35..7aff2321d 100644 --- a/neutron/tests/unit/extensions/test_securitygroup.py +++ b/neutron/tests/unit/extensions/test_securitygroup.py @@ -373,51 +373,36 @@ class TestSecurityGroups(SecurityGroupDBTestCase): self.assertEqual(res.status_int, webob.exc.HTTPConflict.code) def test_list_security_groups(self): - with contextlib.nested(self.security_group(name='sg1', - description='sg'), - self.security_group(name='sg2', - description='sg'), - self.security_group(name='sg3', - description='sg') - ) as security_groups: + with self.security_group(name='sg1', description='sg') as v1,\ + self.security_group(name='sg2', description='sg') as v2,\ + self.security_group(name='sg3', description='sg') as v3: + security_groups = (v1, v2, v3) self._test_list_resources('security-group', security_groups, query_params='description=sg') def test_list_security_groups_with_sort(self): - with contextlib.nested(self.security_group(name='sg1', - description='sg'), - self.security_group(name='sg2', - description='sg'), - self.security_group(name='sg3', - description='sg') - ) as (sg1, sg2, sg3): + with self.security_group(name='sg1', description='sg') as sg1,\ + self.security_group(name='sg2', description='sg') as sg2,\ + self.security_group(name='sg3', description='sg') as sg3: self._test_list_with_sort('security-group', (sg3, sg2, sg1), [('name', 'desc')], query_params='description=sg') def test_list_security_groups_with_pagination(self): - with contextlib.nested(self.security_group(name='sg1', - description='sg'), - self.security_group(name='sg2', - description='sg'), - self.security_group(name='sg3', - description='sg') - ) as (sg1, sg2, sg3): + with self.security_group(name='sg1', description='sg') as sg1,\ + self.security_group(name='sg2', description='sg') as sg2,\ + self.security_group(name='sg3', description='sg') as sg3: self._test_list_with_pagination('security-group', (sg1, sg2, sg3), ('name', 'asc'), 2, 2, query_params='description=sg') def test_list_security_groups_with_pagination_reverse(self): - with contextlib.nested(self.security_group(name='sg1', - description='sg'), - self.security_group(name='sg2', - description='sg'), - self.security_group(name='sg3', - description='sg') - ) as (sg1, sg2, sg3): + with self.security_group(name='sg1', description='sg') as sg1,\ + self.security_group(name='sg2', description='sg') as sg2,\ + self.security_group(name='sg3', description='sg') as sg3: self._test_list_with_pagination_reverse( 'security-group', (sg1, sg2, sg3), ('name', 'asc'), 2, 2, query_params='description=sg') @@ -1069,19 +1054,18 @@ class TestSecurityGroups(SecurityGroupDBTestCase): def test_list_security_group_rules(self): with self.security_group(name='sg') as sg: security_group_id = sg['security_group']['id'] - with contextlib.nested(self.security_group_rule(security_group_id, - direction='egress', - port_range_min=22, - port_range_max=22), - self.security_group_rule(security_group_id, - direction='egress', - port_range_min=23, - port_range_max=23), - self.security_group_rule(security_group_id, - direction='egress', - port_range_min=24, - port_range_max=24) - ) as (sgr1, sgr2, sgr3): + with self.security_group_rule(security_group_id, + direction='egress', + port_range_min=22, + port_range_max=22) as sgr1,\ + self.security_group_rule(security_group_id, + direction='egress', + port_range_min=23, + port_range_max=23) as sgr2,\ + self.security_group_rule(security_group_id, + direction='egress', + port_range_min=24, + port_range_max=24) as sgr3: # Delete default rules as they would fail the following # assertion at the end. @@ -1096,19 +1080,18 @@ class TestSecurityGroups(SecurityGroupDBTestCase): def test_list_security_group_rules_with_sort(self): with self.security_group(name='sg') as sg: security_group_id = sg['security_group']['id'] - with contextlib.nested(self.security_group_rule(security_group_id, - direction='egress', - port_range_min=22, - port_range_max=22), - self.security_group_rule(security_group_id, - direction='egress', - port_range_min=23, - port_range_max=23), - self.security_group_rule(security_group_id, - direction='egress', - port_range_min=24, - port_range_max=24) - ) as (sgr1, sgr2, sgr3): + with self.security_group_rule(security_group_id, + direction='egress', + port_range_min=22, + port_range_max=22) as sgr1,\ + self.security_group_rule(security_group_id, + direction='egress', + port_range_min=23, + port_range_max=23) as sgr2,\ + self.security_group_rule(security_group_id, + direction='egress', + port_range_min=24, + port_range_max=24) as sgr3: # Delete default rules as they would fail the following # assertion at the end. @@ -1124,19 +1107,18 @@ class TestSecurityGroups(SecurityGroupDBTestCase): def test_list_security_group_rules_with_pagination(self): with self.security_group(name='sg') as sg: security_group_id = sg['security_group']['id'] - with contextlib.nested(self.security_group_rule(security_group_id, - direction='egress', - port_range_min=22, - port_range_max=22), - self.security_group_rule(security_group_id, - direction='egress', - port_range_min=23, - port_range_max=23), - self.security_group_rule(security_group_id, - direction='egress', - port_range_min=24, - port_range_max=24) - ) as (sgr1, sgr2, sgr3): + with self.security_group_rule(security_group_id, + direction='egress', + port_range_min=22, + port_range_max=22) as sgr1,\ + self.security_group_rule(security_group_id, + direction='egress', + port_range_min=23, + port_range_max=23) as sgr2,\ + self.security_group_rule(security_group_id, + direction='egress', + port_range_min=24, + port_range_max=24) as sgr3: # Delete default rules as they would fail the following # assertion at the end. @@ -1152,19 +1134,18 @@ class TestSecurityGroups(SecurityGroupDBTestCase): def test_list_security_group_rules_with_pagination_reverse(self): with self.security_group(name='sg') as sg: security_group_id = sg['security_group']['id'] - with contextlib.nested(self.security_group_rule(security_group_id, - direction='egress', - port_range_min=22, - port_range_max=22), - self.security_group_rule(security_group_id, - direction='egress', - port_range_min=23, - port_range_max=23), - self.security_group_rule(security_group_id, - direction='egress', - port_range_min=24, - port_range_max=24) - ) as (sgr1, sgr2, sgr3): + with self.security_group_rule(security_group_id, + direction='egress', + port_range_min=22, + port_range_max=22) as sgr1,\ + self.security_group_rule(security_group_id, + direction='egress', + port_range_min=23, + port_range_max=23) as sgr2,\ + self.security_group_rule(security_group_id, + direction='egress', + port_range_min=24, + port_range_max=24) as sgr3: self._test_list_with_pagination_reverse( 'security-group-rule', (sgr3, sgr2, sgr1), ('port_range_max', 'desc'), 2, 2, diff --git a/neutron/tests/unit/plugins/ibm/test_sdnve_agent.py b/neutron/tests/unit/plugins/ibm/test_sdnve_agent.py index a1e017fd4..a170704c8 100644 --- a/neutron/tests/unit/plugins/ibm/test_sdnve_agent.py +++ b/neutron/tests/unit/plugins/ibm/test_sdnve_agent.py @@ -14,8 +14,6 @@ # License for the specific language governing permissions and limitations # under the License. -import contextlib - import mock from oslo_config import cfg @@ -67,13 +65,12 @@ class TestSdnveNeutronAgent(base.BaseTestCase): def start(self, interval=0): self.f() - with contextlib.nested( - mock.patch('neutron.plugins.ibm.agent.sdnve_neutron_agent.' - 'SdnveNeutronAgent.setup_integration_br', - return_value=mock.Mock()), - mock.patch('neutron.openstack.common.loopingcall.' - 'FixedIntervalLoopingCall', - new=MockFixedIntervalLoopingCall)): + with mock.patch('neutron.plugins.ibm.agent.sdnve_neutron_agent.' + 'SdnveNeutronAgent.setup_integration_br', + return_value=mock.Mock()),\ + mock.patch('neutron.openstack.common.loopingcall.' + 'FixedIntervalLoopingCall', + new=MockFixedIntervalLoopingCall): self.agent = sdnve_neutron_agent.SdnveNeutronAgent(**kwargs) def test_setup_physical_interfaces(self): diff --git a/neutron/tests/unit/plugins/ibm/test_sdnve_plugin.py b/neutron/tests/unit/plugins/ibm/test_sdnve_plugin.py index 525468f15..ff79eafff 100644 --- a/neutron/tests/unit/plugins/ibm/test_sdnve_plugin.py +++ b/neutron/tests/unit/plugins/ibm/test_sdnve_plugin.py @@ -15,7 +15,6 @@ # under the License. -import contextlib import mock from neutron.extensions import portbindings @@ -71,13 +70,10 @@ class MockKeystoneClient(object): class IBMPluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase): def setUp(self): - with contextlib.nested( - mock.patch('neutron.plugins.ibm.sdnve_api.' - 'KeystoneClient', - new=MockKeystoneClient), - mock.patch('neutron.plugins.ibm.sdnve_api.' - 'Client', - new=MockClient)): + with mock.patch('neutron.plugins.ibm.sdnve_api.' 'KeystoneClient', + new=MockKeystoneClient),\ + mock.patch('neutron.plugins.ibm.sdnve_api.' 'Client', + new=MockClient): super(IBMPluginV2TestCase, self).setUp(plugin=_plugin_name) @@ -114,13 +110,10 @@ class TestIBMPortBinding(IBMPluginV2TestCase, class IBMPluginRouterTestCase(test_l3.L3NatDBIntTestCase): def setUp(self): - with contextlib.nested( - mock.patch('neutron.plugins.ibm.sdnve_api.' - 'KeystoneClient', - new=MockKeystoneClient), - mock.patch('neutron.plugins.ibm.sdnve_api.' - 'Client', - new=MockClient)): + with mock.patch('neutron.plugins.ibm.sdnve_api.' 'KeystoneClient', + new=MockKeystoneClient),\ + mock.patch('neutron.plugins.ibm.sdnve_api.' 'Client', + new=MockClient): super(IBMPluginRouterTestCase, self).setUp(plugin=_plugin_name) def test_floating_port_status_not_applicable(self): diff --git a/neutron/tests/unit/plugins/linuxbridge/agent/test_linuxbridge_neutron_agent.py b/neutron/tests/unit/plugins/linuxbridge/agent/test_linuxbridge_neutron_agent.py index 192637f28..11e923fd3 100644 --- a/neutron/tests/unit/plugins/linuxbridge/agent/test_linuxbridge_neutron_agent.py +++ b/neutron/tests/unit/plugins/linuxbridge/agent/test_linuxbridge_neutron_agent.py @@ -12,7 +12,6 @@ # License for the specific language governing permissions and limitations # under the License. -import contextlib import os import mock @@ -106,10 +105,10 @@ class TestLinuxBridgeAgent(base.BaseTestCase): def test_treat_devices_removed_with_existed_device(self): agent = self.agent devices = [DEVICE_1] - with contextlib.nested( - mock.patch.object(agent.plugin_rpc, "update_device_down"), - mock.patch.object(agent.sg_agent, "remove_devices_filter") - ) as (fn_udd, fn_rdf): + with mock.patch.object(agent.plugin_rpc, + "update_device_down") as fn_udd,\ + mock.patch.object(agent.sg_agent, + "remove_devices_filter") as fn_rdf: fn_udd.return_value = {'device': DEVICE_1, 'exists': True} with mock.patch.object(linuxbridge_neutron_agent.LOG, @@ -123,10 +122,10 @@ class TestLinuxBridgeAgent(base.BaseTestCase): def test_treat_devices_removed_with_not_existed_device(self): agent = self.agent devices = [DEVICE_1] - with contextlib.nested( - mock.patch.object(agent.plugin_rpc, "update_device_down"), - mock.patch.object(agent.sg_agent, "remove_devices_filter") - ) as (fn_udd, fn_rdf): + with mock.patch.object(agent.plugin_rpc, + "update_device_down") as fn_udd,\ + mock.patch.object(agent.sg_agent, + "remove_devices_filter") as fn_rdf: fn_udd.return_value = {'device': DEVICE_1, 'exists': False} with mock.patch.object(linuxbridge_neutron_agent.LOG, @@ -140,10 +139,10 @@ class TestLinuxBridgeAgent(base.BaseTestCase): def test_treat_devices_removed_failed(self): agent = self.agent devices = [DEVICE_1] - with contextlib.nested( - mock.patch.object(agent.plugin_rpc, "update_device_down"), - mock.patch.object(agent.sg_agent, "remove_devices_filter") - ) as (fn_udd, fn_rdf): + with mock.patch.object(agent.plugin_rpc, + "update_device_down") as fn_udd,\ + mock.patch.object(agent.sg_agent, + "remove_devices_filter") as fn_rdf: fn_udd.side_effect = Exception() with mock.patch.object(linuxbridge_neutron_agent.LOG, 'debug') as log: @@ -387,11 +386,9 @@ class TestLinuxBridgeManager(base.BaseTestCase): self.assertTrue(listdir_fn.called) def test_get_interfaces_on_bridge(self): - with contextlib.nested( - mock.patch.object(utils, 'execute'), - mock.patch.object(os, 'listdir'), - mock.patch.object(ip_lib, 'device_exists', return_value=True) - ) as (exec_fn, listdir_fn, dev_exists_fn): + with mock.patch.object(utils, 'execute'),\ + mock.patch.object(os, 'listdir') as listdir_fn,\ + mock.patch.object(ip_lib, 'device_exists', return_value=True): listdir_fn.return_value = ["qbr1"] self.assertEqual(self.lbm.get_interfaces_on_bridge("br0"), ["qbr1"]) @@ -408,10 +405,8 @@ class TestLinuxBridgeManager(base.BaseTestCase): self.assertEqual(self.lbm.get_tap_devices_count('br0'), 0) def test_get_interface_by_ip(self): - with contextlib.nested( - mock.patch.object(ip_lib.IPWrapper, 'get_devices'), - mock.patch.object(ip_lib.IpAddrCommand, 'list') - ) as (get_dev_fn, ip_list_fn): + with mock.patch.object(ip_lib.IPWrapper, 'get_devices') as get_dev_fn,\ + mock.patch.object(ip_lib.IpAddrCommand, 'list') as ip_list_fn: device = mock.Mock() device.name = 'dev_name' get_dev_fn.return_value = [device] @@ -420,10 +415,10 @@ class TestLinuxBridgeManager(base.BaseTestCase): 'dev_name') def test_get_bridge_for_tap_device(self): - with contextlib.nested( - mock.patch.object(self.lbm, "get_all_neutron_bridges"), - mock.patch.object(self.lbm, "get_interfaces_on_bridge") - ) as (get_all_qbr_fn, get_if_fn): + with mock.patch.object(self.lbm, + "get_all_neutron_bridges") as get_all_qbr_fn,\ + mock.patch.object(self.lbm, + "get_interfaces_on_bridge") as get_if_fn: get_all_qbr_fn.return_value = ["br-int", "br-ex"] get_if_fn.return_value = ["tap1", "tap2", "tap3"] self.assertEqual(self.lbm.get_bridge_for_tap_device("tap1"), @@ -440,10 +435,9 @@ class TestLinuxBridgeManager(base.BaseTestCase): ) def test_get_interface_details(self): - with contextlib.nested( - mock.patch.object(ip_lib.IpAddrCommand, 'list'), - mock.patch.object(ip_lib.IpRouteCommand, 'get_gateway') - ) as (list_fn, getgw_fn): + with mock.patch.object(ip_lib.IpAddrCommand, 'list') as list_fn,\ + mock.patch.object(ip_lib.IpRouteCommand, + 'get_gateway') as getgw_fn: gwdict = dict(gateway='1.1.1.1') getgw_fn.return_value = gwdict ipdict = dict(cidr='1.1.1.1/24', @@ -459,10 +453,9 @@ class TestLinuxBridgeManager(base.BaseTestCase): self.assertEqual(ret, (ipdict, gwdict)) def test_ensure_flat_bridge(self): - with contextlib.nested( - mock.patch.object(ip_lib.IpAddrCommand, 'list'), - mock.patch.object(ip_lib.IpRouteCommand, 'get_gateway') - ) as (list_fn, getgw_fn): + with mock.patch.object(ip_lib.IpAddrCommand, 'list') as list_fn,\ + mock.patch.object(ip_lib.IpRouteCommand, + 'get_gateway') as getgw_fn: gwdict = dict(gateway='1.1.1.1') getgw_fn.return_value = gwdict ipdict = dict(cidr='1.1.1.1/24', @@ -482,11 +475,10 @@ class TestLinuxBridgeManager(base.BaseTestCase): ipdict, gwdict) def test_ensure_vlan_bridge(self): - with contextlib.nested( - mock.patch.object(self.lbm, 'ensure_vlan'), - mock.patch.object(self.lbm, 'ensure_bridge'), - mock.patch.object(self.lbm, 'get_interface_details'), - ) as (ens_vl_fn, ens, get_int_det_fn): + with mock.patch.object(self.lbm, 'ensure_vlan') as ens_vl_fn,\ + mock.patch.object(self.lbm, 'ensure_bridge') as ens,\ + mock.patch.object(self.lbm, + 'get_interface_details') as get_int_det_fn: ens_vl_fn.return_value = "eth0.1" get_int_det_fn.return_value = (None, None) self.assertEqual(self.lbm.ensure_vlan_bridge("123", "eth0", "1"), @@ -549,19 +541,17 @@ class TestLinuxBridgeManager(base.BaseTestCase): scope='global', ip_version=4, dynamic=False) - with contextlib.nested( - mock.patch.object(ip_lib.IpAddrCommand, 'add'), - mock.patch.object(ip_lib.IpAddrCommand, 'delete') - ) as (add_fn, del_fn): + with mock.patch.object(ip_lib.IpAddrCommand, 'add') as add_fn,\ + mock.patch.object(ip_lib.IpAddrCommand, 'delete') as del_fn: self.lbm.update_interface_ip_details("br0", "eth0", [ipdict], None) self.assertTrue(add_fn.called) self.assertTrue(del_fn.called) - with contextlib.nested( - mock.patch.object(ip_lib.IpRouteCommand, 'add_gateway'), - mock.patch.object(ip_lib.IpRouteCommand, 'delete_gateway') - ) as (addgw_fn, delgw_fn): + with mock.patch.object(ip_lib.IpRouteCommand, + 'add_gateway') as addgw_fn,\ + mock.patch.object(ip_lib.IpRouteCommand, + 'delete_gateway') as delgw_fn: self.lbm.update_interface_ip_details("br0", "eth0", None, gwdict) self.assertTrue(addgw_fn.called) @@ -578,14 +568,16 @@ class TestLinuxBridgeManager(base.BaseTestCase): self.assertFalse(self.lbm._bridge_exists_and_ensure_up("br0")) def test_ensure_bridge(self): - with contextlib.nested( - mock.patch.object(self.lbm, '_bridge_exists_and_ensure_up'), - mock.patch.object(utils, 'execute'), - mock.patch.object(self.lbm, 'update_interface_ip_details'), - mock.patch.object(self.lbm, 'interface_exists_on_bridge'), - mock.patch.object(self.lbm, 'is_device_on_bridge'), - mock.patch.object(self.lbm, 'get_bridge_for_tap_device'), - ) as (de_fn, exec_fn, upd_fn, ie_fn, if_br_fn, get_if_br_fn): + with mock.patch.object(self.lbm, + '_bridge_exists_and_ensure_up') as de_fn,\ + mock.patch.object(utils, 'execute') as exec_fn,\ + mock.patch.object(self.lbm, + 'update_interface_ip_details') as upd_fn,\ + mock.patch.object(self.lbm, + 'interface_exists_on_bridge') as ie_fn,\ + mock.patch.object(self.lbm, 'is_device_on_bridge'),\ + mock.patch.object(self.lbm, + 'get_bridge_for_tap_device') as get_if_br_fn: de_fn.return_value = False exec_fn.return_value = False self.assertEqual(self.lbm.ensure_bridge("br0", None), "br0") @@ -652,11 +644,10 @@ class TestLinuxBridgeManager(base.BaseTestCase): ) de_fn.return_value = True - with contextlib.nested( - mock.patch.object(self.lbm, "ensure_local_bridge"), - mock.patch.object(utils, "execute"), - mock.patch.object(self.lbm, "get_bridge_for_tap_device") - ) as (en_fn, exec_fn, get_br): + with mock.patch.object(self.lbm, "ensure_local_bridge") as en_fn,\ + mock.patch.object(utils, "execute") as exec_fn,\ + mock.patch.object(self.lbm, + "get_bridge_for_tap_device") as get_br: exec_fn.return_value = False get_br.return_value = True self.assertTrue(self.lbm.add_tap_interface("123", @@ -672,11 +663,12 @@ class TestLinuxBridgeManager(base.BaseTestCase): "physnet1", None, "tap1")) - with contextlib.nested( - mock.patch.object(self.lbm, "ensure_physical_in_bridge"), - mock.patch.object(self.lbm, "ensure_tap_mtu"), - mock.patch.object(self.lbm, "get_bridge_for_tap_device") - ) as (ens_fn, en_mtu_fn, get_br): + with mock.patch.object(self.lbm, + "ensure_physical_in_bridge") as ens_fn,\ + mock.patch.object(self.lbm, + "ensure_tap_mtu") as en_mtu_fn,\ + mock.patch.object(self.lbm, + "get_bridge_for_tap_device") as get_br: ens_fn.return_value = False self.assertFalse(self.lbm.add_tap_interface("123", p_const.TYPE_VLAN, @@ -697,16 +689,16 @@ class TestLinuxBridgeManager(base.BaseTestCase): "1", "tap234") def test_delete_vlan_bridge(self): - with contextlib.nested( - mock.patch.object(ip_lib, "device_exists"), - mock.patch.object(self.lbm, "get_interfaces_on_bridge"), - mock.patch.object(self.lbm, "remove_interface"), - mock.patch.object(self.lbm, "get_interface_details"), - mock.patch.object(self.lbm, "update_interface_ip_details"), - mock.patch.object(self.lbm, "delete_vxlan"), - mock.patch.object(utils, "execute") - ) as (de_fn, getif_fn, remif_fn, if_det_fn, - updif_fn, del_vxlan, exec_fn): + with mock.patch.object(ip_lib, "device_exists") as de_fn,\ + mock.patch.object(self.lbm, + "get_interfaces_on_bridge") as getif_fn,\ + mock.patch.object(self.lbm, "remove_interface"),\ + mock.patch.object(self.lbm, + "get_interface_details") as if_det_fn,\ + mock.patch.object(self.lbm, + "update_interface_ip_details") as updif_fn,\ + mock.patch.object(self.lbm, "delete_vxlan") as del_vxlan,\ + mock.patch.object(utils, "execute") as exec_fn: de_fn.return_value = False self.lbm.delete_vlan_bridge("br0") self.assertFalse(getif_fn.called) @@ -720,16 +712,16 @@ class TestLinuxBridgeManager(base.BaseTestCase): del_vxlan.assert_called_with("vxlan-1002") def test_delete_vlan_bridge_with_ip(self): - with contextlib.nested( - mock.patch.object(ip_lib, "device_exists"), - mock.patch.object(self.lbm, "get_interfaces_on_bridge"), - mock.patch.object(self.lbm, "remove_interface"), - mock.patch.object(self.lbm, "get_interface_details"), - mock.patch.object(self.lbm, "update_interface_ip_details"), - mock.patch.object(self.lbm, "delete_vlan"), - mock.patch.object(utils, "execute") - ) as (de_fn, getif_fn, remif_fn, if_det_fn, - updif_fn, del_vlan, exec_fn): + with mock.patch.object(ip_lib, "device_exists") as de_fn,\ + mock.patch.object(self.lbm, + "get_interfaces_on_bridge") as getif_fn,\ + mock.patch.object(self.lbm, "remove_interface"),\ + mock.patch.object(self.lbm, + "get_interface_details") as if_det_fn,\ + mock.patch.object(self.lbm, + "update_interface_ip_details") as updif_fn,\ + mock.patch.object(self.lbm, "delete_vlan") as del_vlan,\ + mock.patch.object(utils, "execute") as exec_fn: de_fn.return_value = True getif_fn.return_value = ["eth0", "eth1.1"] if_det_fn.return_value = ("ips", "gateway") @@ -739,16 +731,16 @@ class TestLinuxBridgeManager(base.BaseTestCase): self.assertFalse(del_vlan.called) def test_delete_vlan_bridge_no_ip(self): - with contextlib.nested( - mock.patch.object(ip_lib, "device_exists"), - mock.patch.object(self.lbm, "get_interfaces_on_bridge"), - mock.patch.object(self.lbm, "remove_interface"), - mock.patch.object(self.lbm, "get_interface_details"), - mock.patch.object(self.lbm, "update_interface_ip_details"), - mock.patch.object(self.lbm, "delete_vlan"), - mock.patch.object(utils, "execute") - ) as (de_fn, getif_fn, remif_fn, if_det_fn, - updif_fn, del_vlan, exec_fn): + with mock.patch.object(ip_lib, "device_exists") as de_fn,\ + mock.patch.object(self.lbm, + "get_interfaces_on_bridge") as getif_fn,\ + mock.patch.object(self.lbm, "remove_interface"),\ + mock.patch.object(self.lbm, + "get_interface_details") as if_det_fn,\ + mock.patch.object(self.lbm, + "update_interface_ip_details") as updif_fn,\ + mock.patch.object(self.lbm, "delete_vlan") as del_vlan,\ + mock.patch.object(utils, "execute") as exec_fn: de_fn.return_value = True getif_fn.return_value = ["eth0", "eth1.1"] exec_fn.return_value = False @@ -764,13 +756,12 @@ class TestLinuxBridgeManager(base.BaseTestCase): lbm = linuxbridge_neutron_agent.LinuxBridgeManager( interface_mappings) - with contextlib.nested( - mock.patch.object(ip_lib, "device_exists"), - mock.patch.object(lbm, "get_interfaces_on_bridge"), - mock.patch.object(lbm, "remove_interface"), - mock.patch.object(lbm, "delete_vxlan"), - mock.patch.object(utils, "execute") - ) as (de_fn, getif_fn, remif_fn, del_vxlan, exec_fn): + with mock.patch.object(ip_lib, "device_exists") as de_fn,\ + mock.patch.object(lbm, + "get_interfaces_on_bridge") as getif_fn,\ + mock.patch.object(lbm, "remove_interface"),\ + mock.patch.object(lbm, "delete_vxlan") as del_vxlan,\ + mock.patch.object(utils, "execute") as exec_fn: de_fn.return_value = False lbm.delete_vlan_bridge("br0") self.assertFalse(getif_fn.called) @@ -787,20 +778,18 @@ class TestLinuxBridgeManager(base.BaseTestCase): def tap_count_side_effect(*args): return 0 if args[0] == 'brqnet1' else 1 - with contextlib.nested( - mock.patch.object(self.lbm, "delete_vlan_bridge"), - mock.patch.object(self.lbm, "get_tap_devices_count", - side_effect=tap_count_side_effect), - ) as (del_br_fn, count_tap_fn): + with mock.patch.object(self.lbm, "delete_vlan_bridge") as del_br_fn,\ + mock.patch.object(self.lbm, + "get_tap_devices_count", + side_effect=tap_count_side_effect): self.lbm.remove_empty_bridges() del_br_fn.assert_called_once_with('brqnet1') def test_remove_interface(self): - with contextlib.nested( - mock.patch.object(ip_lib, "device_exists"), - mock.patch.object(self.lbm, "is_device_on_bridge"), - mock.patch.object(utils, "execute") - ) as (de_fn, isdev_fn, exec_fn): + with mock.patch.object(ip_lib, "device_exists") as de_fn,\ + mock.patch.object(self.lbm, + "is_device_on_bridge") as isdev_fn,\ + mock.patch.object(utils, "execute") as exec_fn: de_fn.return_value = False self.assertFalse(self.lbm.remove_interface("br0", "eth0")) self.assertFalse(isdev_fn.called) @@ -817,10 +806,8 @@ class TestLinuxBridgeManager(base.BaseTestCase): self.assertTrue(self.lbm.remove_interface("br0", "eth0")) def test_delete_vlan(self): - with contextlib.nested( - mock.patch.object(ip_lib, "device_exists"), - mock.patch.object(utils, "execute") - ) as (de_fn, exec_fn): + with mock.patch.object(ip_lib, "device_exists") as de_fn,\ + mock.patch.object(utils, "execute") as exec_fn: de_fn.return_value = False self.lbm.delete_vlan("eth1.1") self.assertFalse(exec_fn.called) @@ -832,11 +819,12 @@ class TestLinuxBridgeManager(base.BaseTestCase): def _check_vxlan_support(self, expected, vxlan_ucast_supported, vxlan_mcast_supported): - with contextlib.nested( - mock.patch.object(self.lbm, 'vxlan_ucast_supported', - return_value=vxlan_ucast_supported), - mock.patch.object(self.lbm, 'vxlan_mcast_supported', - return_value=vxlan_mcast_supported)): + with mock.patch.object(self.lbm, + 'vxlan_ucast_supported', + return_value=vxlan_ucast_supported),\ + mock.patch.object(self.lbm, + 'vxlan_mcast_supported', + return_value=vxlan_mcast_supported): if expected == lconst.VXLAN_NONE: self.assertRaises(exceptions.VxlanNetworkUnsupported, self.lbm.check_vxlan_support) @@ -863,17 +851,20 @@ class TestLinuxBridgeManager(base.BaseTestCase): def _check_vxlan_ucast_supported( self, expected, l2_population, iproute_arg_supported, fdb_append): cfg.CONF.set_override('l2_population', l2_population, 'VXLAN') - with contextlib.nested( + with mock.patch.object(ip_lib, 'device_exists', return_value=False),\ + mock.patch.object(self.lbm, + 'delete_vxlan', + return_value=None),\ + mock.patch.object(self.lbm, + 'ensure_vxlan', + return_value=None),\ mock.patch.object( - ip_lib, 'device_exists', return_value=False), - mock.patch.object(self.lbm, 'delete_vxlan', return_value=None), - mock.patch.object(self.lbm, 'ensure_vxlan', return_value=None), - mock.patch.object( - utils, 'execute', - side_effect=None if fdb_append else RuntimeError()), - mock.patch.object( - ip_lib, 'iproute_arg_supported', - return_value=iproute_arg_supported)): + utils, + 'execute', + side_effect=None if fdb_append else RuntimeError()),\ + mock.patch.object(ip_lib, + 'iproute_arg_supported', + return_value=iproute_arg_supported): self.assertEqual(expected, self.lbm.vxlan_ucast_supported()) def test_vxlan_ucast_supported(self): @@ -940,10 +931,10 @@ class TestLinuxBridgeRpcCallbacks(base.BaseTestCase): ) def test_network_delete(self): - with contextlib.nested( - mock.patch.object(self.lb_rpc.agent.br_mgr, "get_bridge_name"), - mock.patch.object(self.lb_rpc.agent.br_mgr, "delete_vlan_bridge") - ) as (get_br_fn, del_fn): + with mock.patch.object(self.lb_rpc.agent.br_mgr, + "get_bridge_name") as get_br_fn,\ + mock.patch.object(self.lb_rpc.agent.br_mgr, + "delete_vlan_bridge") as del_fn: get_br_fn.return_value = "br0" self.lb_rpc.network_delete("anycontext", network_id="123") get_br_fn.assert_called_with("123") diff --git a/neutron/tests/unit/plugins/ml2/drivers/base_type_tunnel.py b/neutron/tests/unit/plugins/ml2/drivers/base_type_tunnel.py index cd5469fd0..0cc93f02e 100644 --- a/neutron/tests/unit/plugins/ml2/drivers/base_type_tunnel.py +++ b/neutron/tests/unit/plugins/ml2/drivers/base_type_tunnel.py @@ -13,7 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -import contextlib import mock from six import moves import testtools @@ -237,10 +236,10 @@ class TunnelRpcCallbackTestMixin(object): self.driver = self.DRIVER_CLASS() def _test_tunnel_sync(self, kwargs, delete_tunnel=False): - with contextlib.nested( - mock.patch.object(self.notifier, 'tunnel_update'), - mock.patch.object(self.notifier, 'tunnel_delete') - ) as (tunnel_update, tunnel_delete): + with mock.patch.object(self.notifier, + 'tunnel_update') as tunnel_update,\ + mock.patch.object(self.notifier, + 'tunnel_delete') as tunnel_delete: details = self.callbacks.tunnel_sync('fake_context', **kwargs) tunnels = details['tunnels'] for tunnel in tunnels: @@ -253,10 +252,10 @@ class TunnelRpcCallbackTestMixin(object): self.assertFalse(tunnel_delete.called) def _test_tunnel_sync_raises(self, kwargs): - with contextlib.nested( - mock.patch.object(self.notifier, 'tunnel_update'), - mock.patch.object(self.notifier, 'tunnel_delete') - ) as (tunnel_update, tunnel_delete): + with mock.patch.object(self.notifier, + 'tunnel_update') as tunnel_update,\ + mock.patch.object(self.notifier, + 'tunnel_delete') as tunnel_delete: self.assertRaises(exc.InvalidInput, self.callbacks.tunnel_sync, 'fake_context', **kwargs) diff --git a/neutron/tests/unit/plugins/ml2/drivers/l2pop/test_mech_driver.py b/neutron/tests/unit/plugins/ml2/drivers/l2pop/test_mech_driver.py index 22d661f5c..9275689fe 100644 --- a/neutron/tests/unit/plugins/ml2/drivers/l2pop/test_mech_driver.py +++ b/neutron/tests/unit/plugins/ml2/drivers/l2pop/test_mech_driver.py @@ -13,7 +13,6 @@ # License for the specific language governing permissions and limitations # under the License. -import contextlib import testtools import mock @@ -791,13 +790,11 @@ class TestL2PopulationRpcTestCase(test_plugin.Ml2PluginV2TestCase): l2pop_mech = l2pop_mech_driver.L2populationMechanismDriver() l2pop_mech.L2PopulationAgentNotify = mock.Mock() l2pop_mech.rpc_ctx = mock.Mock() - with contextlib.nested( - mock.patch.object(l2pop_mech, - '_update_port_down', - return_value=None), + with mock.patch.object(l2pop_mech, + '_update_port_down', + return_value=None) as upd_port_down,\ mock.patch.object(l2pop_mech.L2PopulationAgentNotify, - 'remove_fdb_entries')) as (upd_port_down, - rem_fdb_entries): + 'remove_fdb_entries'): l2pop_mech.delete_port_postcommit(mock.Mock()) self.assertTrue(upd_port_down.called) @@ -836,16 +833,15 @@ class TestL2PopulationMechDriver(base.BaseTestCase): def agent_ip_side_effect(agent): return agent_ips[agent] - with contextlib.nested( - mock.patch.object(l2pop_db.L2populationDbMixin, - 'get_agent_ip', - side_effect=agent_ip_side_effect), + with mock.patch.object(l2pop_db.L2populationDbMixin, + 'get_agent_ip', + side_effect=agent_ip_side_effect),\ mock.patch.object(l2pop_db.L2populationDbMixin, 'get_nondvr_active_network_ports', - new=fdb_network_ports_query), + new=fdb_network_ports_query),\ mock.patch.object(l2pop_db.L2populationDbMixin, 'get_dvr_active_network_ports', - new=tunnel_network_ports_query)): + new=tunnel_network_ports_query): session = mock.Mock() agent = mock.Mock() agent.host = HOST diff --git a/neutron/tests/unit/plugins/ml2/test_extension_driver_api.py b/neutron/tests/unit/plugins/ml2/test_extension_driver_api.py index 92e4957e9..bff70fecb 100644 --- a/neutron/tests/unit/plugins/ml2/test_extension_driver_api.py +++ b/neutron/tests/unit/plugins/ml2/test_extension_driver_api.py @@ -10,7 +10,6 @@ # License for the specific language governing permissions and limitations # under the License. -import contextlib import mock from neutron import context @@ -86,13 +85,11 @@ class ExtensionDriverTestCase(test_plugin.Ml2PluginV2TestCase): self.assertEqual('Test_Port_Extension_Update_update', val) def test_extend_network_dict(self): - with contextlib.nested( - mock.patch.object(ext_test.TestExtensionDriver, - 'process_update_network'), - mock.patch.object(ext_test.TestExtensionDriver, - 'extend_network_dict'), - self.network() - ) as (ext_update_net, ext_net_dict, network): + with mock.patch.object(ext_test.TestExtensionDriver, + 'process_update_network') as ext_update_net,\ + mock.patch.object(ext_test.TestExtensionDriver, + 'extend_network_dict') as ext_net_dict,\ + self.network() as network: net_id = network['network']['id'] net_data = {'network': {'id': net_id}} self._plugin.update_network(self._ctxt, net_id, net_data) @@ -100,13 +97,11 @@ class ExtensionDriverTestCase(test_plugin.Ml2PluginV2TestCase): self.assertTrue(ext_net_dict.called) def test_extend_subnet_dict(self): - with contextlib.nested( - mock.patch.object(ext_test.TestExtensionDriver, - 'process_update_subnet'), - mock.patch.object(ext_test.TestExtensionDriver, - 'extend_subnet_dict'), - self.subnet() - ) as (ext_update_subnet, ext_subnet_dict, subnet): + with mock.patch.object(ext_test.TestExtensionDriver, + 'process_update_subnet') as ext_update_subnet,\ + mock.patch.object(ext_test.TestExtensionDriver, + 'extend_subnet_dict') as ext_subnet_dict,\ + self.subnet() as subnet: subnet_id = subnet['subnet']['id'] subnet_data = {'subnet': {'id': subnet_id}} self._plugin.update_subnet(self._ctxt, subnet_id, subnet_data) @@ -114,13 +109,11 @@ class ExtensionDriverTestCase(test_plugin.Ml2PluginV2TestCase): self.assertTrue(ext_subnet_dict.called) def test_extend_port_dict(self): - with contextlib.nested( - mock.patch.object(ext_test.TestExtensionDriver, - 'process_update_port'), - mock.patch.object(ext_test.TestExtensionDriver, - 'extend_port_dict'), - self.port() - ) as (ext_update_port, ext_port_dict, port): + with mock.patch.object(ext_test.TestExtensionDriver, + 'process_update_port') as ext_update_port,\ + mock.patch.object(ext_test.TestExtensionDriver, + 'extend_port_dict') as ext_port_dict,\ + self.port() as port: port_id = port['port']['id'] port_data = {'port': {'id': port_id}} self._plugin.update_port(self._ctxt, port_id, port_data) diff --git a/neutron/tests/unit/plugins/ml2/test_plugin.py b/neutron/tests/unit/plugins/ml2/test_plugin.py index 8b49f6152..68b14f92f 100644 --- a/neutron/tests/unit/plugins/ml2/test_plugin.py +++ b/neutron/tests/unit/plugins/ml2/test_plugin.py @@ -13,7 +13,6 @@ # License for the specific language governing permissions and limitations # under the License. -import contextlib import functools import mock import six @@ -461,13 +460,11 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase): def test_create_ports_bulk_with_sec_grp(self): ctx = context.get_admin_context() plugin = manager.NeutronManager.get_plugin() - with contextlib.nested( - self.network(), - mock.patch.object(plugin.notifier, - 'security_groups_member_updated'), - mock.patch.object(plugin.notifier, - 'security_groups_provider_updated') - ) as (net, m_upd, p_upd): + with self.network() as net,\ + mock.patch.object(plugin.notifier, + 'security_groups_member_updated') as m_upd,\ + mock.patch.object(plugin.notifier, + 'security_groups_provider_updated') as p_upd: res = self._create_port_bulk(self.fmt, 3, net['network']['id'], 'test', True, context=ctx) @@ -479,13 +476,11 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase): def test_create_ports_bulk_with_sec_grp_member_provider_update(self): ctx = context.get_admin_context() plugin = manager.NeutronManager.get_plugin() - with contextlib.nested( - self.network(), - mock.patch.object(plugin.notifier, - 'security_groups_member_updated'), - mock.patch.object(plugin.notifier, - 'security_groups_provider_updated') - ) as (net, m_upd, p_upd): + with self.network() as net,\ + mock.patch.object(plugin.notifier, + 'security_groups_member_updated') as m_upd,\ + mock.patch.object(plugin.notifier, + 'security_groups_provider_updated') as p_upd: net_id = net['network']['id'] data = [{ @@ -520,14 +515,16 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase): fake_prefix = '2001:db8::/64' fake_gateway = 'fe80::1' with self.network() as net: - with contextlib.nested( - self.subnet(net, gateway_ip=fake_gateway, - cidr=fake_prefix, ip_version=6), - mock.patch.object( - plugin.notifier, 'security_groups_member_updated'), - mock.patch.object( - plugin.notifier, 'security_groups_provider_updated') - ) as (snet_v6, m_upd, p_upd): + with self.subnet(net, + gateway_ip=fake_gateway, + cidr=fake_prefix, + ip_version=6) as snet_v6,\ + mock.patch.object( + plugin.notifier, + 'security_groups_member_updated') as m_upd,\ + mock.patch.object( + plugin.notifier, + 'security_groups_provider_updated') as p_upd: net_id = net['network']['id'] data = [{ @@ -547,11 +544,11 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase): plugin = manager.NeutronManager.get_plugin() l3plugin = manager.NeutronManager.get_service_plugins().get( p_const.L3_ROUTER_NAT) - with contextlib.nested( - self.port(), - mock.patch.object(l3plugin, 'disassociate_floatingips'), - mock.patch.object(registry, 'notify') - ) as (port, disassociate_floatingips, notify): + with self.port() as port,\ + mock.patch.object( + l3plugin, + 'disassociate_floatingips') as disassociate_floatingips,\ + mock.patch.object(registry, 'notify') as notify: port_id = port['port']['id'] plugin.delete_port(ctx, port_id) @@ -662,18 +659,18 @@ class TestMl2DvrPortsV2(TestMl2PortsV2): if floating_ip: fip_set.add(ns_to_delete['router_id']) - with contextlib.nested( - mock.patch.object(manager.NeutronManager, - 'get_service_plugins', - return_value=self.service_plugins), - self.port(device_owner=device_owner), - mock.patch.object(registry, 'notify'), - mock.patch.object(self.l3plugin, 'disassociate_floatingips', - return_value=fip_set), - mock.patch.object(self.l3plugin, 'dvr_deletens_if_no_port', - return_value=[ns_to_delete]), - ) as (get_service_plugin, port, notify, disassociate_floatingips, - dvr_delns_ifno_port): + with mock.patch.object(manager.NeutronManager, + 'get_service_plugins', + return_value=self.service_plugins),\ + self.port(device_owner=device_owner) as port,\ + mock.patch.object(registry, 'notify') as notify,\ + mock.patch.object(self.l3plugin, + 'disassociate_floatingips', + return_value=fip_set),\ + mock.patch.object( + self.l3plugin, + 'dvr_deletens_if_no_port', + return_value=[ns_to_delete]) as dvr_delns_ifno_port: port_id = port['port']['id'] self.plugin.delete_port(self.context, port_id) @@ -790,12 +787,11 @@ class TestMl2PortBinding(Ml2PluginV2TestCase, plugin, self.context, port['port'], plugin.get_network(self.context, port['port']['network_id']), binding, None) - with contextlib.nested( - mock.patch('neutron.plugins.ml2.plugin.' - 'db.get_locked_port_and_binding', - return_value=(None, None)), - mock.patch('neutron.plugins.ml2.plugin.Ml2Plugin._make_port_dict') - ) as (glpab_mock, mpd_mock): + with mock.patch( + 'neutron.plugins.ml2.plugin.' 'db.get_locked_port_and_binding', + return_value=(None, None)) as glpab_mock,\ + mock.patch('neutron.plugins.ml2.plugin.Ml2Plugin.' + '_make_port_dict') as mpd_mock: plugin._bind_port_if_needed(mech_context) # called during deletion to get port self.assertTrue(glpab_mock.mock_calls) @@ -1489,10 +1485,9 @@ class TestMl2PluginCreateUpdateDeletePort(base.BaseTestCase): return plugin def test_create_port_rpc_outside_transaction(self): - with contextlib.nested( - mock.patch.object(ml2_plugin.Ml2Plugin, '__init__'), - mock.patch.object(base_plugin.NeutronDbPluginV2, 'create_port'), - ) as (init, super_create_port): + with mock.patch.object(ml2_plugin.Ml2Plugin, '__init__') as init,\ + mock.patch.object(base_plugin.NeutronDbPluginV2, + 'create_port'): init.return_value = None new_host_port = mock.Mock() @@ -1505,10 +1500,9 @@ class TestMl2PluginCreateUpdateDeletePort(base.BaseTestCase): plugin, **kwargs) def test_update_port_rpc_outside_transaction(self): - with contextlib.nested( - mock.patch.object(ml2_plugin.Ml2Plugin, '__init__'), - mock.patch.object(base_plugin.NeutronDbPluginV2, 'update_port'), - ) as (init, super_update_port): + with mock.patch.object(ml2_plugin.Ml2Plugin, '__init__') as init,\ + mock.patch.object(base_plugin.NeutronDbPluginV2, + 'update_port'): init.return_value = None new_host_port = mock.Mock() plugin = self._create_plugin_for_create_update_port(new_host_port) @@ -1531,13 +1525,12 @@ class TestMl2PluginCreateUpdateDeletePort(base.BaseTestCase): 'router', constants.L3_AGENT_SCHEDULER_EXT_ALIAS, constants.L3_DISTRIBUTED_EXT_ALIAS ] - with contextlib.nested( - mock.patch.object(ml2_plugin.Ml2Plugin, '__init__', - return_value=None), - mock.patch.object(manager.NeutronManager, - 'get_service_plugins', - return_value={'L3_ROUTER_NAT': l3plugin}), - ): + with mock.patch.object(ml2_plugin.Ml2Plugin, + '__init__', + return_value=None),\ + mock.patch.object(manager.NeutronManager, + 'get_service_plugins', + return_value={'L3_ROUTER_NAT': l3plugin}): plugin = self._create_plugin_for_create_update_port(mock.Mock()) # deleting the port will call registry.notify, which will # run the transaction balancing function defined in this test diff --git a/neutron/tests/unit/plugins/ml2/test_rpc.py b/neutron/tests/unit/plugins/ml2/test_rpc.py index 65c35aa0e..f0e1a3603 100644 --- a/neutron/tests/unit/plugins/ml2/test_rpc.py +++ b/neutron/tests/unit/plugins/ml2/test_rpc.py @@ -18,7 +18,6 @@ Unit Tests for ml2 rpc """ import collections -import contextlib import mock from oslo_config import cfg @@ -202,12 +201,8 @@ class RpcApiTestCase(base.BaseTestCase): expected_version = kwargs.pop('version', None) fanout = kwargs.pop('fanout', False) - with contextlib.nested( - mock.patch.object(rpcapi.client, rpc_method), - mock.patch.object(rpcapi.client, 'prepare'), - ) as ( - rpc_mock, prepare_mock - ): + with mock.patch.object(rpcapi.client, rpc_method) as rpc_mock,\ + mock.patch.object(rpcapi.client, 'prepare') as prepare_mock: prepare_mock.return_value = rpcapi.client rpc_mock.return_value = expected_retval retval = getattr(rpcapi, method)(ctxt, **kwargs) diff --git a/neutron/tests/unit/plugins/ml2/test_security_group.py b/neutron/tests/unit/plugins/ml2/test_security_group.py index 772853938..897cadf58 100644 --- a/neutron/tests/unit/plugins/ml2/test_security_group.py +++ b/neutron/tests/unit/plugins/ml2/test_security_group.py @@ -14,7 +14,6 @@ # License for the specific language governing permissions and limitations # under the License. -import contextlib import math import mock @@ -110,12 +109,11 @@ class TestMl2SecurityGroups(Ml2SecurityGroupsTestCase, max_ports_per_query = 5 ports_to_query = 73 for max_ports_per_query in (1, 2, 5, 7, 9, 31): - with contextlib.nested( - mock.patch('neutron.plugins.ml2.db.MAX_PORTS_PER_QUERY', - new=max_ports_per_query), - mock.patch('neutron.plugins.ml2.db.get_sg_ids_grouped_by_port', - return_value={}), - ) as (max_mock, get_mock): + with mock.patch('neutron.plugins.ml2.db.MAX_PORTS_PER_QUERY', + new=max_ports_per_query),\ + mock.patch( + 'neutron.plugins.ml2.db.get_sg_ids_grouped_by_port', + return_value={}) as get_mock: plugin.get_ports_from_devices(self.ctx, ['%s%s' % (const.TAP_DEVICE_PREFIX, i) for i in range(ports_to_query)]) @@ -139,10 +137,8 @@ class TestMl2SecurityGroups(Ml2SecurityGroupsTestCase, plugin = manager.NeutronManager.get_plugin() # when full UUIDs are provided, the _or statement should only # have one matching 'IN' critiera for all of the IDs - with contextlib.nested( - mock.patch('neutron.plugins.ml2.db.or_'), - mock.patch('sqlalchemy.orm.Session.query') - ) as (or_mock, qmock): + with mock.patch('neutron.plugins.ml2.db.or_') as or_mock,\ + mock.patch('sqlalchemy.orm.Session.query') as qmock: fmock = qmock.return_value.outerjoin.return_value.filter # return no ports to exit the method early since we are mocking # the query diff --git a/neutron/tests/unit/plugins/oneconvergence/test_nvsd_agent.py b/neutron/tests/unit/plugins/oneconvergence/test_nvsd_agent.py index 5835d89a7..769bf4f42 100644 --- a/neutron/tests/unit/plugins/oneconvergence/test_nvsd_agent.py +++ b/neutron/tests/unit/plugins/oneconvergence/test_nvsd_agent.py @@ -12,7 +12,6 @@ # License for the specific language governing permissions and limitations # under the License. -import contextlib import time import mock @@ -37,10 +36,8 @@ class TestOneConvergenceAgentBase(base.BaseTestCase): cfg.CONF.set_default('firewall_driver', 'neutron.agent.firewall.NoopFirewallDriver', group='SECURITYGROUP') - with contextlib.nested( - mock.patch('neutron.openstack.common.loopingcall.' - 'FixedIntervalLoopingCall'), - ) as (loopingcall): + with mock.patch('neutron.openstack.common.loopingcall.' + 'FixedIntervalLoopingCall') as loopingcall: kwargs = {'integ_br': 'integration_bridge', 'polling_interval': 5} context = mock.Mock() @@ -56,10 +53,10 @@ class TestOneConvergenceAgentBase(base.BaseTestCase): class TestOneConvergenceAgentCallback(TestOneConvergenceAgentBase): def test_port_update(self): - with contextlib.nested( - mock.patch.object(ovs_lib.OVSBridge, 'get_vif_port_by_id'), - mock.patch.object(self.sg_agent, 'refresh_firewall') - ) as (get_vif_port_by_id, refresh_firewall): + with mock.patch.object(ovs_lib.OVSBridge, + 'get_vif_port_by_id') as get_vif_port_by_id,\ + mock.patch.object(self.sg_agent, + 'refresh_firewall') as refresh_firewall: context = mock.Mock() vifport = ovs_lib.VifPort('port1', '1', 'id-1', 'mac-1', self.agent.int_br) @@ -129,13 +126,17 @@ class TestNVSDAgent(TestOneConvergenceAgentBase): [] for _i in moves.range(DAEMON_LOOP_COUNT - len(self.vif_ports_scenario))) - with contextlib.nested( - mock.patch.object(time, 'sleep', side_effect=sleep_mock), - mock.patch.object(ovs_lib.OVSBridge, 'get_vif_port_set'), - mock.patch.object(self.agent.sg_agent, 'prepare_devices_filter'), - mock.patch.object(self.agent.sg_agent, 'remove_devices_filter') - ) as (sleep, get_vif_port_set, prepare_devices_filter, - remove_devices_filter): + with mock.patch.object(time, + 'sleep', + side_effect=sleep_mock) as sleep,\ + mock.patch.object(ovs_lib.OVSBridge, + 'get_vif_port_set') as get_vif_port_set,\ + mock.patch.object( + self.agent.sg_agent, + 'prepare_devices_filter') as prepare_devices_filter,\ + mock.patch.object( + self.agent.sg_agent, + 'remove_devices_filter') as remove_devices_filter: get_vif_port_set.side_effect = self.vif_ports_scenario with testtools.ExpectedException(RuntimeError): @@ -158,11 +159,11 @@ class TestNVSDAgent(TestOneConvergenceAgentBase): class TestOneConvergenceAgentMain(base.BaseTestCase): def test_main(self): - with contextlib.nested( - mock.patch.object(nvsd_neutron_agent, 'NVSDNeutronAgent'), - mock.patch.object(nvsd_neutron_agent, 'common_config'), - mock.patch.object(nvsd_neutron_agent, 'config') - ) as (agent, common_config, config): + with mock.patch.object(nvsd_neutron_agent, + 'NVSDNeutronAgent') as agent,\ + mock.patch.object(nvsd_neutron_agent, + 'common_config') as common_config,\ + mock.patch.object(nvsd_neutron_agent, 'config') as config: config.AGENT.integration_bridge = 'br-int-dummy' config.AGENT.polling_interval = 5 diff --git a/neutron/tests/unit/plugins/oneconvergence/test_nvsd_plugin.py b/neutron/tests/unit/plugins/oneconvergence/test_nvsd_plugin.py index f8ac5f480..8058c813e 100644 --- a/neutron/tests/unit/plugins/oneconvergence/test_nvsd_plugin.py +++ b/neutron/tests/unit/plugins/oneconvergence/test_nvsd_plugin.py @@ -15,7 +15,6 @@ """Test Library for OneConvergencePlugin.""" -import contextlib import uuid import mock @@ -79,7 +78,7 @@ class TestOneConvergencePluginPortsV2(test_plugin.TestPortsV2, def test_ports_vif_details(self): cfg.CONF.set_default('allow_overlapping_ips', True) plugin = manager.NeutronManager.get_plugin() - with contextlib.nested(self.port(), self.port()) as (port1, port2): + with self.port(), self.port(): ctx = context.get_admin_context() ports = plugin.get_ports(ctx) self.assertEqual(len(ports), 2) diff --git a/neutron/tests/unit/plugins/openvswitch/agent/test_ovs_neutron_agent.py b/neutron/tests/unit/plugins/openvswitch/agent/test_ovs_neutron_agent.py index 9063df417..d29750dba 100644 --- a/neutron/tests/unit/plugins/openvswitch/agent/test_ovs_neutron_agent.py +++ b/neutron/tests/unit/plugins/openvswitch/agent/test_ovs_neutron_agent.py @@ -12,7 +12,6 @@ # License for the specific language governing permissions and limitations # under the License. -import contextlib import sys import time @@ -114,20 +113,19 @@ class TestOvsNeutronAgent(object): def start(self, interval=0): self.f() - with contextlib.nested( - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'setup_integration_br'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'setup_ancillary_bridges', - return_value=[]), - mock.patch('neutron.agent.linux.utils.get_interface_mac', - return_value='00:00:00:00:00:01'), - mock.patch('neutron.agent.common.ovs_lib.BaseOVS.get_bridges'), - mock.patch('neutron.openstack.common.loopingcall.' - 'FixedIntervalLoopingCall', - new=MockFixedIntervalLoopingCall), - mock.patch('neutron.agent.common.ovs_lib.OVSBridge.' - 'get_vif_ports', return_value=[])): + with mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'setup_integration_br'),\ + mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'setup_ancillary_bridges', + return_value=[]),\ + mock.patch('neutron.agent.linux.utils.get_interface_mac', + return_value='00:00:00:00:00:01'),\ + mock.patch( + 'neutron.agent.common.ovs_lib.BaseOVS.get_bridges'),\ + mock.patch('neutron.openstack.common.loopingcall.' 'FixedIntervalLoopingCall', new=MockFixedIntervalLoopingCall),\ + mock.patch( + 'neutron.agent.common.ovs_lib.OVSBridge.' 'get_vif_ports', + return_value=[]): self.agent = self.mod_agent.OVSNeutronAgent(self._bridge_classes(), **kwargs) # set back to true because initial report state will succeed due @@ -237,12 +235,12 @@ class TestOvsNeutronAgent(object): updated_ports=None, port_tags_dict=None): if port_tags_dict is None: # Because empty dicts evaluate as False. port_tags_dict = {} - with contextlib.nested( - mock.patch.object(self.agent.int_br, 'get_vif_port_set', - return_value=vif_port_set), - mock.patch.object(self.agent.int_br, 'get_port_tag_dict', - return_value=port_tags_dict) - ): + with mock.patch.object(self.agent.int_br, + 'get_vif_port_set', + return_value=vif_port_set),\ + mock.patch.object(self.agent.int_br, + 'get_port_tag_dict', + return_value=port_tags_dict): return self.agent.scan_ports(registered_ports, updated_ports) def test_scan_ports_returns_current_only_for_unchanged_ports(self): @@ -309,21 +307,19 @@ class TestOvsNeutronAgent(object): added=set([3]), current=vif_port_set, removed=set([2]), updated=set([1]) ) - with contextlib.nested( - mock.patch.dict(self.agent.local_vlan_map, local_vlan_map), - mock.patch.object(self.agent, 'tun_br', autospec=True), - ): + with mock.patch.dict(self.agent.local_vlan_map, local_vlan_map),\ + mock.patch.object(self.agent, 'tun_br', autospec=True): actual = self.mock_scan_ports( vif_port_set, registered_ports, port_tags_dict=port_tags_dict) self.assertEqual(expected, actual) def test_treat_devices_added_returns_raises_for_missing_device(self): - with contextlib.nested( - mock.patch.object(self.agent.plugin_rpc, - 'get_devices_details_list', - side_effect=Exception()), - mock.patch.object(self.agent.int_br, 'get_vif_port_by_id', - return_value=mock.Mock())): + with mock.patch.object(self.agent.plugin_rpc, + 'get_devices_details_list', + side_effect=Exception()),\ + mock.patch.object(self.agent.int_br, + 'get_vif_port_by_id', + return_value=mock.Mock()): self.assertRaises( self.mod_agent.DeviceListRetrievalError, self.agent.treat_devices_added_or_updated, [{}], False) @@ -336,16 +332,16 @@ class TestOvsNeutronAgent(object): :param func_name: the function that should be called :returns: whether the named function was called """ - with contextlib.nested( - mock.patch.object(self.agent.plugin_rpc, - 'get_devices_details_list', - return_value=[details]), - mock.patch.object(self.agent.int_br, 'get_vif_port_by_id', - return_value=port), - mock.patch.object(self.agent.plugin_rpc, 'update_device_up'), - mock.patch.object(self.agent.plugin_rpc, 'update_device_down'), - mock.patch.object(self.agent, func_name) - ) as (get_dev_fn, get_vif_func, upd_dev_up, upd_dev_down, func): + with mock.patch.object(self.agent.plugin_rpc, + 'get_devices_details_list', + return_value=[details]),\ + mock.patch.object(self.agent.int_br, + 'get_vif_port_by_id', + return_value=port),\ + mock.patch.object(self.agent.plugin_rpc, 'update_device_up'),\ + mock.patch.object(self.agent.plugin_rpc, + 'update_device_down'),\ + mock.patch.object(self.agent, func_name) as func: skip_devs, need_bound_devices = ( self.agent.treat_devices_added_or_updated([{}], False)) # The function should not raise @@ -365,11 +361,11 @@ class TestOvsNeutronAgent(object): mock.MagicMock(), port, 'port_dead')) def test_treat_devices_added_does_not_process_missing_port(self): - with contextlib.nested( - mock.patch.object(self.agent.plugin_rpc, 'get_device_details'), - mock.patch.object(self.agent.int_br, 'get_vif_port_by_id', - return_value=None) - ) as (get_dev_fn, get_vif_func): + with mock.patch.object(self.agent.plugin_rpc, + 'get_device_details') as get_dev_fn,\ + mock.patch.object(self.agent.int_br, + 'get_vif_port_by_id', + return_value=None): self.assertFalse(get_dev_fn.called) def test_treat_devices_added_updated_updates_known_port(self): @@ -381,14 +377,14 @@ class TestOvsNeutronAgent(object): def test_treat_devices_added_updated_skips_if_port_not_found(self): dev_mock = mock.MagicMock() dev_mock.__getitem__.return_value = 'the_skipped_one' - with contextlib.nested( - mock.patch.object(self.agent.plugin_rpc, - 'get_devices_details_list', - return_value=[dev_mock]), - mock.patch.object(self.agent.int_br, 'get_vif_port_by_id', - return_value=None), - mock.patch.object(self.agent, 'treat_vif_port') - ) as (get_dev_fn, get_vif_func, treat_vif_port): + with mock.patch.object(self.agent.plugin_rpc, + 'get_devices_details_list', + return_value=[dev_mock]),\ + mock.patch.object(self.agent.int_br, + 'get_vif_port_by_id', + return_value=None),\ + mock.patch.object(self.agent, + 'treat_vif_port') as treat_vif_port: skip_devs = self.agent.treat_devices_added_or_updated([{}], False) # The function should return False for resync and no device # processed @@ -408,14 +404,14 @@ class TestOvsNeutronAgent(object): 'device_owner': 'compute:None' } - with contextlib.nested( - mock.patch.object(self.agent.plugin_rpc, - 'get_devices_details_list', - return_value=[fake_details_dict]), - mock.patch.object(self.agent.int_br, 'get_vif_port_by_id', - return_value=mock.MagicMock()), - mock.patch.object(self.agent, 'treat_vif_port') - ) as (get_dev_fn, get_vif_func, treat_vif_port): + with mock.patch.object(self.agent.plugin_rpc, + 'get_devices_details_list', + return_value=[fake_details_dict]),\ + mock.patch.object(self.agent.int_br, + 'get_vif_port_by_id', + return_value=mock.MagicMock()),\ + mock.patch.object(self.agent, + 'treat_vif_port') as treat_vif_port: skip_devs, need_bound_devices = ( self.agent.treat_devices_added_or_updated([{}], False)) # The function should return False for resync @@ -445,13 +441,15 @@ class TestOvsNeutronAgent(object): self.agent._bind_devices([{'network_id': 'non-existent'}]) def _test_process_network_ports(self, port_info): - with contextlib.nested( - mock.patch.object(self.agent.sg_agent, "setup_port_filters"), - mock.patch.object(self.agent, "treat_devices_added_or_updated", - return_value=([], [])), - mock.patch.object(self.agent, "treat_devices_removed", - return_value=False) - ) as (setup_port_filters, device_added_updated, device_removed): + with mock.patch.object(self.agent.sg_agent, + "setup_port_filters") as setup_port_filters,\ + mock.patch.object( + self.agent, + "treat_devices_added_or_updated", + return_value=([], [])) as device_added_updated,\ + mock.patch.object(self.agent, + "treat_devices_removed", + return_value=False) as device_removed: self.assertFalse(self.agent.process_network_ports(port_info, False)) setup_port_filters.assert_called_once_with( @@ -510,10 +508,9 @@ class TestOvsNeutronAgent(object): self.agent.agent_state, True) def test_network_delete(self): - with contextlib.nested( - mock.patch.object(self.agent, "reclaim_local_vlan"), - mock.patch.object(self.agent.tun_br, "cleanup_tunnel_port") - ) as (recl_fn, clean_tun_fn): + with mock.patch.object(self.agent, "reclaim_local_vlan") as recl_fn,\ + mock.patch.object(self.agent.tun_br, + "cleanup_tunnel_port") as clean_tun_fn: self.agent.network_delete("unused_context", network_id="123") self.assertFalse(recl_fn.called) @@ -537,26 +534,23 @@ class TestOvsNeutronAgent(object): def test_port_delete(self): port_id = "123" port_name = "foo" - with contextlib.nested( - mock.patch.object(self.agent.int_br, 'get_vif_port_by_id', - return_value=mock.MagicMock( - port_name=port_name)), - mock.patch.object(self.agent.int_br, "delete_port") - ) as (get_vif_func, del_port_func): + with mock.patch.object( + self.agent.int_br, + 'get_vif_port_by_id', + return_value=mock.MagicMock(port_name=port_name)) as get_vif_func,\ + mock.patch.object(self.agent.int_br, + "delete_port") as del_port_func: self.agent.port_delete("unused_context", port_id=port_id) self.assertTrue(get_vif_func.called) del_port_func.assert_called_once_with(port_name) def test_setup_physical_bridges(self): - with contextlib.nested( - mock.patch.object(ip_lib, "device_exists"), - mock.patch.object(sys, "exit"), - mock.patch.object(utils, "execute"), - mock.patch.object(self.agent, 'br_phys_cls'), - mock.patch.object(self.agent, 'int_br'), - ) as (devex_fn, sysexit_fn, utilsexec_fn, - phys_br_cls, int_br): + with mock.patch.object(ip_lib, "device_exists") as devex_fn,\ + mock.patch.object(sys, "exit"),\ + mock.patch.object(utils, "execute"),\ + mock.patch.object(self.agent, 'br_phys_cls') as phys_br_cls,\ + mock.patch.object(self.agent, 'int_br') as int_br: devex_fn.return_value = True parent = mock.MagicMock() phys_br = phys_br_cls() @@ -593,19 +587,17 @@ class TestOvsNeutronAgent(object): def test_setup_physical_bridges_using_veth_interconnection(self): self.agent.use_veth_interconnection = True - with contextlib.nested( - mock.patch.object(ip_lib, "device_exists"), - mock.patch.object(sys, "exit"), - mock.patch.object(utils, "execute"), - mock.patch.object(self.agent, 'br_phys_cls'), - mock.patch.object(self.agent, 'int_br'), - mock.patch.object(ip_lib.IPWrapper, "add_veth"), - mock.patch.object(ip_lib.IpLinkCommand, "delete"), - mock.patch.object(ip_lib.IpLinkCommand, "set_up"), - mock.patch.object(ip_lib.IpLinkCommand, "set_mtu"), - mock.patch.object(ovs_lib.BaseOVS, "get_bridges") - ) as (devex_fn, sysexit_fn, utilsexec_fn, phys_br_cls, int_br, - addveth_fn, linkdel_fn, linkset_fn, linkmtu_fn, get_br_fn): + with mock.patch.object(ip_lib, "device_exists") as devex_fn,\ + mock.patch.object(sys, "exit"),\ + mock.patch.object(utils, "execute") as utilsexec_fn,\ + mock.patch.object(self.agent, 'br_phys_cls') as phys_br_cls,\ + mock.patch.object(self.agent, 'int_br') as int_br,\ + mock.patch.object(ip_lib.IPWrapper, "add_veth") as addveth_fn,\ + mock.patch.object(ip_lib.IpLinkCommand, + "delete") as linkdel_fn,\ + mock.patch.object(ip_lib.IpLinkCommand, "set_up"),\ + mock.patch.object(ip_lib.IpLinkCommand, "set_mtu"),\ + mock.patch.object(ovs_lib.BaseOVS, "get_bridges") as get_br_fn: devex_fn.return_value = True parent = mock.MagicMock() parent.attach_mock(utilsexec_fn, 'utils_execute') @@ -643,12 +635,13 @@ class TestOvsNeutronAgent(object): def test_setup_tunnel_br(self): self.tun_br = mock.Mock() - with contextlib.nested( - mock.patch.object(self.agent.int_br, "add_patch_port", - return_value=1), - mock.patch.object(self.agent, 'tun_br', autospec=True), - mock.patch.object(sys, "exit") - ) as (intbr_patch_fn, tun_br, exit_fn): + with mock.patch.object(self.agent.int_br, + "add_patch_port", + return_value=1) as intbr_patch_fn,\ + mock.patch.object(self.agent, + 'tun_br', + autospec=True) as tun_br,\ + mock.patch.object(sys, "exit"): tun_br.add_patch_port.return_value = 2 self.agent.reset_tunnel_br(None) self.agent.setup_tunnel_br() @@ -659,11 +652,10 @@ class TestOvsNeutronAgent(object): self.agent.l2_pop = False self.agent.udp_vxlan_port = 8472 self.agent.tun_br_ofports['vxlan'] = {} - with contextlib.nested( - mock.patch.object(self.agent.tun_br, "add_tunnel_port", - return_value='6'), - mock.patch.object(self.agent.tun_br, "add_flow") - ) as (add_tun_port_fn, add_flow_fn): + with mock.patch.object(self.agent.tun_br, + "add_tunnel_port", + return_value='6') as add_tun_port_fn,\ + mock.patch.object(self.agent.tun_br, "add_flow"): self.agent._setup_tunnel_port(self.agent.tun_br, 'portname', '1.2.3.4', 'vxlan') self.assertTrue(add_tun_port_fn.called) @@ -706,12 +698,13 @@ class TestOvsNeutronAgent(object): def test_fdb_ignore_network(self): self._prepare_l2_pop_ofports() fdb_entry = {'net3': {}} - with contextlib.nested( - mock.patch.object(self.agent.tun_br, 'add_flow'), - mock.patch.object(self.agent.tun_br, 'delete_flows'), - mock.patch.object(self.agent, '_setup_tunnel_port'), - mock.patch.object(self.agent, 'cleanup_tunnel_port') - ) as (add_flow_fn, del_flow_fn, add_tun_fn, clean_tun_fn): + with mock.patch.object(self.agent.tun_br, 'add_flow') as add_flow_fn,\ + mock.patch.object(self.agent.tun_br, + 'delete_flows') as del_flow_fn,\ + mock.patch.object(self.agent, + '_setup_tunnel_port') as add_tun_fn,\ + mock.patch.object(self.agent, + 'cleanup_tunnel_port') as clean_tun_fn: self.agent.fdb_add(None, fdb_entry) self.assertFalse(add_flow_fn.called) self.assertFalse(add_tun_fn.called) @@ -747,10 +740,10 @@ class TestOvsNeutronAgent(object): [l2pop_rpc.PortInfo(FAKE_MAC, FAKE_IP1), n_const.FLOODING_ENTRY]}}} - with contextlib.nested( - mock.patch.object(self.agent, 'tun_br', autospec=True), - mock.patch.object(self.agent, '_setup_tunnel_port', autospec=True), - ) as (tun_br, add_tun_fn): + with mock.patch.object(self.agent, 'tun_br', autospec=True) as tun_br,\ + mock.patch.object(self.agent, + '_setup_tunnel_port', + autospec=True) as add_tun_fn: self.agent.fdb_add(None, fdb_entry) self.assertFalse(add_tun_fn.called) deferred_br_call = mock.call.deferred().__enter__() @@ -796,10 +789,9 @@ class TestOvsNeutronAgent(object): 'segment_id': 'tun1', 'ports': {'1.1.1.1': [l2pop_rpc.PortInfo(FAKE_MAC, FAKE_IP1)]}}} - with contextlib.nested( - mock.patch.object(self.agent, 'tun_br', autospec=True), - mock.patch.object(self.agent, '_setup_tunnel_port') - ) as (tun_br, add_tun_fn): + with mock.patch.object(self.agent, 'tun_br', autospec=True) as tun_br,\ + mock.patch.object(self.agent, + '_setup_tunnel_port') as add_tun_fn: self.agent.fdb_add(None, fdb_entry) self.assertFalse(add_tun_fn.called) fdb_entry['net1']['ports']['10.10.10.10'] = [ @@ -815,10 +807,9 @@ class TestOvsNeutronAgent(object): {'network_type': 'gre', 'segment_id': 'tun2', 'ports': {'2.2.2.2': [n_const.FLOODING_ENTRY]}}} - with contextlib.nested( - mock.patch.object(self.agent.tun_br, 'deferred'), - mock.patch.object(self.agent.tun_br, 'delete_port'), - ) as (defer_fn, delete_port_fn): + with mock.patch.object(self.agent.tun_br, 'deferred') as defer_fn,\ + mock.patch.object(self.agent.tun_br, + 'delete_port') as delete_port_fn: self.agent.fdb_remove(None, fdb_entry) deferred_br = defer_fn().__enter__() deferred_br.delete_port.assert_called_once_with('gre-02020202') @@ -845,10 +836,9 @@ class TestOvsNeutronAgent(object): lvm.vlan = 'vlan1' lvm.segmentation_id = 'seg1' lvm.tun_ofports = set(['1', '2']) - with contextlib.nested( - mock.patch.object(self.agent.tun_br, 'mod_flow'), - mock.patch.object(self.agent.tun_br, 'delete_flows') - ) as (mod_flow_fn, delete_flows_fn): + with mock.patch.object(self.agent.tun_br, 'mod_flow') as mod_flow_fn,\ + mock.patch.object(self.agent.tun_br, + 'delete_flows') as delete_flows_fn: self.agent.del_fdb_flow(self.agent.tun_br, n_const.FLOODING_ENTRY, '1.1.1.1', lvm, '3') self.assertFalse(mod_flow_fn.called) @@ -880,11 +870,11 @@ class TestOvsNeutronAgent(object): mock_loop.assert_called_once_with(polling_manager=mock.ANY) def test_setup_tunnel_port_invalid_ofport(self): - with contextlib.nested( - mock.patch.object(self.agent.tun_br, 'add_tunnel_port', - return_value=ovs_lib.INVALID_OFPORT), - mock.patch.object(self.mod_agent.LOG, 'error') - ) as (add_tunnel_port_fn, log_error_fn): + with mock.patch.object( + self.agent.tun_br, + 'add_tunnel_port', + return_value=ovs_lib.INVALID_OFPORT) as add_tunnel_port_fn,\ + mock.patch.object(self.mod_agent.LOG, 'error') as log_error_fn: ofport = self.agent._setup_tunnel_port( self.agent.tun_br, 'gre-1', 'remote_ip', p_const.TYPE_GRE) add_tunnel_port_fn.assert_called_once_with( @@ -896,11 +886,11 @@ class TestOvsNeutronAgent(object): self.assertEqual(ofport, 0) def test_setup_tunnel_port_error_negative_df_disabled(self): - with contextlib.nested( - mock.patch.object(self.agent.tun_br, 'add_tunnel_port', - return_value=ovs_lib.INVALID_OFPORT), - mock.patch.object(self.mod_agent.LOG, 'error') - ) as (add_tunnel_port_fn, log_error_fn): + with mock.patch.object( + self.agent.tun_br, + 'add_tunnel_port', + return_value=ovs_lib.INVALID_OFPORT) as add_tunnel_port_fn,\ + mock.patch.object(self.mod_agent.LOG, 'error') as log_error_fn: self.agent.dont_fragment = False ofport = self.agent._setup_tunnel_port( self.agent.tun_br, 'gre-1', 'remote_ip', p_const.TYPE_GRE) @@ -914,11 +904,12 @@ class TestOvsNeutronAgent(object): def test_tunnel_sync_with_ml2_plugin(self): fake_tunnel_details = {'tunnels': [{'ip_address': '100.101.31.15'}]} - with contextlib.nested( - mock.patch.object(self.agent.plugin_rpc, 'tunnel_sync', - return_value=fake_tunnel_details), - mock.patch.object(self.agent, '_setup_tunnel_port') - ) as (tunnel_sync_rpc_fn, _setup_tunnel_port_fn): + with mock.patch.object(self.agent.plugin_rpc, + 'tunnel_sync', + return_value=fake_tunnel_details),\ + mock.patch.object( + self.agent, + '_setup_tunnel_port') as _setup_tunnel_port_fn: self.agent.tunnel_types = ['vxlan'] self.agent.tunnel_sync() expected_calls = [mock.call(self.agent.tun_br, 'vxlan-64651f0f', @@ -928,11 +919,12 @@ class TestOvsNeutronAgent(object): def test_tunnel_sync_invalid_ip_address(self): fake_tunnel_details = {'tunnels': [{'ip_address': '300.300.300.300'}, {'ip_address': '100.100.100.100'}]} - with contextlib.nested( - mock.patch.object(self.agent.plugin_rpc, 'tunnel_sync', - return_value=fake_tunnel_details), - mock.patch.object(self.agent, '_setup_tunnel_port') - ) as (tunnel_sync_rpc_fn, _setup_tunnel_port_fn): + with mock.patch.object(self.agent.plugin_rpc, + 'tunnel_sync', + return_value=fake_tunnel_details),\ + mock.patch.object( + self.agent, + '_setup_tunnel_port') as _setup_tunnel_port_fn: self.agent.tunnel_types = ['vxlan'] self.agent.tunnel_sync() _setup_tunnel_port_fn.assert_called_once_with(self.agent.tun_br, @@ -973,25 +965,23 @@ class TestOvsNeutronAgent(object): 'added': set([]), 'removed': set(['tap0'])} - with contextlib.nested( - mock.patch.object(async_process.AsyncProcess, "_spawn"), - mock.patch.object(log.KeywordArgumentAdapter, 'exception'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'scan_ports'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'process_network_ports'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'check_ovs_status'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'setup_integration_br'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'setup_physical_bridges'), - mock.patch.object(time, 'sleep'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'update_stale_ofport_rules') - ) as (spawn_fn, log_exception, scan_ports, process_network_ports, - check_ovs_status, setup_int_br, setup_phys_br, time_sleep, - update_stale): + with mock.patch.object(async_process.AsyncProcess, "_spawn"),\ + mock.patch.object(log.KeywordArgumentAdapter, + 'exception') as log_exception,\ + mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'scan_ports') as scan_ports,\ + mock.patch.object( + self.mod_agent.OVSNeutronAgent, + 'process_network_ports') as process_network_ports,\ + mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'check_ovs_status') as check_ovs_status,\ + mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'setup_integration_br') as setup_int_br,\ + mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'setup_physical_bridges') as setup_phys_br,\ + mock.patch.object(time, 'sleep'),\ + mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'update_stale_ofport_rules') as update_stale: log_exception.side_effect = Exception( 'Fake exception to get out of the loop') scan_ports.side_effect = [reply2, reply3] @@ -1182,18 +1172,16 @@ class AncillaryBridgesTest(object): except Exception: return None - with contextlib.nested( - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'setup_integration_br'), - mock.patch('neutron.agent.linux.utils.get_interface_mac', - return_value='00:00:00:00:00:01'), - mock.patch('neutron.agent.common.ovs_lib.BaseOVS.get_bridges', - return_value=bridges), - mock.patch('neutron.agent.common.ovs_lib.BaseOVS.' - 'get_bridge_external_bridge_id', - side_effect=pullup_side_effect), - mock.patch('neutron.agent.common.ovs_lib.OVSBridge.' - 'get_vif_ports', return_value=[])): + with mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'setup_integration_br'),\ + mock.patch('neutron.agent.linux.utils.get_interface_mac', + return_value='00:00:00:00:00:01'),\ + mock.patch('neutron.agent.common.ovs_lib.BaseOVS.get_bridges', + return_value=bridges),\ + mock.patch('neutron.agent.common.ovs_lib.BaseOVS.' 'get_bridge_external_bridge_id', side_effect=pullup_side_effect),\ + mock.patch( + 'neutron.agent.common.ovs_lib.OVSBridge.' 'get_vif_ports', + return_value=[]): self.agent = self.mod_agent.OVSNeutronAgent(self._bridge_classes(), **self.kwargs) self.assertEqual(len(ancillary), len(self.agent.ancillary_brs)) @@ -1240,20 +1228,19 @@ class TestOvsDvrNeutronAgent(object): def start(self, interval=0): self.f() - with contextlib.nested( - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'setup_integration_br'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'setup_ancillary_bridges', - return_value=[]), - mock.patch('neutron.agent.linux.utils.get_interface_mac', - return_value='00:00:00:00:00:01'), - mock.patch('neutron.agent.common.ovs_lib.BaseOVS.get_bridges'), - mock.patch('neutron.openstack.common.loopingcall.' - 'FixedIntervalLoopingCall', - new=MockFixedIntervalLoopingCall), - mock.patch('neutron.agent.common.ovs_lib.OVSBridge.' - 'get_vif_ports', return_value=[])): + with mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'setup_integration_br'),\ + mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'setup_ancillary_bridges', + return_value=[]),\ + mock.patch('neutron.agent.linux.utils.get_interface_mac', + return_value='00:00:00:00:00:01'),\ + mock.patch( + 'neutron.agent.common.ovs_lib.BaseOVS.get_bridges'),\ + mock.patch('neutron.openstack.common.loopingcall.' 'FixedIntervalLoopingCall', new=MockFixedIntervalLoopingCall),\ + mock.patch( + 'neutron.agent.common.ovs_lib.OVSBridge.' 'get_vif_ports', + return_value=[]): self.agent = self.mod_agent.OVSNeutronAgent(self._bridge_classes(), **kwargs) # set back to true because initial report state will succeed due @@ -1346,29 +1333,26 @@ class TestOvsDvrNeutronAgent(object): phys_br = mock.create_autospec(self.br_phys_cls('br-phys')) int_br.set_db_attribute.return_value = True int_br.db_get_val.return_value = {} - with contextlib.nested( - mock.patch.object(self.agent.dvr_agent.plugin_rpc, - 'get_subnet_for_dvr', - return_value={ - 'gateway_ip': gateway_ip, - 'cidr': cidr, - 'ip_version': ip_version, - 'gateway_mac': gateway_mac}), - mock.patch.object(self.agent.dvr_agent.plugin_rpc, - 'get_ports_on_host_by_subnet', - return_value=[]), - mock.patch.object(self.agent.dvr_agent.int_br, - 'get_vif_port_by_id', - return_value=self._port), - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.dict(self.agent.phys_brs, - {physical_network: phys_br}), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - mock.patch.dict(self.agent.dvr_agent.phys_brs, - {physical_network: phys_br}), - ) as (get_subnet_fn, get_cphost_fn, get_vif_fn, _, _, _, _, _, _): + with mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_subnet_for_dvr', + return_value={'gateway_ip': gateway_ip, + 'cidr': cidr, + 'ip_version': ip_version, + 'gateway_mac': gateway_mac}),\ + mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_ports_on_host_by_subnet', + return_value=[]),\ + mock.patch.object(self.agent.dvr_agent.int_br, + 'get_vif_port_by_id', + return_value=self._port),\ + mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.dict(self.agent.phys_brs, + {physical_network: phys_br}),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\ + mock.patch.dict(self.agent.dvr_agent.phys_brs, + {physical_network: phys_br}): self.agent.port_bound( self._port, self._net_uuid, network_type, physical_network, segmentation_id, self._fixed_ips, @@ -1439,29 +1423,26 @@ class TestOvsDvrNeutronAgent(object): phys_br = mock.create_autospec(self.br_phys_cls('br-phys')) int_br.set_db_attribute.return_value = True int_br.db_get_val.return_value = {} - with contextlib.nested( - mock.patch.object(self.agent.dvr_agent.plugin_rpc, - 'get_subnet_for_dvr', - return_value={ - 'gateway_ip': gateway_ip, - 'cidr': cidr, - 'ip_version': ip_version, - 'gateway_mac': gateway_mac}), - mock.patch.object(self.agent.dvr_agent.plugin_rpc, - 'get_ports_on_host_by_subnet', - return_value=[]), - mock.patch.object(self.agent.dvr_agent.int_br, - 'get_vif_port_by_id', - return_value=self._port), - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.dict(self.agent.phys_brs, - {physical_network: phys_br}), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - mock.patch.dict(self.agent.dvr_agent.phys_brs, - {physical_network: phys_br}), - ) as (get_subnet_fn, get_cphost_fn, get_vif_fn, _, _, _, _, _, _): + with mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_subnet_for_dvr', + return_value={'gateway_ip': gateway_ip, + 'cidr': cidr, + 'ip_version': ip_version, + 'gateway_mac': gateway_mac}),\ + mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_ports_on_host_by_subnet', + return_value=[]),\ + mock.patch.object(self.agent.dvr_agent.int_br, + 'get_vif_port_by_id', + return_value=self._port),\ + mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.dict(self.agent.phys_brs, + {physical_network: phys_br}),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\ + mock.patch.dict(self.agent.dvr_agent.phys_brs, + {physical_network: phys_br}): self.agent.port_bound( self._port, self._net_uuid, network_type, physical_network, segmentation_id, self._fixed_ips, @@ -1541,24 +1522,22 @@ class TestOvsDvrNeutronAgent(object): tun_br = mock.create_autospec(self.agent.tun_br) int_br.set_db_attribute.return_value = True int_br.db_get_val.return_value = {} - with contextlib.nested( - mock.patch.object( - self.agent.dvr_agent.plugin_rpc, 'get_subnet_for_dvr', - return_value={'gateway_ip': '1.1.1.1', - 'cidr': '1.1.1.0/24', - 'ip_version': 4, - 'gateway_mac': 'aa:bb:cc:11:22:33'}), - mock.patch.object(self.agent.dvr_agent.plugin_rpc, - 'get_ports_on_host_by_subnet', - return_value=[]), - mock.patch.object(self.agent.dvr_agent.int_br, - 'get_vif_port_by_id', - return_value=self._port), - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - ) as (get_subnet_fn, get_cphost_fn, get_vif_fn, _, _, _, _): + with mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_subnet_for_dvr', + return_value={'gateway_ip': '1.1.1.1', + 'cidr': '1.1.1.0/24', + 'ip_version': 4, + 'gateway_mac': 'aa:bb:cc:11:22:33'}),\ + mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_ports_on_host_by_subnet', + return_value=[]),\ + mock.patch.object(self.agent.dvr_agent.int_br, + 'get_vif_port_by_id', + return_value=self._port),\ + mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br): self.agent.port_bound( self._port, self._net_uuid, 'vxlan', None, None, self._fixed_ips, @@ -1604,24 +1583,22 @@ class TestOvsDvrNeutronAgent(object): tun_br = mock.create_autospec(self.agent.tun_br) int_br.set_db_attribute.return_value = True int_br.db_get_val.return_value = {} - with contextlib.nested( - mock.patch.object( - self.agent.dvr_agent.plugin_rpc, 'get_subnet_for_dvr', - return_value={'gateway_ip': gateway_ip, - 'cidr': cidr, - 'ip_version': ip_version, - 'gateway_mac': gateway_mac}), - mock.patch.object(self.agent.dvr_agent.plugin_rpc, - 'get_ports_on_host_by_subnet', - return_value=[]), - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - mock.patch.object(self.agent.dvr_agent.int_br, - 'get_vif_port_by_id', - return_value=self._port), - ) as (get_subnet_fn, get_cphost_fn, _, _, _, _, get_vif_fn): + with mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_subnet_for_dvr', + return_value={'gateway_ip': gateway_ip, + 'cidr': cidr, + 'ip_version': ip_version, + 'gateway_mac': gateway_mac}),\ + mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_ports_on_host_by_subnet', + return_value=[]),\ + mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\ + mock.patch.object(self.agent.dvr_agent.int_br, + 'get_vif_port_by_id', + return_value=self._port): self.agent.port_bound( self._port, self._net_uuid, 'vxlan', None, None, self._fixed_ips, @@ -1643,15 +1620,14 @@ class TestOvsDvrNeutronAgent(object): int_br.reset_mock() tun_br.reset_mock() - with contextlib.nested( - mock.patch.object(self.agent, 'reclaim_local_vlan'), - mock.patch.object(self.agent.plugin_rpc, 'update_device_down', - return_value=None), - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - ) as (reclaim_vlan_fn, update_dev_down_fn, _, _, _, _): + with mock.patch.object(self.agent, 'reclaim_local_vlan'),\ + mock.patch.object(self.agent.plugin_rpc, + 'update_device_down', + return_value=None),\ + mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br): self.agent.treat_devices_removed([self._port.vif_id]) if ip_version == 4: expected = [ @@ -1686,24 +1662,22 @@ class TestOvsDvrNeutronAgent(object): tun_br = mock.create_autospec(self.agent.tun_br) int_br.set_db_attribute.return_value = True int_br.db_get_val.return_value = {} - with contextlib.nested( - mock.patch.object( - self.agent.dvr_agent.plugin_rpc, 'get_subnet_for_dvr', - return_value={'gateway_ip': gateway_ip, - 'cidr': cidr, - 'ip_version': ip_version, - 'gateway_mac': gateway_mac}), - mock.patch.object(self.agent.dvr_agent.plugin_rpc, - 'get_ports_on_host_by_subnet', - return_value=[]), - mock.patch.object(self.agent.dvr_agent.int_br, - 'get_vif_port_by_id', - return_value=self._port), - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - ) as (get_subnet_fn, get_cphost_fn, get_vif_fn, _, _, _, _): + with mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_subnet_for_dvr', + return_value={'gateway_ip': gateway_ip, + 'cidr': cidr, + 'ip_version': ip_version, + 'gateway_mac': gateway_mac}),\ + mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_ports_on_host_by_subnet', + return_value=[]),\ + mock.patch.object(self.agent.dvr_agent.int_br, + 'get_vif_port_by_id', + return_value=self._port),\ + mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br): self.agent.port_bound( self._port, self._net_uuid, 'vxlan', None, None, self._fixed_ips, @@ -1748,15 +1722,14 @@ class TestOvsDvrNeutronAgent(object): int_br.reset_mock() tun_br.reset_mock() - with contextlib.nested( - mock.patch.object(self.agent, 'reclaim_local_vlan'), - mock.patch.object(self.agent.plugin_rpc, 'update_device_down', - return_value=None), - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - ) as (reclaim_vlan_fn, update_dev_down_fn, _, _, _, _): + with mock.patch.object(self.agent, 'reclaim_local_vlan'),\ + mock.patch.object(self.agent.plugin_rpc, + 'update_device_down', + return_value=None),\ + mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br): self.agent.treat_devices_removed([self._compute_port.vif_id]) int_br.assert_has_calls([ mock.call.delete_dvr_to_src_mac( @@ -1792,24 +1765,22 @@ class TestOvsDvrNeutronAgent(object): tun_br = mock.create_autospec(self.agent.tun_br) int_br.set_db_attribute.return_value = True int_br.db_get_val.return_value = {} - with contextlib.nested( - mock.patch.object( - self.agent.dvr_agent.plugin_rpc, 'get_subnet_for_dvr', - return_value={'gateway_ip': '1.1.1.1', - 'cidr': '1.1.1.0/24', - 'ip_version': 4, - 'gateway_mac': gateway_mac}), - mock.patch.object(self.agent.dvr_agent.plugin_rpc, - 'get_ports_on_host_by_subnet', - return_value=[]), - mock.patch.object(self.agent.dvr_agent.int_br, - 'get_vif_port_by_id', - return_value=self._port), - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - ) as (get_subnet_fn, get_cphost_fn, get_vif_fn, _, _, _, _): + with mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_subnet_for_dvr', + return_value={'gateway_ip': '1.1.1.1', + 'cidr': '1.1.1.0/24', + 'ip_version': 4, + 'gateway_mac': gateway_mac}),\ + mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_ports_on_host_by_subnet', + return_value=[]),\ + mock.patch.object(self.agent.dvr_agent.int_br, + 'get_vif_port_by_id', + return_value=self._port),\ + mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br): self.agent.port_bound( self._port, self._net_uuid, 'vxlan', None, None, self._fixed_ips, @@ -1838,15 +1809,14 @@ class TestOvsDvrNeutronAgent(object): int_br.reset_mock() tun_br.reset_mock() - with contextlib.nested( - mock.patch.object(self.agent, 'reclaim_local_vlan'), - mock.patch.object(self.agent.plugin_rpc, 'update_device_down', - return_value=None), - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - ) as (reclaim_vlan_fn, update_dev_down_fn, _, _, _, _): + with mock.patch.object(self.agent, 'reclaim_local_vlan'),\ + mock.patch.object(self.agent.plugin_rpc, + 'update_device_down', + return_value=None),\ + mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br): self.agent.treat_devices_removed([self._port.vif_id]) expected_on_int_br = [ mock.call.delete_dvr_to_src_mac( @@ -1863,19 +1833,16 @@ class TestOvsDvrNeutronAgent(object): self._setup_for_dvr_test() int_br = mock.create_autospec(self.agent.int_br) tun_br = mock.create_autospec(self.agent.tun_br) - with contextlib.nested( - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - mock.patch.object( - self.agent.dvr_agent.plugin_rpc, - 'get_dvr_mac_address_list', - return_value=[{'host': 'cn1', - 'mac_address': 'aa:bb:cc:dd:ee:ff'}, - {'host': 'cn2', - 'mac_address': '11:22:33:44:55:66'}]) - ) as (_, _, _, _, get_mac_list_fn): + with mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\ + mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_dvr_mac_address_list', + return_value=[{'host': 'cn1', + 'mac_address': 'aa:bb:cc:dd:ee:ff'}, + {'host': 'cn2', + 'mac_address': '11:22:33:44:55:66'}]): self.agent.dvr_agent.setup_dvr_flows_on_integ_br() self.assertTrue(self.agent.dvr_agent.in_distributed_mode()) physical_networks = self.agent.dvr_agent.bridge_mappings.keys() @@ -1913,13 +1880,11 @@ class TestOvsDvrNeutronAgent(object): self._setup_for_dvr_test() self.agent.dvr_agent.dvr_mac_address = None int_br = mock.create_autospec(self.agent.int_br) - with contextlib.nested( - mock.patch.object(self.agent.dvr_agent.plugin_rpc, + with mock.patch.object(self.agent.dvr_agent.plugin_rpc, 'get_dvr_mac_address_by_host', - side_effect=oslo_messaging.RemoteError), - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - ) as (gd_mac, _, _): + side_effect=oslo_messaging.RemoteError),\ + mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br): self.agent.dvr_agent.get_dvr_mac_address() self.assertIsNone(self.agent.dvr_agent.dvr_mac_address) self.assertFalse(self.agent.dvr_agent.in_distributed_mode()) @@ -1948,14 +1913,12 @@ class TestOvsDvrNeutronAgent(object): self._setup_for_dvr_test() self.agent.dvr_agent.dvr_mac_address = None int_br = mock.create_autospec(self.agent.int_br) - with contextlib.nested( - mock.patch.object(self.agent.dvr_agent.plugin_rpc, - 'get_dvr_mac_address_by_host', - side_effect=raise_timeout), - mock.patch.object(utils, "execute"), - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - ) as (rpc_mock, execute_mock, _, _): + with mock.patch.object(self.agent.dvr_agent.plugin_rpc, + 'get_dvr_mac_address_by_host', + side_effect=raise_timeout),\ + mock.patch.object(utils, "execute"),\ + mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br): self.agent.dvr_agent.get_dvr_mac_address() self.assertIsNone(self.agent.dvr_agent.dvr_mac_address) self.assertFalse(self.agent.dvr_agent.in_distributed_mode()) @@ -1970,16 +1933,14 @@ class TestOvsDvrNeutronAgent(object): tun_br = mock.create_autospec(self.agent.tun_br) phys_br = mock.create_autospec(self.br_phys_cls('br-phys')) physical_network = 'physeth1' - with contextlib.nested( - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.dict(self.agent.phys_brs, - {physical_network: phys_br}), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - mock.patch.dict(self.agent.dvr_agent.phys_brs, - {physical_network: phys_br}), - ): + with mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.dict(self.agent.phys_brs, + {physical_network: phys_br}),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\ + mock.patch.dict(self.agent.dvr_agent.phys_brs, + {physical_network: phys_br}): self.agent.dvr_agent.\ dvr_mac_address_update( dvr_macs=[{'host': newhost, @@ -2008,16 +1969,14 @@ class TestOvsDvrNeutronAgent(object): int_br.reset_mock() tun_br.reset_mock() phys_br.reset_mock() - with contextlib.nested( - mock.patch.object(self.agent, 'int_br', new=int_br), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - mock.patch.dict(self.agent.phys_brs, - {physical_network: phys_br}), - mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br), - mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br), - mock.patch.dict(self.agent.dvr_agent.phys_brs, - {physical_network: phys_br}), - ): + with mock.patch.object(self.agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br),\ + mock.patch.dict(self.agent.phys_brs, + {physical_network: phys_br}),\ + mock.patch.object(self.agent.dvr_agent, 'int_br', new=int_br),\ + mock.patch.object(self.agent.dvr_agent, 'tun_br', new=tun_br),\ + mock.patch.dict(self.agent.dvr_agent.phys_brs, + {physical_network: phys_br}): self.agent.dvr_agent.dvr_mac_address_update(dvr_macs=[]) expected_on_int_br = [ mock.call.remove_dvr_mac_vlan( @@ -2047,13 +2006,13 @@ class TestOvsDvrNeutronAgent(object): reset_mocks = [mock.patch.object(self.agent.dvr_agent, method).start() for method in reset_methods] tun_br = mock.create_autospec(self.agent.tun_br) - with contextlib.nested( - mock.patch.object(self.agent, 'check_ovs_status', - return_value=constants.OVS_RESTARTED), - mock.patch.object(self.agent, '_agent_has_updates', - side_effect=TypeError('loop exit')), - mock.patch.object(self.agent, 'tun_br', new=tun_br), - ): + with mock.patch.object(self.agent, + 'check_ovs_status', + return_value=constants.OVS_RESTARTED),\ + mock.patch.object(self.agent, + '_agent_has_updates', + side_effect=TypeError('loop exit')),\ + mock.patch.object(self.agent, 'tun_br', new=tun_br): # block RPC calls and bridge calls self.agent.setup_physical_bridges = mock.Mock() self.agent.setup_integration_br = mock.Mock() diff --git a/neutron/tests/unit/plugins/openvswitch/test_agent_scheduler.py b/neutron/tests/unit/plugins/openvswitch/test_agent_scheduler.py index cd6ea2f7c..2682014d0 100644 --- a/neutron/tests/unit/plugins/openvswitch/test_agent_scheduler.py +++ b/neutron/tests/unit/plugins/openvswitch/test_agent_scheduler.py @@ -13,7 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -import contextlib import datetime import mock @@ -272,8 +271,7 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): def test_network_auto_schedule_with_disabled(self): cfg.CONF.set_override('allow_overlapping_ips', True) - with contextlib.nested(self.subnet(), - self.subnet()): + with self.subnet(), self.subnet(): dhcp_rpc_cb = dhcp_rpc.DhcpRpcCallback() self._register_agent_states() hosta_id = self._get_agent_id(constants.AGENT_TYPE_DHCP, @@ -293,8 +291,7 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): def test_network_auto_schedule_with_no_dhcp(self): cfg.CONF.set_override('allow_overlapping_ips', True) - with contextlib.nested(self.subnet(enable_dhcp=False), - self.subnet(enable_dhcp=False)): + with self.subnet(enable_dhcp=False), self.subnet(enable_dhcp=False): dhcp_rpc_cb = dhcp_rpc.DhcpRpcCallback() self._register_agent_states() hosta_id = self._get_agent_id(constants.AGENT_TYPE_DHCP, @@ -314,8 +311,7 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): def test_network_auto_schedule_with_multiple_agents(self): cfg.CONF.set_override('dhcp_agents_per_network', 2) cfg.CONF.set_override('allow_overlapping_ips', True) - with contextlib.nested(self.subnet(), - self.subnet()): + with self.subnet(), self.subnet(): dhcp_rpc_cb = dhcp_rpc.DhcpRpcCallback() self._register_agent_states() hosta_id = self._get_agent_id(constants.AGENT_TYPE_DHCP, @@ -345,8 +341,7 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): def test_network_auto_schedule_with_hosted(self): # one agent hosts all the networks, other hosts none cfg.CONF.set_override('allow_overlapping_ips', True) - with contextlib.nested(self.subnet(), - self.subnet()) as (sub1, sub2): + with self.subnet() as sub1, self.subnet(): dhcp_rpc_cb = dhcp_rpc.DhcpRpcCallback() self._register_agent_states() dhcp_rpc_cb.get_active_networks(self.adminContext, host=DHCP_HOSTA) @@ -474,10 +469,8 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): is_eligible_agent = ('neutron.db.agentschedulers_db.' 'AgentSchedulerDbMixin.is_eligible_agent') dhcp_mixin = agentschedulers_db.DhcpAgentSchedulerDbMixin() - with contextlib.nested( - mock.patch(agent_startup), - mock.patch(is_eligible_agent) - ) as (startup, elig): + with mock.patch(agent_startup) as startup,\ + mock.patch(is_eligible_agent) as elig: tests = [(True, True), (True, False), (False, True), @@ -698,7 +691,7 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): service_constants.L3_ROUTER_NAT) l3_rpc_cb = l3_rpc.L3RpcCallback() self._register_agent_states() - with contextlib.nested(self.router(), self.router()) as (r1, r2): + with self.router() as r1, self.router() as r2: # schedule the routers to host A l3_rpc_cb.sync_routers(self.adminContext, host=L3_HOSTA) @@ -823,8 +816,7 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): self.assertEqual(L3_HOSTB, l3_agents_2['agents'][0]['host']) def test_router_auto_schedule_with_disabled(self): - with contextlib.nested(self.router(), - self.router()): + with self.router(), self.router(): l3_rpc_cb = l3_rpc.L3RpcCallback() self._register_agent_states() hosta_id = self._get_agent_id(constants.AGENT_TYPE_L3, @@ -844,8 +836,7 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): self.assertEqual(0, num_hosta_routers) def test_router_auto_schedule_with_candidates(self): - with contextlib.nested(self.router(), - self.router()) as (router1, router2): + with self.router() as router1, self.router() as router2: l3_rpc_cb = l3_rpc.L3RpcCallback() agent = helpers.register_l3_agent( host=L3_HOSTA, router_id=router1['router']['id']) @@ -869,9 +860,8 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): ret_a = l3_rpc_cb.sync_routers(self.adminContext, host=L3_HOSTA) self.assertEqual(0, len(ret_a)) - with contextlib.nested(self.router(), - self.router(), - self.router()) as routers: + with self.router() as v1, self.router() as v2, self.router() as v3: + routers = (v1, v2, v3) router_ids = [r['router']['id'] for r in routers] # Get all routers @@ -915,8 +905,11 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): self._register_agent_states() hosta_id = self._get_agent_id(constants.AGENT_TYPE_L3, L3_HOSTA) - with contextlib.nested(self.router(), self.router(), - self.router(), self.router()) as routers: + with self.router() as v1,\ + self.router() as v2,\ + self.router() as v3,\ + self.router() as v4: + routers = (v1, v2, v3, v4) router_ids = [r['router']['id'] for r in routers] # Sync router1 (router1 is scheduled) _sync_router_with_ids([router_ids[0]], 1, 1, hosta_id) @@ -931,13 +924,10 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): _sync_router_with_ids(router_ids, 4, 4, hosta_id) def test_router_schedule_with_candidates(self): - with contextlib.nested(self.router(), - self.router(), - self.subnet(), - self.subnet(cidr='10.0.3.0/24')) as (router1, - router2, - subnet1, - subnet2): + with self.router() as router1,\ + self.router() as router2,\ + self.subnet() as subnet1,\ + self.subnet(cidr='10.0.3.0/24') as subnet2: agent = helpers.register_l3_agent( host=L3_HOSTA, router_id=router1['router']['id']) self._router_interface_action('add', @@ -986,11 +976,9 @@ class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase): self.assertEqual(0, len(l3agents)) def test_router_sync_data(self): - with contextlib.nested( - self.subnet(), - self.subnet(cidr='10.0.2.0/24'), - self.subnet(cidr='10.0.3.0/24') - ) as (s1, s2, s3): + with self.subnet() as s1,\ + self.subnet(cidr='10.0.2.0/24') as s2,\ + self.subnet(cidr='10.0.3.0/24') as s3: self._register_agent_states() self._set_net_external(s1['subnet']['network_id']) data = {'router': {'tenant_id': uuidutils.generate_uuid()}} @@ -1287,15 +1275,13 @@ class OvsDhcpAgentNotifierTestCase(test_l3.L3NatTestCaseMixin, def _is_schedule_network_called(self, device_id): plugin = manager.NeutronManager.get_plugin() notifier = plugin.agent_notifiers[constants.AGENT_TYPE_DHCP] - with contextlib.nested( - self.subnet(), - mock.patch.object(plugin, - 'get_dhcp_agents_hosting_networks', - return_value=[]), - mock.patch.object(notifier, - '_schedule_network', - return_value=[]) - ) as (subnet, _, mock_sched): + with self.subnet() as subnet,\ + mock.patch.object(plugin, + 'get_dhcp_agents_hosting_networks', + return_value=[]),\ + mock.patch.object(notifier, + '_schedule_network', + return_value=[]) as mock_sched: with self.port(subnet=subnet, device_id=device_id): return mock_sched.called @@ -1347,14 +1333,12 @@ class OvsL3AgentNotifierTestCase(test_l3.L3NatTestCaseMixin, l3_plugin = (manager.NeutronManager.get_service_plugins() [service_constants.L3_ROUTER_NAT]) l3_notifier = l3_plugin.agent_notifiers[constants.AGENT_TYPE_L3] - with contextlib.nested( - mock.patch.object(l3_notifier.client, 'prepare', - return_value=l3_notifier.client), - mock.patch.object(l3_notifier.client, 'cast'), - self.router(), - ) as ( - mock_prepare, mock_cast, router1 - ): + with mock.patch.object( + l3_notifier.client, + 'prepare', + return_value=l3_notifier.client) as mock_prepare,\ + mock.patch.object(l3_notifier.client, 'cast') as mock_cast,\ + self.router() as router1: self._register_agent_states() hosta_id = self._get_agent_id(constants.AGENT_TYPE_L3, L3_HOSTA) @@ -1372,14 +1356,12 @@ class OvsL3AgentNotifierTestCase(test_l3.L3NatTestCaseMixin, l3_plugin = (manager.NeutronManager.get_service_plugins() [service_constants.L3_ROUTER_NAT]) l3_notifier = l3_plugin.agent_notifiers[constants.AGENT_TYPE_L3] - with contextlib.nested( - mock.patch.object(l3_notifier.client, 'prepare', - return_value=l3_notifier.client), - mock.patch.object(l3_notifier.client, 'cast'), - self.router(), - ) as ( - mock_prepare, mock_cast, router1 - ): + with mock.patch.object( + l3_notifier.client, + 'prepare', + return_value=l3_notifier.client) as mock_prepare,\ + mock.patch.object(l3_notifier.client, 'cast') as mock_cast,\ + self.router() as router1: self._register_agent_states() hosta_id = self._get_agent_id(constants.AGENT_TYPE_L3, L3_HOSTA) @@ -1399,13 +1381,11 @@ class OvsL3AgentNotifierTestCase(test_l3.L3NatTestCaseMixin, l3_plugin = (manager.NeutronManager.get_service_plugins() [service_constants.L3_ROUTER_NAT]) l3_notifier = l3_plugin.agent_notifiers[constants.AGENT_TYPE_L3] - with contextlib.nested( - mock.patch.object(l3_notifier.client, 'prepare', - return_value=l3_notifier.client), - mock.patch.object(l3_notifier.client, 'cast'), - ) as ( - mock_prepare, mock_cast - ): + with mock.patch.object( + l3_notifier.client, + 'prepare', + return_value=l3_notifier.client) as mock_prepare,\ + mock.patch.object(l3_notifier.client, 'cast') as mock_cast: agent_id = helpers.register_l3_agent(L3_HOSTA).id self._disable_agent(agent_id, admin_state_up=False) diff --git a/neutron/tests/unit/plugins/openvswitch/test_ovs_tunnel.py b/neutron/tests/unit/plugins/openvswitch/test_ovs_tunnel.py index 64d130b4b..fe8d9a00e 100644 --- a/neutron/tests/unit/plugins/openvswitch/test_ovs_tunnel.py +++ b/neutron/tests/unit/plugins/openvswitch/test_ovs_tunnel.py @@ -14,7 +14,6 @@ # under the License. # -import contextlib import time import mock @@ -495,19 +494,18 @@ class TunnelTest(object): self.ovs_bridges[self.INT_BRIDGE].check_canary_table.return_value = \ constants.OVS_NORMAL - with contextlib.nested( - mock.patch.object(log.KeywordArgumentAdapter, 'exception'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'scan_ports'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'process_network_ports'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'tunnel_sync'), - mock.patch.object(time, 'sleep'), - mock.patch.object(self.mod_agent.OVSNeutronAgent, - 'update_stale_ofport_rules') - ) as (log_exception, scan_ports, process_network_ports, - ts, time_sleep, update_stale): + with mock.patch.object(log.KeywordArgumentAdapter, + 'exception') as log_exception,\ + mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'scan_ports') as scan_ports,\ + mock.patch.object( + self.mod_agent.OVSNeutronAgent, + 'process_network_ports') as process_network_ports,\ + mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'tunnel_sync'),\ + mock.patch.object(time, 'sleep'),\ + mock.patch.object(self.mod_agent.OVSNeutronAgent, + 'update_stale_ofport_rules') as update_stale: log_exception.side_effect = Exception( 'Fake exception to get out of the loop') scan_ports.side_effect = [reply2, reply3] diff --git a/neutron/tests/unit/plugins/sriovnicagent/test_eswitch_manager.py b/neutron/tests/unit/plugins/sriovnicagent/test_eswitch_manager.py index 3aacb722c..3cc1b25d8 100644 --- a/neutron/tests/unit/plugins/sriovnicagent/test_eswitch_manager.py +++ b/neutron/tests/unit/plugins/sriovnicagent/test_eswitch_manager.py @@ -14,7 +14,6 @@ # limitations under the License. -import contextlib import os import mock @@ -33,28 +32,23 @@ class TestCreateESwitchManager(base.BaseTestCase): def test_create_eswitch_mgr_fail(self): device_mappings = {'physnet1': 'p6p1'} - with contextlib.nested( - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "PciOsWrapper.scan_vf_devices", - side_effect=exc.InvalidDeviceError(dev_name="p6p1", - reason="device" - " not found")), - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "PciOsWrapper.is_assigned_vf", - return_value=True)): + with mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "PciOsWrapper.scan_vf_devices", + side_effect=exc.InvalidDeviceError( + dev_name="p6p1", reason="device" " not found")),\ + mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "PciOsWrapper.is_assigned_vf", return_value=True): with testtools.ExpectedException(exc.InvalidDeviceError): esm.ESwitchManager(device_mappings, None) def test_create_eswitch_mgr_ok(self): device_mappings = {'physnet1': 'p6p1'} - with contextlib.nested( - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "PciOsWrapper.scan_vf_devices", - return_value=self.SCANNED_DEVICES), - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "PciOsWrapper.is_assigned_vf", - return_value=True)): + with mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "PciOsWrapper.scan_vf_devices", + return_value=self.SCANNED_DEVICES),\ + mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "PciOsWrapper.is_assigned_vf", return_value=True): esm.ESwitchManager(device_mappings, None) @@ -72,13 +66,11 @@ class TestESwitchManagerApi(base.BaseTestCase): def setUp(self): super(TestESwitchManagerApi, self).setUp() device_mappings = {'physnet1': 'p6p1'} - with contextlib.nested( - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "PciOsWrapper.scan_vf_devices", - return_value=self.SCANNED_DEVICES), - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "PciOsWrapper.is_assigned_vf", - return_value=True)): + with mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "PciOsWrapper.scan_vf_devices", + return_value=self.SCANNED_DEVICES),\ + mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "PciOsWrapper.is_assigned_vf", return_value=True): self.eswitch_mgr = esm.ESwitchManager(device_mappings, None) def test_get_assigned_devices(self): @@ -89,37 +81,32 @@ class TestESwitchManagerApi(base.BaseTestCase): self.assertEqual(set([self.ASSIGNED_MAC]), result) def test_get_device_status_true(self): - with contextlib.nested( - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "EmbSwitch.get_pci_device", - return_value=self.ASSIGNED_MAC), - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "EmbSwitch.get_device_state", - return_value=True)): + with mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "EmbSwitch.get_pci_device", + return_value=self.ASSIGNED_MAC),\ + mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "EmbSwitch.get_device_state", return_value=True): result = self.eswitch_mgr.get_device_state(self.ASSIGNED_MAC, self.PCI_SLOT) self.assertTrue(result) def test_get_device_status_false(self): - with contextlib.nested( - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "EmbSwitch.get_pci_device", - return_value=self.ASSIGNED_MAC), - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "EmbSwitch.get_device_state", - return_value=False)): + with mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "EmbSwitch.get_pci_device", + return_value=self.ASSIGNED_MAC),\ + mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "EmbSwitch.get_device_state", + return_value=False): result = self.eswitch_mgr.get_device_state(self.ASSIGNED_MAC, self.PCI_SLOT) self.assertFalse(result) def test_get_device_status_mismatch(self): - with contextlib.nested( - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "EmbSwitch.get_pci_device", - return_value=self.ASSIGNED_MAC), - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "EmbSwitch.get_device_state", - return_value=True)): + with mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "EmbSwitch.get_pci_device", + return_value=self.ASSIGNED_MAC),\ + mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "EmbSwitch.get_device_state", return_value=True): with mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." "LOG.warning") as log_mock: result = self.eswitch_mgr.get_device_state(self.WRONG_MAC, @@ -131,22 +118,20 @@ class TestESwitchManagerApi(base.BaseTestCase): self.assertFalse(result) def test_set_device_status(self): - with contextlib.nested( - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "EmbSwitch.get_pci_device", - return_value=self.ASSIGNED_MAC), - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "EmbSwitch.set_device_state")): + with mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "EmbSwitch.get_pci_device", + return_value=self.ASSIGNED_MAC),\ + mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "EmbSwitch.set_device_state"): self.eswitch_mgr.set_device_state(self.ASSIGNED_MAC, self.PCI_SLOT, True) def test_set_device_status_mismatch(self): - with contextlib.nested( - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "EmbSwitch.get_pci_device", - return_value=self.ASSIGNED_MAC), - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "EmbSwitch.set_device_state")): + with mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "EmbSwitch.get_pci_device", + return_value=self.ASSIGNED_MAC),\ + mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "EmbSwitch.set_device_state"): with mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." "LOG.warning") as log_mock: self.eswitch_mgr.set_device_state(self.WRONG_MAC, @@ -209,13 +194,11 @@ class TestEmbSwitch(base.BaseTestCase): exclude_devices) def test_get_assigned_devices(self): - with contextlib.nested( - mock.patch("neutron.plugins.sriovnicagent.pci_lib." - "PciDeviceIPWrapper.get_assigned_macs", - return_value=[self.ASSIGNED_MAC]), - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "PciOsWrapper.is_assigned_vf", - return_value=True)): + with mock.patch("neutron.plugins.sriovnicagent.pci_lib." + "PciDeviceIPWrapper.get_assigned_macs", + return_value=[self.ASSIGNED_MAC]),\ + mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "PciOsWrapper.is_assigned_vf", return_value=True): result = self.emb_switch.get_assigned_devices() self.assertEqual([self.ASSIGNED_MAC], result) @@ -257,24 +240,20 @@ class TestEmbSwitch(base.BaseTestCase): self.WRONG_PCI_SLOT, True) def test_get_pci_device(self): - with contextlib.nested( - mock.patch("neutron.plugins.sriovnicagent.pci_lib." - "PciDeviceIPWrapper.get_assigned_macs", - return_value=[self.ASSIGNED_MAC]), - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "PciOsWrapper.is_assigned_vf", - return_value=True)): + with mock.patch("neutron.plugins.sriovnicagent.pci_lib." + "PciDeviceIPWrapper.get_assigned_macs", + return_value=[self.ASSIGNED_MAC]),\ + mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "PciOsWrapper.is_assigned_vf", return_value=True): result = self.emb_switch.get_pci_device(self.PCI_SLOT) self.assertEqual(self.ASSIGNED_MAC, result) def test_get_pci_device_fail(self): - with contextlib.nested( - mock.patch("neutron.plugins.sriovnicagent.pci_lib." - "PciDeviceIPWrapper.get_assigned_macs", - return_value=[self.ASSIGNED_MAC]), - mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." - "PciOsWrapper.is_assigned_vf", - return_value=True)): + with mock.patch("neutron.plugins.sriovnicagent.pci_lib." + "PciDeviceIPWrapper.get_assigned_macs", + return_value=[self.ASSIGNED_MAC]),\ + mock.patch("neutron.plugins.sriovnicagent.eswitch_manager." + "PciOsWrapper.is_assigned_vf", return_value=True): result = self.emb_switch.get_pci_device(self.WRONG_PCI_SLOT) self.assertIsNone(result) @@ -313,15 +292,10 @@ class TestPciOsWrapper(base.BaseTestCase): file_name = os.path.basename(file_path) return self.LINKS[file_name] - with contextlib.nested( - mock.patch("os.path.isdir", - return_value=True), - mock.patch("os.listdir", - return_value=self.DIR_CONTENTS), - mock.patch("os.path.islink", - return_value=True), - mock.patch("os.readlink", - side_effect=_get_link),): + with mock.patch("os.path.isdir", return_value=True),\ + mock.patch("os.listdir", return_value=self.DIR_CONTENTS),\ + mock.patch("os.path.islink", return_value=True),\ + mock.patch("os.readlink", side_effect=_get_link): result = esm.PciOsWrapper.scan_vf_devices(self.DEV_NAME) self.assertEqual(self.PCI_SLOTS, result) @@ -332,21 +306,16 @@ class TestPciOsWrapper(base.BaseTestCase): self.DEV_NAME) def test_scan_vf_devices_no_content(self): - with contextlib.nested( - mock.patch("os.path.isdir", - return_value=True), - mock.patch("os.listdir", - return_value=[])): + with mock.patch("os.path.isdir", return_value=True),\ + mock.patch("os.listdir", return_value=[]): self.assertRaises(exc.InvalidDeviceError, esm.PciOsWrapper.scan_vf_devices, self.DEV_NAME) def test_scan_vf_devices_no_match(self): - with contextlib.nested( - mock.patch("os.path.isdir", - return_value=True), - mock.patch("os.listdir", - return_value=self.DIR_CONTENTS_NO_MATCH)): + with mock.patch("os.path.isdir", return_value=True),\ + mock.patch("os.listdir", + return_value=self.DIR_CONTENTS_NO_MATCH): self.assertRaises(exc.InvalidDeviceError, esm.PciOsWrapper.scan_vf_devices, self.DEV_NAME) @@ -368,11 +337,8 @@ class TestPciOsWrapper(base.BaseTestCase): def _glob(file_path): return ["upper_macvtap0"] if macvtap_exists else [] - with contextlib.nested( - mock.patch("os.path.isdir", - return_value=True), - mock.patch("glob.glob", - side_effect=_glob)): + with mock.patch("os.path.isdir", return_value=True),\ + mock.patch("glob.glob", side_effect=_glob): result = esm.PciOsWrapper.is_assigned_vf(self.DEV_NAME, self.VF_INDEX) self.assertEqual(macvtap_exists, result) diff --git a/neutron/tests/unit/scheduler/test_dhcp_agent_scheduler.py b/neutron/tests/unit/scheduler/test_dhcp_agent_scheduler.py index c43e2a220..9faa8ab5b 100644 --- a/neutron/tests/unit/scheduler/test_dhcp_agent_scheduler.py +++ b/neutron/tests/unit/scheduler/test_dhcp_agent_scheduler.py @@ -13,8 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -import contextlib - import mock from oslo_config import cfg from oslo_utils import importutils @@ -185,13 +183,14 @@ class TestNetworksFailover(TestDhcpSchedulerBaseTestCase, self._test_schedule_bind_network([agents[0]], self.network_id) self._save_networks(["foo-network-2"]) self._test_schedule_bind_network([agents[1]], "foo-network-2") - with contextlib.nested( - mock.patch.object(self, 'remove_network_from_dhcp_agent'), - mock.patch.object(self, 'schedule_network', - return_value=[agents[1]]), - mock.patch.object(self, 'get_network', create=True, - return_value={'id': self.network_id}) - ) as (rn, sch, getn): + with mock.patch.object(self, 'remove_network_from_dhcp_agent') as rn,\ + mock.patch.object(self, + 'schedule_network', + return_value=[agents[1]]) as sch,\ + mock.patch.object(self, + 'get_network', + create=True, + return_value={'id': self.network_id}): notifier = mock.MagicMock() self.agent_notifiers[constants.AGENT_TYPE_DHCP] = notifier self.remove_networks_from_down_agents() @@ -204,15 +203,16 @@ class TestNetworksFailover(TestDhcpSchedulerBaseTestCase, def _test_failed_rescheduling(self, rn_side_effect=None): agents = self._create_and_set_agents_down(['host-a'], 1) self._test_schedule_bind_network([agents[0]], self.network_id) - with contextlib.nested( - mock.patch.object( - self, 'remove_network_from_dhcp_agent', - side_effect=rn_side_effect), - mock.patch.object(self, 'schedule_network', - return_value=None), - mock.patch.object(self, 'get_network', create=True, - return_value={'id': self.network_id}) - ) as (rn, sch, getn): + with mock.patch.object(self, + 'remove_network_from_dhcp_agent', + side_effect=rn_side_effect) as rn,\ + mock.patch.object(self, + 'schedule_network', + return_value=None) as sch,\ + mock.patch.object(self, + 'get_network', + create=True, + return_value={'id': self.network_id}): notifier = mock.MagicMock() self.agent_notifiers[constants.AGENT_TYPE_DHCP] = notifier self.remove_networks_from_down_agents() diff --git a/neutron/tests/unit/scheduler/test_l3_agent_scheduler.py b/neutron/tests/unit/scheduler/test_l3_agent_scheduler.py index 07f06db7a..87bbc9e03 100644 --- a/neutron/tests/unit/scheduler/test_l3_agent_scheduler.py +++ b/neutron/tests/unit/scheduler/test_l3_agent_scheduler.py @@ -128,10 +128,10 @@ class L3SchedulerBaseTestCase(base.BaseTestCase): def test_auto_schedule_routers(self): self.plugin.get_enabled_agent_on_host.return_value = [mock.ANY] - with contextlib.nested( - mock.patch.object(self.scheduler, '_get_routers_to_schedule'), - mock.patch.object(self.scheduler, '_get_routers_can_schedule') - ) as (gs, gr): + with mock.patch.object(self.scheduler, + '_get_routers_to_schedule') as gs,\ + mock.patch.object(self.scheduler, + '_get_routers_can_schedule') as gr: result = self.scheduler.auto_schedule_routers( self.plugin, mock.ANY, mock.ANY, mock.ANY) self.assertTrue(self.plugin.get_enabled_agent_on_host.called) @@ -159,10 +159,12 @@ class L3SchedulerBaseTestCase(base.BaseTestCase): def test_auto_schedule_routers_no_target_routers(self): self.plugin.get_enabled_agent_on_host.return_value = [mock.ANY] - with contextlib.nested( - mock.patch.object(self.scheduler, '_get_routers_to_schedule'), - mock.patch.object(self.scheduler, '_get_routers_can_schedule') - ) as (mock_unscheduled_routers, mock_target_routers): + with mock.patch.object( + self.scheduler, + '_get_routers_to_schedule') as mock_unscheduled_routers,\ + mock.patch.object( + self.scheduler, + '_get_routers_can_schedule') as mock_target_routers: mock_unscheduled_routers.return_value = mock.ANY mock_target_routers.return_value = None result = self.scheduler.auto_schedule_routers( @@ -250,12 +252,11 @@ class L3SchedulerBaseTestCase(base.BaseTestCase): def _test__bind_routers_ha(self, has_binding): routers = [{'id': 'foo_router', 'ha': True, 'tenant_id': '42'}] agent = agents_db.Agent(id='foo_agent') - with contextlib.nested( - mock.patch.object(self.scheduler, '_router_has_binding', - return_value=has_binding), - mock.patch.object(self.scheduler, '_create_ha_router_binding') - ) as ( - mock_has_binding, mock_bind): + with mock.patch.object(self.scheduler, + '_router_has_binding', + return_value=has_binding) as mock_has_binding,\ + mock.patch.object(self.scheduler, + '_create_ha_router_binding') as mock_bind: self.scheduler._bind_routers(mock.ANY, mock.ANY, routers, agent) mock_has_binding.assert_called_once_with(mock.ANY, 'foo_router', 'foo_agent') @@ -336,12 +337,11 @@ class L3SchedulerTestBaseMixin(object): router['router']['external_gateway_info'] = external_gw if already_scheduled: self._test_schedule_bind_router(agent, router) - with contextlib.nested( - mock.patch.object(self, "validate_agent_router_combination"), - mock.patch.object(self, "create_router_to_agent_binding"), - mock.patch('neutron.db.l3_db.L3_NAT_db_mixin.get_router', - return_value=router['router']) - ) as (valid, auto_s, gr): + with mock.patch.object(self, "validate_agent_router_combination"),\ + mock.patch.object(self, + "create_router_to_agent_binding") as auto_s,\ + mock.patch('neutron.db.l3_db.L3_NAT_db_mixin.get_router', + return_value=router['router']): self.add_router_to_l3_agent(self.adminContext, agent_id, router['router']['id']) self.assertNotEqual(already_scheduled, auto_s.called) @@ -377,10 +377,9 @@ class L3SchedulerTestBaseMixin(object): expected_exception=None): router = self._create_router_for_l3_agent_dvr_test( distributed=distributed, external_gw=external_gw) - with contextlib.nested( - mock.patch.object(self, "create_router_to_agent_binding"), - mock.patch('neutron.db.l3_db.L3_NAT_db_mixin.get_router', - return_value=router['router'])): + with mock.patch.object(self, "create_router_to_agent_binding"),\ + mock.patch('neutron.db.l3_db.L3_NAT_db_mixin.get_router', + return_value=router['router']): self.assertRaises(expected_exception, self.add_router_to_l3_agent, self.adminContext, agent_id, @@ -416,12 +415,12 @@ class L3SchedulerTestBaseMixin(object): router = self._create_router_for_l3_agent_dvr_test( distributed=True, external_gw=external_gw_info) - with contextlib.nested( - mock.patch.object(self, "validate_agent_router_combination"), - mock.patch.object(self, "create_router_to_agent_binding"), - mock.patch('neutron.db.l3_db.L3_NAT_db_mixin.get_router', - return_value=router['router']) - ) as (valid_agent_rtr, rtr_agent_binding, get_rtr): + with mock.patch.object(self, "validate_agent_router_combination"),\ + mock.patch.object( + self, + "create_router_to_agent_binding") as rtr_agent_binding,\ + mock.patch('neutron.db.l3_db.L3_NAT_db_mixin.get_router', + return_value=router['router']): self.add_router_to_l3_agent(self.adminContext, agent_id, router['router']['id']) @@ -470,10 +469,10 @@ class L3SchedulerTestBaseMixin(object): 'distributed': True } plugin.get_router.return_value = sync_router - with contextlib.nested( - mock.patch.object(scheduler, 'bind_router'), - mock.patch.object( - plugin, 'get_snat_bindings', return_value=False)): + with mock.patch.object(scheduler, 'bind_router'),\ + mock.patch.object(plugin, + 'get_snat_bindings', + return_value=False): scheduler._schedule_router( plugin, self.adminContext, 'foo_router_id', None) expected_calls = [ @@ -954,15 +953,16 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase): 'distributed': True, } - with contextlib.nested( - mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2' - '.get_ports', return_value=[dvr_port]), - mock.patch('neutron.manager.NeutronManager.get_service_plugins', - return_value=mock.Mock()), - mock.patch('neutron.db.l3_db.L3_NAT_db_mixin.get_router', - return_value=r1), - mock.patch('neutron.api.rpc.agentnotifiers.l3_rpc_agent_api' - '.L3AgentNotifyAPI')): + with mock.patch( + 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2' '.get_ports', + return_value=[dvr_port]),\ + mock.patch( + 'neutron.manager.NeutronManager.get_service_plugins', + return_value=mock.Mock()),\ + mock.patch('neutron.db.l3_db.L3_NAT_db_mixin.get_router', + return_value=r1),\ + mock.patch('neutron.api.rpc.agentnotifiers.l3_rpc_agent_api' + '.L3AgentNotifyAPI'): self.dut.dvr_update_router_addvm(self.adminContext, port) def test_get_dvr_routers_by_portid(self): @@ -982,11 +982,11 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase): 'distributed': True, } - with contextlib.nested( - mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2' - '.get_port', return_value=dvr_port), - mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2' - '.get_ports', return_value=[dvr_port])): + with mock.patch( + 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2' '.get_port', + return_value=dvr_port),\ + mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2' + '.get_ports', return_value=[dvr_port]): router_id = self.dut.get_dvr_routers_by_portid(self.adminContext, dvr_port['id']) self.assertEqual(router_id.pop(), r1['id']) @@ -1008,9 +1008,9 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase): 'distributed': True, } - with contextlib.nested( - mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2' - '.get_ports', return_value=[dvr_port])): + with mock.patch( + 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2' '.get_ports', + return_value=[dvr_port]): sub_ids = self.dut.get_subnet_ids_on_router(self.adminContext, r1['id']) self.assertEqual(sub_ids.pop(), @@ -1034,15 +1034,16 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase): 'id': 'r1', 'distributed': True, } - with contextlib.nested( - mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2' - '.get_ports', return_value=[dvr_port]), - mock.patch('neutron.manager.NeutronManager.get_service_plugins', - return_value=mock.Mock()), - mock.patch('neutron.db.l3_db.L3_NAT_db_mixin.get_router', - return_value=r1), - mock.patch('neutron.api.rpc.agentnotifiers.l3_rpc_agent_api' - '.L3AgentNotifyAPI')): + with mock.patch( + 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2' '.get_ports', + return_value=[dvr_port]),\ + mock.patch( + 'neutron.manager.NeutronManager.get_service_plugins', + return_value=mock.Mock()),\ + mock.patch('neutron.db.l3_db.L3_NAT_db_mixin.get_router', + return_value=r1),\ + mock.patch('neutron.api.rpc.agentnotifiers.l3_rpc_agent_api' + '.L3AgentNotifyAPI'): sub_ids = self.dut.get_subnet_ids_on_router(self.adminContext, r1['id']) result = self.dut.check_ports_active_on_host_and_subnet( @@ -1109,16 +1110,16 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase): vm_port_id = vm_port['id'] fakePortDB = FakePortDB([vm_port]) - with contextlib.nested( - mock.patch.object(my_context, 'elevated', - return_value=self.adminContext), - mock.patch('neutron.plugins.ml2.db.' - 'get_port_binding_host', return_value=vm_port_host), - mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' - 'get_ports', side_effect=fakePortDB.get_ports), - mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' - 'get_port', return_value=vm_port)) as ( - _, mock_get_port_binding_host, _, _): + with mock.patch.object(my_context, + 'elevated', + return_value=self.adminContext),\ + mock.patch( + 'neutron.plugins.ml2.db.get_port_binding_host', + return_value=vm_port_host) as mock_get_port_binding_host,\ + mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' + 'get_ports', side_effect=fakePortDB.get_ports),\ + mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' + 'get_port', return_value=vm_port): routers = self.dut.dvr_deletens_if_no_port(my_context, vm_port_id) self.assertEqual([], routers) @@ -1155,20 +1156,21 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase): 'host': vm_port_host } - with contextlib.nested( - mock.patch.object(my_context, 'elevated', - return_value=self.adminContext), - mock.patch('neutron.plugins.ml2.db.get_port_binding_host', - return_value=vm_port_host), - mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' - 'get_port', side_effect=fakePortDB.get_port), - mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' - 'get_ports', side_effect=fakePortDB.get_ports), - mock.patch('neutron.plugins.ml2.db.get_dvr_port_binding_by_host', - return_value=vm_port_binding)) as (_, - mock_get_port_binding_host, _, - mock_get_ports, - mock_get_dvr_port_binding_by_host): + with mock.patch.object(my_context, + 'elevated', + return_value=self.adminContext),\ + mock.patch( + 'neutron.plugins.ml2.db.get_port_binding_host', + return_value=vm_port_host) as mock_get_port_binding_host,\ + mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' + 'get_port', side_effect=fakePortDB.get_port),\ + mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' + 'get_ports', side_effect=fakePortDB.get_ports) as\ + mock_get_ports,\ + mock.patch('neutron.plugins.ml2.db.' + 'get_dvr_port_binding_by_host', + return_value=vm_port_binding) as\ + mock_get_dvr_port_binding_by_host: routers = self.dut.dvr_deletens_if_no_port( my_context, deleted_vm_port_id) @@ -1220,24 +1222,24 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase): vm_port_host, constants.AGENT_TYPE_L3) - with contextlib.nested( - mock.patch.object(my_context, 'elevated', - return_value=self.adminContext), - mock.patch('neutron.plugins.ml2.db.get_port_binding_host', - return_value=vm_port_host), - mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' - 'get_port', side_effect=fakePortDB.get_port), - mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' - 'get_ports', side_effect=fakePortDB.get_ports), - mock.patch('neutron.plugins.ml2.db.get_dvr_port_binding_by_host', - return_value=dvr_port_binding), - mock.patch('neutron.db.agents_db.AgentDbMixin.' - '_get_agent_by_type_and_host', - return_value=l3_agent_on_vm_host)) as (_, - mock_get_port_binding_host, _, - mock_get_ports, - mock_get_dvr_port_binding_by_host, - mock__get_agent_by_type_and_host): + with mock.patch.object(my_context, + 'elevated', + return_value=self.adminContext),\ + mock.patch( + 'neutron.plugins.ml2.db.get_port_binding_host', + return_value=vm_port_host) as mock_get_port_binding_host,\ + mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' + 'get_port', side_effect=fakePortDB.get_port),\ + mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.' + 'get_ports', side_effect=fakePortDB.get_ports) as\ + mock_get_ports,\ + mock.patch('neutron.plugins.ml2.db.' + 'get_dvr_port_binding_by_host', + return_value=dvr_port_binding) as\ + mock_get_dvr_port_binding_by_host,\ + mock.patch('neutron.db.agents_db.AgentDbMixin.' + '_get_agent_by_type_and_host', + return_value=l3_agent_on_vm_host): routers = self.dut.dvr_deletens_if_no_port( my_context, deleted_vm_port_id) @@ -1299,25 +1301,30 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase): def test_schedule_snat_router_duplicate_entry(self): self._prepare_schedule_snat_tests() - with contextlib.nested( - mock.patch.object(self.dut, 'get_l3_agents'), - mock.patch.object(self.dut, 'get_snat_candidates'), - mock.patch.object(self.dut, 'bind_snat_servicenode', - side_effect=db_exc.DBDuplicateEntry()), - mock.patch.object(self.dut, 'bind_dvr_router_servicenode') - ) as (mock_gl3, mock_snat_canidates, mock_bind_snat, mock_bind_dvr): + with mock.patch.object(self.dut, 'get_l3_agents'),\ + mock.patch.object(self.dut, 'get_snat_candidates'),\ + mock.patch.object( + self.dut, + 'bind_snat_servicenode', + side_effect=db_exc.DBDuplicateEntry()) as mock_bind_snat,\ + mock.patch.object( + self.dut, + 'bind_dvr_router_servicenode') as mock_bind_dvr: self.dut.schedule_snat_router(self.adminContext, 'foo', 'bar') self.assertTrue(mock_bind_snat.called) self.assertFalse(mock_bind_dvr.called) def test_schedule_snat_router_return_value(self): agent, router = self._prepare_schedule_snat_tests() - with contextlib.nested( - mock.patch.object(self.dut, 'get_l3_agents'), - mock.patch.object(self.dut, 'get_snat_candidates'), - mock.patch.object(self.dut, 'bind_snat_servicenode'), - mock.patch.object(self.dut, 'bind_dvr_router_servicenode') - ) as (mock_gl3, mock_snat_canidates, mock_bind_snat, mock_bind_dvr): + with mock.patch.object(self.dut, 'get_l3_agents'),\ + mock.patch.object( + self.dut, + 'get_snat_candidates') as mock_snat_canidates,\ + mock.patch.object(self.dut, + 'bind_snat_servicenode') as mock_bind_snat,\ + mock.patch.object( + self.dut, + 'bind_dvr_router_servicenode') as mock_bind_dvr: mock_snat_canidates.return_value = [agent] mock_bind_snat.return_value = [agent] mock_bind_dvr.return_value = [agent] @@ -1330,11 +1337,11 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase): 'id': 'foo_router_id', 'distributed': True } - with contextlib.nested( - mock.patch.object(self.dut, 'get_router'), - mock.patch.object(self.dut, 'get_snat_bindings'), - mock.patch.object(self.dut, 'unbind_snat_servicenode') - ) as (mock_rd, mock_snat_bind, mock_unbind): + with mock.patch.object(self.dut, 'get_router') as mock_rd,\ + mock.patch.object(self.dut, + 'get_snat_bindings') as mock_snat_bind,\ + mock.patch.object(self.dut, + 'unbind_snat_servicenode') as mock_unbind: mock_rd.return_value = router mock_snat_bind.return_value = False self.dut.schedule_snat_router( @@ -1343,15 +1350,14 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase): def test_schedule_snat_router_with_snat_candidates(self): agent, router = self._prepare_schedule_snat_tests() - with contextlib.nested( - mock.patch.object(query.Query, 'first'), - mock.patch.object(self.dut, 'get_l3_agents'), - mock.patch.object(self.dut, 'get_snat_candidates'), - mock.patch.object(self.dut, 'get_router'), - mock.patch.object(self.dut, 'bind_dvr_router_servicenode'), - mock.patch.object(self.dut, 'bind_snat_servicenode')) as ( - mock_query, mock_agents, - mock_candidates, mock_rd, mock_dvr, mock_bind): + with mock.patch.object(query.Query, 'first') as mock_query,\ + mock.patch.object(self.dut, 'get_l3_agents') as mock_agents,\ + mock.patch.object(self.dut, + 'get_snat_candidates') as mock_candidates,\ + mock.patch.object(self.dut, 'get_router') as mock_rd,\ + mock.patch.object(self.dut, 'bind_dvr_router_servicenode'),\ + mock.patch.object(self.dut, + 'bind_snat_servicenode') as mock_bind: mock_rd.return_value = router mock_query.return_value = [] mock_agents.return_value = [agent] @@ -1373,12 +1379,13 @@ class L3DvrSchedulerTestCase(testlib_api.SqlTestCase): binding = l3_dvrscheduler_db.CentralizedSnatL3AgentBinding( router_id=router_id, l3_agent_id='foo_l3_agent_id', l3_agent=agents_db.Agent()) - with contextlib.nested( - mock.patch.object(query.Query, 'one'), - mock.patch.object(self.adminContext.session, 'delete'), - mock.patch.object(query.Query, 'delete'), - mock.patch.object(self.dut, 'get_subnet_ids_on_router')) as ( - mock_query, mock_session, mock_delete, mock_get_subnets): + with mock.patch.object(query.Query, 'one') as mock_query,\ + mock.patch.object(self.adminContext.session, + 'delete') as mock_session,\ + mock.patch.object(query.Query, 'delete') as mock_delete,\ + mock.patch.object( + self.dut, + 'get_subnet_ids_on_router') as mock_get_subnets: mock_query.return_value = binding mock_get_subnets.return_value = ['foo_subnet_id'] self.dut.unbind_snat_servicenode(self.adminContext, router_id) -- 2.45.2