Merge "Add tests for Pod api"

This commit is contained in:
Jenkins 2015-01-29 23:20:59 +00:00 committed by Gerrit Code Review
commit 88c71bc8c5
4 changed files with 393 additions and 75 deletions

View File

@ -31,6 +31,11 @@ from magnum import objects
class PodPatchType(types.JsonPatchType):
@staticmethod
def internal_attrs():
defaults = types.JsonPatchType.internal_attrs()
return defaults + ['/name', '/labels']
@staticmethod
def mandatory_attrs():
return ['/bay_uuid']

View File

@ -85,7 +85,7 @@ class TestListBay(api_base.FunctionalTest):
bm_list.append(bay.uuid)
response = self.get_json('/bays')
self.assertEqual(len(bm_list), len(response['bays']))
uuids = [bm['uuid'] for bm in response['bays']]
uuids = [b['uuid'] for b in response['bays']]
self.assertEqual(sorted(bm_list), sorted(uuids))
def test_links(self):
@ -102,7 +102,7 @@ class TestListBay(api_base.FunctionalTest):
def test_collection_links(self):
for id_ in range(5):
obj_utils.create_test_bay(self.context, id=id_,
uuid=utils.generate_uuid())
uuid=utils.generate_uuid())
response = self.get_json('/bays/?limit=3')
self.assertEqual(3, len(response['bays']))
@ -113,7 +113,7 @@ class TestListBay(api_base.FunctionalTest):
cfg.CONF.set_override('max_limit', 3, 'api')
for id_ in range(5):
obj_utils.create_test_bay(self.context, id=id_,
uuid=utils.generate_uuid())
uuid=utils.generate_uuid())
response = self.get_json('/bays')
self.assertEqual(3, len(response['bays']))
@ -239,22 +239,20 @@ class TestPatch(api_base.FunctionalTest):
self.assertTrue(response.json['error_message'])
def test_remove_ok(self):
bay = obj_utils.create_test_bay(self.context,
uuid=utils.generate_uuid())
response = self.get_json('/bays/%s' % bay.uuid)
response = self.get_json('/bays/%s' % self.bay.uuid)
self.assertIsNotNone(response['name'])
response = self.patch_json('/bays/%s' % bay.uuid,
response = self.patch_json('/bays/%s' % self.bay.uuid,
[{'path': '/name', 'op': 'remove'}])
self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code)
response = self.get_json('/bays/%s' % bay.uuid)
response = self.get_json('/bays/%s' % self.bay.uuid)
self.assertIsNone(response['name'])
# Assert nothing else was changed
self.assertEqual(bay.uuid, response['uuid'])
self.assertEqual(bay.baymodel_id, response['baymodel_id'])
self.assertEqual(bay.node_count, response['node_count'])
self.assertEqual(self.bay.uuid, response['uuid'])
self.assertEqual(self.bay.baymodel_id, response['baymodel_id'])
self.assertEqual(self.bay.node_count, response['node_count'])
def test_remove_uuid(self):
response = self.patch_json('/bays/%s' % self.bay.uuid,

View File

@ -9,75 +9,376 @@
# 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.
from magnum.conductor import api
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 pod as api_pod
from magnum.common import utils
from magnum.conductor import api as rpcapi
from magnum import objects
from magnum.tests.db import base as db_base
from magnum.tests.db import utils as db_utils
from mock import patch
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 TestPodController(db_base.DbTestCase):
def mock_pod_create(self, pod):
class TestPodObject(base.TestCase):
def test_pod_init(self):
pod_dict = apiutils.pod_post_data(bay_uuid=None)
del pod_dict['desc']
pod = api_pod.Pod(**pod_dict)
self.assertEqual(wtypes.Unset, pod.desc)
class TestListPod(api_base.FunctionalTest):
def setUp(self):
super(TestListPod, self).setUp()
obj_utils.create_test_bay(self.context)
def test_empty(self):
response = self.get_json('/pods')
self.assertEqual([], response['pods'])
def _assert_pod_fields(self, pod):
pod_fields = ['name', 'bay_uuid', 'desc', 'images', 'labels', 'status']
for field in pod_fields:
self.assertIn(field, pod)
def test_one(self):
pod = obj_utils.create_test_pod(self.context)
response = self.get_json('/pods')
self.assertEqual(pod.uuid, response['pods'][0]["uuid"])
self._assert_pod_fields(response['pods'][0])
def test_get_one(self):
pod = obj_utils.create_test_pod(self.context)
response = self.get_json('/pods/%s' % pod['uuid'])
self.assertEqual(pod.uuid, response['uuid'])
self._assert_pod_fields(response)
def test_detail(self):
pod = obj_utils.create_test_pod(self.context)
response = self.get_json('/pods/detail')
self.assertEqual(pod.uuid, response['pods'][0]["uuid"])
self._assert_pod_fields(response['pods'][0])
def test_detail_against_single(self):
pod = obj_utils.create_test_pod(self.context)
response = self.get_json('/pods/%s/detail' % pod['uuid'],
expect_errors=True)
self.assertEqual(404, response.status_int)
def test_many(self):
pod_list = []
for id_ in range(5):
pod = obj_utils.create_test_pod(self.context, id=id_,
uuid=utils.generate_uuid())
pod_list.append(pod.uuid)
response = self.get_json('/pods')
self.assertEqual(len(pod_list), len(response['pods']))
uuids = [p['uuid'] for p in response['pods']]
self.assertEqual(sorted(pod_list), sorted(uuids))
def test_links(self):
uuid = utils.generate_uuid()
obj_utils.create_test_pod(self.context, id=1, uuid=uuid)
response = self.get_json('/pods/%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_pod(self.context, id=id_,
uuid=utils.generate_uuid())
response = self.get_json('/pods/?limit=3')
self.assertEqual(3, len(response['pods']))
next_marker = response['pods'][-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_pod(self.context, id=id_,
uuid=utils.generate_uuid())
response = self.get_json('/pods')
self.assertEqual(3, len(response['pods']))
next_marker = response['pods'][-1]['uuid']
self.assertIn(next_marker, response['next'])
class TestPatch(api_base.FunctionalTest):
def setUp(self):
super(TestPatch, self).setUp()
obj_utils.create_test_bay(self.context)
self.pod = obj_utils.create_test_pod(self.context,
desc='pod_example_A_desc',
status='Running')
@mock.patch.object(timeutils, 'utcnow')
def test_replace_ok(self, mock_utcnow):
test_time = datetime.datetime(2000, 1, 1, 0, 0)
mock_utcnow.return_value = test_time
new_desc = 'pod_example_B_desc'
response = self.get_json('/pods/%s' % self.pod.uuid)
self.assertNotEqual(new_desc, response['desc'])
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/desc', 'value': new_desc,
'op': 'replace'}])
self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code)
response = self.get_json('/pods/%s' % self.pod.uuid)
self.assertEqual(new_desc, response['desc'])
return_updated_at = timeutils.parse_isotime(
response['updated_at']).replace(tzinfo=None)
self.assertEqual(test_time, return_updated_at)
def test_replace_bay_uuid(self):
another_bay = obj_utils.create_test_bay(self.context,
uuid=utils.generate_uuid())
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/bay_uuid',
'value': another_bay.uuid,
'op': 'replace'}],
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code)
def test_replace_non_existent_bay_uuid(self):
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/bay_uuid',
'value': utils.generate_uuid(),
'op': 'replace'}],
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_code)
self.assertTrue(response.json['error_message'])
def test_replace_internal_field(self):
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/labels', 'value': {}, 'op': 'replace'}],
expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_code)
self.assertTrue(response.json['error_message'])
def test_replace_non_existent_pod(self):
response = self.patch_json('/pods/%s' % utils.generate_uuid(),
[{'path': '/desc',
'value': 'pod_example_B_desc',
'op': 'replace'}],
expect_errors=True)
self.assertEqual(404, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_add_ok(self):
new_desc = 'pod_example_B_desc'
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/desc', 'value': new_desc, 'op': 'add'}])
self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_int)
response = self.get_json('/pods/%s' % self.pod.uuid)
self.assertEqual(new_desc, response['desc'])
def test_add_multi(self):
new_status = 'Stopped'
new_desc = 'pod_example_B_desc'
response = self.get_json('/pods/%s' % self.pod.uuid)
self.assertNotEqual(new_status, response['status'])
self.assertNotEqual(new_desc, response['desc'])
json = [
{
'path': '/status',
'value': new_status,
'op': 'add'
},
{
'path': '/desc',
'value': new_desc,
'op': 'add'
}
]
response = self.patch_json('/pods/%s' % self.pod.uuid, json)
self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code)
response = self.get_json('/pods/%s' % self.pod.uuid)
self.assertEqual(new_status, response['status'])
self.assertEqual(new_desc, response['desc'])
def test_add_non_existent_property(self):
response = self.patch_json('/pods/%s' % self.pod.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_remove_ok(self):
response = self.get_json('/pods/%s' % self.pod.uuid)
self.assertIsNotNone(response['desc'])
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/desc', 'op': 'remove'}])
self.assertEqual('application/json', response.content_type)
self.assertEqual(200, response.status_code)
response = self.get_json('/pods/%s' % self.pod.uuid)
self.assertIsNone(response['desc'])
def test_remove_uuid(self):
response = self.patch_json('/pods/%s' % self.pod.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'])
def test_remove_bay_uuid(self):
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/bay_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'])
def test_remove_internal_field(self):
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/labels', 'op': 'remove'}],
expect_errors=True)
self.assertEqual(400, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
def test_remove_non_existent_property(self):
response = self.patch_json('/pods/%s' % self.pod.uuid,
[{'path': '/non-existent', 'op': 'remove'}],
expect_errors=True)
self.assertEqual(400, response.status_code)
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()
obj_utils.create_test_bay(self.context)
p = mock.patch.object(rpcapi.API, 'pod_create')
self.mock_pod_create = p.start()
self.mock_pod_create.side_effect = self._simulate_rpc_pod_create
self.addCleanup(p.stop)
def _simulate_rpc_pod_create(self, pod):
pod.create()
return pod
def mock_pod_destroy(self, pod_uuid):
pod = objects.Pod.get_by_uuid({}, pod_uuid)
@mock.patch.object(timeutils, 'utcnow')
def test_create_pod(self, mock_utcnow):
pdict = apiutils.pod_post_data()
test_time = datetime.datetime(2000, 1, 1, 0, 0)
mock_utcnow.return_value = test_time
response = self.post_json('/pods', pdict)
self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int)
# Check location header
self.assertIsNotNone(response.location)
expected_location = '/v1/pods/%s' % pdict['uuid']
self.assertEqual(urlparse.urlparse(response.location).path,
expected_location)
response = self.get_json('/pods/%s' % pdict['uuid'])
self.assertEqual(pdict['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_pod_doesnt_contain_id(self):
with mock.patch.object(self.dbapi, 'create_pod',
wraps=self.dbapi.create_pod) as cc_mock:
pdict = apiutils.pod_post_data(desc='pod_example_A_desc')
self.post_json('/pods', pdict)
response = self.get_json('/pods/%s' % pdict['uuid'])
self.assertEqual(pdict['desc'], response['desc'])
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_pod_generate_uuid(self):
pdict = apiutils.pod_post_data()
del pdict['uuid']
response = self.post_json('/pods', pdict)
self.assertEqual('application/json', response.content_type)
self.assertEqual(201, response.status_int)
response = self.get_json('/pods')
self.assertEqual(pdict['desc'], response['pods'][0]['desc'])
self.assertTrue(utils.is_uuid_like(response['pods'][0]['uuid']))
def test_create_pod_no_bay_uuid(self):
pdict = apiutils.pod_post_data()
del pdict['bay_uuid']
response = self.post_json('/pods', pdict, expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int)
def test_create_pod_with_non_existent_bay_uuid(self):
pdict = apiutils.pod_post_data(bay_uuid=utils.generate_uuid())
response = self.post_json('/pods', pdict, expect_errors=True)
self.assertEqual('application/json', response.content_type)
self.assertEqual(400, response.status_int)
self.assertTrue(response.json['error_message'])
class TestDelete(api_base.FunctionalTest):
def setUp(self):
super(TestDelete, self).setUp()
obj_utils.create_test_bay(self.context)
self.pod = obj_utils.create_test_pod(self.context)
p = mock.patch.object(rpcapi.API, 'pod_delete')
self.mock_pod_delete = p.start()
self.mock_pod_delete.side_effect = self._simulate_rpc_pod_delete
self.addCleanup(p.stop)
def _simulate_rpc_pod_delete(self, pod_uuid):
pod = objects.Pod.get_by_uuid(self.context, pod_uuid)
pod.destroy()
def test_pod_api(self):
with patch.object(api.API, 'pod_create') as mock_method:
mock_method.side_effect = self.mock_pod_create
# Create a bay
bay = db_utils.create_test_bay()
def test_delete_pod(self):
self.delete('/pods/%s' % self.pod.uuid)
response = self.get_json('/pods/%s' % self.pod.uuid,
expect_errors=True)
self.assertEqual(404, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])
# Create a pod
params = '''
{
"desc": "My Pod",
"bay_uuid": "%s",
"images": ["ubuntu"],
"manifest": "{\\"id\\": \\"name_of_pod\\", \
\\"labels\\": {\\"foo\\": \\"foo1\\"} }"
}
''' % bay.uuid
response = self.app.post('/v1/pods',
params=params,
content_type='application/json')
self.assertEqual(response.status_int, 201)
# Get all pods
response = self.app.get('/v1/pods')
self.assertEqual(response.status_int, 200)
self.assertEqual(1, len(response.json))
c = response.json['pods'][0]
self.assertIsNotNone(c.get('uuid'))
self.assertEqual('name_of_pod', c.get('name'))
self.assertEqual('My Pod', c.get('desc'))
self.assertEqual(bay.uuid, c.get('bay_uuid'))
self.assertEqual(['ubuntu'], c.get('images'))
self.assertEqual('foo1', c.get('labels')['foo'])
# Get just the one we created
response = self.app.get('/v1/pods/%s' % c.get('uuid'))
self.assertEqual(response.status_int, 200)
# Update the description
params = [{'path': '/name',
'value': 'pod_example_B',
'op': 'replace'}]
response = self.app.patch_json('/v1/pods/%s' % c.get('uuid'),
params=params)
self.assertEqual(response.status_int, 200)
with patch.object(api.API, 'pod_delete') as mock_method:
mock_method.side_effect = self.mock_pod_destroy
# Delete the pod we created
response = self.app.delete('/v1/pods/%s' % c.get('uuid'))
self.assertEqual(response.status_int, 204)
response = self.app.get('/v1/pods')
self.assertEqual(response.status_int, 200)
c = response.json['pods']
self.assertEqual(0, len(c))
def test_delete_pod_not_found(self):
uuid = utils.generate_uuid()
response = self.delete('/pods/%s' % uuid, expect_errors=True)
self.assertEqual(404, response.status_int)
self.assertEqual('application/json', response.content_type)
self.assertTrue(response.json['error_message'])

View File

@ -17,6 +17,7 @@ Utils for testing the API service.
from magnum.api.controllers.v1 import bay as bay_controller
from magnum.api.controllers.v1 import baymodel as baymodel_controller
from magnum.api.controllers.v1 import pod as pod_controller
from magnum.tests.db import utils
@ -35,4 +36,17 @@ def baymodel_post_data(**kw):
def bay_post_data(**kw):
bay = utils.get_test_bay(**kw)
internal = bay_controller.BayPatchType.internal_attrs()
return remove_internal(bay, internal)
return remove_internal(bay, internal)
def pod_post_data(**kw):
pod = utils.get_test_pod(**kw)
if 'manifest' not in pod:
pod['manifest'] = '''{
"id": "name_of_pod",
"labels": {
"foo": "foo1"
}
}'''
internal = pod_controller.PodPatchType.internal_attrs()
return remove_internal(pod, internal)