"Fossies" - the Fresh Open Source Software Archive

Member "magnum-8.1.0/magnum/tests/unit/api/controllers/v1/test_cluster_template.py" (1 Oct 2019, 65455 Bytes) of package /linux/misc/openstack/magnum-8.1.0.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_cluster_template.py": 8.0.0_vs_8.1.0.

    1 # Licensed under the Apache License, Version 2.0 (the "License");
    2 #    you may not use this file except in compliance with the License.
    3 #    You may obtain 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,
    9 #    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   10 #    See the License for the specific language governing permissions and
   11 #    limitations under the License.
   12 
   13 import datetime
   14 
   15 import mock
   16 from oslo_config import cfg
   17 from oslo_utils import timeutils
   18 from oslo_utils import uuidutils
   19 from six.moves.urllib import parse as urlparse
   20 from webtest.app import AppError
   21 from wsme import types as wtypes
   22 
   23 from magnum.api import attr_validator
   24 from magnum.api.controllers.v1 import cluster_template as api_cluster_template
   25 from magnum.common import exception
   26 from magnum.common import policy as magnum_policy
   27 from magnum.tests import base
   28 from magnum.tests.unit.api import base as api_base
   29 from magnum.tests.unit.api import utils as apiutils
   30 from magnum.tests.unit.objects import utils as obj_utils
   31 
   32 
   33 class TestClusterTemplateObject(base.TestCase):
   34 
   35     def test_cluster_template_init(self):
   36         cluster_template_dict = apiutils.cluster_template_post_data()
   37         del cluster_template_dict['image_id']
   38         del cluster_template_dict['registry_enabled']
   39         del cluster_template_dict['tls_disabled']
   40         del cluster_template_dict['public']
   41         del cluster_template_dict['server_type']
   42         del cluster_template_dict['master_lb_enabled']
   43         del cluster_template_dict['floating_ip_enabled']
   44         del cluster_template_dict['hidden']
   45         cluster_template = api_cluster_template.ClusterTemplate(
   46             **cluster_template_dict)
   47         self.assertEqual(wtypes.Unset, cluster_template.image_id)
   48         self.assertFalse(cluster_template.registry_enabled)
   49         self.assertFalse(cluster_template.tls_disabled)
   50         self.assertFalse(cluster_template.public)
   51         self.assertEqual('vm', cluster_template.server_type)
   52         self.assertFalse(cluster_template.master_lb_enabled)
   53         self.assertTrue(cluster_template.floating_ip_enabled)
   54         self.assertFalse(cluster_template.hidden)
   55 
   56 
   57 class TestListClusterTemplate(api_base.FunctionalTest):
   58 
   59     _cluster_template_attrs = ('name', 'apiserver_port', 'network_driver',
   60                                'coe', 'flavor_id', 'fixed_network',
   61                                'dns_nameserver', 'http_proxy',
   62                                'docker_volume_size', 'server_type',
   63                                'cluster_distro', 'external_network_id',
   64                                'image_id', 'registry_enabled', 'no_proxy',
   65                                'keypair_id', 'https_proxy', 'tls_disabled',
   66                                'public', 'labels', 'master_flavor_id',
   67                                'volume_driver', 'insecure_registry', 'hidden')
   68 
   69     def test_empty(self):
   70         response = self.get_json('/clustertemplates')
   71         self.assertEqual([], response['clustertemplates'])
   72 
   73     def test_one(self):
   74         cluster_template = obj_utils.create_test_cluster_template(self.context)
   75         response = self.get_json('/clustertemplates')
   76         self.assertEqual(cluster_template.uuid,
   77                          response['clustertemplates'][0]["uuid"])
   78         self._verify_attrs(self._cluster_template_attrs,
   79                            response['clustertemplates'][0])
   80 
   81     def test_get_one(self):
   82         cluster_template = obj_utils.create_test_cluster_template(self.context)
   83         response = self.get_json('/clustertemplates/%s' %
   84                                  cluster_template['uuid'])
   85         self.assertEqual(cluster_template.uuid, response['uuid'])
   86         self._verify_attrs(self._cluster_template_attrs, response)
   87 
   88     def test_get_one_by_name(self):
   89         cluster_template = obj_utils.create_test_cluster_template(self.context)
   90         response = self.get_json('/clustertemplates/%s' %
   91                                  cluster_template['name'])
   92         self.assertEqual(cluster_template.uuid, response['uuid'])
   93         self._verify_attrs(self._cluster_template_attrs, response)
   94 
   95     def test_get_one_by_name_not_found(self):
   96         response = self.get_json(
   97             '/clustertemplates/not_found',
   98             expect_errors=True)
   99         self.assertEqual(404, response.status_int)
  100         self.assertEqual('application/json', response.content_type)
  101         self.assertTrue(response.json['errors'])
  102 
  103     def test_get_one_by_uuid(self):
  104         temp_uuid = uuidutils.generate_uuid()
  105         obj_utils.create_test_cluster_template(self.context, uuid=temp_uuid)
  106         response = self.get_json(
  107             '/clustertemplates/%s' % temp_uuid)
  108         self.assertEqual(temp_uuid, response['uuid'])
  109 
  110     def test_get_one_by_uuid_not_found(self):
  111         temp_uuid = uuidutils.generate_uuid()
  112         response = self.get_json(
  113             '/clustertemplates/%s' % temp_uuid,
  114             expect_errors=True)
  115         self.assertEqual(404, response.status_int)
  116         self.assertEqual('application/json', response.content_type)
  117         self.assertTrue(response.json['errors'])
  118 
  119     @mock.patch("magnum.common.policy.enforce")
  120     @mock.patch("magnum.common.context.make_context")
  121     def test_get_one_by_uuid_admin(self, mock_context, mock_policy):
  122         temp_uuid = uuidutils.generate_uuid()
  123         obj_utils.create_test_cluster_template(self.context, uuid=temp_uuid,
  124                                                project_id=temp_uuid)
  125         self.context.is_admin = True
  126         response = self.get_json(
  127             '/clustertemplates/%s' % temp_uuid)
  128         self.assertEqual(temp_uuid, response['uuid'])
  129 
  130     def test_get_one_by_name_multiple_cluster_template(self):
  131         obj_utils.create_test_cluster_template(
  132             self.context, name='test_clustertemplate',
  133             uuid=uuidutils.generate_uuid())
  134         obj_utils.create_test_cluster_template(
  135             self.context, name='test_clustertemplate',
  136             uuid=uuidutils.generate_uuid())
  137         response = self.get_json(
  138             '/clustertemplates/test_clustertemplate',
  139             expect_errors=True)
  140         self.assertEqual(409, response.status_int)
  141         self.assertEqual('application/json', response.content_type)
  142         self.assertTrue(response.json['errors'])
  143 
  144     def test_get_all_with_pagination_marker(self):
  145         bm_list = []
  146         for id_ in range(4):
  147             cluster_template = obj_utils.create_test_cluster_template(
  148                 self.context, id=id_,
  149                 uuid=uuidutils.generate_uuid())
  150             bm_list.append(cluster_template)
  151 
  152         response = self.get_json('/clustertemplates?limit=3&marker=%s'
  153                                  % bm_list[2].uuid)
  154         self.assertEqual(1, len(response['clustertemplates']))
  155         self.assertEqual(bm_list[-1].uuid,
  156                          response['clustertemplates'][0]['uuid'])
  157 
  158     @mock.patch("magnum.common.policy.enforce")
  159     @mock.patch("magnum.common.context.make_context")
  160     def test_get_all_with_all_projects(self, mock_context, mock_policy):
  161         for id_ in range(4):
  162             obj_utils.create_test_cluster_template(
  163                 self.context, id=id_, project_id=id_,
  164                 uuid=uuidutils.generate_uuid())
  165 
  166         self.context.is_admin = True
  167         response = self.get_json('/clustertemplates')
  168         self.assertEqual(4, len(response['clustertemplates']))
  169 
  170     def test_detail(self):
  171         cluster_template = obj_utils.create_test_cluster_template(self.context)
  172         response = self.get_json('/clustertemplates/detail')
  173         self.assertEqual(cluster_template.uuid,
  174                          response['clustertemplates'][0]["uuid"])
  175         self._verify_attrs(self._cluster_template_attrs,
  176                            response['clustertemplates'][0])
  177 
  178     def test_detail_with_pagination_marker(self):
  179         bm_list = []
  180         for id_ in range(4):
  181             cluster_template = obj_utils.create_test_cluster_template(
  182                 self.context, id=id_,
  183                 uuid=uuidutils.generate_uuid())
  184             bm_list.append(cluster_template)
  185 
  186         response = self.get_json('/clustertemplates/detail?limit=3&marker=%s'
  187                                  % bm_list[2].uuid)
  188         self.assertEqual(1, len(response['clustertemplates']))
  189         self.assertEqual(bm_list[-1].uuid,
  190                          response['clustertemplates'][0]['uuid'])
  191         self._verify_attrs(self._cluster_template_attrs,
  192                            response['clustertemplates'][0])
  193 
  194     def test_detail_against_single(self):
  195         cluster_template = obj_utils.create_test_cluster_template(self.context)
  196         response = self.get_json('/clustertemplates/%s/detail' %
  197                                  cluster_template['uuid'],
  198                                  expect_errors=True)
  199         self.assertEqual(404, response.status_int)
  200 
  201     def test_many(self):
  202         bm_list = []
  203         for id_ in range(5):
  204             cluster_template = obj_utils.create_test_cluster_template(
  205                 self.context, id=id_,
  206                 uuid=uuidutils.generate_uuid())
  207             bm_list.append(cluster_template.uuid)
  208         response = self.get_json('/clustertemplates')
  209         self.assertEqual(len(bm_list), len(response['clustertemplates']))
  210         uuids = [bm['uuid'] for bm in response['clustertemplates']]
  211         self.assertEqual(sorted(bm_list), sorted(uuids))
  212 
  213     def test_links(self):
  214         uuid = uuidutils.generate_uuid()
  215         obj_utils.create_test_cluster_template(self.context, id=1, uuid=uuid)
  216         response = self.get_json('/clustertemplates/%s' % uuid)
  217         self.assertIn('links', response.keys())
  218         self.assertEqual(2, len(response['links']))
  219         self.assertIn(uuid, response['links'][0]['href'])
  220         for l in response['links']:
  221             bookmark = l['rel'] == 'bookmark'
  222             self.assertTrue(self.validate_link(l['href'],
  223                                                bookmark=bookmark))
  224 
  225     def test_collection_links(self):
  226         for id_ in range(5):
  227             obj_utils.create_test_cluster_template(
  228                 self.context, id=id_, uuid=uuidutils.generate_uuid())
  229         response = self.get_json('/clustertemplates/?limit=3')
  230         self.assertEqual(3, len(response['clustertemplates']))
  231 
  232         next_marker = response['clustertemplates'][-1]['uuid']
  233         self.assertIn(next_marker, response['next'])
  234 
  235     def test_collection_links_default_limit(self):
  236         cfg.CONF.set_override('max_limit', 3, 'api')
  237         for id_ in range(5):
  238             obj_utils.create_test_cluster_template(
  239                 self.context, id=id_, uuid=uuidutils.generate_uuid())
  240         response = self.get_json('/clustertemplates')
  241         self.assertEqual(3, len(response['clustertemplates']))
  242 
  243         next_marker = response['clustertemplates'][-1]['uuid']
  244         self.assertIn(next_marker, response['next'])
  245 
  246 
  247 class TestPatch(api_base.FunctionalTest):
  248 
  249     def setUp(self):
  250         super(TestPatch, self).setUp()
  251         p = mock.patch.object(attr_validator, 'validate_os_resources')
  252         self.mock_valid_os_res = p.start()
  253         self.addCleanup(p.stop)
  254         self.cluster_template = obj_utils.create_test_cluster_template(
  255             self.context,
  256             name='cluster_model_example_A',
  257             image_id='nerdherd',
  258             apiserver_port=8080,
  259             fixed_network='private',
  260             flavor_id='m1.magnum',
  261             master_flavor_id='m1.magnum',
  262             external_network_id='public',
  263             keypair_id='test',
  264             volume_driver='rexray',
  265             public=False,
  266             docker_volume_size=20,
  267             coe='swarm',
  268             labels={'key1': 'val1', 'key2': 'val2'},
  269             hidden=False
  270         )
  271 
  272     def test_update_not_found(self):
  273         uuid = uuidutils.generate_uuid()
  274         response = self.patch_json('/clustertemplates/%s' % uuid,
  275                                    [{'path': '/name',
  276                                      'value': 'cluster_model_example_B',
  277                                      'op': 'add'}],
  278                                    expect_errors=True)
  279         self.assertEqual(404, response.status_int)
  280         self.assertEqual('application/json', response.content_type)
  281         self.assertTrue(response.json['errors'])
  282 
  283     def test_update_cluster_template_with_cluster(self):
  284         cluster_template = obj_utils.create_test_cluster_template(self.context)
  285         obj_utils.create_test_cluster(
  286             self.context, cluster_template_id=cluster_template.uuid)
  287 
  288         response = self.patch_json('/clustertemplates/%s' %
  289                                    cluster_template.uuid,
  290                                    [{'path': '/network_driver',
  291                                      'value': 'flannel',
  292                                      'op': 'replace'}],
  293                                    expect_errors=True)
  294         self.assertEqual(400, response.status_int)
  295         self.assertEqual('application/json', response.content_type)
  296         self.assertTrue(response.json['errors'])
  297         self.assertIn(cluster_template.uuid,
  298                       response.json['errors'][0]['detail'])
  299 
  300     def test_update_cluster_template_name_with_cluster(self):
  301         cluster_template = obj_utils.create_test_cluster_template(self.context)
  302         obj_utils.create_test_cluster(
  303             self.context, cluster_template_id=cluster_template.uuid)
  304 
  305         response = self.patch_json('/clustertemplates/%s' %
  306                                    cluster_template.uuid,
  307                                    [{'path': '/name',
  308                                      'value': 'cluster_model_example_B',
  309                                      'op': 'replace'}],
  310                                    expect_errors=True)
  311         self.assertEqual(200, response.status_int)
  312 
  313     @mock.patch.object(magnum_policy, 'enforce')
  314     def test_update_public_cluster_template_success(self, mock_policy):
  315         mock_policy.return_value = True
  316         response = self.patch_json('/clustertemplates/%s' %
  317                                    self.cluster_template.uuid,
  318                                    [{'path': '/public', 'value': True,
  319                                      'op': 'replace'}])
  320         self.assertEqual('application/json', response.content_type)
  321         self.assertEqual(200, response.status_code)
  322 
  323         response = self.get_json('/clustertemplates/%s' %
  324                                  self.cluster_template.uuid)
  325         self.assertTrue(response['public'])
  326 
  327     @mock.patch.object(magnum_policy, 'enforce')
  328     def test_update_public_cluster_template_fail(self, mock_policy):
  329         mock_policy.return_value = False
  330         self.assertRaises(AppError, self.patch_json,
  331                           '/clustertemplates/%s' % self.cluster_template.uuid,
  332                           [{'path': '/public', 'value': True,
  333                             'op': 'replace'}])
  334 
  335     @mock.patch.object(magnum_policy, 'enforce')
  336     def test_update_cluster_template_with_cluster_allow_update(self,
  337                                                                mock_policy):
  338         mock_policy.return_value = True
  339         cluster_template = obj_utils.create_test_cluster_template(self.context)
  340         obj_utils.create_test_cluster(
  341             self.context, cluster_template_id=cluster_template.uuid)
  342         response = self.patch_json('/clustertemplates/%s' %
  343                                    cluster_template.uuid,
  344                                    [{'path': '/public',
  345                                      'value': True,
  346                                      'op': 'replace'}],
  347                                    expect_errors=True)
  348         self.assertEqual(200, response.status_int)
  349         response = self.get_json('/clustertemplates/%s' %
  350                                  self.cluster_template.uuid)
  351         self.assertEqual(response['public'], True)
  352 
  353     @mock.patch.object(magnum_policy, 'enforce')
  354     def test_update_hidden_cluster_template_success(self, mock_policy):
  355         mock_policy.return_value = True
  356         response = self.patch_json('/clustertemplates/%s' %
  357                                    self.cluster_template.uuid,
  358                                    [{'path': '/hidden', 'value': True,
  359                                      'op': 'replace'}])
  360         self.assertEqual('application/json', response.content_type)
  361         self.assertEqual(200, response.status_code)
  362 
  363         response = self.get_json('/clustertemplates/%s' %
  364                                  self.cluster_template.uuid)
  365         self.assertTrue(response['hidden'])
  366 
  367     @mock.patch.object(magnum_policy, 'enforce')
  368     def test_update_hidden_cluster_template_fail(self, mock_policy):
  369         mock_policy.return_value = False
  370         self.assertRaises(AppError, self.patch_json,
  371                           '/clustertemplates/%s' % self.cluster_template.uuid,
  372                           [{'path': '/hidden', 'value': True,
  373                             'op': 'replace'}])
  374 
  375     @mock.patch.object(magnum_policy, 'enforce')
  376     def test_update_cluster_template_hidden_with_cluster_allow_update(
  377             self, mock_policy):
  378         mock_policy.return_value = True
  379         cluster_template = obj_utils.create_test_cluster_template(self.context)
  380         obj_utils.create_test_cluster(
  381             self.context, cluster_template_id=cluster_template.uuid)
  382         response = self.patch_json('/clustertemplates/%s' %
  383                                    cluster_template.uuid,
  384                                    [{'path': '/hidden',
  385                                      'value': True,
  386                                      'op': 'replace'}],
  387                                    expect_errors=True)
  388         self.assertEqual(200, response.status_int)
  389         response = self.get_json('/clustertemplates/%s' %
  390                                  self.cluster_template.uuid)
  391         self.assertEqual(response['hidden'], True)
  392 
  393     def test_update_cluster_template_replace_labels_success(self):
  394         cluster_template = obj_utils.create_test_cluster_template(self.context)
  395         response = self.patch_json('/clustertemplates/%s' %
  396                                    cluster_template.uuid,
  397                                    [{'path': '/labels',
  398                                      'value': '{\'etcd_volume_size\': \'1\'}',
  399                                      'op': 'replace'}],
  400                                    expect_errors=True)
  401         self.assertEqual(200, response.status_int)
  402         response = self.get_json('/clustertemplates/%s' %
  403                                  self.cluster_template.uuid)
  404         self.assertEqual(response['labels'], {'etcd_volume_size': '1'})
  405 
  406     def test_update_cluster_template_with_cluster_not_allow_update(self):
  407         cluster_template = obj_utils.create_test_cluster_template(self.context)
  408         obj_utils.create_test_cluster(
  409             self.context, cluster_template_id=cluster_template.uuid)
  410         response = self.patch_json('/clustertemplates/%s' %
  411                                    cluster_template.uuid,
  412                                    [{'path': '/network_driver',
  413                                      'value': 'calico',
  414                                      'op': 'replace'}],
  415                                    expect_errors=True)
  416         self.assertEqual(400, response.status_code)
  417 
  418     @mock.patch('oslo_utils.timeutils.utcnow')
  419     def test_replace_singular(self, mock_utcnow):
  420         name = 'cluster_model_example_B'
  421         test_time = datetime.datetime(2000, 1, 1, 0, 0)
  422 
  423         mock_utcnow.return_value = test_time
  424         response = self.patch_json('/clustertemplates/%s' %
  425                                    self.cluster_template.uuid,
  426                                    [{'path': '/name', 'value': name,
  427                                      'op': 'replace'}])
  428         self.assertEqual('application/json', response.content_type)
  429         self.assertEqual(200, response.status_code)
  430 
  431         response = self.get_json('/clustertemplates/%s' %
  432                                  self.cluster_template.uuid)
  433         self.assertEqual(name, response['name'])
  434         return_updated_at = timeutils.parse_isotime(
  435             response['updated_at']).replace(tzinfo=None)
  436         self.assertEqual(test_time, return_updated_at)
  437         # Assert nothing else was changed
  438         self.assertEqual(self.cluster_template.uuid, response['uuid'])
  439         self.assertEqual(self.cluster_template.image_id, response['image_id'])
  440         self.assertEqual(self.cluster_template.apiserver_port,
  441                          response['apiserver_port'])
  442         self.assertEqual(self.cluster_template.fixed_network,
  443                          response['fixed_network'])
  444         self.assertEqual(self.cluster_template.network_driver,
  445                          response['network_driver'])
  446         self.assertEqual(self.cluster_template.volume_driver,
  447                          response['volume_driver'])
  448         self.assertEqual(self.cluster_template.docker_volume_size,
  449                          response['docker_volume_size'])
  450         self.assertEqual(self.cluster_template.coe,
  451                          response['coe'])
  452         self.assertEqual(self.cluster_template.http_proxy,
  453                          response['http_proxy'])
  454         self.assertEqual(self.cluster_template.https_proxy,
  455                          response['https_proxy'])
  456         self.assertEqual(self.cluster_template.no_proxy,
  457                          response['no_proxy'])
  458         self.assertEqual(self.cluster_template.labels,
  459                          response['labels'])
  460 
  461     def test_replace_cluster_template_with_no_exist_flavor_id(self):
  462         self.mock_valid_os_res.side_effect = exception.FlavorNotFound("aaa")
  463         response = self.patch_json('/clustertemplates/%s' %
  464                                    self.cluster_template.uuid,
  465                                    [{'path': '/flavor_id', 'value': 'aaa',
  466                                      'op': 'replace'}],
  467                                    expect_errors=True)
  468         self.assertEqual('application/json', response.content_type)
  469         self.assertEqual(400, response.status_code)
  470         self.assertTrue(response.json['errors'])
  471 
  472     def test_replace_cluster_template_with_no_exist_keypair_id(self):
  473         self.mock_valid_os_res.side_effect = exception.KeyPairNotFound("aaa")
  474         response = self.patch_json('/clustertemplates/%s' %
  475                                    self.cluster_template.uuid,
  476                                    [{'path': '/keypair_id', 'value': 'aaa',
  477                                      'op': 'replace'}],
  478                                    expect_errors=True)
  479         self.assertEqual('application/json', response.content_type)
  480         self.assertEqual(404, response.status_code)
  481         self.assertTrue(response.json['errors'])
  482 
  483     def test_replace_cluster_template_with_no_exist_external_network_id(self):
  484         self.mock_valid_os_res.side_effect = exception.ExternalNetworkNotFound(
  485             "aaa")
  486         response = self.patch_json('/clustertemplates/%s' %
  487                                    self.cluster_template.uuid,
  488                                    [{'path': '/external_network_id',
  489                                      'value': 'aaa',
  490                                      'op': 'replace'}],
  491                                    expect_errors=True)
  492         self.assertEqual('application/json', response.content_type)
  493         self.assertEqual(400, response.status_code)
  494         self.assertTrue(response.json['errors'])
  495 
  496     def test_replace_cluster_template_with_no_exist_image_id(self):
  497         self.mock_valid_os_res.side_effect = exception.ImageNotFound("aaa")
  498         response = self.patch_json('/clustertemplates/%s' %
  499                                    self.cluster_template.uuid,
  500                                    [{'path': '/image_id', 'value': 'aaa',
  501                                      'op': 'replace'}],
  502                                    expect_errors=True)
  503         self.assertEqual('application/json', response.content_type)
  504         self.assertEqual(400, response.status_code)
  505         self.assertTrue(response.json['errors'])
  506 
  507     def test_create_cluster_template_with_no_os_distro_image(self):
  508         image_exce = exception.OSDistroFieldNotFound('img')
  509         self.mock_valid_os_res.side_effect = image_exce
  510         response = self.patch_json('/clustertemplates/%s' %
  511                                    self.cluster_template.uuid,
  512                                    [{'path': '/image_id', 'value': 'img',
  513                                      'op': 'replace'}],
  514                                    expect_errors=True)
  515         self.assertEqual('application/json', response.content_type)
  516         self.assertEqual(400, response.status_code)
  517         self.assertTrue(response.json['errors'])
  518 
  519     def test_remove_singular(self):
  520         response = self.get_json('/clustertemplates/%s' %
  521                                  self.cluster_template.uuid)
  522         self.assertIsNotNone(response['dns_nameserver'])
  523 
  524         response = self.patch_json('/clustertemplates/%s' %
  525                                    self.cluster_template.uuid,
  526                                    [{'path': '/dns_nameserver',
  527                                      'op': 'remove'}])
  528         self.assertEqual('application/json', response.content_type)
  529         self.assertEqual(200, response.status_code)
  530 
  531         response = self.get_json('/clustertemplates/%s' %
  532                                  self.cluster_template.uuid)
  533         self.assertIsNone(response['dns_nameserver'])
  534         # Assert nothing else was changed
  535         self.assertEqual(self.cluster_template.uuid, response['uuid'])
  536         self.assertEqual(self.cluster_template.name, response['name'])
  537         self.assertEqual(self.cluster_template.apiserver_port,
  538                          response['apiserver_port'])
  539         self.assertEqual(self.cluster_template.image_id,
  540                          response['image_id'])
  541         self.assertEqual(self.cluster_template.fixed_network,
  542                          response['fixed_network'])
  543         self.assertEqual(self.cluster_template.network_driver,
  544                          response['network_driver'])
  545         self.assertEqual(self.cluster_template.volume_driver,
  546                          response['volume_driver'])
  547         self.assertEqual(self.cluster_template.docker_volume_size,
  548                          response['docker_volume_size'])
  549         self.assertEqual(self.cluster_template.coe, response['coe'])
  550         self.assertEqual(self.cluster_template.http_proxy,
  551                          response['http_proxy'])
  552         self.assertEqual(self.cluster_template.https_proxy,
  553                          response['https_proxy'])
  554         self.assertEqual(self.cluster_template.no_proxy, response['no_proxy'])
  555         self.assertEqual(self.cluster_template.labels, response['labels'])
  556 
  557     def test_remove_non_existent_property_fail(self):
  558         response = self.patch_json('/clustertemplates/%s' %
  559                                    self.cluster_template.uuid,
  560                                    [{'path': '/non-existent',
  561                                      'op': 'remove'}],
  562                                    expect_errors=True)
  563         self.assertEqual('application/json', response.content_type)
  564         self.assertEqual(400, response.status_code)
  565         self.assertTrue(response.json['errors'])
  566 
  567     def test_remove_mandatory_property_fail(self):
  568         mandatory_properties = ('/image_id', '/coe',
  569                                 '/external_network_id', '/server_type',
  570                                 '/tls_disabled', '/public',
  571                                 '/registry_enabled',
  572                                 '/cluster_distro', '/network_driver')
  573         for p in mandatory_properties:
  574             response = self.patch_json('/clustertemplates/%s' %
  575                                        self.cluster_template.uuid,
  576                                        [{'path': p, 'op': 'remove'}],
  577                                        expect_errors=True)
  578             self.assertEqual('application/json', response.content_type)
  579             self.assertEqual(400, response.status_code)
  580             self.assertTrue(response.json['errors'])
  581 
  582     def test_add_root_non_existent(self):
  583         response = self.patch_json(
  584             '/clustertemplates/%s' % self.cluster_template.uuid,
  585             [{'path': '/foo', 'value': 'bar', 'op': 'add'}],
  586             expect_errors=True)
  587         self.assertEqual('application/json', response.content_type)
  588         self.assertEqual(400, response.status_int)
  589         self.assertTrue(response.json['errors'])
  590 
  591     def test_remove_uuid(self):
  592         response = self.patch_json('/clustertemplates/%s' %
  593                                    self.cluster_template.uuid,
  594                                    [{'path': '/uuid', 'op': 'remove'}],
  595                                    expect_errors=True)
  596         self.assertEqual(400, response.status_int)
  597         self.assertEqual('application/json', response.content_type)
  598         self.assertTrue(response.json['errors'])
  599 
  600 
  601 class TestPost(api_base.FunctionalTest):
  602 
  603     def setUp(self):
  604         super(TestPost, self).setUp()
  605         p = mock.patch.object(attr_validator, 'validate_os_resources')
  606         self.mock_valid_os_res = p.start()
  607         self.addCleanup(p.stop)
  608 
  609     @mock.patch('magnum.api.attr_validator.validate_image')
  610     @mock.patch('oslo_utils.timeutils.utcnow')
  611     def test_create_cluster_template(self, mock_utcnow,
  612                                      mock_image_data):
  613         bdict = apiutils.cluster_template_post_data()
  614         test_time = datetime.datetime(2000, 1, 1, 0, 0)
  615         mock_utcnow.return_value = test_time
  616         mock_image_data.return_value = {'name': 'mock_name',
  617                                         'os_distro': 'fedora-atomic'}
  618 
  619         response = self.post_json('/clustertemplates', bdict)
  620         self.assertEqual(201, response.status_int)
  621         # Check location header
  622         self.assertIsNotNone(response.location)
  623         expected_location = '/v1/clustertemplates/%s' % bdict['uuid']
  624         self.assertEqual(expected_location,
  625                          urlparse.urlparse(response.location).path)
  626         self.assertEqual(bdict['uuid'], response.json['uuid'])
  627         self.assertNotIn('updated_at', response.json.keys)
  628         return_created_at = timeutils.parse_isotime(
  629             response.json['created_at']).replace(tzinfo=None)
  630         self.assertEqual(test_time, return_created_at)
  631 
  632     @mock.patch('magnum.api.attr_validator.validate_image')
  633     def test_create_cluster_template_set_project_id_and_user_id(
  634             self, mock_image_data):
  635         with mock.patch.object(
  636                 self.dbapi, 'create_cluster_template',
  637                 wraps=self.dbapi.create_cluster_template) as cc_mock:
  638             mock_image_data.return_value = {'name': 'mock_name',
  639                                             'os_distro': 'fedora-atomic'}
  640             bdict = apiutils.cluster_template_post_data()
  641             self.post_json('/clustertemplates', bdict)
  642             cc_mock.assert_called_once_with(mock.ANY)
  643             self.assertEqual(self.context.project_id,
  644                              cc_mock.call_args[0][0]['project_id'])
  645             self.assertEqual(self.context.user_id,
  646                              cc_mock.call_args[0][0]['user_id'])
  647 
  648     @mock.patch('magnum.api.attr_validator.validate_image')
  649     def test_create_cluster_template_doesnt_contain_id(self,
  650                                                        mock_image_data):
  651         with mock.patch.object(
  652                 self.dbapi, 'create_cluster_template',
  653                 wraps=self.dbapi.create_cluster_template) as cc_mock:
  654             mock_image_data.return_value = {'name': 'mock_name',
  655                                             'os_distro': 'fedora-atomic'}
  656             bdict = apiutils.cluster_template_post_data(image_id='my-image')
  657             response = self.post_json('/clustertemplates', bdict)
  658             self.assertEqual(bdict['image_id'], response.json['image_id'])
  659             cc_mock.assert_called_once_with(mock.ANY)
  660             # Check that 'id' is not in first arg of positional args
  661             self.assertNotIn('id', cc_mock.call_args[0][0])
  662 
  663     def _create_model_raises_app_error(self, **kwargs):
  664         # Create mock for db and image data
  665         with mock.patch.object(
  666                 self.dbapi, 'create_cluster_template',
  667                 wraps=self.dbapi.create_cluster_template) as cc_mock,\
  668             mock.patch('magnum.api.attr_validator.validate_image')\
  669                 as mock_image_data:
  670             mock_image_data.return_value = {'name': 'mock_name',
  671                                             'os_distro': 'fedora-atomic'}
  672             bdict = apiutils.cluster_template_post_data(**kwargs)
  673             self.assertRaises(AppError, self.post_json, '/clustertemplates',
  674                               bdict)
  675             self.assertFalse(cc_mock.called)
  676 
  677     def test_create_cluster_template_with_invalid_long_string(self):
  678         fields = ["uuid", "name", "image_id", "flavor_id", "master_flavor_id",
  679                   "dns_nameserver", "keypair_id", "external_network_id",
  680                   "cluster_distro", "fixed_network", "apiserver_port",
  681                   "docker_volume_size", "http_proxy", "https_proxy",
  682                   "no_proxy", "network_driver", "labels", "volume_driver"]
  683         for field in fields:
  684             self._create_model_raises_app_error(**{field: 'i' * 256})
  685 
  686     def test_create_cluster_template_with_invalid_empty_string(self):
  687         fields = ["uuid", "name", "image_id", "flavor_id", "master_flavor_id",
  688                   "dns_nameserver", "keypair_id", "external_network_id",
  689                   "cluster_distro", "fixed_network", "apiserver_port",
  690                   "docker_volume_size", "labels", "http_proxy", "https_proxy",
  691                   "no_proxy", "network_driver", "volume_driver", "coe"]
  692         for field in fields:
  693             self._create_model_raises_app_error(**{field: ''})
  694 
  695     def test_create_cluster_template_with_invalid_coe(self):
  696         self._create_model_raises_app_error(coe='k8s')
  697         self._create_model_raises_app_error(coe='storm')
  698         self._create_model_raises_app_error(coe='meson')
  699         self._create_model_raises_app_error(coe='osomatsu')
  700 
  701     def test_create_cluster_template_with_invalid_docker_volume_size(self):
  702         self._create_model_raises_app_error(docker_volume_size=-1)
  703         self._create_model_raises_app_error(
  704             docker_volume_size=1,
  705             docker_storage_driver="devicemapper")
  706         self._create_model_raises_app_error(
  707             docker_volume_size=2,
  708             docker_storage_driver="devicemapper")
  709         self._create_model_raises_app_error(docker_volume_size='notanint')
  710 
  711     def test_create_cluster_template_with_invalid_dns_nameserver(self):
  712         self._create_model_raises_app_error(dns_nameserver='1.1.2')
  713         self._create_model_raises_app_error(dns_nameserver='1.1..1')
  714         self._create_model_raises_app_error(dns_nameserver='openstack.org')
  715 
  716     def test_create_cluster_template_with_invalid_apiserver_port(self):
  717         self._create_model_raises_app_error(apiserver_port=-12)
  718         self._create_model_raises_app_error(apiserver_port=65536)
  719         self._create_model_raises_app_error(apiserver_port=0)
  720         self._create_model_raises_app_error(apiserver_port=1023)
  721         self._create_model_raises_app_error(apiserver_port='not an int')
  722 
  723     @mock.patch('magnum.api.attr_validator.validate_image')
  724     def test_create_cluster_template_with_labels(self, mock_image_data):
  725         with mock.patch.object(
  726                 self.dbapi, 'create_cluster_template',
  727                 wraps=self.dbapi.create_cluster_template) as cc_mock:
  728             mock_image_data.return_value = {'name': 'mock_name',
  729                                             'os_distro': 'fedora-atomic'}
  730             bdict = apiutils.cluster_template_post_data(
  731                 labels={'key1': 'val1', 'key2': 'val2'})
  732             response = self.post_json('/clustertemplates', bdict)
  733             self.assertEqual(bdict['labels'],
  734                              response.json['labels'])
  735             cc_mock.assert_called_once_with(mock.ANY)
  736             self.assertNotIn('id', cc_mock.call_args[0][0])
  737 
  738     @mock.patch('magnum.api.attr_validator.validate_image')
  739     def test_create_cluster_template_with_docker_volume_size(self,
  740                                                              mock_image_data):
  741         with mock.patch.object(
  742                 self.dbapi, 'create_cluster_template',
  743                 wraps=self.dbapi.create_cluster_template) as cc_mock:
  744             mock_image_data.return_value = {'name': 'mock_name',
  745                                             'os_distro': 'fedora-atomic'}
  746             bdict = apiutils.cluster_template_post_data(docker_volume_size=99)
  747             response = self.post_json('/clustertemplates', bdict)
  748             self.assertEqual(bdict['docker_volume_size'],
  749                              response.json['docker_volume_size'])
  750             cc_mock.assert_called_once_with(mock.ANY)
  751             self.assertNotIn('id', cc_mock.call_args[0][0])
  752 
  753     @mock.patch('magnum.api.attr_validator.validate_image')
  754     def test_create_cluster_template_with_overlay(self, mock_image_data):
  755         with mock.patch.object(
  756                 self.dbapi, 'create_cluster_template',
  757                 wraps=self.dbapi.create_cluster_template) as cc_mock:
  758             mock_image_data.return_value = {'name': 'mock_name',
  759                                             'os_distro': 'fedora-atomic'}
  760             bdict = apiutils.cluster_template_post_data(
  761                 docker_volume_size=1, docker_storage_driver="overlay")
  762             response = self.post_json('/clustertemplates', bdict)
  763             self.assertEqual(bdict['docker_volume_size'],
  764                              response.json['docker_volume_size'])
  765             cc_mock.assert_called_once_with(mock.ANY)
  766             self.assertNotIn('id', cc_mock.call_args[0][0])
  767 
  768     @mock.patch('magnum.api.attr_validator.validate_image')
  769     def _test_create_cluster_template_network_driver_attr(
  770             self,
  771             cluster_template_dict,
  772             cluster_template_config_dict,
  773             expect_errors,
  774             mock_image_data):
  775         mock_image_data.return_value = {'name': 'mock_name',
  776                                         'os_distro': 'fedora-atomic'}
  777         for k, v in cluster_template_config_dict.items():
  778             cfg.CONF.set_override(k, v, 'cluster_template')
  779         with mock.patch.object(
  780                 self.dbapi, 'create_cluster_template',
  781                 wraps=self.dbapi.create_cluster_template) as cc_mock:
  782 
  783             bdict = apiutils.cluster_template_post_data(
  784                 **cluster_template_dict)
  785             response = self.post_json('/clustertemplates', bdict,
  786                                       expect_errors=expect_errors)
  787             if expect_errors:
  788                 self.assertEqual(400, response.status_int)
  789             else:
  790                 expected_driver = bdict.get('network_driver')
  791                 if not expected_driver:
  792                     expected_driver = (
  793                         cfg.CONF.cluster_template.swarm_default_network_driver)
  794                 self.assertEqual(expected_driver,
  795                                  response.json['network_driver'])
  796                 self.assertEqual(bdict['image_id'],
  797                                  response.json['image_id'])
  798                 cc_mock.assert_called_once_with(mock.ANY)
  799                 self.assertNotIn('id', cc_mock.call_args[0][0])
  800                 self.assertTrue(uuidutils.is_uuid_like(response.json['uuid']))
  801 
  802     def test_create_cluster_template_with_network_driver(self):
  803         cluster_template_dict = {'coe': 'kubernetes',
  804                                  'network_driver': 'flannel'}
  805         config_dict = {}    # Default config
  806         expect_errors_flag = False
  807         self._test_create_cluster_template_network_driver_attr(
  808             cluster_template_dict,
  809             config_dict,
  810             expect_errors_flag)
  811 
  812     def test_create_cluster_template_with_no_network_driver(self):
  813         cluster_template_dict = {}
  814         config_dict = {}
  815         expect_errors_flag = False
  816         self._test_create_cluster_template_network_driver_attr(
  817             cluster_template_dict,
  818             config_dict,
  819             expect_errors_flag)
  820 
  821     def test_create_cluster_template_with_network_driver_non_def_config(self):
  822         cluster_template_dict = {'coe': 'kubernetes',
  823                                  'network_driver': 'flannel'}
  824         config_dict = {
  825             'kubernetes_allowed_network_drivers': ['flannel', 'foo']}
  826         expect_errors_flag = False
  827         self._test_create_cluster_template_network_driver_attr(
  828             cluster_template_dict,
  829             config_dict,
  830             expect_errors_flag)
  831 
  832     def test_create_cluster_template_with_invalid_network_driver(self):
  833         cluster_template_dict = {'coe': 'kubernetes',
  834                                  'network_driver': 'bad_driver'}
  835         config_dict = {
  836             'kubernetes_allowed_network_drivers': ['flannel', 'good_driver']}
  837         expect_errors_flag = True
  838         self._test_create_cluster_template_network_driver_attr(
  839             cluster_template_dict,
  840             config_dict,
  841             expect_errors_flag)
  842 
  843     @mock.patch('magnum.api.attr_validator.validate_image')
  844     def test_create_cluster_template_with_volume_driver(self,
  845                                                         mock_image_data):
  846         with mock.patch.object(
  847                 self.dbapi, 'create_cluster_template',
  848                 wraps=self.dbapi.create_cluster_template) as cc_mock:
  849             mock_image_data.return_value = {'name': 'mock_name',
  850                                             'os_distro': 'fedora-atomic'}
  851             bdict = apiutils.cluster_template_post_data(volume_driver='rexray')
  852             response = self.post_json('/clustertemplates', bdict)
  853             self.assertEqual(bdict['volume_driver'],
  854                              response.json['volume_driver'])
  855             cc_mock.assert_called_once_with(mock.ANY)
  856             self.assertNotIn('id', cc_mock.call_args[0][0])
  857 
  858     @mock.patch('magnum.api.attr_validator.validate_image')
  859     def test_create_cluster_template_with_no_volume_driver(self,
  860                                                            mock_image_data):
  861         with mock.patch.object(
  862                 self.dbapi, 'create_cluster_template',
  863                 wraps=self.dbapi.create_cluster_template) as cc_mock:
  864             mock_image_data.return_value = {'name': 'mock_name',
  865                                             'os_distro': 'fedora-atomic'}
  866             bdict = apiutils.cluster_template_post_data()
  867             response = self.post_json('/clustertemplates', bdict)
  868             self.assertEqual(bdict['volume_driver'],
  869                              response.json['volume_driver'])
  870             cc_mock.assert_called_once_with(mock.ANY)
  871             self.assertNotIn('id', cc_mock.call_args[0][0])
  872 
  873     @mock.patch('magnum.api.attr_validator.validate_image')
  874     @mock.patch.object(magnum_policy, 'enforce')
  875     def test_create_cluster_template_public_success(self, mock_policy,
  876                                                     mock_image_data):
  877         with mock.patch.object(
  878                 self.dbapi, 'create_cluster_template',
  879                 wraps=self.dbapi.create_cluster_template) as cc_mock:
  880             mock_policy.return_value = True
  881             mock_image_data.return_value = {'name': 'mock_name',
  882                                             'os_distro': 'fedora-atomic'}
  883             bdict = apiutils.cluster_template_post_data(public=True)
  884             response = self.post_json('/clustertemplates', bdict)
  885             self.assertTrue(response.json['public'])
  886             mock_policy.assert_called_with(mock.ANY,
  887                                            "clustertemplate:publish",
  888                                            None, do_raise=False)
  889             cc_mock.assert_called_once_with(mock.ANY)
  890             self.assertNotIn('id', cc_mock.call_args[0][0])
  891             self.assertTrue(cc_mock.call_args[0][0]['public'])
  892 
  893     @mock.patch('magnum.api.attr_validator.validate_image')
  894     @mock.patch.object(magnum_policy, 'enforce')
  895     def test_create_cluster_template_public_fail(self, mock_policy,
  896                                                  mock_image_data):
  897         with mock.patch.object(self.dbapi, 'create_cluster_template',
  898                                wraps=self.dbapi.create_cluster_template):
  899             # make policy enforcement fail
  900             mock_policy.return_value = False
  901             mock_image_data.return_value = {'name': 'mock_name',
  902                                             'os_distro': 'fedora-atomic'}
  903             bdict = apiutils.cluster_template_post_data(public=True)
  904             self.assertRaises(AppError, self.post_json, '/clustertemplates',
  905                               bdict)
  906 
  907     @mock.patch('magnum.api.attr_validator.validate_image')
  908     @mock.patch.object(magnum_policy, 'enforce')
  909     def test_create_cluster_template_public_not_set(self, mock_policy,
  910                                                     mock_image_data):
  911         with mock.patch.object(
  912                 self.dbapi, 'create_cluster_template',
  913                 wraps=self.dbapi.create_cluster_template) as cc_mock:
  914             mock_image_data.return_value = {'name': 'mock_name',
  915                                             'os_distro': 'fedora-atomic'}
  916             bdict = apiutils.cluster_template_post_data(public=False)
  917             response = self.post_json('/clustertemplates', bdict)
  918             self.assertFalse(response.json['public'])
  919             # policy enforcement is called only once for enforce_wsgi
  920             self.assertEqual(1, mock_policy.call_count)
  921             cc_mock.assert_called_once_with(mock.ANY)
  922             self.assertNotIn('id', cc_mock.call_args[0][0])
  923             self.assertFalse(cc_mock.call_args[0][0]['public'])
  924 
  925     @mock.patch('magnum.api.attr_validator.validate_image')
  926     @mock.patch.object(magnum_policy, 'enforce')
  927     def test_create_cluster_template_hidden_success(self, mock_policy,
  928                                                     mock_image_data):
  929         with mock.patch.object(
  930                 self.dbapi, 'create_cluster_template',
  931                 wraps=self.dbapi.create_cluster_template) as cc_mock:
  932             mock_policy.return_value = True
  933             mock_image_data.return_value = {'name': 'mock_name',
  934                                             'os_distro': 'fedora-atomic'}
  935             bdict = apiutils.cluster_template_post_data(hidden=True)
  936             response = self.post_json('/clustertemplates', bdict)
  937             self.assertTrue(response.json['hidden'])
  938             mock_policy.assert_called_with(mock.ANY,
  939                                            "clustertemplate:publish",
  940                                            None, do_raise=False)
  941             cc_mock.assert_called_once_with(mock.ANY)
  942             self.assertNotIn('id', cc_mock.call_args[0][0])
  943             self.assertTrue(cc_mock.call_args[0][0]['hidden'])
  944 
  945     @mock.patch('magnum.api.attr_validator.validate_image')
  946     @mock.patch.object(magnum_policy, 'enforce')
  947     def test_create_cluster_template_hidden_fail(self, mock_policy,
  948                                                  mock_image_data):
  949         with mock.patch.object(self.dbapi, 'create_cluster_template',
  950                                wraps=self.dbapi.create_cluster_template):
  951             # make policy enforcement fail
  952             mock_policy.return_value = False
  953             mock_image_data.return_value = {'name': 'mock_name',
  954                                             'os_distro': 'fedora-atomic'}
  955             bdict = apiutils.cluster_template_post_data(hidden=True)
  956             self.assertRaises(AppError, self.post_json, '/clustertemplates',
  957                               bdict)
  958 
  959     @mock.patch('magnum.api.attr_validator.validate_image')
  960     def test_create_cluster_template_with_no_os_distro_image(self,
  961                                                              mock_image_data):
  962         mock_image_data.side_effect = exception.OSDistroFieldNotFound('img')
  963         bdict = apiutils.cluster_template_post_data()
  964         del bdict['uuid']
  965         response = self.post_json('/clustertemplates', bdict,
  966                                   expect_errors=True)
  967         self.assertEqual(400, response.status_int)
  968 
  969     @mock.patch('magnum.api.attr_validator.validate_image')
  970     def test_create_cluster_template_with_os_distro_image(self,
  971                                                           mock_image_data):
  972         mock_image_data.return_value = {'name': 'mock_name',
  973                                         'os_distro': 'fedora-atomic'}
  974         bdict = apiutils.cluster_template_post_data()
  975         del bdict['uuid']
  976         response = self.post_json('/clustertemplates', bdict,
  977                                   expect_errors=True)
  978         self.assertEqual(201, response.status_int)
  979 
  980     @mock.patch('magnum.api.attr_validator.validate_image')
  981     def test_create_cluster_template_with_image_name(self,
  982                                                      mock_image_data):
  983         mock_image = {'name': 'mock_name',
  984                       'os_distro': 'fedora-atomic'}
  985         mock_image_data.return_value = mock_image
  986         bdict = apiutils.cluster_template_post_data()
  987         del bdict['uuid']
  988         response = self.post_json('/clustertemplates', bdict,
  989                                   expect_errors=True)
  990         self.assertEqual(201, response.status_int)
  991 
  992     @mock.patch('magnum.api.attr_validator.validate_image')
  993     def test_create_cluster_template_with_no_exist_image_name(self,
  994                                                               mock_image_data):
  995         mock_image_data.side_effect = exception.ResourceNotFound('test-img')
  996         bdict = apiutils.cluster_template_post_data()
  997         del bdict['uuid']
  998         response = self.post_json('/clustertemplates', bdict,
  999                                   expect_errors=True)
 1000         self.assertEqual(404, response.status_int)
 1001 
 1002     @mock.patch('magnum.api.attr_validator.validate_image')
 1003     def test_create_cluster_template_with_multi_image_name(self,
 1004                                                            mock_image_data):
 1005         mock_image_data.side_effect = exception.Conflict('Multiple images')
 1006         bdict = apiutils.cluster_template_post_data()
 1007         del bdict['uuid']
 1008         response = self.post_json('/clustertemplates', bdict,
 1009                                   expect_errors=True)
 1010         self.assertEqual(409, response.status_int)
 1011 
 1012     def test_create_cluster_template_without_image_id(self):
 1013         bdict = apiutils.cluster_template_post_data()
 1014         del bdict['image_id']
 1015         response = self.post_json('/clustertemplates', bdict,
 1016                                   expect_errors=True)
 1017         self.assertEqual(400, response.status_int)
 1018 
 1019     @mock.patch('magnum.api.attr_validator.validate_image')
 1020     def test_create_cluster_template_without_keypair_id(self,
 1021                                                         mock_image_data):
 1022         mock_image_data.return_value = {'name': 'mock_name',
 1023                                         'os_distro': 'fedora-atomic'}
 1024         bdict = apiutils.cluster_template_post_data()
 1025         del bdict['keypair_id']
 1026         response = self.post_json('/clustertemplates', bdict)
 1027         self.assertEqual(201, response.status_int)
 1028 
 1029     @mock.patch('magnum.api.attr_validator.validate_image')
 1030     def test_create_cluster_template_with_dns(self,
 1031                                               mock_image_data):
 1032         mock_image_data.return_value = {'name': 'mock_name',
 1033                                         'os_distro': 'fedora-atomic'}
 1034         bdict = apiutils.cluster_template_post_data()
 1035         response = self.post_json('/clustertemplates', bdict)
 1036         self.assertEqual(201, response.status_int)
 1037         self.assertEqual(bdict['dns_nameserver'],
 1038                          response.json['dns_nameserver'])
 1039 
 1040     @mock.patch('magnum.api.attr_validator.validate_image')
 1041     def test_create_cluster_template_with_no_exist_keypair(self,
 1042                                                            mock_image_data):
 1043         self.mock_valid_os_res.side_effect = exception.KeyPairNotFound("Test")
 1044         mock_image_data.return_value = {'name': 'mock_name',
 1045                                         'os_distro': 'fedora-atomic'}
 1046         bdict = apiutils.cluster_template_post_data()
 1047         response = self.post_json('/clustertemplates', bdict,
 1048                                   expect_errors=True)
 1049         self.assertEqual(404, response.status_int)
 1050 
 1051     @mock.patch('magnum.api.attr_validator.validate_image')
 1052     def test_create_cluster_template_with_flavor(self,
 1053                                                  mock_image_data):
 1054         mock_image_data.return_value = {'name': 'mock_name',
 1055                                         'os_distro': 'fedora-atomic'}
 1056         bdict = apiutils.cluster_template_post_data()
 1057         response = self.post_json('/clustertemplates', bdict)
 1058         self.assertEqual(201, response.status_int)
 1059         self.assertEqual(bdict['flavor_id'],
 1060                          response.json['flavor_id'])
 1061         self.assertEqual(bdict['master_flavor_id'],
 1062                          response.json['master_flavor_id'])
 1063 
 1064     @mock.patch('magnum.api.attr_validator.validate_image')
 1065     def test_create_cluster_template_with_no_exist_flavor(self,
 1066                                                           mock_image_data):
 1067         self.mock_valid_os_res.side_effect = exception.FlavorNotFound("flavor")
 1068         mock_image_data.return_value = {'name': 'mock_name',
 1069                                         'os_distro': 'fedora-atomic'}
 1070         bdict = apiutils.cluster_template_post_data()
 1071         response = self.post_json('/clustertemplates', bdict,
 1072                                   expect_errors=True)
 1073         self.assertEqual(400, response.status_int)
 1074 
 1075     @mock.patch('magnum.api.attr_validator.validate_image')
 1076     def test_create_cluster_template_with_external_network(self,
 1077                                                            mock_image_data):
 1078         mock_image_data.return_value = {'name': 'mock_name',
 1079                                         'os_distro': 'fedora-atomic'}
 1080         bdict = apiutils.cluster_template_post_data()
 1081         response = self.post_json('/clustertemplates', bdict)
 1082         self.assertEqual(201, response.status_int)
 1083         self.assertEqual(bdict['external_network_id'],
 1084                          response.json['external_network_id'])
 1085 
 1086     @mock.patch('magnum.api.attr_validator.validate_image')
 1087     def test_create_cluster_template_no_exist_external_network(
 1088             self, mock_image_data):
 1089         self.mock_valid_os_res.side_effect = exception.ExternalNetworkNotFound(
 1090             "test")
 1091         mock_image_data.return_value = {'name': 'mock_name',
 1092                                         'os_distro': 'fedora-atomic'}
 1093         bdict = apiutils.cluster_template_post_data()
 1094         response = self.post_json('/clustertemplates', bdict,
 1095                                   expect_errors=True)
 1096         self.assertEqual(400, response.status_int)
 1097 
 1098     @mock.patch('magnum.api.attr_validator.validate_image')
 1099     def test_create_cluster_template_without_name(self, mock_image_data):
 1100         with mock.patch.object(self.dbapi, 'create_cluster_template',
 1101                                wraps=self.dbapi.create_cluster_template):
 1102             mock_image_data.return_value = {'name': 'mock_name',
 1103                                             'os_distro': 'fedora-atomic'}
 1104             bdict = apiutils.cluster_template_post_data()
 1105             bdict.pop('name')
 1106             resp = self.post_json('/clustertemplates', bdict)
 1107             self.assertEqual(201, resp.status_int)
 1108             self.assertIsNotNone(resp.json['name'])
 1109 
 1110     def test_create_cluster_with_disabled_driver(self):
 1111         cfg.CONF.set_override('disabled_drivers',
 1112                               ['mesos_ubuntu_v1'],
 1113                               group='drivers')
 1114         bdict = apiutils.cluster_template_post_data(coe="mesos")
 1115         self.assertRaises(AppError, self.post_json, '/clustertemplates',
 1116                           bdict)
 1117 
 1118     @mock.patch('magnum.api.attr_validator.validate_image')
 1119     @mock.patch('oslo_utils.timeutils.utcnow')
 1120     def test_create_cluster_template_with_multi_dns(self, mock_utcnow,
 1121                                                     mock_image_data):
 1122         bdict = apiutils.cluster_template_post_data(
 1123             dns_nameserver="8.8.8.8,114.114.114.114")
 1124         test_time = datetime.datetime(2000, 1, 1, 0, 0)
 1125         mock_utcnow.return_value = test_time
 1126         mock_image_data.return_value = {'name': 'mock_name',
 1127                                         'os_distro': 'fedora-atomic'}
 1128 
 1129         response = self.post_json('/clustertemplates', bdict)
 1130         self.assertEqual(201, response.status_int)
 1131         # Check location header
 1132         self.assertIsNotNone(response.location)
 1133         expected_location = '/v1/clustertemplates/%s' % bdict['uuid']
 1134         self.assertEqual(expected_location,
 1135                          urlparse.urlparse(response.location).path)
 1136         self.assertEqual(bdict['uuid'], response.json['uuid'])
 1137         self.assertNotIn('updated_at', response.json.keys)
 1138         return_created_at = timeutils.parse_isotime(
 1139             response.json['created_at']).replace(tzinfo=None)
 1140         self.assertEqual(test_time, return_created_at)
 1141 
 1142 
 1143 class TestDelete(api_base.FunctionalTest):
 1144 
 1145     def test_delete_cluster_template(self):
 1146         cluster_template = obj_utils.create_test_cluster_template(self.context)
 1147         self.delete('/clustertemplates/%s' % cluster_template.uuid)
 1148         response = self.get_json('/clustertemplates/%s' %
 1149                                  cluster_template.uuid,
 1150                                  expect_errors=True)
 1151         self.assertEqual(404, response.status_int)
 1152         self.assertEqual('application/json', response.content_type)
 1153         self.assertTrue(response.json['errors'])
 1154 
 1155     def test_delete_cluster_template_with_cluster(self):
 1156         cluster_template = obj_utils.create_test_cluster_template(self.context)
 1157         obj_utils.create_test_cluster(
 1158             self.context, cluster_template_id=cluster_template.uuid)
 1159         response = self.delete('/clustertemplates/%s' % cluster_template.uuid,
 1160                                expect_errors=True)
 1161         self.assertEqual(400, response.status_int)
 1162         self.assertEqual('application/json', response.content_type)
 1163         self.assertTrue(response.json['errors'])
 1164         self.assertIn(cluster_template.uuid,
 1165                       response.json['errors'][0]['detail'])
 1166 
 1167     def test_delete_cluster_template_not_found(self):
 1168         uuid = uuidutils.generate_uuid()
 1169         response = self.delete('/clustertemplates/%s' % uuid,
 1170                                expect_errors=True)
 1171         self.assertEqual(404, response.status_int)
 1172         self.assertEqual('application/json', response.content_type)
 1173         self.assertTrue(response.json['errors'])
 1174 
 1175     def test_delete_cluster_template_with_name(self):
 1176         cluster_template = obj_utils.create_test_cluster_template(self.context)
 1177         response = self.delete('/clustertemplates/%s' %
 1178                                cluster_template['name'],
 1179                                expect_errors=True)
 1180         self.assertEqual(204, response.status_int)
 1181 
 1182     def test_delete_cluster_template_with_name_not_found(self):
 1183         response = self.delete('/clustertemplates/not_found',
 1184                                expect_errors=True)
 1185         self.assertEqual(404, response.status_int)
 1186         self.assertEqual('application/json', response.content_type)
 1187         self.assertTrue(response.json['errors'])
 1188 
 1189     def test_delete_multiple_cluster_template_by_name(self):
 1190         obj_utils.create_test_cluster_template(self.context,
 1191                                                name='test_cluster_template',
 1192                                                uuid=uuidutils.generate_uuid())
 1193         obj_utils.create_test_cluster_template(self.context,
 1194                                                name='test_cluster_template',
 1195                                                uuid=uuidutils.generate_uuid())
 1196         response = self.delete('/clustertemplates/test_cluster_template',
 1197                                expect_errors=True)
 1198         self.assertEqual(409, response.status_int)
 1199         self.assertEqual('application/json', response.content_type)
 1200         self.assertTrue(response.json['errors'])
 1201 
 1202     @mock.patch("magnum.common.policy.enforce")
 1203     @mock.patch("magnum.common.context.make_context")
 1204     def test_delete_cluster_template_as_admin(self, mock_context, mock_policy):
 1205         temp_uuid = uuidutils.generate_uuid()
 1206         obj_utils.create_test_cluster_template(self.context, uuid=temp_uuid,
 1207                                                project_id=temp_uuid)
 1208         self.context.is_admin = True
 1209         response = self.delete('/clustertemplates/%s' % temp_uuid,
 1210                                expect_errors=True)
 1211         self.assertEqual(204, response.status_int)
 1212 
 1213 
 1214 class TestClusterTemplatePolicyEnforcement(api_base.FunctionalTest):
 1215 
 1216     def _common_policy_check(self, rule, func, *arg, **kwarg):
 1217         self.policy.set_rules({rule: "project:non_fake"})
 1218         response = func(*arg, **kwarg)
 1219         self.assertEqual(403, response.status_int)
 1220         self.assertEqual('application/json', response.content_type)
 1221         self.assertTrue(
 1222             "Policy doesn't allow %s to be performed." % rule,
 1223             response.json['errors'][0]['detail'])
 1224 
 1225     def test_policy_disallow_get_all(self):
 1226         self._common_policy_check(
 1227             "cluster_template:get_all", self.get_json, '/clustertemplates',
 1228             expect_errors=True)
 1229 
 1230     def test_policy_disallow_get_one(self):
 1231         cluster_template = obj_utils.create_test_cluster_template(self.context)
 1232         self._common_policy_check(
 1233             "cluster_template:get", self.get_json,
 1234             '/clustertemplates/%s' % cluster_template.uuid,
 1235             expect_errors=True)
 1236 
 1237     def test_policy_disallow_detail(self):
 1238         self._common_policy_check(
 1239             "cluster_template:detail", self.get_json,
 1240             '/clustertemplates/%s/detail' % uuidutils.generate_uuid(),
 1241             expect_errors=True)
 1242 
 1243     def test_policy_disallow_update(self):
 1244         cluster_template = obj_utils.create_test_cluster_template(
 1245             self.context,
 1246             name='example_A',
 1247             uuid=uuidutils.generate_uuid())
 1248         self._common_policy_check(
 1249             "cluster_template:update", self.patch_json,
 1250             '/clustertemplates/%s' % cluster_template.name,
 1251             [{'path': '/name', 'value': "new_name", 'op': 'replace'}],
 1252             expect_errors=True)
 1253 
 1254     def test_policy_disallow_create(self):
 1255         bdict = apiutils.cluster_template_post_data(
 1256             name='cluster_model_example_A')
 1257         self._common_policy_check(
 1258             "cluster_template:create", self.post_json, '/clustertemplates',
 1259             bdict, expect_errors=True)
 1260 
 1261     def test_policy_disallow_delete(self):
 1262         cluster_template = obj_utils.create_test_cluster_template(self.context)
 1263         self._common_policy_check(
 1264             "cluster_template:delete", self.delete,
 1265             '/clustertemplates/%s' % cluster_template.uuid, expect_errors=True)
 1266 
 1267     def _owner_check(self, rule, func, *args, **kwargs):
 1268         self.policy.set_rules({rule: "user_id:%(user_id)s"})
 1269         response = func(*args, **kwargs)
 1270         self.assertEqual(403, response.status_int)
 1271         self.assertEqual('application/json', response.content_type)
 1272         self.assertTrue(
 1273             "Policy doesn't allow %s to be performed." % rule,
 1274             response.json['errors'][0]['detail'])
 1275 
 1276     def test_policy_only_owner_get_one(self):
 1277         cluster_template = obj_utils.create_test_cluster_template(
 1278             self.context,
 1279             user_id='another')
 1280         self._owner_check("cluster_template:get", self.get_json,
 1281                           '/clustertemplates/%s' % cluster_template.uuid,
 1282                           expect_errors=True)
 1283 
 1284     def test_policy_only_owner_update(self):
 1285         cluster_template = obj_utils.create_test_cluster_template(
 1286             self.context,
 1287             user_id='another')
 1288         self._owner_check(
 1289             "cluster_template:update", self.patch_json,
 1290             '/clustertemplates/%s' % cluster_template.uuid,
 1291             [{'path': '/name', 'value': "new_name", 'op': 'replace'}],
 1292             expect_errors=True)
 1293 
 1294     def test_policy_only_owner_delete(self):
 1295         cluster_template = obj_utils.create_test_cluster_template(
 1296             self.context,
 1297             user_id='another')
 1298         self._owner_check(
 1299             "cluster_template:delete", self.delete,
 1300             '/clustertemplates/%s' % cluster_template.uuid,
 1301             expect_errors=True)