# vim: tabstop=4 shiftwidth=4 softtabstop=4 # Copyright (c) 2012 OpenStack, LLC. # 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. import os import mock import unittest2 as unittest import webob from webob import exc import webtest from quantum.api.extensions import PluginAwareExtensionManager from quantum.api.v2 import attributes from quantum.api.v2 import base from quantum.api.v2 import router from quantum.common import config from quantum.common import constants from quantum.common import exceptions as q_exc from quantum import context from quantum.manager import QuantumManager from quantum.openstack.common import cfg from quantum.openstack.common.notifier import api as notifer_api from quantum.openstack.common import uuidutils from quantum.tests.unit import testlib_api from quantum import wsgi ROOTDIR = os.path.dirname(os.path.dirname(__file__)) ETCDIR = os.path.join(ROOTDIR, 'etc') EXTDIR = os.path.join(ROOTDIR, 'unit/extensions') _uuid = uuidutils.generate_uuid def etcdir(*p): return os.path.join(ETCDIR, *p) def _get_path(resource, id=None, action=None, fmt=None): path = '/%s' % resource if id is not None: path = path + '/%s' % id if action is not None: path = path + '/%s' % action if fmt is not None: path = path + '.%s' % fmt return path class ResourceIndexTestCase(unittest.TestCase): def test_index_json(self): index = webtest.TestApp(router.Index({'foo': 'bar'})) res = index.get('') self.assertTrue('resources' in res.json) self.assertTrue(len(res.json['resources']) == 1) resource = res.json['resources'][0] self.assertTrue('collection' in resource) self.assertTrue(resource['collection'] == 'bar') self.assertTrue('name' in resource) self.assertTrue(resource['name'] == 'foo') self.assertTrue('links' in resource) self.assertTrue(len(resource['links']) == 1) link = resource['links'][0] self.assertTrue('href' in link) self.assertTrue(link['href'] == 'http://localhost/bar') self.assertTrue('rel' in link) self.assertTrue(link['rel'] == 'self') class APIv2TestBase(unittest.TestCase): def setUp(self): plugin = 'quantum.quantum_plugin_base_v2.QuantumPluginBaseV2' # Ensure 'stale' patched copies of the plugin are never returned QuantumManager._instance = None # Ensure existing ExtensionManager is not used PluginAwareExtensionManager._instance = None # Create the default configurations args = ['--config-file', etcdir('quantum.conf.test')] config.parse(args=args) # Update the plugin cfg.CONF.set_override('core_plugin', plugin) self._plugin_patcher = mock.patch(plugin, autospec=True) self.plugin = self._plugin_patcher.start() api = router.APIRouter() self.api = webtest.TestApp(api) super(APIv2TestBase, self).setUp() def tearDown(self): self._plugin_patcher.stop() self.api = None self.plugin = None cfg.CONF.reset() class APIv2TestCase(APIv2TestBase): # NOTE(jkoelker) This potentially leaks the mock object if the setUp # raises without being caught. Using unittest2 # or dropping 2.6 support so we can use addCleanup # will get around this. def _do_field_list(self, resource, base_fields): attr_info = attributes.RESOURCE_ATTRIBUTE_MAP[resource] policy_attrs = [name for (name, info) in attr_info.items() if info.get('required_by_policy')] fields = base_fields fields.extend(policy_attrs) return fields def test_fields(self): instance = self.plugin.return_value instance.get_networks.return_value = [] self.api.get(_get_path('networks'), {'fields': 'foo'}) fields = self._do_field_list('networks', ['foo']) instance.get_networks.assert_called_once_with(mock.ANY, filters=mock.ANY, fields=fields) def test_fields_multiple(self): instance = self.plugin.return_value instance.get_networks.return_value = [] fields = self._do_field_list('networks', ['foo', 'bar']) self.api.get(_get_path('networks'), {'fields': ['foo', 'bar']}) instance.get_networks.assert_called_once_with(mock.ANY, filters=mock.ANY, fields=fields) def test_fields_multiple_with_empty(self): instance = self.plugin.return_value instance.get_networks.return_value = [] fields = self._do_field_list('networks', ['foo']) self.api.get(_get_path('networks'), {'fields': ['foo', '']}) instance.get_networks.assert_called_once_with(mock.ANY, filters=mock.ANY, fields=fields) def test_fields_empty(self): instance = self.plugin.return_value instance.get_networks.return_value = [] self.api.get(_get_path('networks'), {'fields': ''}) instance.get_networks.assert_called_once_with(mock.ANY, filters=mock.ANY, fields=[]) def test_fields_multiple_empty(self): instance = self.plugin.return_value instance.get_networks.return_value = [] self.api.get(_get_path('networks'), {'fields': ['', '']}) instance.get_networks.assert_called_once_with(mock.ANY, filters=mock.ANY, fields=[]) def test_filters(self): instance = self.plugin.return_value instance.get_networks.return_value = [] self.api.get(_get_path('networks'), {'foo': 'bar'}) filters = {'foo': ['bar']} instance.get_networks.assert_called_once_with(mock.ANY, filters=filters, fields=mock.ANY) def test_filters_empty(self): instance = self.plugin.return_value instance.get_networks.return_value = [] self.api.get(_get_path('networks'), {'foo': ''}) filters = {} instance.get_networks.assert_called_once_with(mock.ANY, filters=filters, fields=mock.ANY) def test_filters_multiple_empty(self): instance = self.plugin.return_value instance.get_networks.return_value = [] self.api.get(_get_path('networks'), {'foo': ['', '']}) filters = {} instance.get_networks.assert_called_once_with(mock.ANY, filters=filters, fields=mock.ANY) def test_filters_multiple_with_empty(self): instance = self.plugin.return_value instance.get_networks.return_value = [] self.api.get(_get_path('networks'), {'foo': ['bar', '']}) filters = {'foo': ['bar']} instance.get_networks.assert_called_once_with(mock.ANY, filters=filters, fields=mock.ANY) def test_filters_multiple_values(self): instance = self.plugin.return_value instance.get_networks.return_value = [] self.api.get(_get_path('networks'), {'foo': ['bar', 'bar2']}) filters = {'foo': ['bar', 'bar2']} instance.get_networks.assert_called_once_with(mock.ANY, filters=filters, fields=mock.ANY) def test_filters_multiple(self): instance = self.plugin.return_value instance.get_networks.return_value = [] self.api.get(_get_path('networks'), {'foo': 'bar', 'foo2': 'bar2'}) filters = {'foo': ['bar'], 'foo2': ['bar2']} instance.get_networks.assert_called_once_with(mock.ANY, filters=filters, fields=mock.ANY) def test_filters_with_fields(self): instance = self.plugin.return_value instance.get_networks.return_value = [] self.api.get(_get_path('networks'), {'foo': 'bar', 'fields': 'foo'}) filters = {'foo': ['bar']} fields = self._do_field_list('networks', ['foo']) instance.get_networks.assert_called_once_with(mock.ANY, filters=filters, fields=fields) def test_filters_with_convert_to(self): instance = self.plugin.return_value instance.get_ports.return_value = [] self.api.get(_get_path('ports'), {'admin_state_up': 'true'}) filters = {'admin_state_up': [True]} instance.get_ports.assert_called_once_with(mock.ANY, filters=filters, fields=mock.ANY) def test_filters_with_convert_list_to(self): instance = self.plugin.return_value instance.get_ports.return_value = [] self.api.get(_get_path('ports'), {'fixed_ips': ['ip_address=foo', 'subnet_id=bar']}) filters = {'fixed_ips': {'ip_address': ['foo'], 'subnet_id': ['bar']}} instance.get_ports.assert_called_once_with(mock.ANY, filters=filters, fields=mock.ANY) # Note: since all resources use the same controller and validation # logic, we actually get really good coverage from testing just networks. class JSONV2TestCase(APIv2TestBase, testlib_api.WebTestCase): def setUp(self): super(JSONV2TestCase, self).setUp() def _test_list(self, req_tenant_id, real_tenant_id): env = {} if req_tenant_id: env = {'quantum.context': context.Context('', req_tenant_id)} input_dict = {'id': uuidutils.generate_uuid(), 'name': 'net1', 'admin_state_up': True, 'status': "ACTIVE", 'tenant_id': real_tenant_id, 'shared': False, 'subnets': []} return_value = [input_dict] instance = self.plugin.return_value instance.get_networks.return_value = return_value res = self.api.get(_get_path('networks', fmt=self.fmt), extra_environ=env) res = self.deserialize(res) self.assertTrue('networks' in res) if not req_tenant_id or req_tenant_id == real_tenant_id: # expect full list returned self.assertEqual(len(res['networks']), 1) output_dict = res['networks'][0] input_dict['shared'] = False self.assertEqual(len(input_dict), len(output_dict)) for k, v in input_dict.iteritems(): self.assertEqual(v, output_dict[k]) else: # expect no results self.assertEqual(len(res['networks']), 0) def test_list_noauth(self): self._test_list(None, _uuid()) def test_list_keystone(self): tenant_id = _uuid() self._test_list(tenant_id, tenant_id) def test_list_keystone_bad(self): tenant_id = _uuid() self._test_list(tenant_id + "bad", tenant_id) def test_create(self): net_id = _uuid() data = {'network': {'name': 'net1', 'admin_state_up': True, 'tenant_id': _uuid()}} return_value = {'subnets': [], 'status': "ACTIVE", 'id': net_id} return_value.update(data['network'].copy()) instance = self.plugin.return_value instance.create_network.return_value = return_value instance.get_networks_count.return_value = 0 res = self.api.post(_get_path('networks', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt) self.assertEqual(res.status_int, exc.HTTPCreated.code) res = self.deserialize(res) self.assertTrue('network' in res) net = res['network'] self.assertEqual(net['id'], net_id) self.assertEqual(net['status'], "ACTIVE") def test_create_use_defaults(self): net_id = _uuid() initial_input = {'network': {'name': 'net1', 'tenant_id': _uuid()}} full_input = {'network': {'admin_state_up': True, 'shared': False}} full_input['network'].update(initial_input['network']) return_value = {'id': net_id, 'status': "ACTIVE"} return_value.update(full_input['network']) instance = self.plugin.return_value instance.create_network.return_value = return_value instance.get_networks_count.return_value = 0 res = self.api.post(_get_path('networks', fmt=self.fmt), self.serialize(initial_input), content_type='application/' + self.fmt) instance.create_network.assert_called_with(mock.ANY, network=full_input) self.assertEqual(res.status_int, exc.HTTPCreated.code) res = self.deserialize(res) self.assertIn('network', res) net = res['network'] self.assertEqual(net['id'], net_id) self.assertEqual(net['admin_state_up'], True) self.assertEqual(net['status'], "ACTIVE") def test_create_no_keystone_env(self): data = {'name': 'net1'} res = self.api.post(_get_path('networks', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt, expect_errors=True) self.assertEqual(res.status_int, exc.HTTPBadRequest.code) def test_create_with_keystone_env(self): tenant_id = _uuid() net_id = _uuid() env = {'quantum.context': context.Context('', tenant_id)} # tenant_id should be fetched from env initial_input = {'network': {'name': 'net1'}} full_input = {'network': {'admin_state_up': True, 'shared': False, 'tenant_id': tenant_id}} full_input['network'].update(initial_input['network']) return_value = {'id': net_id, 'status': "ACTIVE"} return_value.update(full_input['network']) instance = self.plugin.return_value instance.create_network.return_value = return_value instance.get_networks_count.return_value = 0 res = self.api.post(_get_path('networks', fmt=self.fmt), self.serialize(initial_input), content_type='application/' + self.fmt, extra_environ=env) instance.create_network.assert_called_with(mock.ANY, network=full_input) self.assertEqual(res.status_int, exc.HTTPCreated.code) def test_create_bad_keystone_tenant(self): tenant_id = _uuid() data = {'network': {'name': 'net1', 'tenant_id': tenant_id}} env = {'quantum.context': context.Context('', tenant_id + "bad")} res = self.api.post(_get_path('networks', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt, expect_errors=True, extra_environ=env) self.assertEqual(res.status_int, exc.HTTPBadRequest.code) def test_create_no_body(self): data = {'whoa': None} res = self.api.post(_get_path('networks', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt, expect_errors=True) self.assertEqual(res.status_int, exc.HTTPBadRequest.code) def test_create_no_resource(self): data = {} res = self.api.post(_get_path('networks', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt, expect_errors=True) self.assertEqual(res.status_int, exc.HTTPBadRequest.code) def test_create_missing_attr(self): data = {'port': {'what': 'who', 'tenant_id': _uuid()}} res = self.api.post(_get_path('ports', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt, expect_errors=True) self.assertEqual(res.status_int, 400) def test_create_readonly_attr(self): data = {'network': {'name': 'net1', 'tenant_id': _uuid(), 'status': "ACTIVE"}} res = self.api.post(_get_path('networks', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt, expect_errors=True) self.assertEqual(res.status_int, 400) def test_create_bulk(self): data = {'networks': [{'name': 'net1', 'admin_state_up': True, 'tenant_id': _uuid()}, {'name': 'net2', 'admin_state_up': True, 'tenant_id': _uuid()}]} def side_effect(context, network): net = network.copy() net['network'].update({'subnets': []}) return net['network'] instance = self.plugin.return_value instance.create_network.side_effect = side_effect instance.get_networks_count.return_value = 0 res = self.api.post(_get_path('networks', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt) self.assertEqual(res.status_int, exc.HTTPCreated.code) def test_create_bulk_no_networks(self): data = {'networks': []} res = self.api.post(_get_path('networks', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt, expect_errors=True) self.assertEqual(res.status_int, exc.HTTPBadRequest.code) def test_create_bulk_missing_attr(self): data = {'ports': [{'what': 'who', 'tenant_id': _uuid()}]} res = self.api.post(_get_path('ports', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt, expect_errors=True) self.assertEqual(res.status_int, 400) def test_create_bulk_partial_body(self): data = {'ports': [{'device_id': 'device_1', 'tenant_id': _uuid()}, {'tenant_id': _uuid()}]} res = self.api.post(_get_path('ports', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt, expect_errors=True) self.assertEqual(res.status_int, 400) def test_create_attr_not_specified(self): net_id = _uuid() tenant_id = _uuid() device_id = _uuid() initial_input = {'port': {'name': '', 'network_id': net_id, 'tenant_id': tenant_id, 'device_id': device_id, 'admin_state_up': True}} full_input = {'port': {'admin_state_up': True, 'mac_address': attributes.ATTR_NOT_SPECIFIED, 'fixed_ips': attributes.ATTR_NOT_SPECIFIED, 'device_owner': ''}} full_input['port'].update(initial_input['port']) return_value = {'id': _uuid(), 'status': 'ACTIVE', 'admin_state_up': True, 'mac_address': 'ca:fe:de:ad:be:ef', 'device_id': device_id, 'device_owner': ''} return_value.update(initial_input['port']) instance = self.plugin.return_value instance.get_network.return_value = {'tenant_id': unicode(tenant_id)} instance.get_ports_count.return_value = 1 instance.create_port.return_value = return_value res = self.api.post(_get_path('ports', fmt=self.fmt), self.serialize(initial_input), content_type='application/' + self.fmt) instance.create_port.assert_called_with(mock.ANY, port=full_input) self.assertEqual(res.status_int, exc.HTTPCreated.code) res = self.deserialize(res) self.assertIn('port', res) port = res['port'] self.assertEqual(port['network_id'], net_id) self.assertEqual(port['mac_address'], 'ca:fe:de:ad:be:ef') def test_create_return_extra_attr(self): net_id = _uuid() data = {'network': {'name': 'net1', 'admin_state_up': True, 'tenant_id': _uuid()}} return_value = {'subnets': [], 'status': "ACTIVE", 'id': net_id, 'v2attrs:something': "123"} return_value.update(data['network'].copy()) instance = self.plugin.return_value instance.create_network.return_value = return_value instance.get_networks_count.return_value = 0 res = self.api.post(_get_path('networks', fmt=self.fmt), self.serialize(data), content_type='application/' + self.fmt) self.assertEqual(res.status_int, exc.HTTPCreated.code) res = self.deserialize(res) self.assertIn('network', res) net = res['network'] self.assertEqual(net['id'], net_id) self.assertEqual(net['status'], "ACTIVE") self.assertFalse('v2attrs:something' in net) def test_fields(self): return_value = {'name': 'net1', 'admin_state_up': True, 'subnets': []} instance = self.plugin.return_value instance.get_network.return_value = return_value self.api.get(_get_path('networks', id=uuidutils.generate_uuid(), fmt=self.fmt)) def _test_delete(self, req_tenant_id, real_tenant_id, expected_code, expect_errors=False): env = {} if req_tenant_id: env = {'quantum.context': context.Context('', req_tenant_id)} instance = self.plugin.return_value instance.get_network.return_value = {'tenant_id': real_tenant_id, 'shared': False} instance.delete_network.return_value = None res = self.api.delete(_get_path('networks', id=uuidutils.generate_uuid(), fmt=self.fmt), extra_environ=env, expect_errors=expect_errors) self.assertEqual(res.status_int, expected_code) def test_delete_noauth(self): self._test_delete(None, _uuid(), exc.HTTPNoContent.code) def test_delete_keystone(self): tenant_id = _uuid() self._test_delete(tenant_id, tenant_id, exc.HTTPNoContent.code) def test_delete_keystone_bad_tenant(self): tenant_id = _uuid() self._test_delete(tenant_id + "bad", tenant_id, exc.HTTPNotFound.code, expect_errors=True) def _test_get(self, req_tenant_id, real_tenant_id, expected_code, expect_errors=False): env = {} shared = False if req_tenant_id: env = {'quantum.context': context.Context('', req_tenant_id)} if req_tenant_id.endswith('another'): shared = True env['quantum.context'].roles = ['tenant_admin'] data = {'tenant_id': real_tenant_id, 'shared': shared} instance = self.plugin.return_value instance.get_network.return_value = data res = self.api.get(_get_path('networks', id=uuidutils.generate_uuid(), fmt=self.fmt), extra_environ=env, expect_errors=expect_errors) self.assertEqual(res.status_int, expected_code) def test_get_noauth(self): self._test_get(None, _uuid(), 200) def test_get_keystone(self): tenant_id = _uuid() self._test_get(tenant_id, tenant_id, 200) def test_get_keystone_bad_tenant(self): tenant_id = _uuid() self._test_get(tenant_id + "bad", tenant_id, exc.HTTPNotFound.code, expect_errors=True) def test_get_keystone_shared_network(self): tenant_id = _uuid() self._test_get(tenant_id + "another", tenant_id, 200) def _test_update(self, req_tenant_id, real_tenant_id, expected_code, expect_errors=False): env = {} if req_tenant_id: env = {'quantum.context': context.Context('', req_tenant_id)} # leave out 'name' field intentionally data = {'network': {'admin_state_up': True}} return_value = {'subnets': []} return_value.update(data['network'].copy()) instance = self.plugin.return_value instance.get_network.return_value = {'tenant_id': real_tenant_id, 'shared': False} instance.update_network.return_value = return_value res = self.api.put(_get_path('networks', id=uuidutils.generate_uuid(), fmt=self.fmt), self.serialize(data), extra_environ=env, expect_errors=expect_errors) self.assertEqual(res.status_int, expected_code) def test_update_noauth(self): self._test_update(None, _uuid(), 200) def test_update_keystone(self): tenant_id = _uuid() self._test_update(tenant_id, tenant_id, 200) def test_update_keystone_bad_tenant(self): tenant_id = _uuid() self._test_update(tenant_id + "bad", tenant_id, exc.HTTPNotFound.code, expect_errors=True) def test_update_readonly_field(self): data = {'network': {'status': "NANANA"}} res = self.api.put(_get_path('networks', id=_uuid()), self.serialize(data), content_type='application/' + self.fmt, expect_errors=True) self.assertEqual(res.status_int, 400) class SubresourceTest(unittest.TestCase): def setUp(self): plugin = 'quantum.tests.unit.test_api_v2.TestSubresourcePlugin' QuantumManager._instance = None PluginAwareExtensionManager._instance = None args = ['--config-file', etcdir('quantum.conf.test')] config.parse(args=args) cfg.CONF.set_override('core_plugin', plugin) self._plugin_patcher = mock.patch(plugin, autospec=True) self.plugin = self._plugin_patcher.start() router.SUB_RESOURCES['dummy'] = { 'collection_name': 'dummies', 'parent': {'collection_name': 'networks', 'member_name': 'network'} } api = router.APIRouter() self.api = webtest.TestApp(api) def tearDown(self): self._plugin_patcher.stop() self.api = None self.plugin = None cfg.CONF.reset() def test_index_sub_resource(self): instance = self.plugin.return_value self.api.get('/networks/id1/dummies') instance.get_network_dummies.assert_called_once_with(mock.ANY, filters=mock.ANY, fields=mock.ANY, network_id='id1') def test_show_sub_resource(self): instance = self.plugin.return_value dummy_id = _uuid() self.api.get('/networks/id1' + _get_path('dummies', id=dummy_id)) instance.get_network_dummy.assert_called_once_with(mock.ANY, dummy_id, network_id='id1', fields=mock.ANY) def test_create_sub_resource(self): instance = self.plugin.return_value body = {'dummy': {'foo': 'bar', 'tenant_id': _uuid()}} self.api.post_json('/networks/id1/dummies', body) instance.create_network_dummy.assert_called_once_with(mock.ANY, network_id='id1', dummy=body) def test_update_sub_resource(self): instance = self.plugin.return_value dummy_id = _uuid() body = {'dummy': {'foo': 'bar', 'tenant_id': _uuid()}} self.api.put_json('/networks/id1' + _get_path('dummies', id=dummy_id), body) instance.update_network_dummy.assert_called_once_with(mock.ANY, dummy_id, network_id='id1', dummy=body) def test_delete_sub_resource(self): instance = self.plugin.return_value dummy_id = _uuid() self.api.delete('/networks/id1' + _get_path('dummies', id=dummy_id)) instance.delete_network_dummy.assert_called_once_with(mock.ANY, dummy_id, network_id='id1') # Note: since all resources use the same controller and validation # logic, we actually get really good coverage from testing just networks. class XMLV2TestCase(JSONV2TestCase): fmt = 'xml' class V2Views(unittest.TestCase): def _view(self, keys, collection, resource): data = dict((key, 'value') for key in keys) data['fake'] = 'value' attr_info = attributes.RESOURCE_ATTRIBUTE_MAP[collection] controller = base.Controller(None, collection, resource, attr_info) res = controller._view(data) self.assertTrue('fake' not in res) for key in keys: self.assertTrue(key in res) def test_network(self): keys = ('id', 'name', 'subnets', 'admin_state_up', 'status', 'tenant_id') self._view(keys, 'networks', 'network') def test_port(self): keys = ('id', 'network_id', 'mac_address', 'fixed_ips', 'device_id', 'admin_state_up', 'tenant_id', 'status') self._view(keys, 'ports', 'port') def test_subnet(self): keys = ('id', 'network_id', 'tenant_id', 'gateway_ip', 'ip_version', 'cidr', 'enable_dhcp') self._view(keys, 'subnets', 'subnet') class NotificationTest(APIv2TestBase): def _resource_op_notifier(self, opname, resource, expected_errors=False, notification_level='INFO'): initial_input = {resource: {'name': 'myname'}} instance = self.plugin.return_value instance.get_networks.return_value = initial_input instance.get_networks_count.return_value = 0 expected_code = exc.HTTPCreated.code with mock.patch.object(notifer_api, 'notify') as mynotifier: if opname == 'create': initial_input[resource]['tenant_id'] = _uuid() res = self.api.post_json( _get_path('networks'), initial_input, expected_errors) if opname == 'update': res = self.api.put_json( _get_path('networks', id=_uuid()), initial_input, expect_errors=expected_errors) expected_code = exc.HTTPOk.code if opname == 'delete': initial_input[resource]['tenant_id'] = _uuid() res = self.api.delete( _get_path('networks', id=_uuid()), expect_errors=expected_errors) expected_code = exc.HTTPNoContent.code expected = [mock.call(mock.ANY, 'network.' + cfg.CONF.host, resource + "." + opname + ".start", notification_level, mock.ANY), mock.call(mock.ANY, 'network.' + cfg.CONF.host, resource + "." + opname + ".end", notification_level, mock.ANY)] self.assertEqual(expected, mynotifier.call_args_list) self.assertEqual(res.status_int, expected_code) def test_network_create_notifer(self): self._resource_op_notifier('create', 'network') def test_network_delete_notifer(self): self._resource_op_notifier('delete', 'network') def test_network_update_notifer(self): self._resource_op_notifier('update', 'network') def test_network_create_notifer_with_log_level(self): cfg.CONF.set_override('default_notification_level', 'DEBUG') self._resource_op_notifier('create', 'network', notification_level='DEBUG') class QuotaTest(APIv2TestBase): def test_create_network_quota(self): cfg.CONF.set_override('quota_network', 1, group='QUOTAS') initial_input = {'network': {'name': 'net1', 'tenant_id': _uuid()}} full_input = {'network': {'admin_state_up': True, 'subnets': []}} full_input['network'].update(initial_input['network']) instance = self.plugin.return_value instance.get_networks_count.return_value = 1 res = self.api.post_json( _get_path('networks'), initial_input, expect_errors=True) instance.get_networks_count.assert_called_with(mock.ANY, filters=mock.ANY) self.assertTrue("Quota exceeded for resources" in res.json['QuantumError']) def test_create_network_quota_no_counts(self): cfg.CONF.set_override('quota_network', 1, group='QUOTAS') initial_input = {'network': {'name': 'net1', 'tenant_id': _uuid()}} full_input = {'network': {'admin_state_up': True, 'subnets': []}} full_input['network'].update(initial_input['network']) instance = self.plugin.return_value instance.get_networks_count.side_effect = ( q_exc.NotImplementedError()) instance.get_networks.return_value = ["foo"] res = self.api.post_json( _get_path('networks'), initial_input, expect_errors=True) instance.get_networks_count.assert_called_with(mock.ANY, filters=mock.ANY) self.assertTrue("Quota exceeded for resources" in res.json['QuantumError']) def test_create_network_quota_without_limit(self): cfg.CONF.set_override('quota_network', -1, group='QUOTAS') initial_input = {'network': {'name': 'net1', 'tenant_id': _uuid()}} instance = self.plugin.return_value instance.get_networks_count.return_value = 3 res = self.api.post_json( _get_path('networks'), initial_input) self.assertEqual(res.status_int, exc.HTTPCreated.code) class ExtensionTestCase(unittest.TestCase): # NOTE(jkoelker) This potentially leaks the mock object if the setUp # raises without being caught. Using unittest2 # or dropping 2.6 support so we can use addCleanup # will get around this. def setUp(self): plugin = 'quantum.quantum_plugin_base_v2.QuantumPluginBaseV2' # Ensure 'stale' patched copies of the plugin are never returned QuantumManager._instance = None # Ensure existing ExtensionManager is not used PluginAwareExtensionManager._instance = None # Save the global RESOURCE_ATTRIBUTE_MAP self.saved_attr_map = {} for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems(): self.saved_attr_map[resource] = attrs.copy() # Create the default configurations args = ['--config-file', etcdir('quantum.conf.test')] config.parse(args=args) # Update the plugin and extensions path cfg.CONF.set_override('core_plugin', plugin) cfg.CONF.set_override('api_extensions_path', EXTDIR) self._plugin_patcher = mock.patch(plugin, autospec=True) self.plugin = self._plugin_patcher.start() # Instantiate mock plugin and enable the V2attributes extension QuantumManager.get_plugin().supported_extension_aliases = ["v2attrs"] api = router.APIRouter() self.api = webtest.TestApp(api) def tearDown(self): self._plugin_patcher.stop() self.api = None self.plugin = None cfg.CONF.reset() # Restore the global RESOURCE_ATTRIBUTE_MAP attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map def test_extended_create(self): net_id = _uuid() initial_input = {'network': {'name': 'net1', 'tenant_id': _uuid(), 'v2attrs:something_else': "abc"}} data = {'network': {'admin_state_up': True, 'shared': False}} data['network'].update(initial_input['network']) return_value = {'subnets': [], 'status': "ACTIVE", 'id': net_id, 'v2attrs:something': "123"} return_value.update(data['network'].copy()) instance = self.plugin.return_value instance.create_network.return_value = return_value instance.get_networks_count.return_value = 0 res = self.api.post_json(_get_path('networks'), initial_input) instance.create_network.assert_called_with(mock.ANY, network=data) self.assertEqual(res.status_int, exc.HTTPCreated.code) self.assertTrue('network' in res.json) net = res.json['network'] self.assertEqual(net['id'], net_id) self.assertEqual(net['status'], "ACTIVE") self.assertEqual(net['v2attrs:something'], "123") self.assertFalse('v2attrs:something_else' in net) class TestSubresourcePlugin(): def get_network_dummies(self, context, network_id, filters=None, fields=None): return [] def get_network_dummy(self, context, id, network_id, fields=None): return {} def create_network_dummy(self, context, network_id, dummy): return {} def update_network_dummy(self, context, id, network_id, dummy): return {} def delete_network_dummy(self, context, id, network_id): return class FieldsTestCase(unittest.TestCase): def test_with_fields(self): path = '/?fields=4&foo=3&fields=2&bar=1' request = webob.Request.blank(path) expect_val = ['2', '4'] actual_val = base._fields(request) self.assertItemsEqual(actual_val, expect_val) def test_without_fields(self): path = '/?foo=4&bar=3&baz=2&qux=1' request = webob.Request.blank(path) self.assertListEqual([], base._fields(request)) class FiltersTestCase(unittest.TestCase): def test_all_fields(self): path = '/?fields=4&fields=3&fields=2&fields=1' request = webob.Request.blank(path) self.assertDictEqual({}, base._filters(request, None)) def test_blank_values(self): path = '/?foo=&bar=&baz=&qux=' request = webob.Request.blank(path) self.assertDictEqual({}, base._filters(request, {})) def test_no_attr_info(self): path = '/?foo=4&bar=3&baz=2&qux=1' request = webob.Request.blank(path) expect_val = {'foo': ['4'], 'bar': ['3'], 'baz': ['2'], 'qux': ['1']} actual_val = base._filters(request, {}) self.assertDictEqual(actual_val, expect_val) def test_attr_info_without_conversion(self): path = '/?foo=4&bar=3&baz=2&qux=1' request = webob.Request.blank(path) attr_info = {'foo': {'key': 'val'}} expect_val = {'foo': ['4'], 'bar': ['3'], 'baz': ['2'], 'qux': ['1']} actual_val = base._filters(request, attr_info) self.assertDictEqual(actual_val, expect_val) def test_attr_info_with_convert_list_to(self): path = '/?foo=key=4&bar=3&foo=key=2&qux=1' request = webob.Request.blank(path) attr_info = { 'foo': { 'convert_list_to': attributes.convert_kvp_list_to_dict, } } expect_val = {'foo': {'key': ['2', '4']}, 'bar': ['3'], 'qux': ['1']} actual_val = base._filters(request, attr_info) self.assertDictEqual(actual_val, expect_val) def test_attr_info_with_convert_to(self): path = '/?foo=4&bar=3&baz=2&qux=1' request = webob.Request.blank(path) attr_info = {'foo': {'convert_to': attributes.convert_to_int}} expect_val = {'foo': [4], 'bar': ['3'], 'baz': ['2'], 'qux': ['1']} actual_val = base._filters(request, attr_info) self.assertDictEqual(actual_val, expect_val) class CreateResourceTestCase(unittest.TestCase): def test_resource_creation(self): resource = base.create_resource('fakes', 'fake', None, {}) self.assertIsInstance(resource, webob.dec.wsgify)