¡@
OpenStack Study: test_vpnaas_extension.py**** CubicPower OpenStack Study ****
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # # (c) Copyright 2013 Hewlett-Packard Development Company, L.P. # All Rights Reserved. # # 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. # # @author: Swaminathan Vasudevan, Hewlett-Packard. import copy import mock from webob import exc from neutron.extensions import vpnaas from neutron.openstack.common import uuidutils from neutron.plugins.common import constants from neutron.tests.unit import test_api_v2 from neutron.tests.unit import test_api_v2_extension _uuid = uuidutils.generate_uuid _get_path = test_api_v2._get_path
**** CubicPower OpenStack Study **** class VpnaasExtensionTestCase(test_api_v2_extension.ExtensionTestCase): fmt = 'json'
**** CubicPower OpenStack Study **** def setUp(self): super(VpnaasExtensionTestCase, self).setUp() plural_mappings = {'ipsecpolicy': 'ipsecpolicies', 'ikepolicy': 'ikepolicies', 'ipsec_site_connection': 'ipsec-site-connections'} self._setUpExtension( 'neutron.extensions.vpnaas.VPNPluginBase', constants.VPN, vpnaas.RESOURCE_ATTRIBUTE_MAP, vpnaas.Vpnaas, 'vpn', plural_mappings=plural_mappings, use_quota=True)
**** CubicPower OpenStack Study **** def test_ikepolicy_create(self): """Test case to create an ikepolicy.""" ikepolicy_id = _uuid() data = {'ikepolicy': {'name': 'ikepolicy1', 'description': 'myikepolicy1', 'auth_algorithm': 'sha1', 'encryption_algorithm': 'aes-128', 'phase1_negotiation_mode': 'main', 'lifetime': { 'units': 'seconds', 'value': 3600}, 'ike_version': 'v1', 'pfs': 'group5', 'tenant_id': _uuid()}} return_value = copy.copy(data['ikepolicy']) return_value.update({'id': ikepolicy_id}) instance = self.plugin.return_value instance.create_ikepolicy.return_value = return_value res = self.api.post(_get_path('vpn/ikepolicies', fmt=self.fmt), self.serialize(data), content_type='application/%s' % self.fmt) instance.create_ikepolicy.assert_called_with(mock.ANY, ikepolicy=data) self.assertEqual(res.status_int, exc.HTTPCreated.code) res = self.deserialize(res) self.assertIn('ikepolicy', res) self.assertEqual(res['ikepolicy'], return_value)
**** CubicPower OpenStack Study **** def test_ikepolicy_list(self): """Test case to list all ikepolicies.""" ikepolicy_id = _uuid() return_value = [{'name': 'ikepolicy1', 'auth_algorithm': 'sha1', 'encryption_algorithm': 'aes-128', 'pfs': 'group5', 'ike_version': 'v1', 'id': ikepolicy_id}] instance = self.plugin.return_value instance.get_ikepolicies.return_value = return_value res = self.api.get(_get_path('vpn/ikepolicies', fmt=self.fmt)) instance.get_ikepolicies.assert_called_with(mock.ANY, fields=mock.ANY, filters=mock.ANY) self.assertEqual(res.status_int, exc.HTTPOk.code)
**** CubicPower OpenStack Study **** def test_ikepolicy_update(self): """Test case to update an ikepolicy.""" ikepolicy_id = _uuid() update_data = {'ikepolicy': {'name': 'ikepolicy1', 'encryption_algorithm': 'aes-256'}} return_value = {'name': 'ikepolicy1', 'auth_algorithm': 'sha1', 'encryption_algorithm': 'aes-256', 'phase1_negotiation_mode': 'main', 'lifetime': { 'units': 'seconds', 'value': 3600}, 'ike_version': 'v1', 'pfs': 'group5', 'tenant_id': _uuid(), 'id': ikepolicy_id} instance = self.plugin.return_value instance.update_ikepolicy.return_value = return_value res = self.api.put(_get_path('vpn/ikepolicies', id=ikepolicy_id, fmt=self.fmt), self.serialize(update_data)) instance.update_ikepolicy.assert_called_with(mock.ANY, ikepolicy_id, ikepolicy=update_data) self.assertEqual(res.status_int, exc.HTTPOk.code) res = self.deserialize(res) self.assertIn('ikepolicy', res) self.assertEqual(res['ikepolicy'], return_value)
**** CubicPower OpenStack Study **** def test_ikepolicy_get(self): """Test case to get or show an ikepolicy.""" ikepolicy_id = _uuid() return_value = {'name': 'ikepolicy1', 'auth_algorithm': 'sha1', 'encryption_algorithm': 'aes-128', 'phase1_negotiation_mode': 'main', 'lifetime': { 'units': 'seconds', 'value': 3600}, 'ike_version': 'v1', 'pfs': 'group5', 'tenant_id': _uuid(), 'id': ikepolicy_id} instance = self.plugin.return_value instance.get_ikepolicy.return_value = return_value res = self.api.get(_get_path('vpn/ikepolicies', id=ikepolicy_id, fmt=self.fmt)) instance.get_ikepolicy.assert_called_with(mock.ANY, ikepolicy_id, fields=mock.ANY) self.assertEqual(res.status_int, exc.HTTPOk.code) res = self.deserialize(res) self.assertIn('ikepolicy', res) self.assertEqual(res['ikepolicy'], return_value)
**** CubicPower OpenStack Study **** def test_ikepolicy_delete(self): """Test case to delete an ikepolicy.""" self._test_entity_delete('ikepolicy')
**** CubicPower OpenStack Study **** def test_ipsecpolicy_create(self): """Test case to create an ipsecpolicy.""" ipsecpolicy_id = _uuid() data = {'ipsecpolicy': {'name': 'ipsecpolicy1', 'description': 'myipsecpolicy1', 'auth_algorithm': 'sha1', 'encryption_algorithm': 'aes-128', 'encapsulation_mode': 'tunnel', 'lifetime': { 'units': 'seconds', 'value': 3600}, 'transform_protocol': 'esp', 'pfs': 'group5', 'tenant_id': _uuid()}} return_value = copy.copy(data['ipsecpolicy']) return_value.update({'id': ipsecpolicy_id}) instance = self.plugin.return_value instance.create_ipsecpolicy.return_value = return_value res = self.api.post(_get_path('vpn/ipsecpolicies', fmt=self.fmt), self.serialize(data), content_type='application/%s' % self.fmt) instance.create_ipsecpolicy.assert_called_with(mock.ANY, ipsecpolicy=data) self.assertEqual(res.status_int, exc.HTTPCreated.code) res = self.deserialize(res) self.assertIn('ipsecpolicy', res) self.assertEqual(res['ipsecpolicy'], return_value)
**** CubicPower OpenStack Study **** def test_ipsecpolicy_list(self): """Test case to list an ipsecpolicy.""" ipsecpolicy_id = _uuid() return_value = [{'name': 'ipsecpolicy1', 'auth_algorithm': 'sha1', 'encryption_algorithm': 'aes-128', 'pfs': 'group5', 'id': ipsecpolicy_id}] instance = self.plugin.return_value instance.get_ipsecpolicies.return_value = return_value res = self.api.get(_get_path('vpn/ipsecpolicies', fmt=self.fmt)) instance.get_ipsecpolicies.assert_called_with(mock.ANY, fields=mock.ANY, filters=mock.ANY) self.assertEqual(res.status_int, exc.HTTPOk.code)
**** CubicPower OpenStack Study **** def test_ipsecpolicy_update(self): """Test case to update an ipsecpolicy.""" ipsecpolicy_id = _uuid() update_data = {'ipsecpolicy': {'name': 'ipsecpolicy1', 'encryption_algorithm': 'aes-256'}} return_value = {'name': 'ipsecpolicy1', 'auth_algorithm': 'sha1', 'encryption_algorithm': 'aes-128', 'encapsulation_mode': 'tunnel', 'lifetime': { 'units': 'seconds', 'value': 3600}, 'transform_protocol': 'esp', 'pfs': 'group5', 'tenant_id': _uuid(), 'id': ipsecpolicy_id} instance = self.plugin.return_value instance.update_ipsecpolicy.return_value = return_value res = self.api.put(_get_path('vpn/ipsecpolicies', id=ipsecpolicy_id, fmt=self.fmt), self.serialize(update_data)) instance.update_ipsecpolicy.assert_called_with(mock.ANY, ipsecpolicy_id, ipsecpolicy=update_data) self.assertEqual(res.status_int, exc.HTTPOk.code) res = self.deserialize(res) self.assertIn('ipsecpolicy', res) self.assertEqual(res['ipsecpolicy'], return_value)
**** CubicPower OpenStack Study **** def test_ipsecpolicy_get(self): """Test case to get or show an ipsecpolicy.""" ipsecpolicy_id = _uuid() return_value = {'name': 'ipsecpolicy1', 'auth_algorithm': 'sha1', 'encryption_algorithm': 'aes-128', 'encapsulation_mode': 'tunnel', 'lifetime': { 'units': 'seconds', 'value': 3600}, 'transform_protocol': 'esp', 'pfs': 'group5', 'tenant_id': _uuid(), 'id': ipsecpolicy_id} instance = self.plugin.return_value instance.get_ipsecpolicy.return_value = return_value res = self.api.get(_get_path('vpn/ipsecpolicies', id=ipsecpolicy_id, fmt=self.fmt)) instance.get_ipsecpolicy.assert_called_with(mock.ANY, ipsecpolicy_id, fields=mock.ANY) self.assertEqual(res.status_int, exc.HTTPOk.code) res = self.deserialize(res) self.assertIn('ipsecpolicy', res) self.assertEqual(res['ipsecpolicy'], return_value)
**** CubicPower OpenStack Study **** def test_ipsecpolicy_delete(self): """Test case to delete an ipsecpolicy.""" self._test_entity_delete('ipsecpolicy')
**** CubicPower OpenStack Study **** def test_vpnservice_create(self): """Test case to create a vpnservice.""" vpnservice_id = _uuid() data = {'vpnservice': {'name': 'vpnservice1', 'description': 'descr_vpn1', 'subnet_id': _uuid(), 'router_id': _uuid(), 'admin_state_up': True, 'tenant_id': _uuid()}} return_value = copy.copy(data['vpnservice']) return_value.update({'status': "ACTIVE", 'id': vpnservice_id}) instance = self.plugin.return_value instance.create_vpnservice.return_value = return_value res = self.api.post(_get_path('vpn/vpnservices', fmt=self.fmt), self.serialize(data), content_type='application/%s' % self.fmt) instance.create_vpnservice.assert_called_with(mock.ANY, vpnservice=data) self.assertEqual(res.status_int, exc.HTTPCreated.code) res = self.deserialize(res) self.assertIn('vpnservice', res) self.assertEqual(res['vpnservice'], return_value)
**** CubicPower OpenStack Study **** def test_vpnservice_list(self): """Test case to list all vpnservices.""" vpnservice_id = _uuid() return_value = [{'name': 'vpnservice1', 'tenant_id': _uuid(), 'status': 'ACTIVE', 'id': vpnservice_id}] instance = self.plugin.return_value instance.get_vpnservice.return_value = return_value res = self.api.get(_get_path('vpn/vpnservices', fmt=self.fmt)) instance.get_vpnservices.assert_called_with(mock.ANY, fields=mock.ANY, filters=mock.ANY) self.assertEqual(res.status_int, exc.HTTPOk.code)
**** CubicPower OpenStack Study **** def test_vpnservice_update(self): """Test case to update a vpnservice.""" vpnservice_id = _uuid() update_data = {'vpnservice': {'admin_state_up': False}} return_value = {'name': 'vpnservice1', 'admin_state_up': False, 'subnet_id': _uuid(), 'router_id': _uuid(), 'tenant_id': _uuid(), 'status': "ACTIVE", 'id': vpnservice_id} instance = self.plugin.return_value instance.update_vpnservice.return_value = return_value res = self.api.put(_get_path('vpn/vpnservices', id=vpnservice_id, fmt=self.fmt), self.serialize(update_data)) instance.update_vpnservice.assert_called_with(mock.ANY, vpnservice_id, vpnservice=update_data) self.assertEqual(res.status_int, exc.HTTPOk.code) res = self.deserialize(res) self.assertIn('vpnservice', res) self.assertEqual(res['vpnservice'], return_value)
**** CubicPower OpenStack Study **** def test_vpnservice_get(self): """Test case to get or show a vpnservice.""" vpnservice_id = _uuid() return_value = {'name': 'vpnservice1', 'admin_state_up': True, 'subnet_id': _uuid(), 'router_id': _uuid(), 'tenant_id': _uuid(), 'status': "ACTIVE", 'id': vpnservice_id} instance = self.plugin.return_value instance.get_vpnservice.return_value = return_value res = self.api.get(_get_path('vpn/vpnservices', id=vpnservice_id, fmt=self.fmt)) instance.get_vpnservice.assert_called_with(mock.ANY, vpnservice_id, fields=mock.ANY) self.assertEqual(res.status_int, exc.HTTPOk.code) res = self.deserialize(res) self.assertIn('vpnservice', res) self.assertEqual(res['vpnservice'], return_value)
**** CubicPower OpenStack Study **** def test_vpnservice_delete(self): """Test case to delete a vpnservice.""" self._test_entity_delete('vpnservice')
**** CubicPower OpenStack Study **** def test_ipsec_site_connection_create(self): """Test case to create a ipsec_site_connection.""" ipsecsite_con_id = _uuid() ikepolicy_id = _uuid() ipsecpolicy_id = _uuid() data = { 'ipsec_site_connection': {'name': 'connection1', 'description': 'Remote-connection1', 'peer_address': '192.168.1.10', 'peer_id': '192.168.1.10', 'peer_cidrs': ['192.168.2.0/24', '192.168.3.0/24'], 'mtu': 1500, 'psk': 'abcd', 'initiator': 'bi-directional', 'dpd': { 'action': 'hold', 'interval': 30, 'timeout': 120}, 'ikepolicy_id': ikepolicy_id, 'ipsecpolicy_id': ipsecpolicy_id, 'vpnservice_id': _uuid(), 'admin_state_up': True, 'tenant_id': _uuid()} } return_value = copy.copy(data['ipsec_site_connection']) return_value.update({'status': "ACTIVE", 'id': ipsecsite_con_id}) instance = self.plugin.return_value instance.create_ipsec_site_connection.return_value = return_value res = self.api.post(_get_path('vpn/ipsec-site-connections', fmt=self.fmt), self.serialize(data), content_type='application/%s' % self.fmt) instance.create_ipsec_site_connection.assert_called_with( mock.ANY, ipsec_site_connection=data ) self.assertEqual(res.status_int, exc.HTTPCreated.code) res = self.deserialize(res) self.assertIn('ipsec_site_connection', res) self.assertEqual(res['ipsec_site_connection'], return_value)
**** CubicPower OpenStack Study **** def test_ipsec_site_connection_list(self): """Test case to list all ipsec_site_connections.""" ipsecsite_con_id = _uuid() return_value = [{'name': 'connection1', 'peer_address': '192.168.1.10', 'peer_cidrs': ['192.168.2.0/24', '192.168.3.0/24'], 'route_mode': 'static', 'auth_mode': 'psk', 'tenant_id': _uuid(), 'status': 'ACTIVE', 'id': ipsecsite_con_id}] instance = self.plugin.return_value instance.get_ipsec_site_connections.return_value = return_value res = self.api.get( _get_path('vpn/ipsec-site-connections', fmt=self.fmt)) instance.get_ipsec_site_connections.assert_called_with( mock.ANY, fields=mock.ANY, filters=mock.ANY ) self.assertEqual(res.status_int, exc.HTTPOk.code)
**** CubicPower OpenStack Study **** def test_ipsec_site_connection_update(self): """Test case to update a ipsec_site_connection.""" ipsecsite_con_id = _uuid() update_data = {'ipsec_site_connection': {'admin_state_up': False}} return_value = {'name': 'connection1', 'description': 'Remote-connection1', 'peer_address': '192.168.1.10', 'peer_id': '192.168.1.10', 'peer_cidrs': ['192.168.2.0/24', '192.168.3.0/24'], 'mtu': 1500, 'psk': 'abcd', 'initiator': 'bi-directional', 'dpd': { 'action': 'hold', 'interval': 30, 'timeout': 120}, 'ikepolicy_id': _uuid(), 'ipsecpolicy_id': _uuid(), 'vpnservice_id': _uuid(), 'admin_state_up': False, 'tenant_id': _uuid(), 'status': 'ACTIVE', 'id': ipsecsite_con_id} instance = self.plugin.return_value instance.update_ipsec_site_connection.return_value = return_value res = self.api.put(_get_path('vpn/ipsec-site-connections', id=ipsecsite_con_id, fmt=self.fmt), self.serialize(update_data)) instance.update_ipsec_site_connection.assert_called_with( mock.ANY, ipsecsite_con_id, ipsec_site_connection=update_data ) self.assertEqual(res.status_int, exc.HTTPOk.code) res = self.deserialize(res) self.assertIn('ipsec_site_connection', res) self.assertEqual(res['ipsec_site_connection'], return_value)
**** CubicPower OpenStack Study **** def test_ipsec_site_connection_get(self): """Test case to get or show a ipsec_site_connection.""" ipsecsite_con_id = _uuid() return_value = {'name': 'connection1', 'description': 'Remote-connection1', 'peer_address': '192.168.1.10', 'peer_id': '192.168.1.10', 'peer_cidrs': ['192.168.2.0/24', '192.168.3.0/24'], 'mtu': 1500, 'psk': 'abcd', 'initiator': 'bi-directional', 'dpd': { 'action': 'hold', 'interval': 30, 'timeout': 120}, 'ikepolicy_id': _uuid(), 'ipsecpolicy_id': _uuid(), 'vpnservice_id': _uuid(), 'admin_state_up': True, 'tenant_id': _uuid(), 'status': 'ACTIVE', 'id': ipsecsite_con_id} instance = self.plugin.return_value instance.get_ipsec_site_connection.return_value = return_value res = self.api.get(_get_path('vpn/ipsec-site-connections', id=ipsecsite_con_id, fmt=self.fmt)) instance.get_ipsec_site_connection.assert_called_with( mock.ANY, ipsecsite_con_id, fields=mock.ANY ) self.assertEqual(res.status_int, exc.HTTPOk.code) res = self.deserialize(res) self.assertIn('ipsec_site_connection', res) self.assertEqual(res['ipsec_site_connection'], return_value)
**** CubicPower OpenStack Study **** def test_ipsec_site_connection_delete(self): """Test case to delete a ipsec_site_connection.""" self._test_entity_delete('ipsec_site_connection')
**** CubicPower OpenStack Study **** class VpnaasExtensionTestCaseXML(VpnaasExtensionTestCase): fmt = 'xml'
|