¡@
OpenStack Study: test_vpnaas_plugin.py**** CubicPower OpenStack Study ****
# # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import contextlib import copy import webob.exc from neutron.api.v2 import attributes from neutron.db.vpn import vpn_db from neutron.extensions import vpnaas from neutron import manager from neutron.openstack.common import uuidutils from neutron.tests.unit.db.vpn import test_db_vpnaas from neutron.tests.unit.vmware.vshield import test_edge_router _uuid = uuidutils.generate_uuid
**** CubicPower OpenStack Study **** class VPNTestExtensionManager(
test_edge_router.ServiceRouterTestExtensionManager):
**** CubicPower OpenStack Study **** def get_resources(self): # If l3 resources have been loaded and updated by main API # router, update the map in the l3 extension so it will load # the same attributes as the API router resources = super(VPNTestExtensionManager, self).get_resources() vpn_attr_map = copy.deepcopy(vpnaas.RESOURCE_ATTRIBUTE_MAP) for res in vpnaas.RESOURCE_ATTRIBUTE_MAP.keys(): attr_info = attributes.RESOURCE_ATTRIBUTE_MAP.get(res) if attr_info: vpnaas.RESOURCE_ATTRIBUTE_MAP[res] = attr_info vpn_resources = vpnaas.Vpnaas.get_resources() # restore the original resources once the controllers are created vpnaas.RESOURCE_ATTRIBUTE_MAP = vpn_attr_map resources.extend(vpn_resources) return resources
**** CubicPower OpenStack Study **** class TestVpnPlugin(test_db_vpnaas.VPNTestMixin,
test_edge_router.ServiceRouterTest):
**** CubicPower OpenStack Study **** def vcns_vpn_patch(self): instance = self.vcns_instance instance.return_value.update_ipsec_config.side_effect = ( self.fc2.update_ipsec_config) instance.return_value.get_ipsec_config.side_effect = ( self.fc2.get_ipsec_config) instance.return_value.delete_ipsec_config.side_effect = ( self.fc2.delete_ipsec_config)
**** CubicPower OpenStack Study **** def setUp(self): # Save the global RESOURCE_ATTRIBUTE_MAP self.saved_attr_map = {} for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.items(): self.saved_attr_map[resource] = attrs.copy() super(TestVpnPlugin, self).setUp(ext_mgr=VPNTestExtensionManager()) self.vcns_vpn_patch() self.plugin = manager.NeutronManager.get_plugin() self.router_id = None
**** CubicPower OpenStack Study **** def tearDown(self): super(TestVpnPlugin, self).tearDown() # Restore the global RESOURCE_ATTRIBUTE_MAP attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map self.ext_api = None self.plugin = None @contextlib.contextmanager
**** CubicPower OpenStack Study **** def router(self, vlan_id=None): with self._create_l3_ext_network(vlan_id) as net: with self.subnet(cidr='100.0.0.0/24', network=net) as s: data = {'router': {'tenant_id': self._tenant_id}} data['router']['service_router'] = True router_req = self.new_create_request('routers', data, self.fmt) res = router_req.get_response(self.ext_api) router = self.deserialize(self.fmt, res) self._add_external_gateway_to_router( router['router']['id'], s['subnet']['network_id']) router = self._show('routers', router['router']['id']) yield router self._delete('routers', router['router']['id'])
**** CubicPower OpenStack Study **** def test_create_vpnservice(self, **extras): """Test case to create a vpnservice.""" description = 'my-vpn-service' expected = {'name': 'vpnservice1', 'description': 'my-vpn-service', 'admin_state_up': True, 'status': 'ACTIVE', 'tenant_id': self._tenant_id, } expected.update(extras) with self.subnet(cidr='10.2.0.0/24') as subnet: with self.router() as router: expected['router_id'] = router['router']['id'] expected['subnet_id'] = subnet['subnet']['id'] name = expected['name'] with self.vpnservice(name=name, subnet=subnet, router=router, description=description, **extras) as vpnservice: self.assertEqual(dict((k, v) for k, v in vpnservice['vpnservice'].items() if k in expected), expected)
**** CubicPower OpenStack Study **** def test_create_vpnservices_with_same_router(self, **extras): """Test case to create two vpnservices with same router.""" with self.subnet(cidr='10.2.0.0/24') as subnet: with self.router() as router: with self.vpnservice(name='vpnservice1', subnet=subnet, router=router): res = self._create_vpnservice( 'json', 'vpnservice2', True, router_id=(router['router']['id']), subnet_id=(subnet['subnet']['id'])) self.assertEqual( res.status_int, webob.exc.HTTPConflict.code)
**** CubicPower OpenStack Study **** def test_update_vpnservice(self): """Test case to update a vpnservice.""" name = 'new_vpnservice1' expected = [('name', name)] with contextlib.nested( self.subnet(cidr='10.2.0.0/24'), self.router()) as (subnet, router): with self.vpnservice(name=name, subnet=subnet, router=router) as vpnservice: expected.append(('subnet_id', vpnservice['vpnservice']['subnet_id'])) expected.append(('router_id', vpnservice['vpnservice']['router_id'])) data = {'vpnservice': {'name': name, 'admin_state_up': False}} expected.append(('admin_state_up', False)) self._set_active(vpn_db.VPNService, vpnservice['vpnservice']['id']) req = self.new_update_request( 'vpnservices', data, vpnservice['vpnservice']['id']) res = self.deserialize(self.fmt, req.get_response(self.ext_api)) for k, v in expected: self.assertEqual(res['vpnservice'][k], v)
**** CubicPower OpenStack Study **** def _test_create_ipsec_site_connection(self, key_overrides=None, ike_key_overrides=None, ipsec_key_overrides=None, setup_overrides=None, expected_status_int=200): """Create ipsec_site_connection and check results.""" params = {'ikename': 'ikepolicy1', 'ipsecname': 'ipsecpolicy1', 'vpnsname': 'vpnservice1', 'subnet_cidr': '10.2.0.0/24', 'subnet_version': 4} if setup_overrides: params.update(setup_overrides) expected = {'name': 'connection1', 'description': 'my-ipsec-connection', 'peer_address': '192.168.1.10', 'peer_id': '192.168.1.10', 'peer_cidrs': ['192.168.2.0/24', '192.168.3.0/24'], 'initiator': 'bi-directional', 'mtu': 1500, 'tenant_id': self._tenant_id, 'psk': 'abcd', 'status': 'ACTIVE', 'admin_state_up': True} if key_overrides: expected.update(key_overrides) ike_expected = {'name': params['ikename'], 'auth_algorithm': 'sha1', 'encryption_algorithm': 'aes-128', 'ike_version': 'v1', 'pfs': 'group5'} if ike_key_overrides: ike_expected.update(ike_key_overrides) ipsec_expected = {'name': params['ipsecname'], 'auth_algorithm': 'sha1', 'encryption_algorithm': 'aes-128', 'pfs': 'group5'} if ipsec_key_overrides: ipsec_expected.update(ipsec_key_overrides) dpd = {'action': 'hold', 'interval': 40, 'timeout': 120} with contextlib.nested( self.ikepolicy(self.fmt, ike_expected['name'], ike_expected['auth_algorithm'], ike_expected['encryption_algorithm'], ike_version=ike_expected['ike_version'], pfs=ike_expected['pfs']), self.ipsecpolicy(self.fmt, ipsec_expected['name'], ipsec_expected['auth_algorithm'], ipsec_expected['encryption_algorithm'], pfs=ipsec_expected['pfs']), self.subnet(cidr=params['subnet_cidr'], ip_version=params['subnet_version']), self.router()) as ( ikepolicy, ipsecpolicy, subnet, router): with self.vpnservice(name=params['vpnsname'], subnet=subnet, router=router) as vpnservice1: expected['ikepolicy_id'] = ikepolicy['ikepolicy']['id'] expected['ipsecpolicy_id'] = ( ipsecpolicy['ipsecpolicy']['id'] ) expected['vpnservice_id'] = ( vpnservice1['vpnservice']['id'] ) try: with self.ipsec_site_connection( self.fmt, expected['name'], expected['peer_address'], expected['peer_id'], expected['peer_cidrs'], expected['mtu'], expected['psk'], expected['initiator'], dpd['action'], dpd['interval'], dpd['timeout'], vpnservice1, ikepolicy, ipsecpolicy, expected['admin_state_up'], description=expected['description'] ) as ipsec_site_connection: if expected_status_int != 200: self.fail("Expected failure on create") self._check_ipsec_site_connection( ipsec_site_connection['ipsec_site_connection'], expected, dpd) except webob.exc.HTTPClientError as ce: self.assertEqual(ce.code, expected_status_int)
**** CubicPower OpenStack Study **** def test_create_ipsec_site_connection(self, **extras): """Test case to create an ipsec_site_connection.""" self._test_create_ipsec_site_connection(key_overrides=extras)
**** CubicPower OpenStack Study **** def test_create_ipsec_site_connection_invalid_ikepolicy(self): self._test_create_ipsec_site_connection( ike_key_overrides={'ike_version': 'v2'}, expected_status_int=400)
**** CubicPower OpenStack Study **** def test_create_ipsec_site_connection_invalid_ipsecpolicy(self): self._test_create_ipsec_site_connection( ipsec_key_overrides={'encryption_algorithm': 'aes-192'}, expected_status_int=400) self._test_create_ipsec_site_connection( ipsec_key_overrides={'pfs': 'group14'}, expected_status_int=400)
**** CubicPower OpenStack Study **** def _test_update_ipsec_site_connection(self, update={'name': 'new name'}, overrides=None, expected_status_int=200): """Creates and then updates ipsec_site_connection.""" expected = {'name': 'new_ipsec_site_connection', 'ikename': 'ikepolicy1', 'ipsecname': 'ipsecpolicy1', 'vpnsname': 'vpnservice1', 'description': 'my-ipsec-connection', 'peer_address': '192.168.1.10', 'peer_id': '192.168.1.10', 'peer_cidrs': ['192.168.2.0/24', '192.168.3.0/24'], 'initiator': 'bi-directional', 'mtu': 1500, 'tenant_id': self._tenant_id, 'psk': 'abcd', 'status': 'ACTIVE', 'admin_state_up': True, 'action': 'hold', 'interval': 40, 'timeout': 120, 'subnet_cidr': '10.2.0.0/24', 'subnet_version': 4, 'make_active': True} if overrides: expected.update(overrides) with contextlib.nested( self.ikepolicy(name=expected['ikename']), self.ipsecpolicy(name=expected['ipsecname']), self.subnet(cidr=expected['subnet_cidr'], ip_version=expected['subnet_version']), self.router() ) as (ikepolicy, ipsecpolicy, subnet, router): with self.vpnservice(name=expected['vpnsname'], subnet=subnet, router=router) as vpnservice1: expected['vpnservice_id'] = vpnservice1['vpnservice']['id'] expected['ikepolicy_id'] = ikepolicy['ikepolicy']['id'] expected['ipsecpolicy_id'] = ipsecpolicy['ipsecpolicy']['id'] with self.ipsec_site_connection( self.fmt, expected['name'], expected['peer_address'], expected['peer_id'], expected['peer_cidrs'], expected['mtu'], expected['psk'], expected['initiator'], expected['action'], expected['interval'], expected['timeout'], vpnservice1, ikepolicy, ipsecpolicy, expected['admin_state_up'], description=expected['description'] ) as ipsec_site_connection: data = {'ipsec_site_connection': update} if expected.get('make_active'): self._set_active( vpn_db.IPsecSiteConnection, (ipsec_site_connection['ipsec_site_connection'] ['id'])) req = self.new_update_request( 'ipsec-site-connections', data, ipsec_site_connection['ipsec_site_connection']['id']) res = req.get_response(self.ext_api) self.assertEqual(expected_status_int, res.status_int) if expected_status_int == 200: res_dict = self.deserialize(self.fmt, res) for k, v in update.items(): self.assertEqual( res_dict['ipsec_site_connection'][k], v)
**** CubicPower OpenStack Study **** def test_update_ipsec_site_connection(self): """Test case for valid updates to IPSec site connection.""" dpd = {'action': 'hold', 'interval': 40, 'timeout': 120} self._test_update_ipsec_site_connection(update={'dpd': dpd}) self._test_update_ipsec_site_connection(update={'mtu': 2000})
**** CubicPower OpenStack Study **** def test_delete_ipsec_site_connection(self): """Test case to delete a ipsec_site_connection.""" with self.ipsec_site_connection( no_delete=True) as ipsec_site_connection: req = self.new_delete_request( 'ipsec-site-connections', ipsec_site_connection['ipsec_site_connection']['id'] ) res = req.get_response(self.ext_api) self.assertEqual(res.status_int, 204)
|