919 lines
37 KiB
Python
919 lines
37 KiB
Python
# 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 unittest
|
|
|
|
import mock
|
|
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 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
|
|
|
|
|
|
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)
|
|
|
|
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):
|
|
|
|
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'), extra_environ=env)
|
|
self.assertTrue('networks' in res.json)
|
|
if not req_tenant_id or req_tenant_id == real_tenant_id:
|
|
# expect full list returned
|
|
self.assertEqual(len(res.json['networks']), 1)
|
|
output_dict = res.json['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.json['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_json(_get_path('networks'), 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")
|
|
|
|
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_json(_get_path('networks'), initial_input)
|
|
|
|
instance.create_network.assert_called_with(mock.ANY,
|
|
network=full_input)
|
|
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['admin_state_up'], True)
|
|
self.assertEqual(net['status'], "ACTIVE")
|
|
|
|
def test_create_no_keystone_env(self):
|
|
data = {'name': 'net1'}
|
|
res = self.api.post_json(_get_path('networks'), data,
|
|
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_json(_get_path('networks'), initial_input,
|
|
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_json(_get_path('networks'), data,
|
|
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_json(_get_path('networks'), data,
|
|
expect_errors=True)
|
|
self.assertEqual(res.status_int, exc.HTTPBadRequest.code)
|
|
|
|
def test_create_no_resource(self):
|
|
res = self.api.post_json(_get_path('networks'), dict(),
|
|
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_json(_get_path('ports'), data,
|
|
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_json(_get_path('networks'), data,
|
|
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_json(_get_path('networks'), data)
|
|
self.assertEqual(res.status_int, exc.HTTPCreated.code)
|
|
|
|
def test_create_bulk_no_networks(self):
|
|
data = {'networks': []}
|
|
res = self.api.post_json(_get_path('networks'), data,
|
|
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_json(_get_path('ports'), data,
|
|
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_json(_get_path('ports'), data,
|
|
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_json(_get_path('ports'), initial_input)
|
|
|
|
instance.create_port.assert_called_with(mock.ANY, port=full_input)
|
|
self.assertEqual(res.status_int, exc.HTTPCreated.code)
|
|
self.assertTrue('port' in res.json)
|
|
port = res.json['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_json(_get_path('networks'), 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.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()))
|
|
|
|
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()),
|
|
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()),
|
|
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_json(_get_path('networks',
|
|
id=uuidutils.generate_uuid()),
|
|
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_json(_get_path('networks', id=_uuid()), data,
|
|
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')
|
|
|
|
|
|
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(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
|