From 5362c65416684f731b78603b3a0a82df30d82f6f Mon Sep 17 00:00:00 2001 From: Boden R Date: Thu, 30 May 2019 13:09:17 -0600 Subject: [PATCH] update bandit, hacking and flake8 requirements This patch bumps the hacking, bandit and flake8 requirements to match suit with similar work (ex [1]). It also updates the code to fix a few new pep8 errors as well as adds a local tox target for requirements-check-dev. [1] https://review.opendev.org/#/c/658245/ Change-Id: I6caeb52dc1a5842338ec989a742ae5989608e0da --- devstack/tools/nsxp_cleanup.py | 2 +- devstack/tools/nsxv_cleanup.py | 8 +- lower-constraints.txt | 4 +- test-requirements.txt | 7 +- tox.ini | 18 ++- vmware_nsx/api_client/eventlet_client.py | 1 + vmware_nsx/dhcp_meta/combined.py | 9 +- vmware_nsx/nsxlib/mh/__init__.py | 6 +- vmware_nsx/plugins/nsx_p/plugin.py | 1 + .../nsx_v/housekeeper/error_dhcp_edge.py | 16 +- vmware_nsx/plugins/nsx_v/plugin.py | 30 ++-- vmware_nsx/plugins/nsx_v3/plugin.py | 10 +- .../nsx_p/implementation/listener_mgr.py | 5 +- .../nsx_v/implementation/listener_mgr.py | 2 +- .../plugins/nsxv/resources/backup_edges.py | 1 + .../admin/plugins/nsxv/resources/edges.py | 14 +- vmware_nsx/shell/resources.py | 1 + .../unit/extensions/test_portsecurity.py | 2 +- vmware_nsx/tests/unit/nsx_p/test_plugin.py | 141 +++++++++--------- vmware_nsx/tests/unit/nsx_v/test_plugin.py | 112 +++++++------- .../tests/unit/nsx_v/vshield/fake_vcns.py | 6 +- vmware_nsx/tests/unit/nsx_v3/test_plugin.py | 101 +++++++------ .../dynamic_routing/test_nsxv_bgp_driver.py | 14 +- .../services/qos/test_nsxp_notification.py | 16 +- .../services/qos/test_nsxv3_notification.py | 22 +-- .../unit/services/vpnaas/test_nsxv3_vpnaas.py | 55 +++---- 26 files changed, 313 insertions(+), 291 deletions(-) diff --git a/devstack/tools/nsxp_cleanup.py b/devstack/tools/nsxp_cleanup.py index a10575eab2..0439e94d32 100755 --- a/devstack/tools/nsxp_cleanup.py +++ b/devstack/tools/nsxp_cleanup.py @@ -175,7 +175,7 @@ class NSXClient(object): self.cleanup_tier1_nat_rules(rtr['id']) try: self.nsxpolicy.tier1.delete_locale_service(rtr['id']) - except exceptions.ManagerError as e: + except exceptions.ManagerError: # Not always exists pass try: diff --git a/devstack/tools/nsxv_cleanup.py b/devstack/tools/nsxv_cleanup.py index 40908a5941..19adf0250a 100644 --- a/devstack/tools/nsxv_cleanup.py +++ b/devstack/tools/nsxv_cleanup.py @@ -253,7 +253,7 @@ class VSMClient(object): # Query all firewall sections response = self.get() # Get layer3 sections related to security group - if response.status_code is 200: + if response.status_code == 200: l3_sections = response.json()['layer3Sections']['layer3Sections'] # do not delete the default section, or sections created by the # service composer @@ -289,7 +289,7 @@ class VSMClient(object): self.__set_endpoint("/services/securitygroup/scope/globalroot-0") # Query all security groups response = self.get() - if response.status_code is 200: + if response.status_code == 200: sg_all = response.json() else: print("ERROR: wrong response status code! Exiting...") @@ -323,7 +323,7 @@ class VSMClient(object): self.__set_endpoint("/services/spoofguard/policies/") # Query all spoofguard policies response = self.get() - if response.status_code is not 200: + if response.status_code != 200: print("ERROR: Faield to get spoofguard policies") return sgp_all = response.json() @@ -395,7 +395,7 @@ def ceil(a, b): if b == 0: return 0 div = a / b - mod = 0 if a % b is 0 else 1 + mod = 0 if a % b == 0 else 1 return int(div + mod) diff --git a/lower-constraints.txt b/lower-constraints.txt index 010c4a795b..b90ece6c1e 100644 --- a/lower-constraints.txt +++ b/lower-constraints.txt @@ -3,9 +3,9 @@ coverage==4.0 decorator==4.3.0 eventlet==0.24.1 fixtures==3.0.0 -flake8==2.5.5 flake8-import-order==0.12 -hacking==0.12.0 +flake8==2.6.2 +hacking==1.1.0 httplib2==0.9.1 mock==2.0.0 netaddr==0.7.18 diff --git a/test-requirements.txt b/test-requirements.txt index c48eca82b0..16335eaa11 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1,13 +1,13 @@ # The order of packages is significant, because pip processes them in the order # of appearance. Changing the order has an impact on the overall integration # process, which may cause wedges in the gate later. -hacking!=0.13.0,<0.14,>=0.12.0 # Apache-2.0 - +hacking>=1.1.0 # Apache-2.0 +bandit!=1.6.0,>=1.1.0 # Apache-2.0 coverage!=4.4,>=4.0 # Apache-2.0 fixtures>=3.0.0 # Apache-2.0/BSD +flake8>=2.6.0 flake8-import-order==0.12 # LGPLv3 mock>=2.0.0 # BSD - psycopg2>=2.7 # LGPL/ZPL PyMySQL>=0.7.6 # MIT License oslotest>=3.2.0 # Apache-2.0 @@ -18,7 +18,6 @@ testresources>=2.0.0 # Apache-2.0/BSD testtools>=2.2.0 # MIT testscenarios>=0.4 # Apache-2.0/BSD WebTest>=2.0.27 # MIT -bandit>=1.1.0 # Apache-2.0 tempest>=17.1.0 # Apache-2.0 pylint==1.7.6 # GPLv2 python-openstackclient>=3.16.0 # Apache-2.0 diff --git a/tox.ini b/tox.ini index 796e3bf57f..4b176d3331 100644 --- a/tox.ini +++ b/tox.ini @@ -126,7 +126,7 @@ commands = # If it is easier to add a check via a shell script, consider adding it in this file sh ./tools/misc-sanity-checks.sh # Checks for coding and style guidelines - flake8 {toxinidir}/vmware_nsx + flake8 sh ./tools/coding-checks.sh --pylint '{posargs}' neutron-db-manage --subproject vmware-nsx check_migration {[testenv:genconfig]commands} @@ -163,7 +163,8 @@ commands = sphinx-build -b html doc/source doc/build/html # TODO(dougwig) -- uncomment this to test for remaining linkages # N530 direct neutron imports not allowed # N531 translations hints -ignore = E125,E126,E128,E129,E265,H305,H307,H404,H405,H904,N530,N531 +# W504 line break after binary operator +ignore = E125,E126,E128,E129,E265,H305,H307,H404,H405,H904,N530,N531,W504 show-source = true builtins = _ exclude = .venv,.git,.tox,dist,doc,*lib/python*,*egg,build,.ropeproject @@ -205,3 +206,16 @@ deps = [testenv:venv] commands = {posargs} + +[testenv:requirements-check-dev] +basepython = python3 +commands = + pip install -q -e "git+https://opendev.org/openstack/requirements#egg=requirements" + pip freeze + # must have openstack/requirements on latest src/master in ../requirements + {toxinidir}/../requirements/playbooks/files/project-requirements-change.py --reqs={toxinidir}/../requirements {toxinidir} +deps = + -c{toxinidir}/lower-constraints.txt + -r{toxinidir}/test-requirements.txt + -r{toxinidir}/doc/requirements.txt + -r{toxinidir}/requirements.txt diff --git a/vmware_nsx/api_client/eventlet_client.py b/vmware_nsx/api_client/eventlet_client.py index 2d8c05ebfe..25578be331 100644 --- a/vmware_nsx/api_client/eventlet_client.py +++ b/vmware_nsx/api_client/eventlet_client.py @@ -152,5 +152,6 @@ class EventletApiClient(base.ApiClientBase): return cookie + # Register as subclass. base.ApiClientBase.register(EventletApiClient) diff --git a/vmware_nsx/dhcp_meta/combined.py b/vmware_nsx/dhcp_meta/combined.py index 904134af52..37be1fffee 100644 --- a/vmware_nsx/dhcp_meta/combined.py +++ b/vmware_nsx/dhcp_meta/combined.py @@ -45,13 +45,14 @@ class DhcpAgentNotifyAPI(dhcp_rpc_agent_api.DhcpAgentNotifyAPI): if lsn_exists: # if lsn exists, the network is one created with the new model if (resource == 'subnet' and action == 'create' and - const.DEVICE_OWNER_DHCP not in plugin.port_special_owners): + const.DEVICE_OWNER_DHCP not in + plugin.port_special_owners): # network/subnet provisioned in the new model have a plain # nsx lswitch port, no vif attachment - plugin.port_special_owners.append(const.DEVICE_OWNER_DHCP) - treat_dhcp_owner_specially = True + plugin.port_special_owners.append(const.DEVICE_OWNER_DHCP) + treat_dhcp_owner_specially = True if (resource == 'port' and action == 'update' or - resource == 'subnet'): + resource == 'subnet'): self.agentless_notifier.notify(context, data, methodname) elif not lsn_exists and resource in ['port', 'subnet']: # call notifier for the agent-based mode diff --git a/vmware_nsx/nsxlib/mh/__init__.py b/vmware_nsx/nsxlib/mh/__init__.py index 3a48212b00..eacfd006e1 100644 --- a/vmware_nsx/nsxlib/mh/__init__.py +++ b/vmware_nsx/nsxlib/mh/__init__.py @@ -84,9 +84,9 @@ def format_exception(etype, e, exception_locals): """ msg = [_("Error. %(type)s exception: %(exc)s.") % {'type': etype, 'exc': e}] - l = dict((k, v) for k, v in six.iteritems(exception_locals) - if k != 'request') - msg.append(_("locals=[%s]") % str(l)) + lcls = dict((k, v) for k, v in six.iteritems(exception_locals) + if k != 'request') + msg.append(_("locals=[%s]") % str(lcls)) return ' '.join(msg) diff --git a/vmware_nsx/plugins/nsx_p/plugin.py b/vmware_nsx/plugins/nsx_p/plugin.py index 80a6ab4f44..3d90311cb5 100644 --- a/vmware_nsx/plugins/nsx_p/plugin.py +++ b/vmware_nsx/plugins/nsx_p/plugin.py @@ -1555,6 +1555,7 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base): if (actions['remove_router_link_port'] or actions['add_router_link_port']): # GW was changed. update GW and route advertisement + # pylint: disable=unexpected-keyword-arg self.nsxpolicy.tier1.update_route_advertisement( router_id, nat=actions['advertise_route_nat_flag'], diff --git a/vmware_nsx/plugins/nsx_v/housekeeper/error_dhcp_edge.py b/vmware_nsx/plugins/nsx_v/housekeeper/error_dhcp_edge.py index c5b9e2e3e9..28ccafae41 100644 --- a/vmware_nsx/plugins/nsx_v/housekeeper/error_dhcp_edge.py +++ b/vmware_nsx/plugins/nsx_v/housekeeper/error_dhcp_edge.py @@ -296,14 +296,14 @@ class ErrorDhcpEdgeJob(base_job.BaseJob): if vnic['type'] != 'trunk': # reinitialize the interface as it is missing config - vnic['name'] = (vcns_const.INTERNAL_VNIC_NAME + - str(vnic['index'])) - vnic['type'] = 'trunk' - vnic['portgroupId'] = port_group_id - vnic['mtu'] = 1500 - vnic['enableProxyArp'] = False - vnic['enableSendRedirects'] = True - vnic['isConnected'] = True + vnic['name'] = (vcns_const.INTERNAL_VNIC_NAME + + str(vnic['index'])) + vnic['type'] = 'trunk' + vnic['portgroupId'] = port_group_id + vnic['mtu'] = 1500 + vnic['enableProxyArp'] = False + vnic['enableSendRedirects'] = True + vnic['isConnected'] = True def _update_router_bindings(self, context, edge_id): edge_router_binds = nsxv_db.get_nsxv_router_bindings_by_edge( diff --git a/vmware_nsx/plugins/nsx_v/plugin.py b/vmware_nsx/plugins/nsx_v/plugin.py index 121b10a742..3ba9ac435d 100644 --- a/vmware_nsx/plugins/nsx_v/plugin.py +++ b/vmware_nsx/plugins/nsx_v/plugin.py @@ -1000,7 +1000,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin, if self._vcm: try: h, switch = self.nsx_v.vcns.get_vdn_switch(dvs_id) - except Exception as e: + except Exception: LOG.warning('DVS %s not registered on NSX. Unable to ' 'update teaming for network %s', dvs_id, net_id) @@ -1362,9 +1362,9 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin, # requires allowing multiple or cidr-based allowed address pairs # defined per port but doesn't want to disable spoofguard globally sg_policy_id = None - allow_multiple_addresses = (not net_data[psec.PORTSECURITY] - and cfg.CONF.nsxv. - allow_multiple_ip_addresses) + allow_multiple_addresses = (not net_data[psec.PORTSECURITY] and + cfg.CONF. + nsxv.allow_multiple_ip_addresses) if (cfg.CONF.nsxv.spoofguard_enabled and backend_network and not allow_multiple_addresses): # This variable is set as the method below may result in a @@ -1816,10 +1816,10 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin, self._update_vnic_assigned_addresses(context.session, port, vnic_id) except Exception as e: - msg = _('Unable to add port to spoofguard policy error ' - '%s') % e - raise n_exc.BadRequest(resource='spoofguard policy', - msg=msg) + msg = _('Unable to add port to spoofguard policy error ' + '%s') % e + raise n_exc.BadRequest(resource='spoofguard policy', + msg=msg) def update_network(self, context, id, network): net_attrs = network['network'] @@ -2005,17 +2005,17 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin, not network_port_security): for ap in attrs[addr_apidef.ADDRESS_PAIRS]: # Check that the IP address is a subnet - if len(ap['ip_address'].split('/')) > 1: - msg = _('NSXv does not support CIDR as address pairs') - raise n_exc.BadRequest(resource='address_pairs', - msg=msg) + if len(ap['ip_address'].split('/')) > 1: + msg = _('NSXv does not support CIDR as address pairs') + raise n_exc.BadRequest(resource='address_pairs', + msg=msg) # Check that the MAC address is the same as the port for ap in attrs[addr_apidef.ADDRESS_PAIRS]: if ('mac_address' in ap and ap['mac_address'] != db_port['mac_address']): - msg = _('Address pairs should have same MAC as the ' - 'port') - raise n_exc.BadRequest(resource='address_pairs', msg=msg) + msg = _('Address pairs should have same MAC as the ' + 'port') + raise n_exc.BadRequest(resource='address_pairs', msg=msg) def _is_mac_in_use(self, context, network_id, mac_address): # Override this method as the backed doesn't support using the same diff --git a/vmware_nsx/plugins/nsx_v3/plugin.py b/vmware_nsx/plugins/nsx_v3/plugin.py index b42717e50c..7eb4ff7f5f 100644 --- a/vmware_nsx/plugins/nsx_v3/plugin.py +++ b/vmware_nsx/plugins/nsx_v3/plugin.py @@ -660,6 +660,7 @@ class NsxV3Plugin(nsx_plugin_common.NsxPluginV3Base, def _init_mac_learning_profiles(self): with locking.LockManager.get_lock('nsxv3_mac_learning_profile_init'): if not self._get_mac_learning_profile(): + # pylint: disable=unexpected-keyword-arg self.nsxlib.switching_profile.create_mac_learning_profile( NSX_V3_MAC_LEARNING_PROFILE_NAME, 'Neutron MAC Learning Profile', @@ -667,6 +668,7 @@ class NsxV3Plugin(nsx_plugin_common.NsxPluginV3Base, tags=self.nsxlib.build_v3_api_version_tag()) self._get_mac_learning_profile() if not self._get_mac_learning_disabled_profile(): + # pylint: disable=unexpected-keyword-arg self.nsxlib.switching_profile.create_mac_learning_profile( NSX_V3_MAC_DISABLED_PROFILE_NAME, 'Neutron MAC Learning Disabled Profile', @@ -3139,10 +3141,10 @@ class NsxV3Plugin(nsx_plugin_common.NsxPluginV3Base, name = self.nsxlib.ns_group.get_name(secgroup) if self.nsxlib.feature_supported( - nsxlib_consts.FEATURE_DYNAMIC_CRITERIA): - tag_expression = ( - self.nsxlib.ns_group.get_port_tag_expression( - security.PORT_SG_SCOPE, secgroup['id'])) + nsxlib_consts.FEATURE_DYNAMIC_CRITERIA): + tag_expression = ( + self.nsxlib.ns_group.get_port_tag_expression( + security.PORT_SG_SCOPE, secgroup['id'])) else: tag_expression = None diff --git a/vmware_nsx/services/lbaas/nsx_p/implementation/listener_mgr.py b/vmware_nsx/services/lbaas/nsx_p/implementation/listener_mgr.py index c7628d5546..17fe8f9f7e 100644 --- a/vmware_nsx/services/lbaas/nsx_p/implementation/listener_mgr.py +++ b/vmware_nsx/services/lbaas/nsx_p/implementation/listener_mgr.py @@ -104,8 +104,9 @@ class EdgeListenerManagerFromDict(base_mgr.NsxpLoadbalancerBaseManager): ssl_profile_binding = self._upload_certificate( listener['id'], listener['default_tls_container_id'], tags, certificate=certificate) - if (listener['protocol'] == lb_const.LB_PROTOCOL_TERMINATED_HTTPS - and ssl_profile_binding): + if (listener['protocol'] == + lb_const.LB_PROTOCOL_TERMINATED_HTTPS and + ssl_profile_binding): kwargs.update(ssl_profile_binding) waf_profile, mode = self.core_plugin.get_waf_profile_path_and_mode() diff --git a/vmware_nsx/services/lbaas/nsx_v/implementation/listener_mgr.py b/vmware_nsx/services/lbaas/nsx_v/implementation/listener_mgr.py index a0f718e379..d47d0c863a 100644 --- a/vmware_nsx/services/lbaas/nsx_v/implementation/listener_mgr.py +++ b/vmware_nsx/services/lbaas/nsx_v/implementation/listener_mgr.py @@ -291,7 +291,7 @@ class EdgeListenerManagerFromDict(base_mgr.EdgeLoadbalancerBaseManager): with locking.LockManager.get_lock(edge_id): self.vcns.delete_app_profile(edge_id, app_profile_id) except (vcns_exc.ResourceNotFound, vcns_exc.RequestBad): - LOG.error('app profile not found on edge: %s', edge_id) + LOG.error('app profile not found on edge: %s', edge_id) except vcns_exc.VcnsApiException: LOG.error('Failed to delete app profile on Edge: %s', edge_id) diff --git a/vmware_nsx/shell/admin/plugins/nsxv/resources/backup_edges.py b/vmware_nsx/shell/admin/plugins/nsxv/resources/backup_edges.py index 0c85539bd2..5d5a032ea7 100644 --- a/vmware_nsx/shell/admin/plugins/nsxv/resources/backup_edges.py +++ b/vmware_nsx/shell/admin/plugins/nsxv/resources/backup_edges.py @@ -343,6 +343,7 @@ def nsx_fix_name_mismatch(resource, event, trigger, **kwargs): 'Edge %s has no backup prefix on NSX', edge_id) return + registry.subscribe(nsx_list_backup_edges, constants.BACKUP_EDGES, shell.Operations.LIST.value) diff --git a/vmware_nsx/shell/admin/plugins/nsxv/resources/edges.py b/vmware_nsx/shell/admin/plugins/nsxv/resources/edges.py index 4b3af40f6f..885f91e9bb 100644 --- a/vmware_nsx/shell/admin/plugins/nsxv/resources/edges.py +++ b/vmware_nsx/shell/admin/plugins/nsxv/resources/edges.py @@ -327,7 +327,7 @@ def change_edge_ha(ha, edge_id): 'enabled': ha} try: nsxv.enable_ha(edge_id, request) - except nsxv_exceptions.ResourceNotFound as e: + except nsxv_exceptions.ResourceNotFound: LOG.error("Edge %s not found", edge_id) except exceptions.NeutronException as e: LOG.error("%s", str(e)) @@ -353,7 +353,7 @@ def change_edge_syslog(properties): edge_id = properties.get('edge-id') try: nsxv.update_edge_syslog(edge_id, request) - except nsxv_exceptions.ResourceNotFound as e: + except nsxv_exceptions.ResourceNotFound: LOG.error("Edge %s not found", edge_id) except exceptions.NeutronException as e: LOG.error("%s", str(e)) @@ -362,7 +362,7 @@ def change_edge_syslog(properties): def delete_edge_syslog(edge_id): try: nsxv.delete_edge_syslog(edge_id) - except nsxv_exceptions.ResourceNotFound as e: + except nsxv_exceptions.ResourceNotFound: LOG.error("Edge %s not found", edge_id) except exceptions.NeutronException as e: LOG.error("%s", str(e)) @@ -404,7 +404,7 @@ def change_edge_loglevel(properties): try: edge_utils.update_edge_loglevel(nsxv, edge_id, module, level) - except nsxv_exceptions.ResourceNotFound as e: + except nsxv_exceptions.ResourceNotFound: LOG.error("Edge %s not found", edge_id) except exceptions.NeutronException as e: LOG.error("%s", str(e)) @@ -422,7 +422,7 @@ def change_edge_appliance_size(properties): try: nsxv.change_edge_appliance_size( properties.get('edge-id'), size) - except nsxv_exceptions.ResourceNotFound as e: + except nsxv_exceptions.ResourceNotFound: LOG.error("Edge %s not found", properties.get('edge-id')) except exceptions.NeutronException as e: LOG.error("%s", str(e)) @@ -462,7 +462,7 @@ def change_edge_appliance(edge_id): request = {'appliances': appliances, 'applianceSize': size} try: nsxv.change_edge_appliance(edge_id, request) - except nsxv_exceptions.ResourceNotFound as e: + except nsxv_exceptions.ResourceNotFound: LOG.error("Edge %s not found", edge_id) except exceptions.NeutronException as e: LOG.error("%s", str(e)) @@ -503,7 +503,7 @@ def change_edge_appliance_reservations(properties): request = {'appliances': appliances} try: nsxv.change_edge_appliance(edge_id, request) - except nsxv_exceptions.ResourceNotFound as e: + except nsxv_exceptions.ResourceNotFound: LOG.error("Edge %s not found", edge_id) except exceptions.NeutronException as e: LOG.error("%s", str(e)) diff --git a/vmware_nsx/shell/resources.py b/vmware_nsx/shell/resources.py index dca147be0f..645cd43999 100644 --- a/vmware_nsx/shell/resources.py +++ b/vmware_nsx/shell/resources.py @@ -71,6 +71,7 @@ class Operations(enum.Enum): VALIDATE = 'validate' REUSE = 'reuse' + ops = [op.value for op in Operations] diff --git a/vmware_nsx/tests/unit/extensions/test_portsecurity.py b/vmware_nsx/tests/unit/extensions/test_portsecurity.py index a496a615d5..648af32e81 100644 --- a/vmware_nsx/tests/unit/extensions/test_portsecurity.py +++ b/vmware_nsx/tests/unit/extensions/test_portsecurity.py @@ -29,7 +29,7 @@ class PortSecurityTestCaseNSXv2(psec.PortSecurityDBTestCase, class TestPortSecurityNSXv2(PortSecurityTestCaseNSXv2, psec.TestPortSecurity): - pass + pass class TestPortSecurityNSXv3(psec.TestPortSecurity, diff --git a/vmware_nsx/tests/unit/nsx_p/test_plugin.py b/vmware_nsx/tests/unit/nsx_p/test_plugin.py index 3982cdd7fc..d37eef6729 100644 --- a/vmware_nsx/tests/unit/nsx_p/test_plugin.py +++ b/vmware_nsx/tests/unit/nsx_p/test_plugin.py @@ -552,11 +552,11 @@ class NsxPTestNetworks(test_db_base_plugin_v2.TestNetworksV2, 'provider:physical_network': 'xxx', 'qos_policy_id': policy_id, 'port_security_enabled': False}} - with mock_ens, mock_tz, mock_tt,\ - mock.patch.object(self.plugin, '_validate_qos_policy_id'): - self.assertRaises(n_exc.InvalidInput, - self.plugin.create_network, - context.get_admin_context(), data) + with mock_ens, mock_tz, mock_tt, mock.patch.object( + self.plugin, '_validate_qos_policy_id'): + self.assertRaises(n_exc.InvalidInput, + self.plugin.create_network, + context.get_admin_context(), data) def test_update_ens_network_with_qos(self): cfg.CONF.set_override('ens_support', True, 'nsx_v3') @@ -1660,38 +1660,38 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest, "NsxPolicySegmentApi.update") as seg_update: with self.router() as r, self.network() as n: - with self.subnet(network=n, cidr='fd00::0/64', - gateway_ip='fd00::1', ip_version=6, - enable_dhcp=False) as s6, \ - self.subnet(network=n, cidr='2.0.0.0/24', - gateway_ip='2.0.0.1') as s4: - - subnets = [] - if s6_first: - self._router_interface_action('add', - r['router']['id'], - s6['subnet']['id'], - None) - subnets.append(s6['subnet']['cidr']) + with self.subnet( + network=n, cidr='fd00::0/64', gateway_ip='fd00::1', + ip_version=6, enable_dhcp=False) as s6, self.subnet( + network=n, cidr='2.0.0.0/24', + gateway_ip='2.0.0.1') as s4: + subnets = [] + if s6_first: self._router_interface_action('add', r['router']['id'], - s4['subnet']['id'], + s6['subnet']['id'], None) - subnets.append(s4['subnet']['cidr']) + subnets.append(s6['subnet']['cidr']) - if not s6_first: - self._router_interface_action('add', - r['router']['id'], - s6['subnet']['id'], - None) - subnets.append(s6['subnet']['cidr']) + self._router_interface_action('add', + r['router']['id'], + s4['subnet']['id'], + None) + subnets.append(s4['subnet']['cidr']) - # We expect two subnet objects on segment - seg_update.assert_called_with( - n['network']['id'], - subnets=[mock.ANY, mock.ANY], - tier1_id=r['router']['id']) + if not s6_first: + self._router_interface_action('add', + r['router']['id'], + s6['subnet']['id'], + None) + subnets.append(s6['subnet']['cidr']) + + # We expect two subnet objects on segment + seg_update.assert_called_with( + n['network']['id'], + subnets=[mock.ANY, mock.ANY], + tier1_id=r['router']['id']) def test_router_add_v4_v6_subnets(self): self._test_router_add_dual_stack_subnets() @@ -2031,21 +2031,20 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest, path_prefix = ("/infra/sites/default/enforcement-points/default/" "edge-clusters/") # create a router and external network - with self.router() as r,\ - self._create_l3_ext_network() as ext_net,\ - self.subnet(network=ext_net, cidr='10.0.1.0/24', - enable_dhcp=False) as s,\ - mock.patch("vmware_nsxlib.v3.policy.core_resources." - "NsxPolicyTier1Api.get_edge_cluster_path", - return_value=False),\ - mock.patch("vmware_nsxlib.v3.policy.core_resources." - "NsxPolicyTier1Api.set_edge_cluster_path" - ) as add_srv_router: - self._add_external_gateway_to_router( - r['router']['id'], - s['subnet']['network_id']) - add_srv_router.assert_called_once_with( - mock.ANY, '%s%s' % (path_prefix, edge_cluster)) + with self.router() as r, self._create_l3_ext_network() as ext_net, \ + self.subnet( + network=ext_net, cidr='10.0.1.0/24', + enable_dhcp=False) as s, mock.patch( + "vmware_nsxlib.v3.policy.core_resources." + "NsxPolicyTier1Api.get_edge_cluster_path", + return_value=False), mock.patch( + "vmware_nsxlib.v3.policy.core_resources." + "NsxPolicyTier1Api.set_edge_cluster_path") as add_srv_router: + self._add_external_gateway_to_router( + r['router']['id'], + s['subnet']['network_id']) + add_srv_router.assert_called_once_with( + mock.ANY, '%s%s' % (path_prefix, edge_cluster)) def test_router_add_interface_cidr_overlapped_with_gateway(self): with self.router() as r,\ @@ -2127,22 +2126,22 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest, with self.subnet(cidr='30.0.0.0/24', gateway_ip=None) as private_sub: with self.port( - subnet=private_sub, - device_owner=constants.DEVICE_OWNER_ROUTER_INTF) as p: + subnet=private_sub, + device_owner=constants.DEVICE_OWNER_ROUTER_INTF) as p: port_id = p['port']['id'] with self.router() as r: self._router_interface_action('add', r['router']['id'], None, port_id) - with self.external_network() as public_net,\ - self.subnet(network=public_net, cidr='12.0.0.0/24', - enable_dhcp=False) as public_sub: - self._add_external_gateway_to_router( - r['router']['id'], - public_sub['subnet']['network_id']) - self._make_floatingip( - self.fmt, public_sub['subnet']['network_id'], - port_id=port_id, - http_status=exc.HTTPBadRequest.code) + with self.external_network() as public_net, self.subnet( + network=public_net, cidr='12.0.0.0/24', + enable_dhcp=False) as public_sub: + self._add_external_gateway_to_router( + r['router']['id'], + public_sub['subnet']['network_id']) + self._make_floatingip( + self.fmt, public_sub['subnet']['network_id'], + port_id=port_id, + http_status=exc.HTTPBadRequest.code) def test_assign_floatingip_to_router_interface_device_owner_fail(self): # This tests that an error is raised when trying to assign a router @@ -2150,20 +2149,20 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest, with self.subnet(cidr='30.0.0.0/24', gateway_ip=None) as private_sub: with self.port( - subnet=private_sub, - device_owner=constants.DEVICE_OWNER_ROUTER_INTF) as p: + subnet=private_sub, + device_owner=constants.DEVICE_OWNER_ROUTER_INTF) as p: port_id = p['port']['id'] with self.router() as r: self._router_interface_action('add', r['router']['id'], None, port_id) - with self.external_network() as public_net,\ - self.subnet(network=public_net, cidr='12.0.0.0/24', - enable_dhcp=False) as public_sub: - self._add_external_gateway_to_router( - r['router']['id'], - public_sub['subnet']['network_id']) - fip = self._make_floatingip(self.fmt, public_sub[ - 'subnet']['network_id']) - self._update('floatingips', fip['floatingip'][ - 'id'], {'floatingip': {'port_id': port_id}}, - expected_code=exc.HTTPBadRequest.code) + with self.external_network() as public_net, self.subnet( + network=public_net, cidr='12.0.0.0/24', + enable_dhcp=False) as public_sub: + self._add_external_gateway_to_router( + r['router']['id'], + public_sub['subnet']['network_id']) + fip = self._make_floatingip(self.fmt, public_sub[ + 'subnet']['network_id']) + self._update('floatingips', fip['floatingip'][ + 'id'], {'floatingip': {'port_id': port_id}}, + expected_code=exc.HTTPBadRequest.code) diff --git a/vmware_nsx/tests/unit/nsx_v/test_plugin.py b/vmware_nsx/tests/unit/nsx_v/test_plugin.py index cde7ffc3f2..ae31923951 100644 --- a/vmware_nsx/tests/unit/nsx_v/test_plugin.py +++ b/vmware_nsx/tests/unit/nsx_v/test_plugin.py @@ -2697,31 +2697,33 @@ class L3NatTestCaseBase(test_l3_plugin.L3NatTestCaseMixin): to a router places them all on the same router interface. """ with self.router() as r, self.network() as n: - with (self.subnet(network=n, cidr='fd00::1/64', - enable_dhcp=False, ip_version=6) - ) as s1, self.subnet(network=n, cidr='fd01::1/64', - ip_version=6, enable_dhcp=False) as s2: - body = self._router_interface_action('add', - r['router']['id'], - s1['subnet']['id'], - None) - pid1 = body['port_id'] - body = self._router_interface_action('add', - r['router']['id'], - s2['subnet']['id'], - None) - pid2 = body['port_id'] - self.assertEqual(pid1, pid2) - port = self._show('ports', pid1) - self.assertEqual(2, len(port['port']['fixed_ips'])) - port_subnet_ids = [fip['subnet_id'] for fip in - port['port']['fixed_ips']] - self.assertIn(s1['subnet']['id'], port_subnet_ids) - self.assertIn(s2['subnet']['id'], port_subnet_ids) - self._router_interface_action('remove', r['router']['id'], - s1['subnet']['id'], None) - self._router_interface_action('remove', r['router']['id'], - s2['subnet']['id'], None) + with self.subnet( + network=n, cidr='fd00::1/64', + enable_dhcp=False, ip_version=6) as s1, self.subnet( + network=n, cidr='fd01::1/64', + ip_version=6, enable_dhcp=False) as s2: + + body = self._router_interface_action('add', + r['router']['id'], + s1['subnet']['id'], + None) + pid1 = body['port_id'] + body = self._router_interface_action('add', + r['router']['id'], + s2['subnet']['id'], + None) + pid2 = body['port_id'] + self.assertEqual(pid1, pid2) + port = self._show('ports', pid1) + self.assertEqual(2, len(port['port']['fixed_ips'])) + port_subnet_ids = [fip['subnet_id'] for fip in + port['port']['fixed_ips']] + self.assertIn(s1['subnet']['id'], port_subnet_ids) + self.assertIn(s2['subnet']['id'], port_subnet_ids) + self._router_interface_action('remove', r['router']['id'], + s1['subnet']['id'], None) + self._router_interface_action('remove', r['router']['id'], + s2['subnet']['id'], None) def test_router_add_interface_ipv6_port_existing_network_returns_400(self): """Ensure unique IPv6 router ports per network id. @@ -5379,36 +5381,36 @@ class TestSharedRouterTestCase(L3NatTest, L3NatTestCaseBase, ext_subnet['subnet']['network_id']) def test_routers_with_interface_on_same_edge(self): - with self.router() as r1, self.router() as r2,\ - self.subnet(cidr='11.0.0.0/24') as s11,\ - self.subnet(cidr='12.0.0.0/24') as s12: - self._router_interface_action('add', - r1['router']['id'], - s11['subnet']['id'], - None) - self._router_interface_action('add', - r2['router']['id'], - s12['subnet']['id'], - None) - routers_expected = [r1['router']['id'], r2['router']['id']] - routers_1 = (self.plugin_instance.edge_manager. - get_routers_on_same_edge( - context.get_admin_context(), - r1['router']['id'])) - self.assertEqual(set(routers_expected), set(routers_1)) - routers_2 = (self.plugin_instance.edge_manager. - get_routers_on_same_edge( - context.get_admin_context(), - r2['router']['id'])) - self.assertEqual(set(routers_expected), set(routers_2)) - self._router_interface_action('remove', - r1['router']['id'], - s11['subnet']['id'], - None) - self._router_interface_action('remove', - r2['router']['id'], - s12['subnet']['id'], - None) + with self.router() as r1, self.router() as r2,\ + self.subnet(cidr='11.0.0.0/24') as s11,\ + self.subnet(cidr='12.0.0.0/24') as s12: + self._router_interface_action('add', + r1['router']['id'], + s11['subnet']['id'], + None) + self._router_interface_action('add', + r2['router']['id'], + s12['subnet']['id'], + None) + routers_expected = [r1['router']['id'], r2['router']['id']] + routers_1 = (self.plugin_instance.edge_manager. + get_routers_on_same_edge( + context.get_admin_context(), + r1['router']['id'])) + self.assertEqual(set(routers_expected), set(routers_1)) + routers_2 = (self.plugin_instance.edge_manager. + get_routers_on_same_edge( + context.get_admin_context(), + r2['router']['id'])) + self.assertEqual(set(routers_expected), set(routers_2)) + self._router_interface_action('remove', + r1['router']['id'], + s11['subnet']['id'], + None) + self._router_interface_action('remove', + r2['router']['id'], + s12['subnet']['id'], + None) def test_routers_with_overlap_interfaces(self): with self.router() as r1, self.router() as r2,\ diff --git a/vmware_nsx/tests/unit/nsx_v/vshield/fake_vcns.py b/vmware_nsx/tests/unit/nsx_v/vshield/fake_vcns.py index 9b366794b9..5967861cfb 100644 --- a/vmware_nsx/tests/unit/nsx_v/vshield/fake_vcns.py +++ b/vmware_nsx/tests/unit/nsx_v/vshield/fake_vcns.py @@ -995,9 +995,9 @@ class FakeVcns(object): return self._section_not_found(section_id) _section = self._sections[section_id] if (_section['name'] != section_name and - section_name in self._sections['names']): - # Theres a section with this name already - headers, response = self._unknown_error() + section_name in self._sections['names']): + # There's a section with this name already + headers, response = self._unknown_error() else: # Different Etag every successful update _section['etag'] = ('Etag-1' if _section['etag'] == 'Etag-0' diff --git a/vmware_nsx/tests/unit/nsx_v3/test_plugin.py b/vmware_nsx/tests/unit/nsx_v3/test_plugin.py index b5fc0d1b18..0bf511f77d 100644 --- a/vmware_nsx/tests/unit/nsx_v3/test_plugin.py +++ b/vmware_nsx/tests/unit/nsx_v3/test_plugin.py @@ -569,7 +569,7 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxV3PluginTestCaseMixin): 'provider:physical_network': 'xxx', 'port_security_enabled': True}} with mock_ens, mock_tz, mock_tt: - self.plugin.create_network(context.get_admin_context(), data) + self.plugin.create_network(context.get_admin_context(), data) def test_create_ens_network_with_qos(self): cfg.CONF.set_override('ens_support', True, 'nsx_v3') @@ -590,11 +590,11 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxV3PluginTestCaseMixin): 'provider:physical_network': 'xxx', 'qos_policy_id': policy_id, 'port_security_enabled': False}} - with mock_ens, mock_tz, mock_tt,\ - mock.patch.object(self.plugin, '_validate_qos_policy_id'): - self.assertRaises(n_exc.InvalidInput, - self.plugin.create_network, - context.get_admin_context(), data) + with mock_ens, mock_tz, mock_tt, mock.patch.object( + self.plugin, '_validate_qos_policy_id'): + self.assertRaises(n_exc.InvalidInput, + self.plugin.create_network, + context.get_admin_context(), data) def test_update_ens_network_with_qos(self): cfg.CONF.set_override('ens_support', True, 'nsx_v3') @@ -720,13 +720,14 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxV3PluginTestCaseMixin): pnet.PHYSICAL_NETWORK: 'fb69d878-958e-4f32-84e4-50286f26226b'} - with mock.patch('vmware_nsxlib.v3.core_resources.NsxLibTransportZone.' - 'get_transport_type', return_value='VLAN'): + gtt_path = "vmware_nsxlib.v3.core_resources." \ + "NsxLibTransportZone.get_transport_type" + with mock.patch(gtt_path, return_value='VLAN'): with self.network(name=name, providernet_args=providernet_args, arg_list=(pnet.NETWORK_TYPE, pnet.PHYSICAL_NETWORK)) as net: for k, v in expected: - self.assertEqual(net['network'][k], v) + self.assertEqual(net['network'][k], v) def test_create_phys_vlan_generate(self): cfg.CONF.set_override('network_vlan_ranges', @@ -1069,27 +1070,27 @@ class TestPortsV2(common_v3.NsxV3SubnetMixin, self.plugin.update_port, self.ctx, port['id'], data) def test_fail_create_allowed_address_pairs_over_limit(self): - with self.network() as network,\ - self.subnet(network=network, enable_dhcp=True) as s1: - data = {'port': { - 'network_id': network['network']['id'], - 'tenant_id': self._tenant_id, - 'name': 'pair_port', - 'admin_state_up': True, - 'device_id': 'fake_device', - 'device_owner': 'fake_owner', - 'fixed_ips': [{'subnet_id': s1['subnet']['id']}] - } - } - count = 1 - address_pairs = [] - while count < 129: - address_pairs.append({'ip_address': '10.0.0.%s' % - count}) - count += 1 - data['port']['allowed_address_pairs'] = address_pairs - self.assertRaises(n_exc.InvalidInput, - self.plugin.create_port, self.ctx, data) + with self.network() as network, self.subnet( + network=network, enable_dhcp=True) as s1: + data = { + 'port': { + 'network_id': network['network']['id'], + 'tenant_id': self._tenant_id, + 'name': 'pair_port', + 'admin_state_up': True, + 'device_id': 'fake_device', + 'device_owner': 'fake_owner', + 'fixed_ips': [{'subnet_id': s1['subnet']['id']}] + } + } + count = 1 + address_pairs = [] + while count < 129: + address_pairs.append({'ip_address': '10.0.0.%s' % count}) + count += 1 + data['port']['allowed_address_pairs'] = address_pairs + self.assertRaises(n_exc.InvalidInput, + self.plugin.create_port, self.ctx, data) def test_fail_update_lb_port_with_fixed_ip(self): with self.network() as network: @@ -3171,21 +3172,20 @@ class TestL3NatTestCase(L3NatTest, with self.subnet(cidr='30.0.0.0/24', gateway_ip=None) as private_sub: with self.port( subnet=private_sub, - device_owner=constants.DEVICE_OWNER_ROUTER_INTF) as p: + device_owner=constants.DEVICE_OWNER_ROUTER_INTF) as p: port_id = p['port']['id'] with self.router() as r: self._router_interface_action('add', r['router']['id'], None, port_id) - with self.external_network() as public_net,\ - self.subnet( + with self.external_network() as public_net, self.subnet( network=public_net, cidr='12.0.0.0/24') as public_sub: - self._add_external_gateway_to_router( - r['router']['id'], - public_sub['subnet']['network_id']) - self._make_floatingip( - self.fmt, public_sub['subnet']['network_id'], - port_id=port_id, - http_status=exc.HTTPBadRequest.code) + self._add_external_gateway_to_router( + r['router']['id'], + public_sub['subnet']['network_id']) + self._make_floatingip( + self.fmt, public_sub['subnet']['network_id'], + port_id=port_id, + http_status=exc.HTTPBadRequest.code) def test_assign_floatingip_to_router_interface_device_owner_fail(self): # This tests that an error is raised when trying to assign a router @@ -3194,22 +3194,21 @@ class TestL3NatTestCase(L3NatTest, with self.subnet(cidr='30.0.0.0/24', gateway_ip=None) as private_sub: with self.port( subnet=private_sub, - device_owner=constants.DEVICE_OWNER_ROUTER_INTF) as p: + device_owner=constants.DEVICE_OWNER_ROUTER_INTF) as p: port_id = p['port']['id'] with self.router() as r: self._router_interface_action('add', r['router']['id'], None, port_id) - with self.external_network() as public_net,\ - self.subnet( + with self.external_network() as public_net, self.subnet( network=public_net, cidr='12.0.0.0/24') as public_sub: - self._add_external_gateway_to_router( - r['router']['id'], - public_sub['subnet']['network_id']) - fip = self._make_floatingip(self.fmt, public_sub[ - 'subnet']['network_id']) - self._update('floatingips', fip['floatingip'][ - 'id'], {'floatingip': {'port_id': port_id}}, - expected_code=exc.HTTPBadRequest.code) + self._add_external_gateway_to_router( + r['router']['id'], + public_sub['subnet']['network_id']) + fip = self._make_floatingip(self.fmt, public_sub[ + 'subnet']['network_id']) + self._update('floatingips', fip['floatingip'][ + 'id'], {'floatingip': {'port_id': port_id}}, + expected_code=exc.HTTPBadRequest.code) class ExtGwModeTestCase(test_ext_gw_mode.ExtGwModeIntTestCase, diff --git a/vmware_nsx/tests/unit/services/dynamic_routing/test_nsxv_bgp_driver.py b/vmware_nsx/tests/unit/services/dynamic_routing/test_nsxv_bgp_driver.py index 973b3d2c2a..4fc0831651 100644 --- a/vmware_nsx/tests/unit/services/dynamic_routing/test_nsxv_bgp_driver.py +++ b/vmware_nsx/tests/unit/services/dynamic_routing/test_nsxv_bgp_driver.py @@ -297,11 +297,11 @@ class TestNSXvBgpPlugin(test_plugin.NsxVPluginV2TestCase, self.skipTest('No SNAT with floating ips not supported') def test_add_bgp_peer_with_bad_id(self): - with self.subnetpool_with_address_scope( + with self.subnetpool_with_address_scope( 4, prefixes=['8.0.0.0/8']) as sp: - with self.bgp_speaker(sp['ip_version'], 1234) as speaker: - self.assertRaises(ext_bgp.BgpPeerNotFound, - self.bgp_plugin.add_bgp_peer, - self.context, - speaker['id'], - {'bgp_peer_id': 'aaa'}) + with self.bgp_speaker(sp['ip_version'], 1234) as speaker: + self.assertRaises(ext_bgp.BgpPeerNotFound, + self.bgp_plugin.add_bgp_peer, + self.context, + speaker['id'], + {'bgp_peer_id': 'aaa'}) diff --git a/vmware_nsx/tests/unit/services/qos/test_nsxp_notification.py b/vmware_nsx/tests/unit/services/qos/test_nsxp_notification.py index 4f7de22cce..254c0b15e0 100644 --- a/vmware_nsx/tests/unit/services/qos/test_nsxp_notification.py +++ b/vmware_nsx/tests/unit/services/qos/test_nsxp_notification.py @@ -252,14 +252,14 @@ class TestQosNsxPNotification(base.BaseQosTestCase, self.ctxt, **self.policy_data['policy']) # add a rule to the policy setattr(_policy, "rules", [rule]) - with mock.patch.object(QoSPolicy, 'get_object', - return_value=_policy),\ - mock.patch('neutron.objects.db.api.update_object', - return_value=rule_data): - self.assertRaises( - exceptions.DriverCallError, - self.qos_plugin.update_policy_bandwidth_limit_rule, - self.ctxt, rule.id, _policy.id, rule_data) + with mock.patch.object( + QoSPolicy, 'get_object', return_value=_policy), mock.patch( + 'neutron.objects.db.api.update_object', + return_value=rule_data): + self.assertRaises( + exceptions.DriverCallError, + self.qos_plugin.update_policy_bandwidth_limit_rule, + self.ctxt, rule.id, _policy.id, rule_data) @mock.patch.object(QoSPolicy, '_reload_rules') def test_dscp_rule_create_profile(self, *mocks): diff --git a/vmware_nsx/tests/unit/services/qos/test_nsxv3_notification.py b/vmware_nsx/tests/unit/services/qos/test_nsxv3_notification.py index e9075547b1..c8148ee172 100644 --- a/vmware_nsx/tests/unit/services/qos/test_nsxv3_notification.py +++ b/vmware_nsx/tests/unit/services/qos/test_nsxv3_notification.py @@ -266,14 +266,14 @@ class TestQosNsxV3Notification(base.BaseQosTestCase, self.ctxt, **self.policy_data['policy']) # add a rule to the policy setattr(_policy, "rules", [rule]) - with mock.patch.object(QoSPolicy, 'get_object', - return_value=_policy),\ - mock.patch('neutron.objects.db.api.update_object', - return_value=rule_data): - self.assertRaises( - exceptions.DriverCallError, - self.qos_plugin.update_policy_bandwidth_limit_rule, - self.ctxt, rule.id, _policy.id, rule_data) + with mock.patch.object( + QoSPolicy, 'get_object', return_value=_policy), mock.patch( + 'neutron.objects.db.api.update_object', + return_value=rule_data): + self.assertRaises( + exceptions.DriverCallError, + self.qos_plugin.update_policy_bandwidth_limit_rule, + self.ctxt, rule.id, _policy.id, rule_data) @mock.patch.object(QoSPolicy, '_reload_rules') def test_dscp_rule_create_profile(self, *mocks): @@ -325,9 +325,9 @@ class TestQosNsxV3Notification(base.BaseQosTestCase, # add a rule to the policy setattr(policy, "rules", [min_bw_rule]) with mock.patch.object( - QoSPolicy, 'get_object', return_value=policy),\ - mock.patch('neutron.objects.db.api.' - 'update_object', return_value=self.dscp_rule_data): + QoSPolicy, 'get_object', return_value=policy), mock.patch( + 'neutron.objects.db.api.update_object', + return_value=self.dscp_rule_data): self.assertRaises( exceptions.DriverCallError, self.qos_plugin.update_policy_minimum_bandwidth_rule, diff --git a/vmware_nsx/tests/unit/services/vpnaas/test_nsxv3_vpnaas.py b/vmware_nsx/tests/unit/services/vpnaas/test_nsxv3_vpnaas.py index c16c016d82..b7b61de8a7 100644 --- a/vmware_nsx/tests/unit/services/vpnaas/test_nsxv3_vpnaas.py +++ b/vmware_nsx/tests/unit/services/vpnaas/test_nsxv3_vpnaas.py @@ -315,31 +315,31 @@ class TestDriverValidation(base.BaseTestCase): else: return connections - with mock.patch.object(self.validator.vpn_plugin, '_get_vpnservice', - side_effect=mock_get_service),\ - mock.patch.object(self.validator._core_plugin, 'get_routers', - side_effect=mock_get_routers),\ - mock.patch.object(self.validator._core_plugin, - '_find_router_subnets_cidrs', - return_value=router_subnets),\ - mock.patch.object(self.validator.vpn_plugin, - 'get_ipsec_site_connections', - side_effect=mock_get_connections): - ipsec_sitecon = {'id': '1', - 'vpnservice_id': '1', - 'mtu': 1500, - 'peer_address': self.peer_address, - 'peer_cidrs': [self.peer_cidr]} - if conn_params: - ipsec_sitecon.update(conn_params) - if success: - self.validator.validate_ipsec_site_connection( - self.context, ipsec_sitecon) - else: - self.assertRaises( - nsx_exc.NsxVpnValidationError, - self.validator.validate_ipsec_site_connection, - self.context, ipsec_sitecon) + with mock.patch.object( + self.validator.vpn_plugin, '_get_vpnservice', + side_effect=mock_get_service), mock.patch.object( + self.validator._core_plugin, 'get_routers', + side_effect=mock_get_routers), mock.patch.object( + self.validator._core_plugin, '_find_router_subnets_cidrs', + return_value=router_subnets), mock.patch.object( + self.validator.vpn_plugin, 'get_ipsec_site_connections', + side_effect=mock_get_connections): + + ipsec_sitecon = {'id': '1', + 'vpnservice_id': '1', + 'mtu': 1500, + 'peer_address': self.peer_address, + 'peer_cidrs': [self.peer_cidr]} + if conn_params: + ipsec_sitecon.update(conn_params) + if success: + self.validator.validate_ipsec_site_connection( + self.context, ipsec_sitecon) + else: + self.assertRaises( + nsx_exc.NsxVpnValidationError, + self.validator.validate_ipsec_site_connection, + self.context, ipsec_sitecon) def test_dpd_validation(self): params = {'dpd': {'action': 'hold', @@ -705,8 +705,9 @@ class TestVpnaasDriver(test_plugin.NsxV3PluginTestCaseMixin): with mock.patch.object( self.nsxlib_vpn.service, 'list', return_value={'results': nsx_services}),\ - mock.patch.object(self.service_plugin, 'get_vpnservices', - return_value=[{'id': 'dummy', 'router_id': 'dummy'}]),\ + mock.patch.object( + self.service_plugin, 'get_vpnservices', + return_value=[{'id': 'dummy', 'router_id': 'dummy'}]),\ mock.patch.object(self.nsxlib_vpn.service, 'delete') as delete_service: self.driver.delete_vpnservice(