1134 lines
49 KiB
Python
1134 lines
49 KiB
Python
# 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 mock
|
|
from neutron_lib import constants as n_const
|
|
from neutron_lib import context
|
|
from neutron_lib.plugins import constants as plugin_constants
|
|
from neutron_lib.plugins import directory
|
|
from oslo_config import cfg
|
|
from oslo_db import exception as db_exc
|
|
from oslo_policy import policy as oslo_policy
|
|
from oslo_serialization import jsonutils
|
|
from oslo_utils import uuidutils
|
|
import pecan
|
|
from pecan import request
|
|
|
|
from neutron.api import extensions
|
|
from neutron.conf import quota as qconf
|
|
from neutron import manager
|
|
from neutron.pecan_wsgi.controllers import root as controllers
|
|
from neutron.pecan_wsgi.controllers import utils as controller_utils
|
|
from neutron import policy
|
|
from neutron.tests.common import helpers
|
|
from neutron.tests.functional.pecan_wsgi import test_functional
|
|
from neutron.tests.functional.pecan_wsgi import utils as pecan_utils
|
|
|
|
_SERVICE_PLUGIN_RESOURCE = 'serviceplugin'
|
|
_SERVICE_PLUGIN_COLLECTION = _SERVICE_PLUGIN_RESOURCE + 's'
|
|
_SERVICE_PLUGIN_INDEX_BODY = {_SERVICE_PLUGIN_COLLECTION: []}
|
|
|
|
|
|
class FakeServicePluginController(controller_utils.NeutronPecanController):
|
|
resource = _SERVICE_PLUGIN_RESOURCE
|
|
collection = _SERVICE_PLUGIN_COLLECTION
|
|
|
|
@pecan.expose(generic=True,
|
|
content_type='application/json',
|
|
template='json')
|
|
def index(self):
|
|
return _SERVICE_PLUGIN_INDEX_BODY
|
|
|
|
|
|
class TestRootController(test_functional.PecanFunctionalTest):
|
|
"""Test version listing on root URI."""
|
|
|
|
base_url = '/'
|
|
|
|
def setUp(self):
|
|
super(TestRootController, self).setUp()
|
|
self.setup_service_plugin()
|
|
self.plugin = directory.get_plugin()
|
|
self.ctx = context.get_admin_context()
|
|
|
|
def setup_service_plugin(self):
|
|
manager.NeutronManager.set_controller_for_resource(
|
|
_SERVICE_PLUGIN_COLLECTION,
|
|
FakeServicePluginController(_SERVICE_PLUGIN_COLLECTION,
|
|
_SERVICE_PLUGIN_RESOURCE))
|
|
|
|
def _test_method_returns_code(self, method, code=200):
|
|
api_method = getattr(self.app, method)
|
|
response = api_method(self.base_url, expect_errors=True)
|
|
self.assertEqual(response.status_int, code)
|
|
|
|
def test_get(self):
|
|
response = self.app.get(self.base_url)
|
|
self.assertEqual(response.status_int, 200)
|
|
json_body = jsonutils.loads(response.body)
|
|
versions = json_body.get('versions')
|
|
self.assertEqual(1, len(versions))
|
|
for (attr, value) in controllers.V2Controller.version_info.items():
|
|
self.assertIn(attr, versions[0])
|
|
self.assertEqual(value, versions[0][attr])
|
|
|
|
def test_methods(self):
|
|
self._test_method_returns_code('post', 405)
|
|
self._test_method_returns_code('patch', 405)
|
|
self._test_method_returns_code('delete', 405)
|
|
self._test_method_returns_code('head', 405)
|
|
self._test_method_returns_code('put', 405)
|
|
|
|
|
|
class TestV2Controller(TestRootController):
|
|
|
|
base_url = '/v2.0/'
|
|
|
|
def test_get(self):
|
|
"""Verify current version info are returned."""
|
|
response = self.app.get(self.base_url)
|
|
self.assertEqual(response.status_int, 200)
|
|
json_body = jsonutils.loads(response.body)
|
|
self.assertIn('resources', json_body)
|
|
self.assertIsInstance(json_body['resources'], list)
|
|
for r in json_body['resources']:
|
|
self.assertIn("links", r)
|
|
self.assertIn("name", r)
|
|
self.assertIn("collection", r)
|
|
self.assertIn(self.base_url, r['links'][0]['href'])
|
|
|
|
def test_get_no_trailing_slash(self):
|
|
response = self.app.get(self.base_url[:-1], expect_errors=True)
|
|
self.assertEqual(response.status_int, 404)
|
|
|
|
def test_routing_successs(self):
|
|
"""Test dispatch to controller for existing resource."""
|
|
response = self.app.get('%sports.json' % self.base_url)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
def test_routing_failure(self):
|
|
"""Test dispatch to controller for non-existing resource."""
|
|
response = self.app.get('%sidonotexist.json' % self.base_url,
|
|
expect_errors=True)
|
|
self.assertEqual(response.status_int, 404)
|
|
|
|
def test_methods(self):
|
|
self._test_method_returns_code('post', 405)
|
|
self._test_method_returns_code('put', 405)
|
|
self._test_method_returns_code('patch', 405)
|
|
self._test_method_returns_code('delete', 405)
|
|
self._test_method_returns_code('head', 405)
|
|
self._test_method_returns_code('delete', 405)
|
|
|
|
|
|
class TestExtensionsController(TestRootController):
|
|
"""Test extension listing and detail reporting."""
|
|
|
|
base_url = '/v2.0/extensions'
|
|
|
|
def _get_supported_extensions(self):
|
|
ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
|
|
return ext_mgr.get_supported_extension_aliases()
|
|
|
|
def test_index(self):
|
|
response = self.app.get(self.base_url)
|
|
self.assertEqual(response.status_int, 200)
|
|
json_body = jsonutils.loads(response.body)
|
|
returned_aliases = [ext['alias'] for ext in json_body['extensions']]
|
|
supported_extensions = self._get_supported_extensions()
|
|
self.assertEqual(supported_extensions, set(returned_aliases))
|
|
|
|
def test_get(self):
|
|
# Fetch any extension supported by plugins
|
|
test_alias = self._get_supported_extensions().pop()
|
|
response = self.app.get('%s/%s' % (self.base_url, test_alias))
|
|
self.assertEqual(response.status_int, 200)
|
|
json_body = jsonutils.loads(response.body)
|
|
self.assertEqual(test_alias, json_body['extension']['alias'])
|
|
|
|
def test_methods(self):
|
|
self._test_method_returns_code('post', 404)
|
|
self._test_method_returns_code('put', 404)
|
|
self._test_method_returns_code('patch', 404)
|
|
self._test_method_returns_code('delete', 404)
|
|
self._test_method_returns_code('head', 404)
|
|
self._test_method_returns_code('delete', 404)
|
|
|
|
|
|
class TestQuotasController(test_functional.PecanFunctionalTest):
|
|
"""Test quota management API controller."""
|
|
|
|
base_url = '/v2.0/quotas'
|
|
default_expected_limits = {
|
|
'network': qconf.DEFAULT_QUOTA_NETWORK,
|
|
'port': qconf.DEFAULT_QUOTA_PORT,
|
|
'subnet': qconf.DEFAULT_QUOTA_SUBNET}
|
|
|
|
def _verify_limits(self, response, limits):
|
|
for resource, limit in limits.items():
|
|
self.assertEqual(limit, response['quota'][resource])
|
|
|
|
def _verify_default_limits(self, response):
|
|
self._verify_limits(response, self.default_expected_limits)
|
|
|
|
def _verify_after_update(self, response, updated_limits):
|
|
expected_limits = self.default_expected_limits.copy()
|
|
expected_limits.update(updated_limits)
|
|
self._verify_limits(response, expected_limits)
|
|
|
|
def test_index_admin(self):
|
|
# NOTE(salv-orlando): The quota controller has an hardcoded check for
|
|
# admin-ness for this operation, which is supposed to return quotas for
|
|
# all tenants. Such check is "vestigial" from the home-grown WSGI and
|
|
# shall be removed
|
|
response = self.app.get('%s.json' % self.base_url,
|
|
headers={'X-Project-Id': 'admin',
|
|
'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
|
|
def test_index(self):
|
|
response = self.app.get('%s.json' % self.base_url, expect_errors=True)
|
|
self.assertEqual(403, response.status_int)
|
|
|
|
def test_get_admin(self):
|
|
response = self.app.get('%s/foo.json' % self.base_url,
|
|
headers={'X-Project-Id': 'admin',
|
|
'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
# As quota limits have not been updated, expect default values
|
|
json_body = jsonutils.loads(response.body)
|
|
self._verify_default_limits(json_body)
|
|
|
|
def test_get(self):
|
|
# It is not ok to access another tenant's limits
|
|
url = '%s/foo.json' % self.base_url
|
|
response = self.app.get(url, expect_errors=True)
|
|
self.assertEqual(403, response.status_int)
|
|
# It is however ok to retrieve your own limits
|
|
response = self.app.get(url, headers={'X-Project-Id': 'foo'})
|
|
self.assertEqual(200, response.status_int)
|
|
json_body = jsonutils.loads(response.body)
|
|
self._verify_default_limits(json_body)
|
|
|
|
def test_put_get_delete(self):
|
|
# PUT and DELETE actions are in the same test as a meaningful DELETE
|
|
# test would require a put anyway
|
|
url = '%s/foo.json' % self.base_url
|
|
response = self.app.put_json(url,
|
|
params={'quota': {'network': 99}},
|
|
headers={'X-Project-Id': 'admin',
|
|
'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
json_body = jsonutils.loads(response.body)
|
|
self._verify_after_update(json_body, {'network': 99})
|
|
|
|
response = self.app.get(url, headers={'X-Project-Id': 'foo'})
|
|
self.assertEqual(200, response.status_int)
|
|
json_body = jsonutils.loads(response.body)
|
|
self._verify_after_update(json_body, {'network': 99})
|
|
|
|
response = self.app.delete(url, headers={'X-Project-Id': 'admin',
|
|
'X-Roles': 'admin'})
|
|
self.assertEqual(204, response.status_int)
|
|
self.assertFalse(response.body)
|
|
# As DELETE does not return a body we need another GET
|
|
response = self.app.get(url, headers={'X-Project-Id': 'foo'})
|
|
self.assertEqual(200, response.status_int)
|
|
json_body = jsonutils.loads(response.body)
|
|
self._verify_default_limits(json_body)
|
|
|
|
def test_update_list_delete(self):
|
|
# PUT and DELETE actions are in the same test as a meaningful DELETE
|
|
# test would require a put anyway
|
|
url = '%s/foo.json' % self.base_url
|
|
response = self.app.put_json(url,
|
|
params={'quota': {'network': 99}},
|
|
headers={'X-Project-Id': 'admin',
|
|
'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
json_body = jsonutils.loads(response.body)
|
|
self._verify_after_update(json_body, {'network': 99})
|
|
|
|
response = self.app.get(self.base_url,
|
|
headers={'X-Project-Id': 'admin',
|
|
'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
json_body = jsonutils.loads(response.body)
|
|
found = False
|
|
for qs in json_body['quotas']:
|
|
if qs['tenant_id'] == 'foo':
|
|
found = True
|
|
self.assertTrue(found)
|
|
|
|
response = self.app.delete(url, headers={'X-Project-Id': 'admin',
|
|
'X-Roles': 'admin'})
|
|
self.assertEqual(204, response.status_int)
|
|
self.assertFalse(response.body)
|
|
response = self.app.get(self.base_url,
|
|
headers={'X-Project-Id': 'admin',
|
|
'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
json_body = jsonutils.loads(response.body)
|
|
for qs in json_body['quotas']:
|
|
self.assertNotEqual('foo', qs['tenant_id'])
|
|
|
|
def test_quotas_get_defaults(self):
|
|
response = self.app.get('%s/foo/default.json' % self.base_url,
|
|
headers={'X-Project-Id': 'admin',
|
|
'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
# As quota limits have not been updated, expect default values
|
|
json_body = jsonutils.loads(response.body)
|
|
self._verify_default_limits(json_body)
|
|
|
|
def test_get_tenant_info(self):
|
|
response = self.app.get('%s/tenant.json' % self.base_url,
|
|
headers={'X-Project-Id': 'admin',
|
|
'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
json_body = jsonutils.loads(response.body)
|
|
self.assertEqual('admin', json_body['tenant']['tenant_id'])
|
|
|
|
|
|
class TestResourceController(TestRootController):
|
|
"""Test generic controller"""
|
|
# TODO(salv-orlando): This test case must not explicitly test the 'port'
|
|
# resource. Also it should validate correct plugin/resource association
|
|
base_url = '/v2.0'
|
|
|
|
def setUp(self):
|
|
super(TestResourceController, self).setUp()
|
|
policy.init()
|
|
self.addCleanup(policy.reset)
|
|
self._gen_port()
|
|
|
|
def _gen_port(self):
|
|
network_id = self.plugin.create_network(context.get_admin_context(), {
|
|
'network':
|
|
{'name': 'pecannet', 'tenant_id': 'tenid', 'shared': False,
|
|
'admin_state_up': True, 'status': 'ACTIVE'}})['id']
|
|
self.port = self.plugin.create_port(context.get_admin_context(), {
|
|
'port':
|
|
{'tenant_id': 'tenid', 'network_id': network_id,
|
|
'fixed_ips': n_const.ATTR_NOT_SPECIFIED,
|
|
'mac_address': '00:11:22:33:44:55',
|
|
'admin_state_up': True, 'device_id': 'FF',
|
|
'device_owner': 'pecan', 'name': 'pecan'}})
|
|
|
|
def test_get(self):
|
|
response = self.app.get('/v2.0/ports.json')
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
def _check_item(self, expected, item):
|
|
for attribute in expected:
|
|
self.assertIn(attribute, item)
|
|
self.assertEqual(len(expected), len(item))
|
|
|
|
def _test_get_collection_with_fields_selector(self, fields=None):
|
|
fields = fields or []
|
|
query_params = ['fields=%s' % field for field in fields]
|
|
url = '/v2.0/ports.json'
|
|
if query_params:
|
|
url = '%s?%s' % (url, '&'.join(query_params))
|
|
list_resp = self.app.get(url, headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(200, list_resp.status_int)
|
|
for item in jsonutils.loads(list_resp.body).get('ports', []):
|
|
for field in fields:
|
|
self.assertIn(field, item)
|
|
if fields:
|
|
self.assertEqual(len(fields), len(item))
|
|
else:
|
|
for field in ('id', 'name', 'device_owner'):
|
|
self.assertIn(field, item)
|
|
|
|
def test_get_collection_with_multiple_fields_selector(self):
|
|
self._test_get_collection_with_fields_selector(fields=['id', 'name'])
|
|
|
|
def test_get_collection_with_single_fields_selector(self):
|
|
self._test_get_collection_with_fields_selector(fields=['name'])
|
|
|
|
def test_get_collection_without_fields_selector(self):
|
|
self._test_get_collection_with_fields_selector(fields=[])
|
|
|
|
def test_project_id_in_mandatory_fields(self):
|
|
# ports only specifies that tenant_id is mandatory, but project_id
|
|
# should still be passed to the plugin.
|
|
mock_get = mock.patch.object(self.plugin, 'get_ports',
|
|
return_value=[]).start()
|
|
self.app.get(
|
|
'/v2.0/ports.json?fields=id',
|
|
headers={'X-Project-Id': 'tenid'}
|
|
)
|
|
self.assertIn('project_id', mock_get.mock_calls[-1][2]['fields'])
|
|
|
|
def test_get_item_with_fields_selector(self):
|
|
item_resp = self.app.get(
|
|
'/v2.0/ports/%s.json?fields=id&fields=name' % self.port['id'],
|
|
headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(200, item_resp.status_int)
|
|
self._check_item(['id', 'name'],
|
|
jsonutils.loads(item_resp.body)['port'])
|
|
# Explicitly require an attribute which is also 'required_by_policy'.
|
|
# The attribute should not be stripped while generating the response
|
|
item_resp = self.app.get(
|
|
'/v2.0/ports/%s.json?fields=id&fields=tenant_id' % self.port['id'],
|
|
headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(200, item_resp.status_int)
|
|
self._check_item(['id', 'tenant_id'],
|
|
jsonutils.loads(item_resp.body)['port'])
|
|
|
|
def test_duped_and_empty_fields_stripped(self):
|
|
mock_get = mock.patch.object(self.plugin, 'get_ports',
|
|
return_value=[]).start()
|
|
self.app.get(
|
|
'/v2.0/ports.json?fields=id&fields=name&fields=&fields=name',
|
|
headers={'X-Project-Id': 'tenid'}
|
|
)
|
|
received = mock_get.mock_calls[-1][2]['fields']
|
|
self.assertNotIn('', received)
|
|
self.assertEqual(len(received), len(set(received)))
|
|
|
|
def test_post(self):
|
|
response = self.app.post_json(
|
|
'/v2.0/ports.json',
|
|
params={'port': {'network_id': self.port['network_id'],
|
|
'admin_state_up': True,
|
|
'tenant_id': 'tenid'}},
|
|
headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(response.status_int, 201)
|
|
|
|
def test_post_with_retry(self):
|
|
self._create_failed = False
|
|
orig = self.plugin.create_port
|
|
|
|
def new_create(*args, **kwargs):
|
|
if not self._create_failed:
|
|
self._create_failed = True
|
|
raise db_exc.RetryRequest(ValueError())
|
|
return orig(*args, **kwargs)
|
|
|
|
with mock.patch.object(self.plugin, 'create_port',
|
|
new=new_create):
|
|
response = self.app.post_json(
|
|
'/v2.0/ports.json',
|
|
params={'port': {'network_id': self.port['network_id'],
|
|
'admin_state_up': True,
|
|
'tenant_id': 'tenid'}},
|
|
headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(201, response.status_int)
|
|
|
|
def test_put(self):
|
|
response = self.app.put_json('/v2.0/ports/%s.json' % self.port['id'],
|
|
params={'port': {'name': 'test'}},
|
|
headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(response.status_int, 200)
|
|
json_body = jsonutils.loads(response.body)
|
|
self.assertEqual(1, len(json_body))
|
|
self.assertIn('port', json_body)
|
|
self.assertEqual('test', json_body['port']['name'])
|
|
self.assertEqual('tenid', json_body['port']['tenant_id'])
|
|
|
|
def test_delete(self):
|
|
response = self.app.delete('/v2.0/ports/%s.json' % self.port['id'],
|
|
headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(response.status_int, 204)
|
|
self.assertFalse(response.body)
|
|
|
|
def test_delete_disallows_body(self):
|
|
response = self.app.delete_json(
|
|
'/v2.0/ports/%s.json' % self.port['id'],
|
|
params={'port': {'name': 'test'}},
|
|
headers={'X-Project-Id': 'tenid'},
|
|
expect_errors=True)
|
|
self.assertEqual(response.status_int, 400)
|
|
|
|
def test_plugin_initialized(self):
|
|
self.assertIsNotNone(manager.NeutronManager._instance)
|
|
|
|
def test_methods(self):
|
|
self._test_method_returns_code('post', 405)
|
|
self._test_method_returns_code('put', 405)
|
|
self._test_method_returns_code('patch', 405)
|
|
self._test_method_returns_code('delete', 405)
|
|
self._test_method_returns_code('head', 405)
|
|
self._test_method_returns_code('delete', 405)
|
|
|
|
def test_post_with_empty_body(self):
|
|
response = self.app.post_json(
|
|
'/v2.0/ports.json',
|
|
headers={'X-Project-Id': 'tenid'},
|
|
params={},
|
|
expect_errors=True)
|
|
self.assertEqual(response.status_int, 400)
|
|
|
|
def test_post_with_unsupported_json_type(self):
|
|
response = self.app.post_json(
|
|
'/v2.0/ports.json',
|
|
headers={'X-Project-Id': 'tenid'},
|
|
params=[1, 2, 3],
|
|
expect_errors=True)
|
|
self.assertEqual(response.status_int, 400)
|
|
|
|
def test_bulk_create(self):
|
|
response = self.app.post_json(
|
|
'/v2.0/ports.json',
|
|
params={'ports': [{'network_id': self.port['network_id'],
|
|
'admin_state_up': True,
|
|
'tenant_id': 'tenid'},
|
|
{'network_id': self.port['network_id'],
|
|
'admin_state_up': True,
|
|
'tenant_id': 'tenid'}]
|
|
},
|
|
headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(response.status_int, 201)
|
|
json_body = jsonutils.loads(response.body)
|
|
self.assertIn('ports', json_body)
|
|
self.assertEqual(2, len(json_body['ports']))
|
|
|
|
def test_emulated_bulk_create(self):
|
|
self.plugin._FORCE_EMULATED_BULK = True
|
|
response = self.app.post_json(
|
|
'/v2.0/ports.json',
|
|
params={'ports': [{'network_id': self.port['network_id'],
|
|
'admin_state_up': True,
|
|
'tenant_id': 'tenid'},
|
|
{'network_id': self.port['network_id'],
|
|
'admin_state_up': True,
|
|
'tenant_id': 'tenid'}]
|
|
},
|
|
headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(response.status_int, 201)
|
|
json_body = jsonutils.loads(response.body)
|
|
self.assertIn('ports', json_body)
|
|
self.assertEqual(2, len(json_body['ports']))
|
|
|
|
def test_emulated_bulk_create_rollback(self):
|
|
self.plugin._FORCE_EMULATED_BULK = True
|
|
response = self.app.post_json(
|
|
'/v2.0/ports.json',
|
|
params={'ports': [{'network_id': self.port['network_id'],
|
|
'admin_state_up': True,
|
|
'tenant_id': 'tenid'},
|
|
{'network_id': self.port['network_id'],
|
|
'admin_state_up': True,
|
|
'tenant_id': 'tenid'},
|
|
{'network_id': 'bad_net_id',
|
|
'admin_state_up': True,
|
|
'tenant_id': 'tenid'}]
|
|
},
|
|
headers={'X-Project-Id': 'tenid'},
|
|
expect_errors=True)
|
|
self.assertEqual(response.status_int, 400)
|
|
response = self.app.get(
|
|
'/v2.0/ports.json',
|
|
headers={'X-Project-Id': 'tenid'})
|
|
# all ports should be rolled back from above so we are just left
|
|
# with the one created in setup
|
|
self.assertEqual(1, len(jsonutils.loads(response.body)['ports']))
|
|
|
|
def test_bulk_create_one_item(self):
|
|
response = self.app.post_json(
|
|
'/v2.0/ports.json',
|
|
params={'ports': [{'network_id': self.port['network_id'],
|
|
'admin_state_up': True,
|
|
'tenant_id': 'tenid'}]
|
|
},
|
|
headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(response.status_int, 201)
|
|
json_body = jsonutils.loads(response.body)
|
|
self.assertIn('ports', json_body)
|
|
self.assertEqual(1, len(json_body['ports']))
|
|
|
|
|
|
class TestPaginationAndSorting(test_functional.PecanFunctionalTest):
|
|
|
|
RESOURCE_COUNT = 6
|
|
|
|
def setUp(self):
|
|
super(TestPaginationAndSorting, self).setUp()
|
|
policy.init()
|
|
self.addCleanup(policy.reset)
|
|
self.plugin = directory.get_plugin()
|
|
self.ctx = context.get_admin_context()
|
|
self._create_networks(self.RESOURCE_COUNT)
|
|
self.networks = self._get_collection()['networks']
|
|
|
|
def _create_networks(self, count=1):
|
|
network_ids = []
|
|
for index in range(count):
|
|
network = {'name': 'pecannet-%d' % index, 'tenant_id': 'tenid',
|
|
'shared': False, 'admin_state_up': True,
|
|
'status': 'ACTIVE'}
|
|
network_id = self.plugin.create_network(
|
|
self.ctx, {'network': network})['id']
|
|
network_ids.append(network_id)
|
|
return network_ids
|
|
|
|
def _get_collection(self, collection=None, limit=None, marker=None,
|
|
fields=None, page_reverse=False, sort_key=None,
|
|
sort_dir=None):
|
|
collection = collection or 'networks'
|
|
fields = fields or []
|
|
query_params = []
|
|
if limit:
|
|
query_params.append('limit=%d' % limit)
|
|
if marker:
|
|
query_params.append('marker=%s' % marker)
|
|
if page_reverse:
|
|
query_params.append('page_reverse=True')
|
|
if sort_key:
|
|
query_params.append('sort_key=%s' % sort_key)
|
|
if sort_dir:
|
|
query_params.append('sort_dir=%s' % sort_dir)
|
|
query_params.extend(['%s%s' % ('fields=', field) for field in fields])
|
|
url = '/v2.0/%s.json' % collection
|
|
if query_params:
|
|
url = '%s?%s' % (url, '&'.join(query_params))
|
|
list_resp = self.app.get(url, headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(200, list_resp.status_int)
|
|
return list_resp.json
|
|
|
|
def _test_get_collection_with_pagination(self, expected_list,
|
|
collection=None,
|
|
limit=None, marker=None,
|
|
fields=None, page_reverse=False,
|
|
sort_key=None, sort_dir=None):
|
|
expected_list = expected_list or []
|
|
collection = collection or 'networks'
|
|
list_resp = self._get_collection(collection=collection, limit=limit,
|
|
marker=marker, fields=fields,
|
|
page_reverse=page_reverse,
|
|
sort_key=sort_key, sort_dir=sort_dir)
|
|
if limit and marker:
|
|
links_key = '%s_links' % collection
|
|
self.assertIn(links_key, list_resp)
|
|
if not fields or 'id' in fields:
|
|
list_resp_ids = [item['id'] for item in list_resp[collection]]
|
|
self.assertEqual(expected_list, list_resp_ids)
|
|
if fields:
|
|
for item in list_resp[collection]:
|
|
for field in fields:
|
|
self.assertIn(field, item)
|
|
|
|
def test_get_collection_with_pagination_limit(self):
|
|
self._test_get_collection_with_pagination([self.networks[0]['id']],
|
|
limit=1)
|
|
|
|
def test_get_collection_with_pagination_fields_no_pk(self):
|
|
self._test_get_collection_with_pagination([self.networks[0]['id']],
|
|
limit=1, fields=['name'])
|
|
|
|
def test_get_collection_with_pagination_limit_over_count(self):
|
|
expected_ids = [network['id'] for network in self.networks]
|
|
self._test_get_collection_with_pagination(
|
|
expected_ids, limit=self.RESOURCE_COUNT + 1)
|
|
|
|
def test_get_collection_with_pagination_marker(self):
|
|
marker = self.networks[2]['id']
|
|
expected_ids = [network['id'] for network in self.networks[3:]]
|
|
self._test_get_collection_with_pagination(expected_ids, limit=3,
|
|
marker=marker)
|
|
|
|
def test_get_collection_with_pagination_marker_without_limit(self):
|
|
marker = self.networks[2]['id']
|
|
expected_ids = [network['id'] for network in self.networks]
|
|
self._test_get_collection_with_pagination(expected_ids, marker=marker)
|
|
|
|
def test_get_collection_with_pagination_and_fields(self):
|
|
expected_ids = [network['id'] for network in self.networks[:2]]
|
|
self._test_get_collection_with_pagination(
|
|
expected_ids, limit=2, fields=['id', 'name'])
|
|
|
|
def test_get_collection_with_pagination_page_reverse(self):
|
|
marker = self.networks[2]['id']
|
|
expected_ids = [network['id'] for network in self.networks[:2]]
|
|
self._test_get_collection_with_pagination(expected_ids, limit=3,
|
|
marker=marker,
|
|
page_reverse=True)
|
|
|
|
def test_get_collection_with_sorting_desc(self):
|
|
nets = sorted(self.networks, key=lambda net: net['name'], reverse=True)
|
|
expected_ids = [network['id'] for network in nets]
|
|
self._test_get_collection_with_pagination(expected_ids,
|
|
sort_key='name',
|
|
sort_dir='desc')
|
|
|
|
def test_get_collection_with_sorting_asc(self):
|
|
nets = sorted(self.networks, key=lambda net: net['name'])
|
|
expected_ids = [network['id'] for network in nets]
|
|
self._test_get_collection_with_pagination(expected_ids,
|
|
sort_key='name',
|
|
sort_dir='asc')
|
|
|
|
|
|
class TestRequestProcessing(TestRootController):
|
|
|
|
def setUp(self):
|
|
super(TestRequestProcessing, self).setUp()
|
|
mock.patch('neutron.pecan_wsgi.hooks.notifier.registry').start()
|
|
# request.context is thread-local storage so it has to be accessed by
|
|
# the controller. We can capture it into a list here to assert on after
|
|
# the request finishes.
|
|
|
|
def capture_request_details(*args, **kwargs):
|
|
self.captured_context = request.context
|
|
self.request_params = kwargs
|
|
|
|
mock.patch('neutron.pecan_wsgi.controllers.resource.'
|
|
'CollectionsController.get',
|
|
side_effect=capture_request_details).start()
|
|
mock.patch('neutron.pecan_wsgi.controllers.resource.'
|
|
'CollectionsController.create',
|
|
side_effect=capture_request_details).start()
|
|
mock.patch('neutron.pecan_wsgi.controllers.resource.'
|
|
'ItemController.get',
|
|
side_effect=capture_request_details).start()
|
|
# TODO(kevinbenton): add context tests for X-Roles etc
|
|
|
|
def test_context_set_in_request(self):
|
|
self.app.get('/v2.0/ports.json',
|
|
headers={'X-Project-Id': 'tenant_id'})
|
|
self.assertEqual('tenant_id',
|
|
self.captured_context['neutron_context'].tenant_id)
|
|
|
|
def test_core_resource_identified(self):
|
|
self.app.get('/v2.0/ports.json')
|
|
self.assertEqual('port', self.captured_context['resource'])
|
|
self.assertEqual('ports', self.captured_context['collection'])
|
|
|
|
def test_lookup_identifies_resource_id(self):
|
|
# We now this will return a 404 but that's not the point as it is
|
|
# mocked
|
|
self.app.get('/v2.0/ports/reina.json')
|
|
self.assertEqual('port', self.captured_context['resource'])
|
|
self.assertEqual('ports', self.captured_context['collection'])
|
|
self.assertEqual('reina', self.captured_context['resource_id'])
|
|
|
|
def test_resource_processing_post(self):
|
|
self.app.post_json(
|
|
'/v2.0/networks.json',
|
|
params={'network': {'name': 'the_net',
|
|
'admin_state_up': True}},
|
|
headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual('network', self.captured_context['resource'])
|
|
self.assertEqual('networks', self.captured_context['collection'])
|
|
resources = self.captured_context['resources']
|
|
is_bulk = self.captured_context['is_bulk']
|
|
self.assertEqual(1, len(resources))
|
|
self.assertEqual('the_net', resources[0]['name'])
|
|
self.assertTrue(resources[0]['admin_state_up'])
|
|
self.assertFalse(is_bulk)
|
|
|
|
def test_resource_processing_post_bulk(self):
|
|
self.app.post_json(
|
|
'/v2.0/networks.json',
|
|
params={'networks': [{'name': 'the_net_1',
|
|
'admin_state_up': True},
|
|
{'name': 'the_net_2',
|
|
'admin_state_up': False}]},
|
|
headers={'X-Project-Id': 'tenid'})
|
|
resources = self.captured_context['resources']
|
|
is_bulk = self.captured_context['is_bulk']
|
|
self.assertEqual(2, len(resources))
|
|
self.assertTrue(resources[0]['admin_state_up'])
|
|
self.assertEqual('the_net_1', resources[0]['name'])
|
|
self.assertFalse(resources[1]['admin_state_up'])
|
|
self.assertEqual('the_net_2', resources[1]['name'])
|
|
self.assertTrue(is_bulk)
|
|
|
|
def test_resource_processing_post_bulk_one_item(self):
|
|
self.app.post_json(
|
|
'/v2.0/networks.json',
|
|
params={'networks': [{'name': 'the_net_1',
|
|
'admin_state_up': True}]},
|
|
headers={'X-Project-Id': 'tenid'})
|
|
resources = self.captured_context['resources']
|
|
is_bulk = self.captured_context['is_bulk']
|
|
self.assertEqual(1, len(resources))
|
|
self.assertTrue(is_bulk)
|
|
|
|
def test_resource_processing_post_unknown_attribute_returns_400(self):
|
|
response = self.app.post_json(
|
|
'/v2.0/networks.json',
|
|
params={'network': {'name': 'the_net',
|
|
'alien': 'E.T.',
|
|
'admin_state_up': True}},
|
|
headers={'X-Project-Id': 'tenid'},
|
|
expect_errors=True)
|
|
self.assertEqual(400, response.status_int)
|
|
|
|
def test_resource_processing_post_validation_error_returns_400(self):
|
|
response = self.app.post_json(
|
|
'/v2.0/networks.json',
|
|
params={'network': {'name': 'the_net',
|
|
'admin_state_up': 'invalid_value'}},
|
|
headers={'X-Project-Id': 'tenid'},
|
|
expect_errors=True)
|
|
self.assertEqual(400, response.status_int)
|
|
|
|
def test_service_plugin_identified(self):
|
|
# TODO(kevinbenton): fix the unit test setup to include an l3 plugin
|
|
self.skipTest("A dummy l3 plugin needs to be setup")
|
|
self.app.get('/v2.0/routers.json')
|
|
self.assertEqual('router', self.req_stash['resource_type'])
|
|
# make sure the core plugin was identified as the handler for ports
|
|
self.assertEqual(
|
|
directory.get_plugin(plugin_constants.L3),
|
|
self.req_stash['plugin'])
|
|
|
|
def test_service_plugin_uri(self):
|
|
nm = manager.NeutronManager.get_instance()
|
|
nm.path_prefix_resource_mappings['dummy'] = [
|
|
_SERVICE_PLUGIN_COLLECTION]
|
|
response = self.do_request('/v2.0/dummy/serviceplugins.json')
|
|
self.assertEqual(200, response.status_int)
|
|
self.assertEqual(_SERVICE_PLUGIN_INDEX_BODY, response.json_body)
|
|
|
|
|
|
class TestRouterController(TestResourceController):
|
|
"""Specialized tests for the router resource controller
|
|
|
|
This test class adds tests specific for the router controller in
|
|
order to verify the 'member_action' functionality, which this
|
|
controller uses for adding and removing router interfaces.
|
|
"""
|
|
|
|
def setUp(self):
|
|
cfg.CONF.set_override(
|
|
'service_plugins',
|
|
['neutron.services.l3_router.l3_router_plugin.L3RouterPlugin',
|
|
'neutron.services.flavors.flavors_plugin.FlavorsPlugin'])
|
|
super(TestRouterController, self).setUp()
|
|
policy.init()
|
|
self.addCleanup(policy.reset)
|
|
plugin = directory.get_plugin()
|
|
ctx = context.get_admin_context()
|
|
l3_plugin = directory.get_plugin(plugin_constants.L3)
|
|
network_id = pecan_utils.create_network(ctx, plugin)['id']
|
|
self.subnet = pecan_utils.create_subnet(ctx, plugin, network_id)
|
|
self.router = pecan_utils.create_router(ctx, l3_plugin)
|
|
|
|
def test_member_actions_processing(self):
|
|
response = self.app.put_json(
|
|
'/v2.0/routers/%s/add_router_interface.json' % self.router['id'],
|
|
params={'subnet_id': self.subnet['id']},
|
|
headers={'X-Project-Id': 'tenid'})
|
|
self.assertEqual(200, response.status_int)
|
|
|
|
def test_non_existing_member_action_returns_404(self):
|
|
response = self.app.put_json(
|
|
'/v2.0/routers/%s/do_meh.json' % self.router['id'],
|
|
params={'subnet_id': 'doesitevenmatter'},
|
|
headers={'X-Project-Id': 'tenid'},
|
|
expect_errors=True)
|
|
self.assertEqual(404, response.status_int)
|
|
|
|
def test_unsupported_method_member_action(self):
|
|
response = self.app.post_json(
|
|
'/v2.0/routers/%s/add_router_interface.json' % self.router['id'],
|
|
params={'subnet_id': self.subnet['id']},
|
|
headers={'X-Project-Id': 'tenid'},
|
|
expect_errors=True)
|
|
self.assertEqual(405, response.status_int)
|
|
|
|
response = self.app.get(
|
|
'/v2.0/routers/%s/add_router_interface.json' % self.router['id'],
|
|
headers={'X-Project-Id': 'tenid'},
|
|
expect_errors=True)
|
|
self.assertEqual(405, response.status_int)
|
|
|
|
|
|
class TestDHCPAgentShimControllers(test_functional.PecanFunctionalTest):
|
|
|
|
def setUp(self):
|
|
super(TestDHCPAgentShimControllers, self).setUp()
|
|
policy.init()
|
|
policy._ENFORCER.set_rules(
|
|
oslo_policy.Rules.from_dict(
|
|
{'get_dhcp-agents': 'role:admin',
|
|
'get_dhcp-networks': 'role:admin',
|
|
'create_dhcp-networks': 'role:admin',
|
|
'delete_dhcp-networks': 'role:admin'}),
|
|
overwrite=False)
|
|
plugin = directory.get_plugin()
|
|
ctx = context.get_admin_context()
|
|
self.network = pecan_utils.create_network(ctx, plugin)
|
|
self.agent = helpers.register_dhcp_agent()
|
|
# NOTE(blogan): Not sending notifications because this test is for
|
|
# testing the shim controllers
|
|
plugin.agent_notifiers[n_const.AGENT_TYPE_DHCP] = None
|
|
|
|
def test_list_dhcp_agents_hosting_network(self):
|
|
response = self.app.get(
|
|
'/v2.0/networks/%s/dhcp-agents.json' % self.network['id'],
|
|
headers={'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
|
|
def test_list_networks_on_dhcp_agent(self):
|
|
response = self.app.get(
|
|
'/v2.0/agents/%s/dhcp-networks.json' % self.agent.id,
|
|
headers={'X-Project-Id': 'tenid', 'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
|
|
def test_add_remove_dhcp_agent(self):
|
|
headers = {'X-Project-Id': 'tenid', 'X-Roles': 'admin'}
|
|
self.app.post_json(
|
|
'/v2.0/agents/%s/dhcp-networks.json' % self.agent.id,
|
|
headers=headers, params={'network_id': self.network['id']})
|
|
response = self.app.get(
|
|
'/v2.0/networks/%s/dhcp-agents.json' % self.network['id'],
|
|
headers=headers)
|
|
self.assertIn(self.agent.id,
|
|
[a['id'] for a in response.json['agents']])
|
|
self.app.delete('/v2.0/agents/%(a)s/dhcp-networks/%(n)s.json' % {
|
|
'a': self.agent.id, 'n': self.network['id']}, headers=headers)
|
|
response = self.app.get(
|
|
'/v2.0/networks/%s/dhcp-agents.json' % self.network['id'],
|
|
headers=headers)
|
|
self.assertNotIn(self.agent.id,
|
|
[a['id'] for a in response.json['agents']])
|
|
|
|
|
|
class TestL3AgentShimControllers(test_functional.PecanFunctionalTest):
|
|
|
|
def setUp(self):
|
|
cfg.CONF.set_override(
|
|
'service_plugins',
|
|
['neutron.services.l3_router.l3_router_plugin.L3RouterPlugin',
|
|
'neutron.services.flavors.flavors_plugin.FlavorsPlugin'])
|
|
super(TestL3AgentShimControllers, self).setUp()
|
|
policy.init()
|
|
policy._ENFORCER.set_rules(
|
|
oslo_policy.Rules.from_dict(
|
|
{'get_l3-agents': 'role:admin',
|
|
'get_l3-routers': 'role:admin'}),
|
|
overwrite=False)
|
|
ctx = context.get_admin_context()
|
|
l3_plugin = directory.get_plugin(plugin_constants.L3)
|
|
self.router = pecan_utils.create_router(ctx, l3_plugin)
|
|
self.agent = helpers.register_l3_agent()
|
|
# NOTE(blogan): Not sending notifications because this test is for
|
|
# testing the shim controllers
|
|
l3_plugin.agent_notifiers[n_const.AGENT_TYPE_L3] = None
|
|
|
|
def test_list_l3_agents_hosting_router(self):
|
|
response = self.app.get(
|
|
'/v2.0/routers/%s/l3-agents.json' % self.router['id'],
|
|
headers={'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
|
|
def test_list_routers_on_l3_agent(self):
|
|
response = self.app.get(
|
|
'/v2.0/agents/%s/l3-routers.json' % self.agent.id,
|
|
headers={'X-Roles': 'admin'})
|
|
self.assertEqual(200, response.status_int)
|
|
|
|
def test_add_remove_l3_agent(self):
|
|
headers = {'X-Project-Id': 'tenid', 'X-Roles': 'admin'}
|
|
response = self.app.post_json(
|
|
'/v2.0/agents/%s/l3-routers.json' % self.agent.id,
|
|
headers=headers, params={'router_id': self.router['id']})
|
|
self.assertEqual(201, response.status_int)
|
|
response = self.app.get(
|
|
'/v2.0/routers/%s/l3-agents.json' % self.router['id'],
|
|
headers=headers)
|
|
self.assertIn(self.agent.id,
|
|
[a['id'] for a in response.json['agents']])
|
|
response = self.app.delete(
|
|
'/v2.0/agents/%(a)s/l3-routers/%(n)s.json' % {
|
|
'a': self.agent.id, 'n': self.router['id']}, headers=headers)
|
|
self.assertEqual(204, response.status_int)
|
|
self.assertFalse(response.body)
|
|
response = self.app.get(
|
|
'/v2.0/routers/%s/l3-agents.json' % self.router['id'],
|
|
headers=headers)
|
|
self.assertNotIn(self.agent.id,
|
|
[a['id'] for a in response.json['agents']])
|
|
|
|
|
|
class TestShimControllers(test_functional.PecanFunctionalTest):
|
|
|
|
def setUp(self):
|
|
fake_ext = pecan_utils.FakeExtension()
|
|
fake_plugin = pecan_utils.FakePlugin()
|
|
plugins = {pecan_utils.FakePlugin.PLUGIN_TYPE: fake_plugin}
|
|
new_extensions = {fake_ext.get_alias(): fake_ext}
|
|
super(TestShimControllers, self).setUp(
|
|
service_plugins=plugins, extensions=new_extensions)
|
|
policy.init()
|
|
policy._ENFORCER.set_rules(
|
|
oslo_policy.Rules.from_dict(
|
|
{'get_meh_meh': '',
|
|
'get_meh_mehs': '',
|
|
'get_fake_subresources': ''}),
|
|
overwrite=False)
|
|
self.addCleanup(policy.reset)
|
|
|
|
def test_hyphenated_resource_controller_not_shimmed(self):
|
|
collection = pecan_utils.FakeExtension.HYPHENATED_COLLECTION.replace(
|
|
'_', '-')
|
|
resource = pecan_utils.FakeExtension.HYPHENATED_RESOURCE
|
|
url = '/v2.0/{}/something.json'.format(collection)
|
|
resp = self.app.get(url)
|
|
self.assertEqual(200, resp.status_int)
|
|
self.assertEqual({resource: {'fake': 'something'}}, resp.json)
|
|
|
|
def test_hyphenated_collection_controller_not_shimmed(self):
|
|
body_collection = pecan_utils.FakeExtension.HYPHENATED_COLLECTION
|
|
uri_collection = body_collection.replace('_', '-')
|
|
url = '/v2.0/{}.json'.format(uri_collection)
|
|
resp = self.app.get(url)
|
|
self.assertEqual(200, resp.status_int)
|
|
self.assertEqual({body_collection: [{'fake': 'fake'}]}, resp.json)
|
|
|
|
def test_hyphenated_collection_subresource_controller_not_shimmed(self):
|
|
body_collection = pecan_utils.FakeExtension.HYPHENATED_COLLECTION
|
|
uri_collection = body_collection.replace('_', '-')
|
|
# there is only one subresource so far
|
|
sub_resource_collection = (
|
|
pecan_utils.FakeExtension.FAKE_SUB_RESOURCE_COLLECTION)
|
|
temp_id = uuidutils.generate_uuid()
|
|
url = '/v2.0/{0}/{1}/{2}'.format(
|
|
uri_collection,
|
|
temp_id,
|
|
sub_resource_collection.replace('_', '-'))
|
|
resp = self.app.get(url)
|
|
self.assertEqual(200, resp.status_int)
|
|
self.assertEqual({sub_resource_collection: {'foo': temp_id}},
|
|
resp.json)
|
|
|
|
|
|
class TestMemberActionController(test_functional.PecanFunctionalTest):
|
|
def setUp(self):
|
|
fake_ext = pecan_utils.FakeExtension()
|
|
fake_plugin = pecan_utils.FakePlugin()
|
|
plugins = {pecan_utils.FakePlugin.PLUGIN_TYPE: fake_plugin}
|
|
new_extensions = {fake_ext.get_alias(): fake_ext}
|
|
super(TestMemberActionController, self).setUp(
|
|
service_plugins=plugins, extensions=new_extensions)
|
|
hyphen_collection = pecan_utils.FakeExtension.HYPHENATED_COLLECTION
|
|
self.collection = hyphen_collection.replace('_', '-')
|
|
|
|
def test_get_member_action_controller(self):
|
|
url = '/v2.0/{}/something/boo_meh.json'.format(self.collection)
|
|
resp = self.app.get(url)
|
|
self.assertEqual(200, resp.status_int)
|
|
self.assertEqual({'boo_yah': 'something'}, resp.json)
|
|
|
|
def test_put_member_action_controller(self):
|
|
url = '/v2.0/{}/something/put_meh.json'.format(self.collection)
|
|
resp = self.app.put_json(url, params={'it_matters_not': 'ok'})
|
|
self.assertEqual(200, resp.status_int)
|
|
self.assertEqual({'poo_yah': 'something'}, resp.json)
|
|
|
|
def test_get_member_action_does_not_exist(self):
|
|
url = '/v2.0/{}/something/are_you_still_there.json'.format(
|
|
self.collection)
|
|
resp = self.app.get(url, expect_errors=True)
|
|
self.assertEqual(404, resp.status_int)
|
|
|
|
def test_put_member_action_does_not_exist(self):
|
|
url = '/v2.0/{}/something/are_you_still_there.json'.format(
|
|
self.collection)
|
|
resp = self.app.put_json(url, params={'it_matters_not': 'ok'},
|
|
expect_errors=True)
|
|
self.assertEqual(404, resp.status_int)
|
|
|
|
def test_put_on_get_member_action(self):
|
|
url = '/v2.0/{}/something/boo_meh.json'.format(self.collection)
|
|
resp = self.app.put_json(url, params={'it_matters_not': 'ok'},
|
|
expect_errors=True)
|
|
self.assertEqual(405, resp.status_int)
|
|
|
|
def test_get_on_put_member_action(self):
|
|
url = '/v2.0/{}/something/put_meh.json'.format(self.collection)
|
|
resp = self.app.get(url, expect_errors=True)
|
|
self.assertEqual(405, resp.status_int)
|
|
|
|
|
|
class TestParentSubresourceController(test_functional.PecanFunctionalTest):
|
|
def setUp(self):
|
|
fake_ext = pecan_utils.FakeExtension()
|
|
fake_plugin = pecan_utils.FakePlugin()
|
|
plugins = {pecan_utils.FakePlugin.PLUGIN_TYPE: fake_plugin}
|
|
new_extensions = {fake_ext.get_alias(): fake_ext}
|
|
super(TestParentSubresourceController, self).setUp(
|
|
service_plugins=plugins, extensions=new_extensions)
|
|
policy.init()
|
|
policy._ENFORCER.set_rules(
|
|
oslo_policy.Rules.from_dict(
|
|
{'get_fake_duplicate': '',
|
|
'get_meh_meh_fake_duplicate': ''}),
|
|
overwrite=False)
|
|
self.addCleanup(policy.reset)
|
|
hyphen_collection = pecan_utils.FakeExtension.HYPHENATED_COLLECTION
|
|
self.collection = hyphen_collection.replace('_', '-')
|
|
self.fake_collection = (pecan_utils.FakeExtension.
|
|
FAKE_PARENT_SUBRESOURCE_COLLECTION)
|
|
|
|
def test_get_duplicate_parent_resource(self):
|
|
url = '/v2.0/{}'.format(self.fake_collection)
|
|
resp = self.app.get(url)
|
|
self.assertEqual(200, resp.status_int)
|
|
self.assertEqual({'fake_duplicates': [{'fake': 'fakeduplicates'}]},
|
|
resp.json)
|
|
|
|
def test_get_duplicate_parent_resource_item(self):
|
|
url = '/v2.0/{}/something'.format(self.fake_collection)
|
|
resp = self.app.get(url)
|
|
self.assertEqual(200, resp.status_int)
|
|
self.assertEqual({'fake_duplicate': {'fake': 'something'}}, resp.json)
|
|
|
|
def test_get_parent_resource_and_duplicate_subresources(self):
|
|
url = '/v2.0/{0}/something/{1}'.format(self.collection,
|
|
self.fake_collection)
|
|
resp = self.app.get(url)
|
|
self.assertEqual(200, resp.status_int)
|
|
self.assertEqual({'fake_duplicates': [{'fake': 'something'}]},
|
|
resp.json)
|
|
|
|
def test_get_child_resource_policy_check(self):
|
|
policy.reset()
|
|
policy.init()
|
|
policy._ENFORCER.set_rules(
|
|
oslo_policy.Rules.from_dict(
|
|
{'get_meh_meh_fake_duplicate': ''}
|
|
)
|
|
)
|
|
url = '/v2.0/{0}/something/{1}'.format(self.collection,
|
|
self.fake_collection)
|
|
resp = self.app.get(url)
|
|
self.assertEqual(200, resp.status_int)
|
|
self.assertEqual({'fake_duplicates': [{'fake': 'something'}]},
|
|
resp.json)
|
|
|
|
|
|
class TestExcludeAttributePolicy(test_functional.PecanFunctionalTest):
|
|
|
|
def setUp(self):
|
|
super(TestExcludeAttributePolicy, self).setUp()
|
|
policy.init()
|
|
self.addCleanup(policy.reset)
|
|
plugin = directory.get_plugin()
|
|
ctx = context.get_admin_context()
|
|
self.network_id = pecan_utils.create_network(ctx, plugin)['id']
|
|
mock.patch('neutron.pecan_wsgi.controllers.resource.'
|
|
'CollectionsController.get').start()
|
|
|
|
def test_get_networks(self):
|
|
response = self.app.get('/v2.0/networks/%s.json' % self.network_id,
|
|
headers={'X-Project-Id': 'tenid'})
|
|
json_body = jsonutils.loads(response.body)
|
|
self.assertEqual(response.status_int, 200)
|
|
self.assertEqual('tenid', json_body['network']['project_id'])
|
|
self.assertEqual('tenid', json_body['network']['tenant_id'])
|