x/vmware-nsx-tempest-plugin
Code Issues Proposed changes

Enhance VPN testcases

Browse Source 
Change-Id: I7d17b25ba761c51cc80c920451ec3c6206c4e935
This commit is contained in:
dkhairnar 2020-04-08 06:17:50 +00:00
parent 08cba39042
commit 0be66bd560
3 changed files with 146 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
vmware_nsx_tempest_plugin/services/nsxp_client.py
View File

@ -255,10 +255,10 @@ class NSXPClient(object):
locale_service = self.get_logical_router_local_services(os_name,
os_uuid)
vpn_service = self.get_vpn_service(os_name, os_uuid)
endpoint = "tier-1s/%s/locale-services/%s/ipsec-vpn-services/%s/ \
sessions" % (router["id"], locale_service[0].get("id"),
u = "tier-1s/%s/locale-services/%s/ipsec-vpn-services/%s/sessions" % \
(router["id"], locale_service[0].get("id"),
vpn_service[0].get("id"))
return self.get_logical_resources(endpoint)
return self.get_logical_resources(u)
def get_vpn_service(self, os_name, os_uuid):
"""

7
vmware_nsx_tempest_plugin/services/openstack_network_clients.py
View File

@ -144,6 +144,9 @@ class VPNClient(base.BaseNetworkClient):
"""Creates a new VPNService."""
return self.create_resource(self.vpnservices_path, kwargs)
def create_endpoint_group(self, **kwargs):
return self.create_resource(self.endpoint_groups_path, kwargs)
def update_vpnservice(self, vpnservice_id, **kwargs):
"""Updates a VPNService."""
uri = self.vpnservice_path % vpnservice_id
@ -197,6 +200,10 @@ class VPNClient(base.BaseNetworkClient):
uri = self.ipsecpolicy_path % (ipsecpolicy_id)
self.delete_resource(uri)
def delete_endpoint_group(self, endpoint_group_id):
uri = self.endpoint_group_path % (endpoint_group_id)
self.delete_resource(uri)
def list_ipsec_site_connections(self, retrieve_all=True, **_params):
"""Fetches all configured IPsecSiteConnections for a tenant."""
return self.list('ipsec_site_connections',

145
vmware_nsx_tempest_plugin/tests/api/test_vpn.py
View File

@ -94,6 +94,94 @@ class TestVpnOps(feature_manager.FeatureManager):
)
return dict(network=network, subnet=subnet, router=router)
def create_vpn_basic_topo_endpoint_group(
self, network_topology, name=None, ike=None, pfs=constants.PFS,
encryption_algorithm=constants.ENCRYPTION_ALGO,
lifetime=constants.LIFETIME,
peer_address=constants.PEER_ADDRESS,
peer_id=constants.PEER_ID,
site_connection_state=constants.SITE_CONNECTION_STATE):
# Create network topo
kwargs = {}
subnet = network_topology['subnet']
router = network_topology['router']
kwargs['vpnservice'] = dict(router_id=router['id'],
admin_state_up=site_connection_state,
name="vpn")
vpn_service = self.vpnaas_client.create_vpnservice(**kwargs)
self.addCleanup(
test_utils.call_and_ignore_notfound_exc,
self.vpnaas_client.delete_vpnservice,
vpn_service.get('vpnservice')['id'])
self.vpnaas_client.list_vpnservices()
if ike is None:
kwargs = {}
if lifetime is not None:
kwargs[
'ikepolicy'] = \
dict(name=data_utils.rand_name("ike-policy-"), pfs=pfs,
encryption_algorithm=encryption_algorithm,
lifetime=lifetime)
ike = self.vpnaas_client.create_ikepolicy(**kwargs)
self.addCleanup(
test_utils.call_and_ignore_notfound_exc,
self.vpnaas_client.delete_ikepolicy,
ike.get('ikepolicy')['id'])
kwargs[
'ipsecpolicy'] = dict(name=data_utils.rand_name("ipsec-policy-"),
pfs=pfs)
ipsec = self.vpnaas_client.create_ipsecpolicy(**kwargs)
self.addCleanup(
test_utils.call_and_ignore_notfound_exc,
self.vpnaas_client.delete_ipsecpolicy,
ipsec.get('ipsecpolicy')['id'])
kwargs = {}
name = data_utils.rand_name("local-endpoint-")
kwargs['endpoint_group'] = dict(endpoints=[subnet['id']],
type="subnet",
name=name)
local_endpoint = self.vpnaas_client.create_endpoint_group(**kwargs)
self.addCleanup(
test_utils.call_and_ignore_notfound_exc,
self.vpnaas_client.delete_endpoint_group,
local_endpoint['endpoint_group']['id'])
kwargs = {}
name = data_utils.rand_name("remote_endpoint-")
kwargs['endpoint_group'] = dict(endpoints=["40.20.1.0/24"],
type="cidr",
name=name)
remote_endpoint = self.vpnaas_client.create_endpoint_group(**kwargs)
self.addCleanup(
test_utils.call_and_ignore_notfound_exc,
self.vpnaas_client.delete_endpoint_group,
remote_endpoint['endpoint_group']['id'])
kwargs = {}
if name is not None:
name = "site-conn-" + name
else:
name = "site-conn"
kwargs[
"ipsec_site_connection"] = \
dict(vpnservice_id=vpn_service.get('vpnservice')['id'],
psk="secret",
admin_state_up=site_connection_state,
peer_ep_group_id=remote_endpoint['endpoint_group']['id'],
local_ep_group_id=local_endpoint['endpoint_group']['id'],
ikepolicy_id=ike.get(
'ikepolicy')['id'],
ipsecpolicy_id=ipsec.get(
'ipsecpolicy')['id'],
peer_address=peer_address,
peer_id=peer_id, name=name)
endpoint = self.vpnaas_client.create_ipsec_site_connection(**kwargs)
self.addCleanup(
test_utils.call_and_ignore_notfound_exc,
self.vpnaas_client.delete_ipsec_site_connection,
endpoint.get("ipsec_site_connection")['id'])
return dict(endpoint=endpoint, vpn_service=vpn_service,
ike=ike, ipsec=ipsec)
def create_vpn_basic_topo(
self, network_topology, name=None, ike=None, pfs=constants.PFS,
encryption_algorithm=constants.ENCRYPTION_ALGO,
@ -210,7 +298,7 @@ class TestVpnOps(feature_manager.FeatureManager):
if vpn['logical_router_id'] == rtr["id"]:
self.assertEqual((vpn.get('tags')[0]).get('tag'),
rtr["id"])
self.assertEqual(vpn['resource_type'].encode(),
self.assertEqual(vpn['resource_type'],
'IPSecVPNService')
break
break
@ -326,7 +414,7 @@ class TestVpnOps(feature_manager.FeatureManager):
if CONF.network.backend == 'nsxp':
local_endpoint = self.nsxp_client.get_local_endpoint(
router["name"], router["id"])
if local_endpoint[0].get('tags')[0].get("tag").encode() == \
if local_endpoint[0].get('tags')[0].get("tag") == \
router["id"]:
self.assertIsNotNone(local_endpoint[0].get("local_address"))
self.assertIsNotNone(local_endpoint[0].get("local_id"))
@ -388,9 +476,9 @@ class TestVpnOps(feature_manager.FeatureManager):
router["id"])
vpn_service = self.nsxp_client.get_vpn_service(router["name"],
router["id"])
self.assertEqual(vpn_service[0].get('tags')[0].get("tag").encode(),
self.assertEqual(vpn_service[0].get('tags')[0].get("tag"),
nsx_router["id"])
self.assertEqual(vpn_service[0].get('resource_type').encode(),
self.assertEqual(vpn_service[0].get('resource_type'),
'IPSecVpnService')
else:
routers = self.nsx_client.get_logical_routers()
@ -399,7 +487,7 @@ class TestVpnOps(feature_manager.FeatureManager):
for vpn in vpn_services:
if vpn['logical_router_id'] == rtr["id"]:
self.assertEqual(vpn['logical_router_id'], rtr["id"])
self.assertEqual(vpn['resource_type'].encode(),
self.assertEqual(vpn['resource_type'],
'IPSecVPNService')
break
break
@ -427,7 +515,7 @@ class TestVpnOps(feature_manager.FeatureManager):
if vpn['logical_router_id'] == rtr["id"]:
self.assertEqual(vpn['logical_router_id'],
rtr["id"])
self.assertEqual(vpn['resource_type'].encode(),
self.assertEqual(vpn['resource_type'],
'IPSecVPNService')
flag = 1
break
@ -547,7 +635,7 @@ class TestVpnOps(feature_manager.FeatureManager):
if CONF.network.backend == 'nsxp':
ipsec_session = self.nsxp_client.get_ipsec_session(router["name"],
router["id"])
self.assertEqual(ipsec_session[0].get('resource_type').encode(),
self.assertEqual(ipsec_session[0].get('resource_type'),
'PolicyBasedIPSecVpnSession')
self.assertEqual(ipsec_session[0].get('enabled'), False)
else:
@ -579,9 +667,9 @@ class TestVpnOps(feature_manager.FeatureManager):
router["id"])
vpn_service = self.nsxp_client.get_vpn_service(router["name"],
router["id"])
self.assertEqual(vpn_service[0].get('tags')[0].get("tag").encode(),
self.assertEqual(vpn_service[0].get('tags')[0].get("tag"),
nsx_router["id"])
self.assertEqual(vpn_service[0].get('resource_type').encode(),
self.assertEqual(vpn_service[0].get('resource_type'),
'IPSecVpnService')
else:
routers = self.nsx_client.get_logical_routers()
@ -654,7 +742,7 @@ class TestVpnOps(feature_manager.FeatureManager):
if CONF.network.backend == 'nsxp':
local_endpoint = self.nsxp_client.get_local_endpoint(
router["name"], router["id"])
if local_endpoint[0].get('tags')[0].get("tag").encode() == \
if local_endpoint[0].get('tags')[0].get("tag") == \
router["id"]:
self.assertIsNotNone(local_endpoint[0].get("local_address"))
self.assertIsNotNone(local_endpoint[0].get("local_id"))
@ -812,3 +900,40 @@ class TestVpnOps(feature_manager.FeatureManager):
LOG.info(
"IKEPolicy is in use by existing IPsecSiteConnection and "
" can't be updated or deleted")
@decorators.idempotent_id('d576c487-e7d5-4698-8a17-ea4521907675')
def test_vpn_endpoint_group(self):
network_topo = self.create_network_topo(cidr="37.0.0.0/24")
router = network_topo['router']
self.create_vpn_basic_topo_endpoint_group(network_topo)
ipsec_session = self.nsxp_client.get_ipsec_session(router["name"],
router["id"])
self.assertEqual(ipsec_session[0].get('resource_type'),
'PolicyBasedIPSecVpnSession')
nsx_router = self.nsxp_client.get_logical_router(router["name"],
router["id"])
vpn_service = self.nsxp_client.get_vpn_service(router["name"],
router["id"])
self.assertEqual(vpn_service[0].get('tags')[0].get("tag"),
nsx_router["id"])
self.assertEqual(vpn_service[0].get('resource_type'),
'IPSecVpnService')
@decorators.idempotent_id('d576c487-e7d5-4698-8a17-fa4521907675')
def test_vpn_endpoint_group_snat(self):
network_topo = self.create_network_topo(cidr="37.0.0.0/24",
enable_snat="True")
router = network_topo['router']
self.create_vpn_basic_topo_endpoint_group(network_topo)
ipsec_session = self.nsxp_client.get_ipsec_session(router["name"],
router["id"])
self.assertEqual(ipsec_session[0].get('resource_type'),
'PolicyBasedIPSecVpnSession')
nsx_router = self.nsxp_client.get_logical_router(router["name"],
router["id"])
vpn_service = self.nsxp_client.get_vpn_service(router["name"],
router["id"])
self.assertEqual(vpn_service[0].get('tags')[0].get("tag"),
nsx_router["id"])
self.assertEqual(vpn_service[0].get('resource_type'),
'IPSecVpnService')