# Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import datetime import mock from oslo.config import cfg from oslo.utils import timeutils from six.moves.urllib import parse as urlparse from wsme import types as wtypes from magnum.api.controllers.v1 import baymodel as api_baymodel from magnum.common import context from magnum.common import utils from magnum.tests.api import base as api_base from magnum.tests.api import utils as apiutils from magnum.tests import base from magnum.tests.objects import utils as obj_utils class TestBayModelObject(base.TestCase): def test_baymodel_init(self): baymodel_dict = apiutils.baymodel_post_data() del baymodel_dict['image_id'] baymodel = api_baymodel.BayModel(**baymodel_dict) self.assertEqual(wtypes.Unset, baymodel.image_id) class TestListBayModel(api_base.FunctionalTest): def test_empty(self): response = self.get_json('/baymodels') self.assertEqual([], response['baymodels']) def test_one(self): baymodel = obj_utils.create_test_baymodel(self.context) response = self.get_json('/baymodels') self.assertEqual(baymodel.uuid, response['baymodels'][0]["uuid"]) self.assertNotIn('flavor_id', response['baymodels'][0]) self.assertNotIn('dns_nameserver', response['baymodels'][0]) self.assertNotIn('keypair_id', response['baymodels'][0]) self.assertNotIn('external_network_id', response['baymodels'][0]) def test_get_one(self): baymodel = obj_utils.create_test_baymodel(self.context) response = self.get_json('/baymodels/%s' % baymodel['uuid']) self.assertEqual(baymodel.uuid, response['uuid']) self.assertIn('flavor_id', response) self.assertIn('dns_nameserver', response) self.assertIn('keypair_id', response) self.assertIn('external_network_id', response) def test_detail(self): baymodel = obj_utils.create_test_baymodel(self.context) response = self.get_json('/baymodels/detail') self.assertEqual(baymodel.uuid, response['baymodels'][0]["uuid"]) self.assertIn('flavor_id', response['baymodels'][0]) self.assertIn('dns_nameserver', response['baymodels'][0]) self.assertIn('keypair_id', response['baymodels'][0]) self.assertIn('external_network_id', response['baymodels'][0]) def test_detail_against_single(self): baymodel = obj_utils.create_test_baymodel(self.context) response = self.get_json('/baymodels/%s/detail' % baymodel['uuid'], expect_errors=True) self.assertEqual(404, response.status_int) def test_many(self): bm_list = [] for id_ in range(5): baymodel = obj_utils.create_test_baymodel(self.context, id=id_, uuid=utils.generate_uuid()) bm_list.append(baymodel.uuid) response = self.get_json('/baymodels') self.assertEqual(len(bm_list), len(response['baymodels'])) uuids = [bm['uuid'] for bm in response['baymodels']] self.assertEqual(sorted(bm_list), sorted(uuids)) def test_links(self): uuid = utils.generate_uuid() obj_utils.create_test_baymodel(self.context, id=1, uuid=uuid) response = self.get_json('/baymodels/%s' % uuid) self.assertIn('links', response.keys()) self.assertEqual(2, len(response['links'])) self.assertIn(uuid, response['links'][0]['href']) for l in response['links']: bookmark = l['rel'] == 'bookmark' self.assertTrue(self.validate_link(l['href'], bookmark=bookmark)) def test_collection_links(self): for id_ in range(5): obj_utils.create_test_baymodel(self.context, id=id_, uuid=utils.generate_uuid()) response = self.get_json('/baymodels/?limit=3') self.assertEqual(3, len(response['baymodels'])) next_marker = response['baymodels'][-1]['uuid'] self.assertIn(next_marker, response['next']) def test_collection_links_default_limit(self): cfg.CONF.set_override('max_limit', 3, 'api') for id_ in range(5): obj_utils.create_test_baymodel(self.context, id=id_, uuid=utils.generate_uuid()) response = self.get_json('/baymodels') self.assertEqual(3, len(response['baymodels'])) next_marker = response['baymodels'][-1]['uuid'] self.assertIn(next_marker, response['next']) class TestPatch(api_base.FunctionalTest): def setUp(self): super(TestPatch, self).setUp() self.baymodel = obj_utils.create_test_baymodel(self.context, name='bay_model_example_A', image_id='nerdherd', apiserver_port=8080) def test_update_not_found(self): uuid = utils.generate_uuid() response = self.patch_json('/baymodels/%s' % uuid, [{'path': '/name', 'value': 'bay_model_example_B', 'op': 'add'}], expect_errors=True) self.assertEqual(404, response.status_int) self.assertEqual('application/json', response.content_type) self.assertTrue(response.json['error_message']) @mock.patch.object(timeutils, 'utcnow') def test_replace_singular(self, mock_utcnow): name = 'bay_model_example_B' test_time = datetime.datetime(2000, 1, 1, 0, 0) mock_utcnow.return_value = test_time response = self.patch_json('/baymodels/%s' % self.baymodel.uuid, [{'path': '/name', 'value': name, 'op': 'replace'}]) self.assertEqual('application/json', response.content_type) self.assertEqual(200, response.status_code) response = self.get_json('/baymodels/%s' % self.baymodel.uuid) self.assertEqual(name, response['name']) return_updated_at = timeutils.parse_isotime( response['updated_at']).replace(tzinfo=None) self.assertEqual(test_time, return_updated_at) # Assert nothing else was changed self.assertEqual(self.baymodel.uuid, response['uuid']) self.assertEqual(self.baymodel.image_id, response['image_id']) self.assertEqual(self.baymodel.apiserver_port, response['apiserver_port']) def test_remove_singular(self): baymodel = obj_utils.create_test_baymodel(self.context, uuid=utils.generate_uuid()) response = self.get_json('/baymodels/%s' % baymodel.uuid) self.assertIsNotNone(response['image_id']) response = self.patch_json('/baymodels/%s' % baymodel.uuid, [{'path': '/image_id', 'op': 'remove'}]) self.assertEqual('application/json', response.content_type) self.assertEqual(200, response.status_code) response = self.get_json('/baymodels/%s' % baymodel.uuid) self.assertIsNone(response['image_id']) # Assert nothing else was changed self.assertEqual(baymodel.uuid, response['uuid']) self.assertEqual(baymodel.name, response['name']) self.assertEqual(baymodel.apiserver_port, response['apiserver_port']) def test_remove_non_existent_property_fail(self): response = self.patch_json('/baymodels/%s' % self.baymodel.uuid, [{'path': '/non-existent', 'op': 'remove'}], expect_errors=True) self.assertEqual('application/json', response.content_type) self.assertEqual(400, response.status_code) self.assertTrue(response.json['error_message']) def test_add_root(self): name = 'bay_model_example_B' response = self.patch_json('/baymodels/%s' % self.baymodel.uuid, [{'path': '/name', 'value': name, 'op': 'add'}]) self.assertEqual('application/json', response.content_type) self.assertEqual(200, response.status_int) response = self.get_json('/baymodels/%s' % self.baymodel.uuid) self.assertEqual(name, response['name']) # Assert nothing else was changed self.assertEqual(self.baymodel.uuid, response['uuid']) self.assertEqual(self.baymodel.image_id, response['image_id']) self.assertEqual(self.baymodel.apiserver_port, response['apiserver_port']) def test_add_root_non_existent(self): response = self.patch_json('/baymodels/%s' % self.baymodel.uuid, [{'path': '/foo', 'value': 'bar', 'op': 'add'}], expect_errors=True) self.assertEqual('application/json', response.content_type) self.assertEqual(400, response.status_int) self.assertTrue(response.json['error_message']) def test_add_multi(self): json = [ { 'path': '/name', 'value': 'bay_model_example_B', 'op': 'add' }, { 'path': '/image_id', 'value': 'my-image', 'op': 'add' } ] response = self.patch_json('/baymodels/%s' % self.baymodel.uuid, json) self.assertEqual('application/json', response.content_type) self.assertEqual(200, response.status_code) response = self.get_json('/baymodels/%s' % self.baymodel.uuid) self.assertEqual('bay_model_example_B', response['name']) self.assertEqual('my-image', response['image_id']) # Assert nothing else was changed self.assertEqual(self.baymodel.uuid, response['uuid']) self.assertEqual(self.baymodel.apiserver_port, response['apiserver_port']) def test_remove_uuid(self): response = self.patch_json('/baymodels/%s' % self.baymodel.uuid, [{'path': '/uuid', 'op': 'remove'}], expect_errors=True) self.assertEqual(400, response.status_int) self.assertEqual('application/json', response.content_type) self.assertTrue(response.json['error_message']) class TestPost(api_base.FunctionalTest): def setUp(self): super(TestPost, self).setUp() p = mock.patch.object(context, 'RequestContext') self.mock_request_context = p.start() mock_auth_token = self.mock_request_context.auth_token_info['token'] mock_auth_token['project']['id'].return_value = 'fake_project' mock_auth_token['user']['id'].return_value = 'fake_user' self.addCleanup(p.stop) @mock.patch.object(timeutils, 'utcnow') def test_create_baymodel(self, mock_utcnow): cdict = apiutils.baymodel_post_data() test_time = datetime.datetime(2000, 1, 1, 0, 0) mock_utcnow.return_value = test_time response = self.post_json('/baymodels', cdict) self.assertEqual(201, response.status_int) # Check location header self.assertIsNotNone(response.location) expected_location = '/v1/baymodels/%s' % cdict['uuid'] self.assertEqual(urlparse.urlparse(response.location).path, expected_location) response = self.get_json('/baymodels/%s' % cdict['uuid']) self.assertEqual(cdict['uuid'], response['uuid']) self.assertFalse(response['updated_at']) return_created_at = timeutils.parse_isotime( response['created_at']).replace(tzinfo=None) self.assertEqual(test_time, return_created_at) def test_create_baymodel_doesnt_contain_id(self): with mock.patch.object(self.dbapi, 'create_baymodel', wraps=self.dbapi.create_baymodel) as cc_mock: cdict = apiutils.baymodel_post_data(image_id='my-image') self.post_json('/baymodels', cdict) response = self.get_json('/baymodels/%s' % cdict['uuid']) self.assertEqual(cdict['image_id'], response['image_id']) cc_mock.assert_called_once_with(mock.ANY) # Check that 'id' is not in first arg of positional args self.assertNotIn('id', cc_mock.call_args[0][0]) def test_create_baymodel_generate_uuid(self): cdict = apiutils.baymodel_post_data() del cdict['uuid'] self.post_json('/baymodels', cdict) response = self.get_json('/baymodels') self.assertEqual(cdict['image_id'], response['baymodels'][0]['image_id']) self.assertTrue(utils.is_uuid_like(response['baymodels'][0]['uuid'])) class TestDelete(api_base.FunctionalTest): def test_delete_baymodel(self): baymodel = obj_utils.create_test_baymodel(self.context) self.delete('/baymodels/%s' % baymodel.uuid) response = self.get_json('/baymodels/%s' % baymodel.uuid, expect_errors=True) self.assertEqual(404, response.status_int) self.assertEqual('application/json', response.content_type) self.assertTrue(response.json['error_message']) def test_delete_baymodel_with_bay(self): baymodel = obj_utils.create_test_baymodel(self.context) obj_utils.create_test_bay(self.context, baymodel_id=baymodel.uuid) response = self.delete('/baymodels/%s' % baymodel.uuid, expect_errors=True) self.assertEqual(400, response.status_int) self.assertEqual('application/json', response.content_type) self.assertTrue(response.json['error_message']) self.assertIn(baymodel.uuid, response.json['error_message']) def test_delete_baymodel_not_found(self): uuid = utils.generate_uuid() response = self.delete('/baymodels/%s' % uuid, expect_errors=True) self.assertEqual(404, response.status_int) self.assertEqual('application/json', response.content_type) self.assertTrue(response.json['error_message'])