openstack/neutron-vpnaas-dashboard
Code Issues Proposed changes

Migrate from neutronclient to SDK

Browse Source 
Related-Bug: #1999774
Change-Id: I934c7af83a3ab2e59b6a8c85974e88d3e44e5931
Signed-off-by: lajoskatona <lajos.katona@est.tech>
This commit is contained in:
lajoskatona
2025-09-10 16:44:33 +02:00
committed by Lajos Katona
parent c4bbd69944
commit 358bd706d1
4 changed files with 251 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
neutron_vpnaas_dashboard/api/vpn.py
+88 -93
View File
@@ -18,7 +18,7 @@ from horizon.utils.memoized import memoized
from openstack_dashboard.api import neutron
from openstack_dashboard.contrib.developer.profiler import api as profiler
neutronclient = neutron.neutronclient
networkclient = neutron.networkclient
class IKEPolicy(neutron.NeutronAPIDictWrapper):
@@ -52,18 +52,14 @@ def vpnservice_create(request, **kwargs):
:param router_id: router id for router of VPN service
:param subnet_id: subnet id for subnet of VPN service
"""
body = {'vpnservice':
{'admin_state_up': kwargs['admin_state_up'],
'name': kwargs['name'],
'description': kwargs['description'],
'router_id': kwargs['router_id']
}
}
body = {'admin_state_up': kwargs['admin_state_up'],
'name': kwargs['name'],
'description': kwargs['description'],
'router_id': kwargs['router_id']}
if kwargs.get('subnet_id'):
body['vpnservice']['subnet_id'] = kwargs['subnet_id']
vpnservice = neutronclient(request).create_vpnservice(body).get(
'vpnservice')
return VPNService(vpnservice)
body['subnet_id'] = kwargs['subnet_id']
vpnservice = networkclient(request).create_vpn_service(**body)
return VPNService(vpnservice.to_dict())
@profiler.trace
@@ -75,12 +71,12 @@ def vpnservice_list(request, **kwargs):
def _vpnservice_list(request, expand_subnet=False, expand_router=False,
expand_conns=False, **kwargs):
vpnservices = neutronclient(request).list_vpnservices(
**kwargs).get('vpnservices')
vpnservices = networkclient(request).vpn_services(**kwargs)
vpns_dicts = [vpn.to_dict() for vpn in vpnservices]
if expand_subnet:
subnets = neutron.subnet_list(request)
subnet_dict = OrderedDict((s.id, s) for s in subnets)
for s in vpnservices:
for s in vpns_dicts:
if s.get('subnet_id'):
s['subnet_name'] = subnet_dict.get(s['subnet_id']).cidr
if expand_router:
@@ -93,7 +89,7 @@ def _vpnservice_list(request, expand_subnet=False, expand_router=False,
for s in vpnservices:
s['ipsecsiteconns'] = [c.id for c in ipsecsiteconns
if c.vpnservice_id == s['id']]
return [VPNService(v) for v in vpnservices]
return [VPNService(v) for v in vpns_dicts]
@profiler.trace
@@ -104,8 +100,8 @@ def vpnservice_get(request, vpnservice_id):
def _vpnservice_get(request, vpnservice_id, expand_subnet=False,
expand_router=False, expand_conns=False):
vpnservice = neutronclient(request).show_vpnservice(vpnservice_id).get(
'vpnservice')
vpnservice = networkclient(request).get_vpn_service(
vpnservice_id).to_dict()
if expand_subnet and ('subnet_id' in vpnservice):
if vpnservice['subnet_id'] is not None:
vpnservice['subnet'] = neutron.subnet_get(
@@ -122,14 +118,14 @@ def _vpnservice_get(request, vpnservice_id, expand_subnet=False,
@profiler.trace
def vpnservice_update(request, vpnservice_id, **kwargs):
vpnservice = neutronclient(request).update_vpnservice(
vpnservice_id, kwargs).get('vpnservice')
vpnservice = networkclient(request).update_vpn_service(
vpnservice_id, **kwargs).to_dict()
return VPNService(vpnservice)
@profiler.trace
def vpnservice_delete(request, vpnservice_id):
neutronclient(request).delete_vpnservice(vpnservice_id)
networkclient(request).delete_vpn_service(vpnservice_id)
@profiler.trace
@@ -142,15 +138,13 @@ def endpointgroup_create(request, **kwargs):
:param type: type of Endpoint Group
:param endpoints: endpoint(s) of Endpoint Group
"""
body = {'endpoint_group':
{'name': kwargs['name'],
'description': kwargs['description'],
'type': kwargs['type'],
'endpoints': kwargs['endpoints']}
body = {'name': kwargs['name'],
'description': kwargs['description'],
'type': kwargs['type'],
'endpoints': kwargs['endpoints']
}
endpointgroup = neutronclient(request).create_endpoint_group(body).get(
'endpoint_group')
return EndpointGroup(endpointgroup)
endpointgroup = networkclient(request).create_vpn_endpoint_group(**body)
return EndpointGroup(endpointgroup.to_dict())
@profiler.trace
@@ -159,16 +153,16 @@ def endpointgroup_list(request, **kwargs):
def _endpointgroup_list(request, expand_conns=False, **kwargs):
endpointgroups = neutronclient(request).list_endpoint_groups(
**kwargs).get('endpoint_groups')
endpointgroups = networkclient(request).vpn_endpoint_groups(**kwargs)
vpn_epg_dicts = [epg.to_dict() for epg in endpointgroups]
if expand_conns:
ipsecsiteconns = _ipsecsiteconnection_list(request)
for g in endpointgroups:
for g in vpn_epg_dicts:
g['ipsecsiteconns'] = [
c.id for c in ipsecsiteconns
if (c.get('local_ep_group_id') == g['id'] or
c.get('peer_ep_group_id') == g['id'])]
return [EndpointGroup(v) for v in endpointgroups]
return [EndpointGroup(v) for v in vpn_epg_dicts]
@profiler.trace
@@ -177,8 +171,8 @@ def endpointgroup_get(request, endpoint_group_id):
def _endpointgroup_get(request, endpoint_group_id, expand_conns=False):
endpointgroup = neutronclient(request).show_endpoint_group(
endpoint_group_id).get('endpoint_group')
endpointgroup = networkclient(request).get_vpn_endpoint_group(
endpoint_group_id).to_dict()
if expand_conns:
ipsecsiteconns = _ipsecsiteconnection_list(request)
endpointgroup['ipsecsiteconns'] = [
@@ -190,14 +184,14 @@ def _endpointgroup_get(request, endpoint_group_id, expand_conns=False):
@profiler.trace
def endpointgroup_update(request, endpoint_group_id, **kwargs):
endpointgroup = neutronclient(request).update_endpoint_group(
endpoint_group_id, kwargs).get('endpoint_group')
endpointgroup = networkclient(request).update_vpn_endpoint_group(
endpoint_group_id, **kwargs).to_dict()
return EndpointGroup(endpointgroup)
@profiler.trace
def endpointgroup_delete(request, endpoint_group_id):
neutronclient(request).delete_endpoint_group(endpoint_group_id)
networkclient(request).delete_vpn_endpoint_group(endpoint_group_id)
@profiler.trace
@@ -214,18 +208,15 @@ def ikepolicy_create(request, **kwargs):
:param pfs: Perfect Forward Secrecy for IKE policy
:param phase1_negotiation_mode: IKE Phase1 negotiation mode for IKE policy
"""
body = {'ikepolicy':
{'name': kwargs['name'],
'description': kwargs['description'],
'auth_algorithm': kwargs['auth_algorithm'],
'encryption_algorithm': kwargs['encryption_algorithm'],
'ike_version': kwargs['ike_version'],
'lifetime': kwargs['lifetime'],
'pfs': kwargs['pfs'],
'phase1_negotiation_mode': kwargs['phase1_negotiation_mode']}
}
ikepolicy = neutronclient(request).create_ikepolicy(body).get(
'ikepolicy')
body = {'name': kwargs['name'],
'description': kwargs['description'],
'auth_algorithm': kwargs['auth_algorithm'],
'encryption_algorithm': kwargs['encryption_algorithm'],
'ike_version': kwargs['ike_version'],
'lifetime': kwargs['lifetime'],
'pfs': kwargs['pfs'],
'phase1_negotiation_mode': kwargs['phase1_negotiation_mode']}
ikepolicy = networkclient(request).create_vpn_ike_policy(**body).to_dict()
return IKEPolicy(ikepolicy)
@@ -235,14 +226,15 @@ def ikepolicy_list(request, **kwargs):
def _ikepolicy_list(request, expand_conns=False, **kwargs):
ikepolicies = neutronclient(request).list_ikepolicies(
**kwargs).get('ikepolicies')
ikepolicies = networkclient(request).vpn_ike_policies(
**kwargs)
vpn_ikep_dicts = [ikep.to_dict() for ikep in ikepolicies]
if expand_conns:
ipsecsiteconns = _ipsecsiteconnection_list(request, **kwargs)
for p in ikepolicies:
for p in vpn_ikep_dicts:
p['ipsecsiteconns'] = [c.id for c in ipsecsiteconns
if c.ikepolicy_id == p['id']]
return [IKEPolicy(v) for v in ikepolicies]
return [IKEPolicy(v) for v in vpn_ikep_dicts]
@profiler.trace
@@ -251,8 +243,8 @@ def ikepolicy_get(request, ikepolicy_id):
def _ikepolicy_get(request, ikepolicy_id, expand_conns=False):
ikepolicy = neutronclient(request).show_ikepolicy(
ikepolicy_id).get('ikepolicy')
ikepolicy = networkclient(request).get_vpn_ike_policy(
ikepolicy_id).to_dict()
if expand_conns:
ipsecsiteconns = _ipsecsiteconnection_list(request)
ikepolicy['ipsecsiteconns'] = [c for c in ipsecsiteconns
@@ -262,14 +254,14 @@ def _ikepolicy_get(request, ikepolicy_id, expand_conns=False):
@profiler.trace
def ikepolicy_update(request, ikepolicy_id, **kwargs):
ikepolicy = neutronclient(request).update_ikepolicy(
ikepolicy_id, kwargs).get('ikepolicy')
ikepolicy = networkclient(request).update_vpn_ike_policy(
ikepolicy_id, **kwargs).to_dict()
return IKEPolicy(ikepolicy)
@profiler.trace
def ikepolicy_delete(request, ikepolicy_id):
neutronclient(request).delete_ikepolicy(ikepolicy_id)
networkclient(request).delete_vpn_ike_policy(ikepolicy_id)
@profiler.trace
@@ -286,19 +278,17 @@ def ipsecpolicy_create(request, **kwargs):
:param pfs: Perfect Forward Secrecy for IPsec policy
:param transform_protocol: Transform Protocol for IPsec policy
"""
body = {'ipsecpolicy':
{'name': kwargs['name'],
'description': kwargs['description'],
'auth_algorithm': kwargs['auth_algorithm'],
'encapsulation_mode': kwargs['encapsulation_mode'],
'encryption_algorithm': kwargs['encryption_algorithm'],
'lifetime': kwargs['lifetime'],
'pfs': kwargs['pfs'],
'transform_protocol': kwargs['transform_protocol']}
body = {'name': kwargs['name'],
'description': kwargs['description'],
'auth_algorithm': kwargs['auth_algorithm'],
'encapsulation_mode': kwargs['encapsulation_mode'],
'encryption_algorithm': kwargs['encryption_algorithm'],
'lifetime': kwargs['lifetime'],
'pfs': kwargs['pfs'],
'transform_protocol': kwargs['transform_protocol']
}
ipsecpolicy = neutronclient(request).create_ipsecpolicy(body).get(
'ipsecpolicy')
return IPsecPolicy(ipsecpolicy)
ipsecpolicy = networkclient(request).create_vpn_ipsec_policy(**body)
return IPsecPolicy(ipsecpolicy.to_dict())
@profiler.trace
@@ -307,14 +297,15 @@ def ipsecpolicy_list(request, **kwargs):
def _ipsecpolicy_list(request, expand_conns=False, **kwargs):
ipsecpolicies = neutronclient(request).list_ipsecpolicies(
**kwargs).get('ipsecpolicies')
ipsecpolicies = networkclient(request).vpn_ipsec_policies(
**kwargs)
ipsec_dicts = [ipsec.to_dict() for ipsec in ipsecpolicies]
if expand_conns:
ipsecsiteconns = _ipsecsiteconnection_list(request, **kwargs)
for p in ipsecpolicies:
for p in ipsec_dicts:
p['ipsecsiteconns'] = [c.id for c in ipsecsiteconns
if c.ipsecpolicy_id == p['id']]
return [IPsecPolicy(v) for v in ipsecpolicies]
return [IPsecPolicy(v) for v in ipsec_dicts]
@profiler.trace
@@ -323,8 +314,8 @@ def ipsecpolicy_get(request, ipsecpolicy_id):
def _ipsecpolicy_get(request, ipsecpolicy_id, expand_conns=False):
ipsecpolicy = neutronclient(request).show_ipsecpolicy(
ipsecpolicy_id).get('ipsecpolicy')
ipsecpolicy = networkclient(request).get_vpn_ipsec_policy(
ipsecpolicy_id).to_dict()
if expand_conns:
ipsecsiteconns = _ipsecsiteconnection_list(request)
ipsecpolicy['ipsecsiteconns'] = [c for c in ipsecsiteconns
@@ -335,14 +326,14 @@ def _ipsecpolicy_get(request, ipsecpolicy_id, expand_conns=False):
@profiler.trace
def ipsecpolicy_update(request, ipsecpolicy_id, **kwargs):
ipsecpolicy = neutronclient(request).update_ipsecpolicy(
ipsecpolicy_id, kwargs).get('ipsecpolicy')
ipsecpolicy = networkclient(request).update_vpn_ipsec_policy(
ipsecpolicy_id, **kwargs).to_dict()
return IPsecPolicy(ipsecpolicy)
@profiler.trace
def ipsecpolicy_delete(request, ipsecpolicy_id):
neutronclient(request).delete_ipsecpolicy(ipsecpolicy_id)
networkclient(request).delete_vpn_ipsec_policy(ipsecpolicy_id)
@profiler.trace
@@ -384,8 +375,8 @@ def ipsecsiteconnection_create(request, **kwargs):
body['peer_ep_group_id'] = kwargs['peer_ep_group_id']
else:
body['peer_cidrs'] = kwargs['peer_cidrs']
ipsecsiteconnection = neutronclient(request).create_ipsec_site_connection(
{'ipsec_site_connection': body}).get('ipsec_site_connection')
ipsecsiteconnection = networkclient(
request).create_vpn_ipsec_site_connection(**body).to_dict()
return IPsecSiteConnection(ipsecsiteconnection)
@@ -401,26 +392,27 @@ def ipsecsiteconnection_list(request, **kwargs):
def _ipsecsiteconnection_list(request, expand_ikepolicies=False,
expand_ipsecpolicies=False,
expand_vpnservices=False, **kwargs):
ipsecsiteconnections = neutronclient(request).list_ipsec_site_connections(
**kwargs).get('ipsec_site_connections')
ipsecsiteconnections = networkclient(request).vpn_ipsec_site_connections(
**kwargs)
ipsec_dicts = [ipsec.to_dict() for ipsec in ipsecsiteconnections]
if expand_ikepolicies:
ikepolicies = _ikepolicy_list(request)
policy_dict = OrderedDict((p.id, p) for p in ikepolicies)
for c in ipsecsiteconnections:
for c in ipsec_dicts:
c['ikepolicy_name'] = policy_dict.get(c['ikepolicy_id']).name_or_id
if expand_ipsecpolicies:
ipsecpolicies = _ipsecpolicy_list(request)
policy_dict = OrderedDict((p.id, p) for p in ipsecpolicies)
for c in ipsecsiteconnections:
for c in ipsec_dicts:
c['ipsecpolicy_name'] = policy_dict.get(c['ipsecpolicy_id']
).name_or_id
if expand_vpnservices:
vpnservices = _vpnservice_list(request)
service_dict = OrderedDict((s.id, s) for s in vpnservices)
for c in ipsecsiteconnections:
for c in ipsec_dicts:
c['vpnservice_name'] = service_dict.get(c['vpnservice_id']
).name_or_id
return [IPsecSiteConnection(v) for v in ipsecsiteconnections]
return [IPsecSiteConnection(v) for v in ipsec_dicts]
@profiler.trace
@@ -434,8 +426,9 @@ def ipsecsiteconnection_get(request, ipsecsiteconnection_id):
def _ipsecsiteconnection_get(request, ipsecsiteconnection_id,
expand_ikepolicies, expand_ipsecpolicies,
expand_vpnservices):
ipsecsiteconnection = neutronclient(request).show_ipsec_site_connection(
ipsecsiteconnection_id).get('ipsec_site_connection')
ipsecsiteconnection = networkclient(
request).get_vpn_ipsec_site_connection(
ipsecsiteconnection_id).to_dict()
if expand_ikepolicies:
ipsecsiteconnection['ikepolicy'] = _ikepolicy_get(
request, ipsecsiteconnection['ikepolicy_id'])
@@ -450,11 +443,13 @@ def _ipsecsiteconnection_get(request, ipsecsiteconnection_id,
@profiler.trace
def ipsecsiteconnection_update(request, ipsecsiteconnection_id, **kwargs):
ipsecsiteconnection = neutronclient(request).update_ipsec_site_connection(
ipsecsiteconnection_id, kwargs).get('ipsec_site_connection')
ipsecsiteconnection = networkclient(
request).update_vpn_ipsec_site_connection(
ipsecsiteconnection_id, **kwargs).to_dict()
return IPsecSiteConnection(ipsecsiteconnection)
@profiler.trace
def ipsecsiteconnection_delete(request, ipsecsiteconnection_id):
neutronclient(request).delete_ipsec_site_connection(ipsecsiteconnection_id)
networkclient(request).delete_vpn_ipsec_site_connection(
ipsecsiteconnection_id)
neutron_vpnaas_dashboard/test/api_tests/test_vpnaas.py
+134 -167
View File
@@ -10,21 +10,19 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutronclient.v2_0 import client
from openstack.network.v2._proxy import Proxy as networkclient
from openstack_dashboard import api
from openstack_dashboard.test import helpers
from neutron_vpnaas_dashboard.api import vpn as api_vpn
from neutron_vpnaas_dashboard.test import helpers as test
neutronclient = client.Client
class VPNaasApiTests(test.APITestCase):
@helpers.create_mocks({neutronclient: ('create_vpnservice',)})
@helpers.create_mocks({networkclient: ('create_vpn_service',)})
def test_vpnservice_create(self):
vpnservice1 = self.api_vpnservices.first()
vpnservice1 = self.api_vpnservices[0]
form_data = {
'name': vpnservice1['name'],
'description': vpnservice1['description'],
@@ -33,70 +31,63 @@ class VPNaasApiTests(test.APITestCase):
'admin_state_up': vpnservice1['admin_state_up']
}
vpnservice = {'vpnservice': self.api_vpnservices.first()}
self.mock_create_vpnservice.return_value = vpnservice
vpnservice = self.api_vpnservices[0]
self.mock_create_vpn_service.return_value = vpnservice
ret_val = api_vpn.vpnservice_create(self.request, **form_data)
self.assertIsInstance(ret_val, api_vpn.VPNService)
self.mock_create_vpnservice.assert_called_once_with(
{'vpnservice': form_data})
self.mock_create_vpn_service.assert_called_once_with(**form_data)
@helpers.create_mocks({neutronclient: ('list_vpnservices',
'list_ipsec_site_connections'),
@helpers.create_mocks({networkclient: ('vpn_services',
'vpn_ipsec_site_connections'),
api.neutron: ('subnet_list', 'router_list')})
def test_vpnservice_list(self):
vpnservices = {'vpnservices': self.vpnservices.list()}
vpnservices_dict = {'vpnservices': self.api_vpnservices.list()}
vpnservices = self.api_vpnservices
subnets = self.subnets.list()
routers = self.routers.list()
ipsecsiteconnections_dict = {
'ipsec_site_connections': self.api_ipsecsiteconnections.list()}
self.mock_list_vpnservices.return_value = vpnservices_dict
self.mock_vpn_services.return_value = vpnservices
self.mock_subnet_list.return_value = subnets
self.mock_router_list.return_value = routers
self.mock_list_ipsec_site_connections.return_value = \
ipsecsiteconnections_dict
self.mock_vpn_ipsec_site_connections.return_value = (
self.api_ipsecsiteconnections)
ret_val = api_vpn.vpnservice_list(self.request)
for (v, d) in zip(ret_val, vpnservices['vpnservices']):
for (v, d) in zip(ret_val, self.vpnservices.list()):
self.assertIsInstance(v, api_vpn.VPNService)
self.assertTrue(v.name, d.name)
self.assertTrue(v.id)
self.mock_list_vpnservices.assert_called_once_with()
self.mock_vpn_services.assert_called_once_with()
self.mock_subnet_list.assert_called_once_with(self.request)
self.mock_router_list.assert_called_once_with(self.request)
self.mock_list_ipsec_site_connections.assert_called_once_with()
self.mock_vpn_ipsec_site_connections.assert_called_once_with()
@helpers.create_mocks({neutronclient: ('show_vpnservice',
'list_ipsec_site_connections'),
@helpers.create_mocks({networkclient: ('get_vpn_service',
'vpn_ipsec_site_connections'),
api.neutron: ('subnet_get', 'router_get')})
def test_vpnservice_get(self):
vpnservice = self.vpnservices.first()
vpnservice_dict = {'vpnservice': self.api_vpnservices.first()}
vpnservice = self.api_vpnservices[0]
subnet = self.subnets.first()
router = self.routers.first()
ipsecsiteconnections_dict = {
'ipsec_site_connections': self.api_ipsecsiteconnections.list()}
self.mock_show_vpnservice.return_value = vpnservice_dict
self.mock_get_vpn_service.return_value = vpnservice
self.mock_subnet_get.return_value = subnet
self.mock_router_get.return_value = router
self.mock_list_ipsec_site_connections.return_value = \
ipsecsiteconnections_dict
self.mock_vpn_ipsec_site_connections.return_value = (
self.api_ipsecsiteconnections)
ret_val = api_vpn.vpnservice_get(self.request, vpnservice.id)
self.assertIsInstance(ret_val, api_vpn.VPNService)
self.mock_show_vpnservice.assert_called_once_with(vpnservice.id)
self.mock_get_vpn_service.assert_called_once_with(vpnservice.id)
self.mock_subnet_get.assert_called_once_with(self.request, subnet.id)
self.mock_router_get.assert_called_once_with(self.request, router.id)
self.mock_list_ipsec_site_connections.assert_called_once_with()
self.mock_vpn_ipsec_site_connections.assert_called_once_with()
@helpers.create_mocks({neutronclient: ('create_endpoint_group',)})
@helpers.create_mocks({networkclient: ('create_vpn_endpoint_group',)})
def test_endpointgroup_create(self):
endpointgroup = self.api_endpointgroups.first()
endpointgroup = self.api_endpointgroups[0].to_dict()
form_data = {
'name': endpointgroup['name'],
'description': endpointgroup['description'],
@@ -104,57 +95,49 @@ class VPNaasApiTests(test.APITestCase):
'endpoints': endpointgroup['endpoints']
}
endpoint_group = {'endpoint_group': self.api_endpointgroups.first()}
self.mock_create_endpoint_group.return_value = endpoint_group
self.mock_create_vpn_endpoint_group.return_value = (
self.api_endpointgroups[0])
ret_val = api_vpn.endpointgroup_create(self.request, **form_data)
self.assertIsInstance(ret_val, api_vpn.EndpointGroup)
self.mock_create_endpoint_group.assert_called_once_with(
{'endpoint_group': form_data})
self.mock_create_vpn_endpoint_group.assert_called_once_with(
**form_data)
@helpers.create_mocks({neutronclient: ('list_endpoint_groups',
'list_ipsec_site_connections')})
@helpers.create_mocks({networkclient: ('vpn_endpoint_groups',
'vpn_ipsec_site_connections')})
def test_endpointgroup_list(self):
endpointgroups = {'endpoint_groups': self.endpointgroups.list()}
endpointgroups_dict = {
'endpoint_groups': self.api_endpointgroups.list()}
ipsecsiteconnections_dict = {
'ipsec_site_connections': self.api_ipsecsiteconnections.list()}
endpointgroups = self.api_endpointgroups
self.mock_list_endpoint_groups.return_value = endpointgroups_dict
self.mock_list_ipsec_site_connections.return_value = \
ipsecsiteconnections_dict
self.mock_vpn_endpoint_groups.return_value = endpointgroups
self.mock_vpn_ipsec_site_connections.return_value = \
self.api_ipsecsiteconnections
ret_val = api_vpn.endpointgroup_list(self.request)
for (v, d) in zip(ret_val, endpointgroups['endpoint_groups']):
for (v, d) in zip(ret_val, self.endpointgroups.list()):
self.assertIsInstance(v, api_vpn.EndpointGroup)
self.assertTrue(v.name, d.name)
self.assertTrue(v.id)
self.mock_list_endpoint_groups.assert_called_once_with()
self.mock_list_ipsec_site_connections.assert_called_once_with()
self.mock_vpn_endpoint_groups.assert_called_once_with()
self.mock_vpn_ipsec_site_connections.assert_called_once_with()
@helpers.create_mocks({neutronclient: ('show_endpoint_group',
'list_ipsec_site_connections')})
@helpers.create_mocks({networkclient: ('get_vpn_endpoint_group',
'vpn_ipsec_site_connections')})
def test_endpointgroup_get(self):
endpoint_group = self.endpointgroups.first()
endpoint_group_dict = {
'endpoint_group': self.api_endpointgroups.first()}
ipsecsiteconnections_dict = {
'ipsec_site_connections': self.api_ipsecsiteconnections.list()}
endpoint_group = self.api_endpointgroups[0]
self.mock_show_endpoint_group.return_value = endpoint_group_dict
self.mock_list_ipsec_site_connections.return_value = \
ipsecsiteconnections_dict
self.mock_get_vpn_endpoint_group.return_value = endpoint_group
self.mock_vpn_ipsec_site_connections.return_value = \
self.api_ipsecsiteconnections
ret_val = api_vpn.endpointgroup_get(self.request, endpoint_group.id)
self.assertIsInstance(ret_val, api_vpn.EndpointGroup)
self.mock_show_endpoint_group.assert_called_once_with(
self.mock_get_vpn_endpoint_group.assert_called_once_with(
endpoint_group.id)
self.mock_list_ipsec_site_connections.assert_called_once_with()
self.mock_vpn_ipsec_site_connections.assert_called_once_with()
@helpers.create_mocks({neutronclient: ('create_ikepolicy',)})
@helpers.create_mocks({networkclient: ('create_vpn_ike_policy',)})
def test_ikepolicy_create(self):
ikepolicy1 = self.api_ikepolicies.first()
ikepolicy1 = self.api_ikepolicies[0].to_dict()
form_data = {
'name': ikepolicy1['name'],
'description': ikepolicy1['description'],
@@ -166,54 +149,49 @@ class VPNaasApiTests(test.APITestCase):
'pfs': ikepolicy1['pfs']
}
ikepolicy = {'ikepolicy': self.api_ikepolicies.first()}
self.mock_create_ikepolicy.return_value = ikepolicy
ikepolicy = self.api_ikepolicies[0]
self.mock_create_vpn_ike_policy.return_value = ikepolicy
ret_val = api_vpn.ikepolicy_create(self.request, **form_data)
self.assertIsInstance(ret_val, api_vpn.IKEPolicy)
self.mock_create_ikepolicy.assert_called_once_with(
{'ikepolicy': form_data})
self.mock_create_vpn_ike_policy.assert_called_once_with(**form_data)
@helpers.create_mocks({neutronclient: ('list_ikepolicies',
'list_ipsec_site_connections')})
@helpers.create_mocks({networkclient: ('vpn_ike_policies',
'vpn_ipsec_site_connections')})
def test_ikepolicy_list(self):
ikepolicies = {'ikepolicies': self.ikepolicies.list()}
ikepolicies_dict = {'ikepolicies': self.api_ikepolicies.list()}
ipsecsiteconnections_dict = {
'ipsec_site_connections': self.api_ipsecsiteconnections.list()}
ikepolicies = self.api_ikepolicies
ipsecsiteconnections = self.api_ipsecsiteconnections
self.mock_list_ikepolicies.return_value = ikepolicies_dict
self.mock_list_ipsec_site_connections.return_value = \
ipsecsiteconnections_dict
self.mock_vpn_ike_policies.return_value = ikepolicies
self.mock_vpn_ipsec_site_connections.return_value = \
ipsecsiteconnections
ret_val = api_vpn.ikepolicy_list(self.request)
for (v, d) in zip(ret_val, ikepolicies['ikepolicies']):
for (v, d) in zip(ret_val, self.ikepolicies.list()):
self.assertIsInstance(v, api_vpn.IKEPolicy)
self.assertTrue(v.name, d.name)
self.assertTrue(v.id)
self.mock_list_ikepolicies.assert_called_once_with()
self.mock_list_ipsec_site_connections.assert_called_once_with()
self.mock_vpn_ike_policies.assert_called_once_with()
self.mock_vpn_ipsec_site_connections.assert_called_once_with()
@helpers.create_mocks({neutronclient: ('show_ikepolicy',
'list_ipsec_site_connections')})
@helpers.create_mocks({networkclient: ('get_vpn_ike_policy',
'vpn_ipsec_site_connections')})
def test_ikepolicy_get(self):
ikepolicy = self.ikepolicies.first()
ikepolicy_dict = {'ikepolicy': self.api_ikepolicies.first()}
ipsecsiteconnections_dict = {
'ipsec_site_connections': self.api_ipsecsiteconnections.list()}
ikepolicy = self.api_ikepolicies[0]
ipsecsiteconnections = self.api_ipsecsiteconnections
self.mock_show_ikepolicy.return_value = ikepolicy_dict
self.mock_list_ipsec_site_connections.return_value = \
ipsecsiteconnections_dict
self.mock_get_vpn_ike_policy.return_value = ikepolicy
self.mock_vpn_ipsec_site_connections.return_value = \
ipsecsiteconnections
ret_val = api_vpn.ikepolicy_get(self.request, ikepolicy.id)
self.assertIsInstance(ret_val, api_vpn.IKEPolicy)
self.mock_show_ikepolicy.assert_called_once_with(ikepolicy.id)
self.mock_list_ipsec_site_connections.assert_called_once_with()
self.mock_get_vpn_ike_policy.assert_called_once_with(ikepolicy.id)
self.mock_vpn_ipsec_site_connections.assert_called_once_with()
@helpers.create_mocks({neutronclient: ('create_ipsecpolicy',)})
@helpers.create_mocks({networkclient: ('create_vpn_ipsec_policy',)})
def test_ipsecpolicy_create(self):
ipsecpolicy1 = self.api_ipsecpolicies.first()
ipsecpolicy1 = self.ipsecpolicies.first()
form_data = {
'name': ipsecpolicy1['name'],
'description': ipsecpolicy1['description'],
@@ -225,54 +203,49 @@ class VPNaasApiTests(test.APITestCase):
'transform_protocol': ipsecpolicy1['transform_protocol']
}
ipsecpolicy = {'ipsecpolicy': self.api_ipsecpolicies.first()}
self.mock_create_ipsecpolicy.return_value = ipsecpolicy
ipsecpolicy = self.api_ipsecpolicies[0]
self.mock_create_vpn_ipsec_policy.return_value = ipsecpolicy
ret_val = api_vpn.ipsecpolicy_create(self.request, **form_data)
self.assertIsInstance(ret_val, api_vpn.IPsecPolicy)
self.mock_create_ipsecpolicy.assert_called_once_with(
{'ipsecpolicy': form_data})
self.mock_create_vpn_ipsec_policy.assert_called_once_with(**form_data)
@helpers.create_mocks({neutronclient: ('list_ipsecpolicies',
'list_ipsec_site_connections')})
@helpers.create_mocks({networkclient: ('vpn_ipsec_policies',
'vpn_ipsec_site_connections')})
def test_ipsecpolicy_list(self):
ipsecpolicies = {'ipsecpolicies': self.ipsecpolicies.list()}
ipsecpolicies_dict = {'ipsecpolicies': self.api_ipsecpolicies.list()}
ipsecsiteconnections_dict = {
'ipsec_site_connections': self.api_ipsecsiteconnections.list()}
ipsecpolicies = self.api_ipsecpolicies
self.mock_list_ipsecpolicies.return_value = ipsecpolicies_dict
self.mock_list_ipsec_site_connections.return_value = \
ipsecsiteconnections_dict
self.mock_vpn_ipsec_policies.return_value = ipsecpolicies
self.mock_vpn_ipsec_site_connections.return_value = \
self.api_ipsecsiteconnections
ret_val = api_vpn.ipsecpolicy_list(self.request)
for (v, d) in zip(ret_val, ipsecpolicies['ipsecpolicies']):
for (v, d) in zip(ret_val, self.ipsecpolicies.list()):
self.assertIsInstance(v, api_vpn.IPsecPolicy)
self.assertTrue(v.name, d.name)
self.assertTrue(v.id)
self.mock_list_ipsecpolicies.assert_called_once_with()
self.mock_list_ipsec_site_connections.assert_called_once_with()
self.mock_vpn_ipsec_policies.assert_called_once_with()
self.mock_vpn_ipsec_site_connections.assert_called_once_with()
@helpers.create_mocks({neutronclient: ('show_ipsecpolicy',
'list_ipsec_site_connections')})
@helpers.create_mocks({networkclient: ('get_vpn_ipsec_policy',
'vpn_ipsec_site_connections')})
def test_ipsecpolicy_get(self):
ipsecpolicy = self.ipsecpolicies.first()
ipsecpolicy_dict = {'ipsecpolicy': self.api_ipsecpolicies.first()}
ipsecsiteconnections_dict = {
'ipsec_site_connections': self.api_ipsecsiteconnections.list()}
ipsecpolicy = self.api_ipsecpolicies[0]
ipsecsiteconnections = self.api_ipsecsiteconnections
self.mock_show_ipsecpolicy.return_value = ipsecpolicy_dict
self.mock_list_ipsec_site_connections.return_value = \
ipsecsiteconnections_dict
self.mock_get_vpn_ipsec_policy.return_value = ipsecpolicy
self.mock_vpn_ipsec_site_connections.return_value = \
ipsecsiteconnections
ret_val = api_vpn.ipsecpolicy_get(self.request, ipsecpolicy.id)
self.assertIsInstance(ret_val, api_vpn.IPsecPolicy)
self.mock_show_ipsecpolicy.assert_called_once_with(ipsecpolicy.id)
self.mock_list_ipsec_site_connections.assert_called_once_with()
self.mock_get_vpn_ipsec_policy.assert_called_once_with(ipsecpolicy.id)
self.mock_vpn_ipsec_site_connections.assert_called_once_with()
@helpers.create_mocks({neutronclient: ('create_ipsec_site_connection',)})
@helpers.create_mocks({networkclient:
('create_vpn_ipsec_site_connection',)})
def test_ipsecsiteconnection_create(self):
ipsecsiteconnection1 = self.api_ipsecsiteconnections.first()
ipsecsiteconnection1 = self.ipsecsiteconnections.first()
form_data = {
'name': ipsecsiteconnection1['name'],
'description': ipsecsiteconnection1['description'],
@@ -289,73 +262,67 @@ class VPNaasApiTests(test.APITestCase):
'admin_state_up': ipsecsiteconnection1['admin_state_up']
}
ipsecsiteconnection = {'ipsec_site_connection':
self.api_ipsecsiteconnections.first()}
self.mock_create_ipsec_site_connection.return_value = \
ipsecsiteconnection = self.api_ipsecsiteconnections[0]
self.mock_create_vpn_ipsec_site_connection.return_value = \
ipsecsiteconnection
ret_val = api_vpn.ipsecsiteconnection_create(
self.request, **form_data)
self.assertIsInstance(ret_val, api_vpn.IPsecSiteConnection)
self.mock_create_ipsec_site_connection.assert_called_once_with(
{'ipsec_site_connection': form_data})
self.mock_create_vpn_ipsec_site_connection.assert_called_once_with(
**form_data)
@helpers.create_mocks({neutronclient: ('list_ipsec_site_connections',
'list_ikepolicies',
'list_ipsecpolicies',
'list_vpnservices')})
@helpers.create_mocks({networkclient: ('vpn_ipsec_site_connections',
'vpn_ike_policies',
'vpn_ipsec_policies',
'vpn_services')})
def test_ipsecsiteconnection_list(self):
ipsecsiteconnections = {
'ipsec_site_connections': self.ipsecsiteconnections.list()}
ipsecsiteconnections_dict = {
'ipsec_site_connections': self.api_ipsecsiteconnections.list()}
ikepolicies_dict = {'ikepolicies': self.api_ikepolicies.list()}
ipsecpolicies_dict = {'ipsecpolicies': self.api_ipsecpolicies.list()}
vpnservices_dict = {'vpnservices': self.api_vpnservices.list()}
ipsecsiteconnections = self.api_ipsecsiteconnections
ikepolicies = self.api_ikepolicies
ipsecpolicies = self.api_ipsecpolicies
vpnservices = self.api_vpnservices
self.mock_list_ipsec_site_connections.return_value = \
ipsecsiteconnections_dict
self.mock_list_ikepolicies.return_value = ikepolicies_dict
self.mock_list_ipsecpolicies.return_value = ipsecpolicies_dict
self.mock_list_vpnservices.return_value = vpnservices_dict
self.mock_vpn_ipsec_site_connections.return_value = \
ipsecsiteconnections
self.mock_vpn_ike_policies.return_value = ikepolicies
self.mock_vpn_ipsec_policies.return_value = ipsecpolicies
self.mock_vpn_services.return_value = vpnservices
ret_val = api_vpn.ipsecsiteconnection_list(self.request)
for (v, d) in zip(ret_val,
ipsecsiteconnections['ipsec_site_connections']):
for (v, d) in zip(ret_val, self.ipsecsiteconnections.list()):
self.assertIsInstance(v, api_vpn.IPsecSiteConnection)
self.assertTrue(v.name, d.name)
self.assertTrue(v.id)
self.mock_list_ipsec_site_connections.assert_called_once_with()
self.mock_list_ikepolicies.assert_called_once_with()
self.mock_list_ipsecpolicies.assert_called_once_with()
self.mock_list_vpnservices.assert_called_once_with()
self.mock_vpn_ipsec_site_connections.assert_called_once_with()
self.mock_vpn_ike_policies.assert_called_once_with()
self.mock_vpn_ipsec_policies.assert_called_once_with()
self.mock_vpn_services.assert_called_once_with()
@helpers.create_mocks({neutronclient: ('show_ipsec_site_connection',
'show_ikepolicy',
'show_ipsecpolicy',
'show_vpnservice')})
@helpers.create_mocks({networkclient: ('get_vpn_ipsec_site_connection',
'get_vpn_ike_policy',
'get_vpn_ipsec_policy',
'get_vpn_service')})
def test_ipsecsiteconnection_get(self):
ipsecsiteconnection = self.ipsecsiteconnections.first()
connection_dict = {'ipsec_site_connection':
self.api_ipsecsiteconnections.first()}
ikepolicy_dict = {'ikepolicy': self.api_ikepolicies.first()}
ipsecpolicy_dict = {'ipsecpolicy': self.api_ipsecpolicies.first()}
vpnservice_dict = {'vpnservice': self.api_vpnservices.first()}
connection = self.api_ipsecsiteconnections[0]
ikepolicy = self.api_ikepolicies[0]
ipsecpolicy = self.api_ipsecpolicies[0]
vpnservice = self.api_vpnservices[0]
self.mock_show_ipsec_site_connection.return_value = connection_dict
self.mock_show_ikepolicy.return_value = ikepolicy_dict
self.mock_show_ipsecpolicy.return_value = ipsecpolicy_dict
self.mock_show_vpnservice.return_value = vpnservice_dict
self.mock_get_vpn_ipsec_site_connection.return_value = connection
self.mock_get_vpn_ike_policy.return_value = ikepolicy
self.mock_get_vpn_ipsec_policy.return_value = ipsecpolicy
self.mock_get_vpn_service.return_value = vpnservice
ret_val = api_vpn.ipsecsiteconnection_get(self.request,
ipsecsiteconnection.id)
self.assertIsInstance(ret_val, api_vpn.IPsecSiteConnection)
self.mock_show_ipsec_site_connection.assert_called_once_with(
self.mock_get_vpn_ipsec_site_connection.assert_called_once_with(
ipsecsiteconnection.id)
self.mock_show_ikepolicy.assert_called_once_with(
self.mock_get_vpn_ike_policy.assert_called_once_with(
ipsecsiteconnection.ikepolicy_id)
self.mock_show_ipsecpolicy.assert_called_once_with(
self.mock_get_vpn_ipsec_policy.assert_called_once_with(
ipsecsiteconnection.ipsecpolicy_id)
self.mock_show_vpnservice.assert_called_once_with(
self.mock_get_vpn_service.assert_called_once_with(
ipsecsiteconnection.vpnservice_id)
neutron_vpnaas_dashboard/test/test_data/vpnaas_data.py
+28 -16
View File
@@ -12,6 +12,11 @@
# License for the specific language governing permissions and limitations
# under the License.
from openstack.network.v2 import vpn_endpoint_group
from openstack.network.v2 import vpn_ike_policy
from openstack.network.v2 import vpn_ipsec_policy
from openstack.network.v2 import vpn_ipsec_site_connection
from openstack.network.v2 import vpn_service
from openstack_dashboard.test.test_data import utils
from neutron_vpnaas_dashboard.api import vpn
@@ -25,12 +30,12 @@ def data(TEST):
TEST.ipsecsiteconnections = utils.TestDataContainer()
TEST.endpointgroups = utils.TestDataContainer()
# Data return by neutronclient.
TEST.api_vpnservices = utils.TestDataContainer()
TEST.api_ikepolicies = utils.TestDataContainer()
TEST.api_ipsecpolicies = utils.TestDataContainer()
TEST.api_ipsecsiteconnections = utils.TestDataContainer()
TEST.api_endpointgroups = utils.TestDataContainer()
# Data return by OpenstackSDK.
TEST.api_vpnservices = list()
TEST.api_ikepolicies = list()
TEST.api_ipsecpolicies = list()
TEST.api_ipsecsiteconnections = list()
TEST.api_endpointgroups = list()
# 1st VPN service.
vpnservice_dict = {'id': '09a26949-6231-4f72-942a-0c8c0ddd4d61',
@@ -45,7 +50,7 @@ def data(TEST):
'status': 'Active',
'ipsecsiteconns': TEST.ipsecsiteconnections.list()
}
TEST.api_vpnservices.add(vpnservice_dict)
TEST.api_vpnservices.append(vpn_service.VpnService(**vpnservice_dict))
TEST.vpnservices.add(vpn.VPNService(vpnservice_dict))
# 2nd VPN service.
@@ -63,7 +68,7 @@ def data(TEST):
'external_v4_ip': '10.0.0.0/24',
'external_v6_ip': 'fd4c:a535:831c::/64'
}
TEST.api_vpnservices.add(vpnservice_dict)
TEST.api_vpnservices.append(vpn_service.VpnService(**vpnservice_dict))
TEST.vpnservices.add(vpn.VPNService(vpnservice_dict))
# 1st Endpoint Group
@@ -74,7 +79,8 @@ def data(TEST):
'type': 'subnet',
'endpoints': [TEST.subnets.first().id]
}
TEST.api_endpointgroups.add(endpointgroup_dict)
TEST.api_endpointgroups.append(vpn_endpoint_group.VpnEndpointGroup(
**endpointgroup_dict))
TEST.endpointgroups.add(vpn.EndpointGroup(endpointgroup_dict))
# 1st IKE policy
@@ -89,7 +95,7 @@ def data(TEST):
'phase1_negotiation_mode': 'main',
'pfs': 'group5',
'ipsecsiteconns': TEST.ipsecsiteconnections.list()}
TEST.api_ikepolicies.add(ikepolicy_dict)
TEST.api_ikepolicies.append(vpn_ike_policy.VpnIkePolicy(**ikepolicy_dict))
TEST.ikepolicies.add(vpn.IKEPolicy(ikepolicy_dict))
# 2nd IKE policy
@@ -104,7 +110,7 @@ def data(TEST):
'phase1_negotiation_mode': 'aggressive',
'pfs': 'group5',
'ipsecsiteconns': []}
TEST.api_ikepolicies.add(ikepolicy_dict)
TEST.api_ikepolicies.append(vpn_ike_policy.VpnIkePolicy(**ikepolicy_dict))
TEST.ikepolicies.add(vpn.IKEPolicy(ikepolicy_dict))
# 3rd IKE policy
@@ -119,7 +125,7 @@ def data(TEST):
'phase1_negotiation_mode': 'main',
'pfs': 'group5',
'ipsecsiteconns': []}
TEST.api_ikepolicies.add(ikepolicy_dict)
TEST.api_ikepolicies.append(vpn_ike_policy.VpnIkePolicy(**ikepolicy_dict))
TEST.ikepolicies.add(vpn.IKEPolicy(ikepolicy_dict))
# 1st IPsec policy
@@ -134,7 +140,8 @@ def data(TEST):
'pfs': 'group5',
'transform_protocol': 'esp',
'ipsecsiteconns': TEST.ipsecsiteconnections.list()}
TEST.api_ipsecpolicies.add(ipsecpolicy_dict)
TEST.api_ipsecpolicies.append(vpn_ipsec_policy.VpnIpsecPolicy(
**ipsecpolicy_dict))
TEST.ipsecpolicies.add(vpn.IPsecPolicy(ipsecpolicy_dict))
# 2nd IPsec policy
@@ -149,7 +156,8 @@ def data(TEST):
'pfs': 'group5',
'transform_protocol': 'esp',
'ipsecsiteconns': []}
TEST.api_ipsecpolicies.add(ipsecpolicy_dict)
TEST.api_ipsecpolicies.append(vpn_ipsec_policy.VpnIpsecPolicy(
**ipsecpolicy_dict))
TEST.ipsecpolicies.add(vpn.IPsecPolicy(ipsecpolicy_dict))
# 1st IPsec site connection
@@ -173,7 +181,9 @@ def data(TEST):
'vpnservice_id': vpnservice_dict['id'],
'admin_state_up': True,
'status': 'Active'}
TEST.api_ipsecsiteconnections.add(ipsecsiteconnection_dict)
TEST.api_ipsecsiteconnections.append(
vpn_ipsec_site_connection.VpnIPSecSiteConnection(
**ipsecsiteconnection_dict))
TEST.ipsecsiteconnections.add(
vpn.IPsecSiteConnection(ipsecsiteconnection_dict))
@@ -196,6 +206,8 @@ def data(TEST):
'vpnservice_id': vpnservice_dict['id'],
'admin_state_up': True,
'status': 'Active'}
TEST.api_ipsecsiteconnections.add(ipsecsiteconnection_dict)
TEST.api_ipsecsiteconnections.append(
vpn_ipsec_site_connection.VpnIPSecSiteConnection(
**ipsecsiteconnection_dict))
TEST.ipsecsiteconnections.add(
vpn.IPsecSiteConnection(ipsecsiteconnection_dict))
requirements.txt
+1 -1
View File
@@ -4,4 +4,4 @@
pbr!=2.1.0,>=2.0.0 # Apache-2.0
horizon>=24.0.0 # Apache-2.0
python-neutronclient>=6.7.0 # Apache-2.0
openstacksdk>=4.5.0 # Apache-2.0