Created simple test case for server creation, so that we can have something to attach to...

This commit is contained in:
Justin Santa Barbara 2011-03-24 01:13:20 -07:00
parent 143a8387fc
commit 694c2cfd2a
4 changed files with 352 additions and 0 deletions

View File

@ -16,3 +16,6 @@ nova/vcsversion.py
*.DS_Store *.DS_Store
.project .project
.pydevproject .pydevproject
clean.sqlite
run_tests.log
tests.sqlite

109
nova/image/fake.py Normal file
View File

@ -0,0 +1,109 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 Justin Santa Barbara
# All Rights Reserved.
#
# 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.
"""Implementation of an fake image service"""
from nova import exception
from nova import flags
from nova import log as logging
from nova.image import service
LOG = logging.getLogger('nova.image.fake')
FLAGS = flags.FLAGS
class MockImageService(service.BaseImageService):
"""Mock (fake) image service for unit testing"""
def __init__(self):
self.images = {}
# NOTE(justinsb): The OpenStack API can't upload an image???
# So, make sure we've got one..
image = {'id': '123456',
'status': 'active',
'type': 'machine',
'disk_format': 'ami',
'properties': {'kernel_id': FLAGS.null_kernel,
'ramdisk_id': FLAGS.null_kernel,
}
}
self.create(None, image)
super(MockImageService, self).__init__()
def index(self, context):
"""Returns list of images"""
return self.images.values()
def detail(self, context):
"""Return list of detailed image information"""
return self.images.values()
def show(self, context, image_id):
"""
Returns a dict containing image data for the given opaque image id.
"""
image_id = int(image_id)
image = self.images.get(image_id)
if image:
return image
LOG.warn("Unable to find image id %s. Have images: %s",
image_id, self.images)
raise exception.NotFound
def create(self, context, data):
"""
Store the image data and return the new image id.
:raises AlreadyExists if the image already exist.
"""
image_id = int(data['id'])
if self.images.get(image_id):
#TODO(justinsb): Where is this AlreadyExists exception??
raise exception.Error("AlreadyExists")
self.images[image_id] = data
def update(self, context, image_id, data):
"""Replace the contents of the given image with the new data.
:raises NotFound if the image does not exist.
"""
image_id = int(image_id)
if not self.images.get(image_id):
raise exception.NotFound
self.images[image_id] = data
def delete(self, context, image_id):
"""
Delete the given image.
:raises NotFound if the image does not exist.
"""
image_id = int(image_id)
removed = self.images.pop(image_id, None)
if not removed:
raise exception.NotFound
def delete_all(self):
"""
Clears out all images
"""
self.images.clear()

View File

@ -73,6 +73,28 @@ class TestUser(object):
self.secret, self.secret,
self.auth_url) self.auth_url)
def get_unused_server_name(self):
servers = self.openstack_api.get_servers()
server_names = [server['name'] for server in servers]
return generate_new_element(server_names, 'server')
def get_invalid_image(self):
images = self.openstack_api.get_images()
image_ids = [image['id'] for image in images]
return generate_new_element(image_ids, '', numeric=True)
def get_valid_image(self, create=False):
images = self.openstack_api.get_images()
if create and not images:
# TODO(justinsb): No way to create an image through API???
#created_image = self.openstack_api.post_image(image)
#images.append(created_image)
raise exception.Error("No way to create an image through API??")
if images:
return images[0]
return None
class IntegratedUnitTestContext(object): class IntegratedUnitTestContext(object):
__INSTANCE = None __INSTANCE = None

View File

@ -0,0 +1,218 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 Justin Santa Barbara
# All Rights Reserved.
#
# 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 time
import unittest
from nova import flags
from nova import test
from nova.log import logging
from nova.tests.integrated import integrated_helpers
from nova.tests.integrated.api import client
LOG = logging.getLogger('nova.tests.integrated')
FLAGS = flags.FLAGS
FLAGS.verbose = True
class ServersTest(test.TestCase):
def setUp(self):
super(ServersTest, self).setUp()
self.flags(image_service='nova.image.fake.MockImageService')
context = integrated_helpers.IntegratedUnitTestContext.startup()
self.user = context.test_user
self.api = self.user.openstack_api
def tearDown(self):
integrated_helpers.IntegratedUnitTestContext.shutdown()
super(ServersTest, self).tearDown()
def test_get_servers(self):
"""Simple check that listing servers works."""
servers = self.api.get_servers()
for server in servers:
LOG.debug("server: %s" % server)
def test_create_and_delete_server(self):
"""Creates and deletes a server"""
# Create server
# Build the server data gradually, checking errors along the way
server = {}
good_server = self._build_minimal_create_server_request()
post = {'server': server}
# Without an imageId, this throws 500.
# TODO(justinsb): Check whatever the spec says should be thrown here
self.assertRaises(client.OpenStackApiException,
self.api.post_server, post)
# With an invalid imageId, this throws 500.
server['imageId'] = self.user.get_invalid_image()
# TODO(justinsb): Check whatever the spec says should be thrown here
self.assertRaises(client.OpenStackApiException,
self.api.post_server, post)
# Add a valid imageId
server['imageId'] = good_server['imageId']
# Without flavorId, this throws 500
# TODO(justinsb): Check whatever the spec says should be thrown here
self.assertRaises(client.OpenStackApiException,
self.api.post_server, post)
# Set a valid flavorId
server['flavorId'] = good_server['flavorId']
# Without a name, this throws 500
# TODO(justinsb): Check whatever the spec says should be thrown here
self.assertRaises(client.OpenStackApiException,
self.api.post_server, post)
# Set a valid server name
server['name'] = good_server['name']
created_server = self.api.post_server(post)
LOG.debug("created_server: %s" % created_server)
self.assertTrue(created_server['id'])
created_server_id = created_server['id']
# Check it's there
found_server = self.api.get_server(created_server_id)
self.assertEqual(created_server_id, found_server['id'])
# It should also be in the all-servers list
servers = self.api.get_servers()
server_ids = [server['id'] for server in servers]
self.assertTrue(created_server_id in server_ids)
# Wait (briefly) for creation
retries = 0
while found_server['status'] == 'build':
LOG.debug("found server: %s" % found_server)
time.sleep(1)
found_server = self.api.get_server(created_server_id)
retries = retries + 1
if retries > 5:
break
# It should be available...
# TODO(justinsb): Mock doesn't yet do this...
#self.assertEqual('available', found_server['status'])
self._delete_server(created_server_id)
def _delete_server(self, server_id):
# Delete the server
self.api.delete_server(server_id)
# Wait (briefly) for deletion
for _retries in range(5):
try:
found_server = self.api.get_server(server_id)
except client.OpenStackApiNotFoundException:
found_server = None
LOG.debug("Got 404, proceeding")
break
LOG.debug("Found_server=%s" % found_server)
# TODO(justinsb): Mock doesn't yet do accurate state changes
#if found_server['status'] != 'deleting':
# break
time.sleep(1)
# Should be gone
self.assertFalse(found_server)
def _build_minimal_create_server_request(self):
server = {}
image = self.user.get_valid_image(create=True)
image_id = image['id']
#TODO(justinsb): This is FUBAR
image_id = abs(hash(image_id))
# We now have a valid imageId
server['imageId'] = image_id
# Set a valid flavorId
flavor = self.api.get_flavors()[0]
LOG.debug("Using flavor: %s" % flavor)
server['flavorId'] = flavor['id']
# Set a valid server name
server_name = self.user.get_unused_server_name()
server['name'] = server_name
return server
# TODO(justinsb): Enable this unit test when the metadata bug is fixed
# def test_create_server_with_metadata(self):
# """Creates a server with metadata"""
#
# # Build the server data gradually, checking errors along the way
# server = self._build_minimal_create_server_request()
#
# for metadata_count in range(30):
# metadata = {}
# for i in range(metadata_count):
# metadata['key_%s' % i] = 'value_%s' % i
# server['metadata'] = metadata
#
# post = {'server': server}
# created_server = self.api.post_server(post)
# LOG.debug("created_server: %s" % created_server)
# self.assertTrue(created_server['id'])
# created_server_id = created_server['id']
# # Reenable when bug fixed
# # self.assertEqual(metadata, created_server.get('metadata'))
#
# # Check it's there
# found_server = self.api.get_server(created_server_id)
# self.assertEqual(created_server_id, found_server['id'])
# self.assertEqual(metadata, found_server.get('metadata'))
#
# # The server should also be in the all-servers details list
# servers = self.api.get_servers(detail=True)
# server_map = dict((server['id'], server) for server in servers)
# found_server = server_map.get(created_server_id)
# self.assertTrue(found_server)
# # Details do include metadata
# self.assertEqual(metadata, found_server.get('metadata'))
#
# # The server should also be in the all-servers summary list
# servers = self.api.get_servers(detail=False)
# server_map = dict((server['id'], server) for server in servers)
# found_server = server_map.get(created_server_id)
# self.assertTrue(found_server)
# # Summary should not include metadata
# self.assertFalse(found_server.get('metadata'))
#
# # Cleanup
# self._delete_server(created_server_id)
if __name__ == "__main__":
unittest.main()