"Fossies" - the Fresh Open Source Software Archive

Member "neutron-14.0.3/neutron/tests/functional/pecan_wsgi/test_controllers.py" (22 Oct 2019, 51921 Bytes) of package /linux/misc/openstack/neutron-14.0.3.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_controllers.py": 14.0.2_vs_14.0.3.

    1 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    2 #    not use this file except in compliance with the License. You may obtain
    3 #    a copy of the License at
    4 #
    5 #         http://www.apache.org/licenses/LICENSE-2.0
    6 #
    7 #    Unless required by applicable law or agreed to in writing, software
    8 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    9 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   10 #    License for the specific language governing permissions and limitations
   11 #    under the License.
   12 
   13 import mock
   14 from neutron_lib import constants as n_const
   15 from neutron_lib import context
   16 from neutron_lib.plugins import constants as plugin_constants
   17 from neutron_lib.plugins import directory
   18 from oslo_config import cfg
   19 from oslo_db import exception as db_exc
   20 from oslo_policy import policy as oslo_policy
   21 from oslo_serialization import jsonutils
   22 from oslo_utils import uuidutils
   23 import pecan
   24 from pecan import request
   25 
   26 from neutron.api import extensions
   27 from neutron.conf import quota as qconf
   28 from neutron import manager
   29 from neutron.pecan_wsgi.controllers import root as controllers
   30 from neutron.pecan_wsgi.controllers import utils as controller_utils
   31 from neutron import policy
   32 from neutron.tests.common import helpers
   33 from neutron.tests.functional.pecan_wsgi import test_functional
   34 from neutron.tests.functional.pecan_wsgi import utils as pecan_utils
   35 from neutron.tests.unit import dummy_plugin
   36 
   37 
   38 _SERVICE_PLUGIN_RESOURCE = 'serviceplugin'
   39 _SERVICE_PLUGIN_COLLECTION = _SERVICE_PLUGIN_RESOURCE + 's'
   40 _SERVICE_PLUGIN_INDEX_BODY = {_SERVICE_PLUGIN_COLLECTION: []}
   41 
   42 
   43 class FakeServicePluginController(controller_utils.NeutronPecanController):
   44     resource = _SERVICE_PLUGIN_RESOURCE
   45     collection = _SERVICE_PLUGIN_COLLECTION
   46 
   47     @pecan.expose(generic=True,
   48                   content_type='application/json',
   49                   template='json')
   50     def index(self):
   51         return _SERVICE_PLUGIN_INDEX_BODY
   52 
   53 
   54 class TestRootController(test_functional.PecanFunctionalTest):
   55     """Test version listing on root URI."""
   56 
   57     base_url = '/'
   58 
   59     def setUp(self):
   60         super(TestRootController, self).setUp()
   61         self.setup_service_plugin()
   62         self.plugin = directory.get_plugin()
   63         self.ctx = context.get_admin_context()
   64 
   65     def setup_service_plugin(self):
   66         manager.NeutronManager.set_controller_for_resource(
   67             _SERVICE_PLUGIN_COLLECTION,
   68             FakeServicePluginController(_SERVICE_PLUGIN_COLLECTION,
   69                                         _SERVICE_PLUGIN_RESOURCE,
   70                                         resource_info={'foo': {}}))
   71 
   72     def _test_method_returns_code(self, method, code=200):
   73         api_method = getattr(self.app, method)
   74         response = api_method(self.base_url, expect_errors=True)
   75         self.assertEqual(response.status_int, code)
   76 
   77     def test_get(self):
   78         response = self.app.get(self.base_url)
   79         self.assertEqual(response.status_int, 200)
   80         json_body = jsonutils.loads(response.body)
   81         versions = json_body.get('versions')
   82         self.assertEqual(1, len(versions))
   83         for (attr, value) in controllers.V2Controller.version_info.items():
   84             self.assertIn(attr, versions[0])
   85             self.assertEqual(value, versions[0][attr])
   86 
   87     def test_methods(self):
   88         self._test_method_returns_code('post', 405)
   89         self._test_method_returns_code('patch', 405)
   90         self._test_method_returns_code('delete', 405)
   91         self._test_method_returns_code('head', 405)
   92         self._test_method_returns_code('put', 405)
   93 
   94 
   95 class TestV2Controller(TestRootController):
   96 
   97     base_url = '/v2.0/'
   98 
   99     def test_get(self):
  100         """Verify current version info are returned."""
  101         response = self.app.get(self.base_url)
  102         self.assertEqual(response.status_int, 200)
  103         json_body = jsonutils.loads(response.body)
  104         self.assertIn('resources', json_body)
  105         self.assertIsInstance(json_body['resources'], list)
  106         for r in json_body['resources']:
  107             self.assertIn("links", r)
  108             self.assertIn("name", r)
  109             self.assertIn("collection", r)
  110             self.assertIn(self.base_url, r['links'][0]['href'])
  111 
  112     def test_get_no_trailing_slash(self):
  113         response = self.app.get(self.base_url[:-1], expect_errors=True)
  114         self.assertEqual(response.status_int, 404)
  115 
  116     def test_routing_successs(self):
  117         """Test dispatch to controller for existing resource."""
  118         response = self.app.get('%sports.json' % self.base_url)
  119         self.assertEqual(response.status_int, 200)
  120 
  121     def test_routing_failure(self):
  122         """Test dispatch to controller for non-existing resource."""
  123         response = self.app.get('%sidonotexist.json' % self.base_url,
  124                                 expect_errors=True)
  125         self.assertEqual(response.status_int, 404)
  126 
  127     def test_methods(self):
  128         self._test_method_returns_code('post', 405)
  129         self._test_method_returns_code('put', 405)
  130         self._test_method_returns_code('patch', 405)
  131         self._test_method_returns_code('delete', 405)
  132         self._test_method_returns_code('head', 405)
  133         self._test_method_returns_code('delete', 405)
  134 
  135 
  136 class TestExtensionsController(TestRootController):
  137     """Test extension listing and detail reporting."""
  138 
  139     base_url = '/v2.0/extensions'
  140 
  141     def _get_supported_extensions(self):
  142         ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
  143         return ext_mgr.get_supported_extension_aliases()
  144 
  145     def test_index(self):
  146         response = self.app.get(self.base_url)
  147         self.assertEqual(response.status_int, 200)
  148         json_body = jsonutils.loads(response.body)
  149         returned_aliases = [ext['alias'] for ext in json_body['extensions']]
  150         supported_extensions = self._get_supported_extensions()
  151         self.assertEqual(supported_extensions, set(returned_aliases))
  152 
  153     def test_get(self):
  154         # Fetch any extension supported by plugins
  155         test_alias = self._get_supported_extensions().pop()
  156         response = self.app.get('%s/%s' % (self.base_url, test_alias))
  157         self.assertEqual(response.status_int, 200)
  158         json_body = jsonutils.loads(response.body)
  159         self.assertEqual(test_alias, json_body['extension']['alias'])
  160 
  161     def test_methods(self):
  162         self._test_method_returns_code('post', 404)
  163         self._test_method_returns_code('put', 404)
  164         self._test_method_returns_code('patch', 404)
  165         self._test_method_returns_code('delete', 404)
  166         self._test_method_returns_code('head', 404)
  167         self._test_method_returns_code('delete', 404)
  168 
  169 
  170 class TestQuotasController(test_functional.PecanFunctionalTest):
  171     """Test quota management API controller."""
  172 
  173     base_url = '/v2.0/quotas'
  174     default_expected_limits = {
  175         'network': qconf.DEFAULT_QUOTA_NETWORK,
  176         'port': qconf.DEFAULT_QUOTA_PORT,
  177         'subnet': qconf.DEFAULT_QUOTA_SUBNET}
  178 
  179     def _verify_limits(self, response, limits):
  180         for resource, limit in limits.items():
  181             self.assertEqual(limit, response['quota'][resource])
  182 
  183     def _verify_default_limits(self, response):
  184         self._verify_limits(response, self.default_expected_limits)
  185 
  186     def _verify_after_update(self, response, updated_limits):
  187         expected_limits = self.default_expected_limits.copy()
  188         expected_limits.update(updated_limits)
  189         self._verify_limits(response, expected_limits)
  190 
  191     def test_index_admin(self):
  192         # NOTE(salv-orlando): The quota controller has an hardcoded check for
  193         # admin-ness for this operation, which is supposed to return quotas for
  194         # all tenants. Such check is "vestigial" from the home-grown WSGI and
  195         # shall be removed
  196         response = self.app.get('%s.json' % self.base_url,
  197                                 headers={'X-Project-Id': 'admin',
  198                                          'X-Roles': 'admin'})
  199         self.assertEqual(200, response.status_int)
  200 
  201     def test_index(self):
  202         response = self.app.get('%s.json' % self.base_url, expect_errors=True)
  203         self.assertEqual(403, response.status_int)
  204 
  205     def test_get_admin(self):
  206         response = self.app.get('%s/foo.json' % self.base_url,
  207                                 headers={'X-Project-Id': 'admin',
  208                                          'X-Roles': 'admin'})
  209         self.assertEqual(200, response.status_int)
  210         # As quota limits have not been updated, expect default values
  211         json_body = jsonutils.loads(response.body)
  212         self._verify_default_limits(json_body)
  213 
  214     def test_get(self):
  215         # It is not ok to access another tenant's limits
  216         url = '%s/foo.json' % self.base_url
  217         response = self.app.get(url, expect_errors=True)
  218         self.assertEqual(403, response.status_int)
  219         # It is however ok to retrieve your own limits
  220         response = self.app.get(url, headers={'X-Project-Id': 'foo'})
  221         self.assertEqual(200, response.status_int)
  222         json_body = jsonutils.loads(response.body)
  223         self._verify_default_limits(json_body)
  224 
  225     def test_put_get_delete(self):
  226         # PUT and DELETE actions are in the same test as a meaningful DELETE
  227         # test would require a put anyway
  228         url = '%s/foo.json' % self.base_url
  229         response = self.app.put_json(url,
  230                                      params={'quota': {'network': 99}},
  231                                      headers={'X-Project-Id': 'admin',
  232                                               'X-Roles': 'admin'})
  233         self.assertEqual(200, response.status_int)
  234         json_body = jsonutils.loads(response.body)
  235         self._verify_after_update(json_body, {'network': 99})
  236 
  237         response = self.app.get(url, headers={'X-Project-Id': 'foo'})
  238         self.assertEqual(200, response.status_int)
  239         json_body = jsonutils.loads(response.body)
  240         self._verify_after_update(json_body, {'network': 99})
  241 
  242         response = self.app.delete(url, headers={'X-Project-Id': 'admin',
  243                                                  'X-Roles': 'admin'})
  244         self.assertEqual(204, response.status_int)
  245         self.assertFalse(response.body)
  246         # As DELETE does not return a body we need another GET
  247         response = self.app.get(url, headers={'X-Project-Id': 'foo'})
  248         self.assertEqual(200, response.status_int)
  249         json_body = jsonutils.loads(response.body)
  250         self._verify_default_limits(json_body)
  251 
  252     def test_update_list_delete(self):
  253         # PUT and DELETE actions are in the same test as a meaningful DELETE
  254         # test would require a put anyway
  255         url = '%s/foo.json' % self.base_url
  256         response = self.app.put_json(url,
  257                                      params={'quota': {'network': 99}},
  258                                      headers={'X-Project-Id': 'admin',
  259                                               'X-Roles': 'admin'})
  260         self.assertEqual(200, response.status_int)
  261         json_body = jsonutils.loads(response.body)
  262         self._verify_after_update(json_body, {'network': 99})
  263 
  264         response = self.app.get(self.base_url,
  265                                 headers={'X-Project-Id': 'admin',
  266                                          'X-Roles': 'admin'})
  267         self.assertEqual(200, response.status_int)
  268         json_body = jsonutils.loads(response.body)
  269         found = False
  270         for qs in json_body['quotas']:
  271             if qs['tenant_id'] == 'foo':
  272                 found = True
  273         self.assertTrue(found)
  274 
  275         response = self.app.delete(url, headers={'X-Project-Id': 'admin',
  276                                                  'X-Roles': 'admin'})
  277         self.assertEqual(204, response.status_int)
  278         self.assertFalse(response.body)
  279         response = self.app.get(self.base_url,
  280                                 headers={'X-Project-Id': 'admin',
  281                                          'X-Roles': 'admin'})
  282         self.assertEqual(200, response.status_int)
  283         json_body = jsonutils.loads(response.body)
  284         for qs in json_body['quotas']:
  285             self.assertNotEqual('foo', qs['tenant_id'])
  286 
  287     def test_quotas_get_defaults(self):
  288         response = self.app.get('%s/foo/default.json' % self.base_url,
  289                                 headers={'X-Project-Id': 'admin',
  290                                          'X-Roles': 'admin'})
  291         self.assertEqual(200, response.status_int)
  292         # As quota limits have not been updated, expect default values
  293         json_body = jsonutils.loads(response.body)
  294         self._verify_default_limits(json_body)
  295 
  296     def test_get_tenant_info(self):
  297         response = self.app.get('%s/tenant.json' % self.base_url,
  298                                 headers={'X-Project-Id': 'admin',
  299                                          'X-Roles': 'admin'})
  300         self.assertEqual(200, response.status_int)
  301         json_body = jsonutils.loads(response.body)
  302         self.assertEqual('admin', json_body['tenant']['tenant_id'])
  303 
  304 
  305 class TestResourceController(TestRootController):
  306     """Test generic controller"""
  307     # TODO(salv-orlando): This test case must not explicitly test the 'port'
  308     # resource. Also it should validate correct plugin/resource association
  309     base_url = '/v2.0'
  310 
  311     def setUp(self):
  312         super(TestResourceController, self).setUp()
  313         policy.init()
  314         self.addCleanup(policy.reset)
  315         self._gen_port()
  316 
  317     def _gen_port(self):
  318         network_id = self.plugin.create_network(context.get_admin_context(), {
  319             'network':
  320             {'name': 'pecannet', 'tenant_id': 'tenid', 'shared': False,
  321              'admin_state_up': True, 'status': 'ACTIVE'}})['id']
  322         self.port = self.plugin.create_port(context.get_admin_context(), {
  323             'port':
  324             {'tenant_id': 'tenid', 'network_id': network_id,
  325              'fixed_ips': n_const.ATTR_NOT_SPECIFIED,
  326              'mac_address': '00:11:22:33:44:55',
  327              'admin_state_up': True, 'device_id': 'FF',
  328              'device_owner': 'pecan', 'name': 'pecan'}})
  329 
  330     def test_get(self):
  331         response = self.app.get('/v2.0/ports.json')
  332         self.assertEqual(response.status_int, 200)
  333 
  334     def _check_item(self, expected, item):
  335         for attribute in expected:
  336             self.assertIn(attribute, item)
  337         self.assertEqual(len(expected), len(item))
  338 
  339     def _test_get_collection_with_fields_selector(self, fields=None):
  340         fields = fields or []
  341         query_params = ['fields=%s' % field for field in fields]
  342         url = '/v2.0/ports.json'
  343         if query_params:
  344             url = '%s?%s' % (url, '&'.join(query_params))
  345         list_resp = self.app.get(url, headers={'X-Project-Id': 'tenid'})
  346         self.assertEqual(200, list_resp.status_int)
  347         for item in jsonutils.loads(list_resp.body).get('ports', []):
  348             for field in fields:
  349                 self.assertIn(field, item)
  350             if fields:
  351                 self.assertEqual(len(fields), len(item))
  352             else:
  353                 for field in ('id', 'name', 'device_owner'):
  354                     self.assertIn(field, item)
  355 
  356     def test_get_collection_with_multiple_fields_selector(self):
  357         self._test_get_collection_with_fields_selector(fields=['id', 'name'])
  358 
  359     def test_get_collection_with_single_fields_selector(self):
  360         self._test_get_collection_with_fields_selector(fields=['name'])
  361 
  362     def test_get_collection_without_fields_selector(self):
  363         self._test_get_collection_with_fields_selector(fields=[])
  364 
  365     def test_project_id_in_mandatory_fields(self):
  366         # ports only specifies that tenant_id is mandatory, but project_id
  367         # should still be passed to the plugin.
  368         mock_get = mock.patch.object(self.plugin, 'get_ports',
  369                                      return_value=[]).start()
  370         self.app.get(
  371             '/v2.0/ports.json?fields=id',
  372             headers={'X-Project-Id': 'tenid'}
  373         )
  374         self.assertIn('project_id', mock_get.mock_calls[-1][2]['fields'])
  375 
  376     def test_get_item_with_fields_selector(self):
  377         item_resp = self.app.get(
  378             '/v2.0/ports/%s.json?fields=id&fields=name' % self.port['id'],
  379             headers={'X-Project-Id': 'tenid'})
  380         self.assertEqual(200, item_resp.status_int)
  381         self._check_item(['id', 'name'],
  382                          jsonutils.loads(item_resp.body)['port'])
  383         # Explicitly require an attribute which is also 'required_by_policy'.
  384         # The attribute should not be stripped while generating the response
  385         item_resp = self.app.get(
  386             '/v2.0/ports/%s.json?fields=id&fields=tenant_id' % self.port['id'],
  387             headers={'X-Project-Id': 'tenid'})
  388         self.assertEqual(200, item_resp.status_int)
  389         self._check_item(['id', 'tenant_id'],
  390                          jsonutils.loads(item_resp.body)['port'])
  391 
  392     def test_duped_and_empty_fields_stripped(self):
  393         mock_get = mock.patch.object(self.plugin, 'get_ports',
  394                                      return_value=[]).start()
  395         self.app.get(
  396             '/v2.0/ports.json?fields=id&fields=name&fields=&fields=name',
  397             headers={'X-Project-Id': 'tenid'}
  398         )
  399         received = mock_get.mock_calls[-1][2]['fields']
  400         self.assertNotIn('', received)
  401         self.assertEqual(len(received), len(set(received)))
  402 
  403     def test_post(self):
  404         response = self.app.post_json(
  405             '/v2.0/ports.json',
  406             params={'port': {'network_id': self.port['network_id'],
  407                              'admin_state_up': True,
  408                              'tenant_id': 'tenid'}},
  409             headers={'X-Project-Id': 'tenid'})
  410         self.assertEqual(response.status_int, 201)
  411 
  412     def test_post_with_retry(self):
  413         self._create_failed = False
  414         orig = self.plugin.create_port
  415 
  416         def new_create(*args, **kwargs):
  417             if not self._create_failed:
  418                 self._create_failed = True
  419                 raise db_exc.RetryRequest(ValueError())
  420             return orig(*args, **kwargs)
  421 
  422         with mock.patch.object(self.plugin, 'create_port',
  423                                new=new_create):
  424             response = self.app.post_json(
  425                 '/v2.0/ports.json',
  426                 params={'port': {'network_id': self.port['network_id'],
  427                                  'admin_state_up': True,
  428                                  'tenant_id': 'tenid'}},
  429                 headers={'X-Project-Id': 'tenid'})
  430             self.assertEqual(201, response.status_int)
  431 
  432     def test_put(self):
  433         response = self.app.put_json('/v2.0/ports/%s.json' % self.port['id'],
  434                                      params={'port': {'name': 'test'}},
  435                                      headers={'X-Project-Id': 'tenid'})
  436         self.assertEqual(response.status_int, 200)
  437         json_body = jsonutils.loads(response.body)
  438         self.assertEqual(1, len(json_body))
  439         self.assertIn('port', json_body)
  440         self.assertEqual('test', json_body['port']['name'])
  441         self.assertEqual('tenid', json_body['port']['tenant_id'])
  442 
  443     def test_delete(self):
  444         response = self.app.delete('/v2.0/ports/%s.json' % self.port['id'],
  445                                    headers={'X-Project-Id': 'tenid'})
  446         self.assertEqual(response.status_int, 204)
  447         self.assertFalse(response.body)
  448 
  449     def test_delete_disallows_body(self):
  450         response = self.app.delete_json(
  451             '/v2.0/ports/%s.json' % self.port['id'],
  452             params={'port': {'name': 'test'}},
  453             headers={'X-Project-Id': 'tenid'},
  454             expect_errors=True)
  455         self.assertEqual(response.status_int, 400)
  456 
  457     def test_plugin_initialized(self):
  458         self.assertIsNotNone(manager.NeutronManager._instance)
  459 
  460     def test_methods(self):
  461         self._test_method_returns_code('post', 405)
  462         self._test_method_returns_code('put', 405)
  463         self._test_method_returns_code('patch', 405)
  464         self._test_method_returns_code('delete', 405)
  465         self._test_method_returns_code('head', 405)
  466         self._test_method_returns_code('delete', 405)
  467 
  468     def test_post_with_empty_body(self):
  469         response = self.app.post_json(
  470             '/v2.0/ports.json',
  471             headers={'X-Project-Id': 'tenid'},
  472             params={},
  473             expect_errors=True)
  474         self.assertEqual(response.status_int, 400)
  475 
  476     def test_post_with_unsupported_json_type(self):
  477         response = self.app.post_json(
  478             '/v2.0/ports.json',
  479             headers={'X-Project-Id': 'tenid'},
  480             params=[1, 2, 3],
  481             expect_errors=True)
  482         self.assertEqual(response.status_int, 400)
  483 
  484     def test_bulk_create(self):
  485         response = self.app.post_json(
  486             '/v2.0/ports.json',
  487             params={'ports': [{'network_id': self.port['network_id'],
  488                              'admin_state_up': True,
  489                              'tenant_id': 'tenid'},
  490                              {'network_id': self.port['network_id'],
  491                               'admin_state_up': True,
  492                               'tenant_id': 'tenid'}]
  493                     },
  494             headers={'X-Project-Id': 'tenid'})
  495         self.assertEqual(201, response.status_int)
  496         json_body = jsonutils.loads(response.body)
  497         self.assertIn('ports', json_body)
  498         ports = json_body['ports']
  499         self.assertEqual(2, len(ports))
  500         for port in ports:
  501             self.assertEqual(1, len(port['security_groups']))
  502 
  503     def test_bulk_create_with_sg(self):
  504         sg_response = self.app.post_json(
  505                 '/v2.0/security-groups.json',
  506                 params={'security_group': {
  507                     "name": "functest",
  508                     "description": "Functional test"}},
  509                 headers={'X-Project-Id': 'tenid'})
  510         self.assertEqual(201, sg_response.status_int)
  511         sg_json_body = jsonutils.loads(sg_response.body)
  512         self.assertIn('security_group', sg_json_body)
  513         sg_id = sg_json_body['security_group']['id']
  514 
  515         port_response = self.app.post_json(
  516                 '/v2.0/ports.json',
  517                 params={'ports': [{'network_id': self.port['network_id'],
  518                                  'admin_state_up': True,
  519                                  'security_groups': [sg_id],
  520                                  'tenant_id': 'tenid'},
  521                                  {'network_id': self.port['network_id'],
  522                                   'admin_state_up': True,
  523                                  'security_groups': [sg_id],
  524                                   'tenant_id': 'tenid'}]
  525                         },
  526                 headers={'X-Project-Id': 'tenid'})
  527         self.assertEqual(201, port_response.status_int)
  528         json_body = jsonutils.loads(port_response.body)
  529         self.assertIn('ports', json_body)
  530         ports = json_body['ports']
  531         self.assertEqual(2, len(ports))
  532         for port in ports:
  533             self.assertEqual(1, len(port['security_groups']))
  534 
  535     def test_emulated_bulk_create(self):
  536         self.plugin._FORCE_EMULATED_BULK = True
  537         response = self.app.post_json(
  538             '/v2.0/ports.json',
  539             params={'ports': [{'network_id': self.port['network_id'],
  540                              'admin_state_up': True,
  541                              'tenant_id': 'tenid'},
  542                              {'network_id': self.port['network_id'],
  543                               'admin_state_up': True,
  544                               'tenant_id': 'tenid'}]
  545                     },
  546             headers={'X-Project-Id': 'tenid'})
  547         self.assertEqual(response.status_int, 201)
  548         json_body = jsonutils.loads(response.body)
  549         self.assertIn('ports', json_body)
  550         self.assertEqual(2, len(json_body['ports']))
  551 
  552     def test_emulated_bulk_create_rollback(self):
  553         self.plugin._FORCE_EMULATED_BULK = True
  554         response = self.app.post_json(
  555             '/v2.0/ports.json',
  556             params={'ports': [{'network_id': self.port['network_id'],
  557                              'admin_state_up': True,
  558                              'tenant_id': 'tenid'},
  559                              {'network_id': self.port['network_id'],
  560                               'admin_state_up': True,
  561                               'tenant_id': 'tenid'},
  562                              {'network_id': 'bad_net_id',
  563                               'admin_state_up': True,
  564                               'tenant_id': 'tenid'}]
  565                     },
  566             headers={'X-Project-Id': 'tenid'},
  567             expect_errors=True)
  568         self.assertEqual(response.status_int, 400)
  569         response = self.app.get(
  570             '/v2.0/ports.json',
  571             headers={'X-Project-Id': 'tenid'})
  572         # all ports should be rolled back from above so we are just left
  573         # with the one created in setup
  574         self.assertEqual(1, len(jsonutils.loads(response.body)['ports']))
  575 
  576     def test_bulk_create_one_item(self):
  577         response = self.app.post_json(
  578             '/v2.0/ports.json',
  579             params={'ports': [{'network_id': self.port['network_id'],
  580                                'admin_state_up': True,
  581                                'tenant_id': 'tenid'}]
  582                     },
  583             headers={'X-Project-Id': 'tenid'})
  584         self.assertEqual(response.status_int, 201)
  585         json_body = jsonutils.loads(response.body)
  586         self.assertIn('ports', json_body)
  587         self.assertEqual(1, len(json_body['ports']))
  588 
  589 
  590 class TestPaginationAndSorting(test_functional.PecanFunctionalTest):
  591 
  592     RESOURCE_COUNT = 6
  593 
  594     def setUp(self):
  595         super(TestPaginationAndSorting, self).setUp()
  596         policy.init()
  597         self.addCleanup(policy.reset)
  598         self.plugin = directory.get_plugin()
  599         self.ctx = context.get_admin_context()
  600         self._create_networks(self.RESOURCE_COUNT)
  601         self.networks = self._get_collection()['networks']
  602 
  603     def _create_networks(self, count=1):
  604         network_ids = []
  605         for index in range(count):
  606             network = {'name': 'pecannet-%d' % index, 'tenant_id': 'tenid',
  607                        'shared': False, 'admin_state_up': True,
  608                        'status': 'ACTIVE'}
  609             network_id = self.plugin.create_network(
  610                 self.ctx, {'network': network})['id']
  611             network_ids.append(network_id)
  612         return network_ids
  613 
  614     def _get_collection(self, collection=None, limit=None, marker=None,
  615                         fields=None, page_reverse=False, sort_key=None,
  616                         sort_dir=None):
  617         collection = collection or 'networks'
  618         fields = fields or []
  619         query_params = []
  620         if limit:
  621             query_params.append('limit=%d' % limit)
  622         if marker:
  623             query_params.append('marker=%s' % marker)
  624         if page_reverse:
  625             query_params.append('page_reverse=True')
  626         if sort_key:
  627             query_params.append('sort_key=%s' % sort_key)
  628         if sort_dir:
  629             query_params.append('sort_dir=%s' % sort_dir)
  630         query_params.extend(['%s%s' % ('fields=', field) for field in fields])
  631         url = '/v2.0/%s.json' % collection
  632         if query_params:
  633             url = '%s?%s' % (url, '&'.join(query_params))
  634         list_resp = self.app.get(url, headers={'X-Project-Id': 'tenid'})
  635         self.assertEqual(200, list_resp.status_int)
  636         return list_resp.json
  637 
  638     def _test_get_collection_with_pagination(self, expected_list,
  639                                              collection=None,
  640                                              limit=None, marker=None,
  641                                              fields=None, page_reverse=False,
  642                                              sort_key=None, sort_dir=None):
  643         expected_list = expected_list or []
  644         collection = collection or 'networks'
  645         list_resp = self._get_collection(collection=collection, limit=limit,
  646                                          marker=marker, fields=fields,
  647                                          page_reverse=page_reverse,
  648                                          sort_key=sort_key, sort_dir=sort_dir)
  649         if limit and marker:
  650             links_key = '%s_links' % collection
  651             self.assertIn(links_key, list_resp)
  652         if not fields or 'id' in fields:
  653             list_resp_ids = [item['id'] for item in list_resp[collection]]
  654             self.assertEqual(expected_list, list_resp_ids)
  655         if fields:
  656             for item in list_resp[collection]:
  657                 for field in fields:
  658                     self.assertIn(field, item)
  659 
  660     def test_get_collection_with_pagination_limit(self):
  661         self._test_get_collection_with_pagination([self.networks[0]['id']],
  662                                                   limit=1)
  663 
  664     def test_get_collection_with_pagination_fields_no_pk(self):
  665         self._test_get_collection_with_pagination([self.networks[0]['id']],
  666                                                   limit=1, fields=['name'])
  667 
  668     def test_get_collection_with_pagination_limit_over_count(self):
  669         expected_ids = [network['id'] for network in self.networks]
  670         self._test_get_collection_with_pagination(
  671             expected_ids, limit=self.RESOURCE_COUNT + 1)
  672 
  673     def test_get_collection_with_pagination_marker(self):
  674         marker = self.networks[2]['id']
  675         expected_ids = [network['id'] for network in self.networks[3:]]
  676         self._test_get_collection_with_pagination(expected_ids, limit=3,
  677                                                   marker=marker)
  678 
  679     def test_get_collection_with_pagination_marker_without_limit(self):
  680         marker = self.networks[2]['id']
  681         expected_ids = [network['id'] for network in self.networks]
  682         self._test_get_collection_with_pagination(expected_ids, marker=marker)
  683 
  684     def test_get_collection_with_pagination_and_fields(self):
  685         expected_ids = [network['id'] for network in self.networks[:2]]
  686         self._test_get_collection_with_pagination(
  687             expected_ids, limit=2, fields=['id', 'name'])
  688 
  689     def test_get_collection_with_pagination_page_reverse(self):
  690         marker = self.networks[2]['id']
  691         expected_ids = [network['id'] for network in self.networks[:2]]
  692         self._test_get_collection_with_pagination(expected_ids, limit=3,
  693                                                   marker=marker,
  694                                                   page_reverse=True)
  695 
  696     def test_get_collection_with_sorting_desc(self):
  697         nets = sorted(self.networks, key=lambda net: net['name'], reverse=True)
  698         expected_ids = [network['id'] for network in nets]
  699         self._test_get_collection_with_pagination(expected_ids,
  700                                                   sort_key='name',
  701                                                   sort_dir='desc')
  702 
  703     def test_get_collection_with_sorting_asc(self):
  704         nets = sorted(self.networks, key=lambda net: net['name'])
  705         expected_ids = [network['id'] for network in nets]
  706         self._test_get_collection_with_pagination(expected_ids,
  707                                                   sort_key='name',
  708                                                   sort_dir='asc')
  709 
  710 
  711 class TestRequestProcessing(TestRootController):
  712 
  713     def setUp(self):
  714         super(TestRequestProcessing, self).setUp()
  715         mock.patch('neutron.pecan_wsgi.hooks.notifier.registry').start()
  716         # request.context is thread-local storage so it has to be accessed by
  717         # the controller. We can capture it into a list here to assert on after
  718         # the request finishes.
  719 
  720         def capture_request_details(*args, **kwargs):
  721             self.captured_context = request.context
  722             self.request_params = kwargs
  723 
  724         mock.patch('neutron.pecan_wsgi.controllers.resource.'
  725                    'CollectionsController.get',
  726                    side_effect=capture_request_details).start()
  727         mock.patch('neutron.pecan_wsgi.controllers.resource.'
  728                    'CollectionsController.create',
  729                    side_effect=capture_request_details).start()
  730         mock.patch('neutron.pecan_wsgi.controllers.resource.'
  731                    'ItemController.get',
  732                    side_effect=capture_request_details).start()
  733     # TODO(kevinbenton): add context tests for X-Roles etc
  734 
  735     def test_context_set_in_request(self):
  736         self.app.get('/v2.0/ports.json',
  737                      headers={'X-Project-Id': 'tenant_id'})
  738         self.assertEqual('tenant_id',
  739                          self.captured_context['neutron_context'].tenant_id)
  740 
  741     def test_core_resource_identified(self):
  742         self.app.get('/v2.0/ports.json')
  743         self.assertEqual('port', self.captured_context['resource'])
  744         self.assertEqual('ports', self.captured_context['collection'])
  745 
  746     def test_lookup_identifies_resource_id(self):
  747         # We now this will return a 404 but that's not the point as it is
  748         # mocked
  749         self.app.get('/v2.0/ports/reina.json')
  750         self.assertEqual('port', self.captured_context['resource'])
  751         self.assertEqual('ports', self.captured_context['collection'])
  752         self.assertEqual('reina', self.captured_context['resource_id'])
  753 
  754     def test_resource_processing_post(self):
  755         self.app.post_json(
  756             '/v2.0/networks.json',
  757             params={'network': {'name': 'the_net',
  758                                 'admin_state_up': True}},
  759             headers={'X-Project-Id': 'tenid'})
  760         self.assertEqual('network', self.captured_context['resource'])
  761         self.assertEqual('networks', self.captured_context['collection'])
  762         resources = self.captured_context['resources']
  763         is_bulk = self.captured_context['is_bulk']
  764         self.assertEqual(1, len(resources))
  765         self.assertEqual('the_net', resources[0]['name'])
  766         self.assertTrue(resources[0]['admin_state_up'])
  767         self.assertFalse(is_bulk)
  768 
  769     def test_resource_processing_post_bulk(self):
  770         self.app.post_json(
  771             '/v2.0/networks.json',
  772             params={'networks': [{'name': 'the_net_1',
  773                                   'admin_state_up': True},
  774                                  {'name': 'the_net_2',
  775                                   'admin_state_up': False}]},
  776             headers={'X-Project-Id': 'tenid'})
  777         resources = self.captured_context['resources']
  778         is_bulk = self.captured_context['is_bulk']
  779         self.assertEqual(2, len(resources))
  780         self.assertTrue(resources[0]['admin_state_up'])
  781         self.assertEqual('the_net_1', resources[0]['name'])
  782         self.assertFalse(resources[1]['admin_state_up'])
  783         self.assertEqual('the_net_2', resources[1]['name'])
  784         self.assertTrue(is_bulk)
  785 
  786     def test_resource_processing_post_bulk_one_item(self):
  787         self.app.post_json(
  788             '/v2.0/networks.json',
  789             params={'networks': [{'name': 'the_net_1',
  790                                   'admin_state_up': True}]},
  791             headers={'X-Project-Id': 'tenid'})
  792         resources = self.captured_context['resources']
  793         is_bulk = self.captured_context['is_bulk']
  794         self.assertEqual(1, len(resources))
  795         self.assertTrue(is_bulk)
  796 
  797     def test_resource_processing_post_unknown_attribute_returns_400(self):
  798         response = self.app.post_json(
  799             '/v2.0/networks.json',
  800             params={'network': {'name': 'the_net',
  801                                 'alien': 'E.T.',
  802                                 'admin_state_up': True}},
  803             headers={'X-Project-Id': 'tenid'},
  804             expect_errors=True)
  805         self.assertEqual(400, response.status_int)
  806 
  807     def test_resource_processing_post_validation_error_returns_400(self):
  808         response = self.app.post_json(
  809             '/v2.0/networks.json',
  810             params={'network': {'name': 'the_net',
  811                                 'admin_state_up': 'invalid_value'}},
  812             headers={'X-Project-Id': 'tenid'},
  813             expect_errors=True)
  814         self.assertEqual(400, response.status_int)
  815 
  816     def test_service_plugin_identified(self):
  817         # TODO(kevinbenton): fix the unit test setup to include an l3 plugin
  818         self.skipTest("A dummy l3 plugin needs to be setup")
  819         self.app.get('/v2.0/routers.json')
  820         self.assertEqual('router', self.req_stash['resource_type'])
  821         # make sure the core plugin was identified as the handler for ports
  822         self.assertEqual(
  823             directory.get_plugin(plugin_constants.L3),
  824             self.req_stash['plugin'])
  825 
  826     def test_service_plugin_uri(self):
  827         nm = manager.NeutronManager.get_instance()
  828         nm.path_prefix_resource_mappings[dummy_plugin.RESOURCE_NAME] = [
  829             _SERVICE_PLUGIN_COLLECTION]
  830         response = self.do_request('/v2.0/dummy/serviceplugins.json')
  831         self.assertEqual(200, response.status_int)
  832         self.assertEqual(_SERVICE_PLUGIN_INDEX_BODY, response.json_body)
  833 
  834 
  835 class TestRouterController(TestResourceController):
  836     """Specialized tests for the router resource controller
  837 
  838     This test class adds tests specific for the router controller in
  839     order to verify the 'member_action' functionality, which this
  840     controller uses for adding and removing router interfaces.
  841     """
  842 
  843     def setUp(self):
  844         cfg.CONF.set_override(
  845             'service_plugins',
  846             ['neutron.services.l3_router.l3_router_plugin.L3RouterPlugin',
  847              'neutron.services.flavors.flavors_plugin.FlavorsPlugin'])
  848         super(TestRouterController, self).setUp()
  849         policy.init()
  850         self.addCleanup(policy.reset)
  851         plugin = directory.get_plugin()
  852         ctx = context.get_admin_context()
  853         l3_plugin = directory.get_plugin(plugin_constants.L3)
  854         network_id = pecan_utils.create_network(ctx, plugin)['id']
  855         self.subnet = pecan_utils.create_subnet(ctx, plugin, network_id)
  856         self.router = pecan_utils.create_router(ctx, l3_plugin)
  857 
  858     def test_member_actions_processing(self):
  859         response = self.app.put_json(
  860             '/v2.0/routers/%s/add_router_interface.json' % self.router['id'],
  861             params={'subnet_id': self.subnet['id']},
  862             headers={'X-Project-Id': 'tenid'})
  863         self.assertEqual(200, response.status_int)
  864 
  865     def test_non_existing_member_action_returns_404(self):
  866         response = self.app.put_json(
  867             '/v2.0/routers/%s/do_meh.json' % self.router['id'],
  868             params={'subnet_id': 'doesitevenmatter'},
  869             headers={'X-Project-Id': 'tenid'},
  870             expect_errors=True)
  871         self.assertEqual(404, response.status_int)
  872 
  873     def test_unsupported_method_member_action(self):
  874         response = self.app.post_json(
  875             '/v2.0/routers/%s/add_router_interface.json' % self.router['id'],
  876             params={'subnet_id': self.subnet['id']},
  877             headers={'X-Project-Id': 'tenid'},
  878             expect_errors=True)
  879         self.assertEqual(405, response.status_int)
  880 
  881         response = self.app.get(
  882             '/v2.0/routers/%s/add_router_interface.json' % self.router['id'],
  883             headers={'X-Project-Id': 'tenid'},
  884             expect_errors=True)
  885         self.assertEqual(405, response.status_int)
  886 
  887 
  888 class TestDHCPAgentShimControllers(test_functional.PecanFunctionalTest):
  889 
  890     def setUp(self):
  891         super(TestDHCPAgentShimControllers, self).setUp()
  892         policy.init()
  893         policy._ENFORCER.set_rules(
  894             oslo_policy.Rules.from_dict(
  895                 {'get_dhcp-agents': 'role:admin',
  896                  'get_dhcp-networks': 'role:admin',
  897                  'create_dhcp-networks': 'role:admin',
  898                  'delete_dhcp-networks': 'role:admin'}),
  899             overwrite=False)
  900         plugin = directory.get_plugin()
  901         ctx = context.get_admin_context()
  902         self.network = pecan_utils.create_network(ctx, plugin)
  903         self.agent = helpers.register_dhcp_agent()
  904         # NOTE(blogan): Not sending notifications because this test is for
  905         # testing the shim controllers
  906         plugin.agent_notifiers[n_const.AGENT_TYPE_DHCP] = None
  907 
  908     def test_list_dhcp_agents_hosting_network(self):
  909         response = self.app.get(
  910             '/v2.0/networks/%s/dhcp-agents.json' % self.network['id'],
  911             headers={'X-Roles': 'admin'})
  912         self.assertEqual(200, response.status_int)
  913 
  914     def test_list_networks_on_dhcp_agent(self):
  915         response = self.app.get(
  916             '/v2.0/agents/%s/dhcp-networks.json' % self.agent.id,
  917             headers={'X-Project-Id': 'tenid', 'X-Roles': 'admin'})
  918         self.assertEqual(200, response.status_int)
  919 
  920     def test_add_remove_dhcp_agent(self):
  921         headers = {'X-Project-Id': 'tenid', 'X-Roles': 'admin'}
  922         self.app.post_json(
  923             '/v2.0/agents/%s/dhcp-networks.json' % self.agent.id,
  924             headers=headers, params={'network_id': self.network['id']})
  925         response = self.app.get(
  926             '/v2.0/networks/%s/dhcp-agents.json' % self.network['id'],
  927             headers=headers)
  928         self.assertIn(self.agent.id,
  929                       [a['id'] for a in response.json['agents']])
  930         self.app.delete('/v2.0/agents/%(a)s/dhcp-networks/%(n)s.json' % {
  931             'a': self.agent.id, 'n': self.network['id']}, headers=headers)
  932         response = self.app.get(
  933             '/v2.0/networks/%s/dhcp-agents.json' % self.network['id'],
  934             headers=headers)
  935         self.assertNotIn(self.agent.id,
  936                          [a['id'] for a in response.json['agents']])
  937 
  938 
  939 class TestL3AgentShimControllers(test_functional.PecanFunctionalTest):
  940 
  941     def setUp(self):
  942         cfg.CONF.set_override(
  943             'service_plugins',
  944             ['neutron.services.l3_router.l3_router_plugin.L3RouterPlugin',
  945              'neutron.services.flavors.flavors_plugin.FlavorsPlugin'])
  946         super(TestL3AgentShimControllers, self).setUp()
  947         policy.init()
  948         policy._ENFORCER.set_rules(
  949             oslo_policy.Rules.from_dict(
  950                 {'get_l3-agents': 'role:admin',
  951                  'get_l3-routers': 'role:admin'}),
  952             overwrite=False)
  953         ctx = context.get_admin_context()
  954         l3_plugin = directory.get_plugin(plugin_constants.L3)
  955         self.router = pecan_utils.create_router(ctx, l3_plugin)
  956         self.agent = helpers.register_l3_agent()
  957         # NOTE(blogan): Not sending notifications because this test is for
  958         # testing the shim controllers
  959         l3_plugin.agent_notifiers[n_const.AGENT_TYPE_L3] = None
  960 
  961     def test_list_l3_agents_hosting_router(self):
  962         response = self.app.get(
  963             '/v2.0/routers/%s/l3-agents.json' % self.router['id'],
  964             headers={'X-Roles': 'admin'})
  965         self.assertEqual(200, response.status_int)
  966 
  967     def test_list_routers_on_l3_agent(self):
  968         response = self.app.get(
  969             '/v2.0/agents/%s/l3-routers.json' % self.agent.id,
  970             headers={'X-Roles': 'admin'})
  971         self.assertEqual(200, response.status_int)
  972 
  973     def test_add_remove_l3_agent(self):
  974         headers = {'X-Project-Id': 'tenid', 'X-Roles': 'admin'}
  975         response = self.app.post_json(
  976             '/v2.0/agents/%s/l3-routers.json' % self.agent.id,
  977             headers=headers, params={'router_id': self.router['id']})
  978         self.assertEqual(201, response.status_int)
  979         response = self.app.get(
  980             '/v2.0/routers/%s/l3-agents.json' % self.router['id'],
  981             headers=headers)
  982         self.assertIn(self.agent.id,
  983                       [a['id'] for a in response.json['agents']])
  984         response = self.app.delete(
  985             '/v2.0/agents/%(a)s/l3-routers/%(n)s.json' % {
  986                 'a': self.agent.id, 'n': self.router['id']}, headers=headers)
  987         self.assertEqual(204, response.status_int)
  988         self.assertFalse(response.body)
  989         response = self.app.get(
  990             '/v2.0/routers/%s/l3-agents.json' % self.router['id'],
  991             headers=headers)
  992         self.assertNotIn(self.agent.id,
  993                          [a['id'] for a in response.json['agents']])
  994 
  995 
  996 class TestShimControllers(test_functional.PecanFunctionalTest):
  997 
  998     def setUp(self):
  999         fake_ext = pecan_utils.FakeExtension()
 1000         fake_plugin = pecan_utils.FakePlugin()
 1001         plugins = {pecan_utils.FakePlugin.PLUGIN_TYPE: fake_plugin}
 1002         new_extensions = {fake_ext.get_alias(): fake_ext}
 1003         super(TestShimControllers, self).setUp(
 1004             service_plugins=plugins, extensions=new_extensions)
 1005         policy.init()
 1006         policy._ENFORCER.set_rules(
 1007             oslo_policy.Rules.from_dict(
 1008                 {'get_meh_meh': '',
 1009                  'get_meh_mehs': '',
 1010                  'get_fake_subresources': ''}),
 1011             overwrite=False)
 1012         self.addCleanup(policy.reset)
 1013 
 1014     def test_hyphenated_resource_controller_not_shimmed(self):
 1015         collection = pecan_utils.FakeExtension.HYPHENATED_COLLECTION.replace(
 1016             '_', '-')
 1017         resource = pecan_utils.FakeExtension.HYPHENATED_RESOURCE
 1018         url = '/v2.0/{}/something.json'.format(collection)
 1019         resp = self.app.get(url)
 1020         self.assertEqual(200, resp.status_int)
 1021         self.assertEqual({resource: {'fake': 'something'}}, resp.json)
 1022 
 1023     def test_hyphenated_collection_controller_not_shimmed(self):
 1024         body_collection = pecan_utils.FakeExtension.HYPHENATED_COLLECTION
 1025         uri_collection = body_collection.replace('_', '-')
 1026         url = '/v2.0/{}.json'.format(uri_collection)
 1027         resp = self.app.get(url)
 1028         self.assertEqual(200, resp.status_int)
 1029         self.assertEqual({body_collection: [{'fake': 'fake'}]}, resp.json)
 1030 
 1031     def test_hyphenated_collection_subresource_controller_not_shimmed(self):
 1032         body_collection = pecan_utils.FakeExtension.HYPHENATED_COLLECTION
 1033         uri_collection = body_collection.replace('_', '-')
 1034         # there is only one subresource so far
 1035         sub_resource_collection = (
 1036             pecan_utils.FakeExtension.FAKE_SUB_RESOURCE_COLLECTION)
 1037         temp_id = uuidutils.generate_uuid()
 1038         url = '/v2.0/{0}/{1}/{2}'.format(
 1039             uri_collection,
 1040             temp_id,
 1041             sub_resource_collection.replace('_', '-'))
 1042         resp = self.app.get(url)
 1043         self.assertEqual(200, resp.status_int)
 1044         self.assertEqual({sub_resource_collection: {'foo': temp_id}},
 1045                          resp.json)
 1046 
 1047 
 1048 class TestMemberActionController(test_functional.PecanFunctionalTest):
 1049     def setUp(self):
 1050         fake_ext = pecan_utils.FakeExtension()
 1051         fake_plugin = pecan_utils.FakePlugin()
 1052         plugins = {pecan_utils.FakePlugin.PLUGIN_TYPE: fake_plugin}
 1053         new_extensions = {fake_ext.get_alias(): fake_ext}
 1054         super(TestMemberActionController, self).setUp(
 1055             service_plugins=plugins, extensions=new_extensions)
 1056         hyphen_collection = pecan_utils.FakeExtension.HYPHENATED_COLLECTION
 1057         self.collection = hyphen_collection.replace('_', '-')
 1058 
 1059     def test_get_member_action_controller(self):
 1060         url = '/v2.0/{}/something/boo_meh.json'.format(self.collection)
 1061         resp = self.app.get(url)
 1062         self.assertEqual(200, resp.status_int)
 1063         self.assertEqual({'boo_yah': 'something'}, resp.json)
 1064 
 1065     def test_put_member_action_controller(self):
 1066         url = '/v2.0/{}/something/put_meh.json'.format(self.collection)
 1067         resp = self.app.put_json(url, params={'it_matters_not': 'ok'})
 1068         self.assertEqual(200, resp.status_int)
 1069         self.assertEqual({'poo_yah': 'something'}, resp.json)
 1070 
 1071     def test_get_member_action_does_not_exist(self):
 1072         url = '/v2.0/{}/something/are_you_still_there.json'.format(
 1073             self.collection)
 1074         resp = self.app.get(url, expect_errors=True)
 1075         self.assertEqual(404, resp.status_int)
 1076 
 1077     def test_put_member_action_does_not_exist(self):
 1078         url = '/v2.0/{}/something/are_you_still_there.json'.format(
 1079             self.collection)
 1080         resp = self.app.put_json(url, params={'it_matters_not': 'ok'},
 1081                                  expect_errors=True)
 1082         self.assertEqual(404, resp.status_int)
 1083 
 1084     def test_put_on_get_member_action(self):
 1085         url = '/v2.0/{}/something/boo_meh.json'.format(self.collection)
 1086         resp = self.app.put_json(url, params={'it_matters_not': 'ok'},
 1087                                  expect_errors=True)
 1088         self.assertEqual(405, resp.status_int)
 1089 
 1090     def test_get_on_put_member_action(self):
 1091         url = '/v2.0/{}/something/put_meh.json'.format(self.collection)
 1092         resp = self.app.get(url, expect_errors=True)
 1093         self.assertEqual(405, resp.status_int)
 1094 
 1095 
 1096 class TestParentSubresourceController(test_functional.PecanFunctionalTest):
 1097     def setUp(self):
 1098         fake_ext = pecan_utils.FakeExtension()
 1099         fake_plugin = pecan_utils.FakePlugin()
 1100         plugins = {pecan_utils.FakePlugin.PLUGIN_TYPE: fake_plugin}
 1101         new_extensions = {fake_ext.get_alias(): fake_ext}
 1102         super(TestParentSubresourceController, self).setUp(
 1103             service_plugins=plugins, extensions=new_extensions)
 1104         policy.init()
 1105         policy._ENFORCER.set_rules(
 1106             oslo_policy.Rules.from_dict(
 1107                 {'get_fake_duplicate': '',
 1108                  'get_meh_meh_fake_duplicate': ''}),
 1109             overwrite=False)
 1110         self.addCleanup(policy.reset)
 1111         hyphen_collection = pecan_utils.FakeExtension.HYPHENATED_COLLECTION
 1112         self.collection = hyphen_collection.replace('_', '-')
 1113         self.fake_collection = (pecan_utils.FakeExtension.
 1114                                 FAKE_PARENT_SUBRESOURCE_COLLECTION)
 1115 
 1116     def test_get_duplicate_parent_resource(self):
 1117         url = '/v2.0/{}'.format(self.fake_collection)
 1118         resp = self.app.get(url)
 1119         self.assertEqual(200, resp.status_int)
 1120         self.assertEqual({'fake_duplicates': [{'fake': 'fakeduplicates'}]},
 1121                          resp.json)
 1122 
 1123     def test_get_duplicate_parent_resource_item(self):
 1124         url = '/v2.0/{}/something'.format(self.fake_collection)
 1125         resp = self.app.get(url)
 1126         self.assertEqual(200, resp.status_int)
 1127         self.assertEqual({'fake_duplicate': {'fake': 'something'}}, resp.json)
 1128 
 1129     def test_get_parent_resource_and_duplicate_subresources(self):
 1130         url = '/v2.0/{0}/something/{1}'.format(self.collection,
 1131                                                self.fake_collection)
 1132         resp = self.app.get(url)
 1133         self.assertEqual(200, resp.status_int)
 1134         self.assertEqual({'fake_duplicates': [{'fake': 'something'}]},
 1135                          resp.json)
 1136 
 1137     def test_get_child_resource_policy_check(self):
 1138         policy.reset()
 1139         policy.init()
 1140         policy._ENFORCER.set_rules(
 1141             oslo_policy.Rules.from_dict(
 1142                 {'get_meh_meh_fake_duplicate': ''}
 1143             )
 1144         )
 1145         url = '/v2.0/{0}/something/{1}'.format(self.collection,
 1146                                                self.fake_collection)
 1147         resp = self.app.get(url)
 1148         self.assertEqual(200, resp.status_int)
 1149         self.assertEqual({'fake_duplicates': [{'fake': 'something'}]},
 1150                          resp.json)
 1151 
 1152 
 1153 class TestExcludeAttributePolicy(test_functional.PecanFunctionalTest):
 1154 
 1155     def setUp(self):
 1156         super(TestExcludeAttributePolicy, self).setUp()
 1157         policy.init()
 1158         self.addCleanup(policy.reset)
 1159         plugin = directory.get_plugin()
 1160         ctx = context.get_admin_context()
 1161         self.network_id = pecan_utils.create_network(ctx, plugin)['id']
 1162         mock.patch('neutron.pecan_wsgi.controllers.resource.'
 1163                    'CollectionsController.get').start()
 1164 
 1165     def test_get_networks(self):
 1166         response = self.app.get('/v2.0/networks/%s.json' % self.network_id,
 1167                 headers={'X-Project-Id': 'tenid'})
 1168         json_body = jsonutils.loads(response.body)
 1169         self.assertEqual(response.status_int, 200)
 1170         self.assertEqual('tenid', json_body['network']['project_id'])
 1171         self.assertEqual('tenid', json_body['network']['tenant_id'])