Ports the Tornado version of an S3 server to eventlet and wsgi, first step in deprecating the twistd-based objectstore.

This is a trivial implementation, never meant for production, it exists to provide an s3-look-alike objectstore for use when developing/testing things related to the amazon APIs (eucatools, etc), any production deployment would be expected to use Swift + an S3 interface.

In later patches I expect to be able to remove the old objectstore code entirely.

5 files changed, 154 insertions, 406 deletions

diff --git a/nova/tests/integrated/integrated_helpers.py b/nova/tests/integrated/integrated_helpers.py
index 47093636e..cc7326e73 100644
--- a/nova/tests/integrated/integrated_helpers.py
+++ b/nova/tests/integrated/integrated_helpers.py
@@ -75,8 +75,6 @@ class TestUser(object):
 
 
 class IntegratedUnitTestContext(object):
-    __INSTANCE = None
-
     def __init__(self):
         self.auth_manager = manager.AuthManager()
 
@@ -92,7 +90,6 @@ class IntegratedUnitTestContext(object):
 
     def setup(self):
         self._start_services()
-
         self._create_test_user()
 
     def _create_test_user(self):
@@ -109,14 +106,6 @@ class IntegratedUnitTestContext(object):
             self._start_api_service()
 
     def cleanup(self):
-        for service in self.services:
-            service.kill()
-        self.services = []
-        # TODO(justinsb): Shutdown WSGI & anything else we startup
-        # bug731668
-        # WSGI shutdown broken :-(
-        # self.wsgi_server.terminate()
-        # self.wsgi_server = None
         self.test_user = None
 
     def _create_unittest_user(self):
@@ -150,39 +139,8 @@ class IntegratedUnitTestContext(object):
         if not api_service:
             raise Exception("API Service was None")
 
-        # WSGI shutdown broken :-(
-        #self.services.append(volume_service)
         self.api_service = api_service
 
         self.auth_url = 'http://localhost:8774/v1.0'
 
         return api_service
-
-    # WSGI shutdown broken :-(
-    # bug731668
-    #@staticmethod
-    #def get():
-    #    if not IntegratedUnitTestContext.__INSTANCE:
-    #        IntegratedUnitTestContext.startup()
-    #        #raise Error("Must call IntegratedUnitTestContext::startup")
-    #    return IntegratedUnitTestContext.__INSTANCE
-
-    @staticmethod
-    def startup():
-        # Because WSGI shutdown is broken at the moment, we have to recycle
-        # bug731668
-        if IntegratedUnitTestContext.__INSTANCE:
-            #raise Error("Multiple calls to IntegratedUnitTestContext.startup")
-            IntegratedUnitTestContext.__INSTANCE.setup()
-        else:
-            IntegratedUnitTestContext.__INSTANCE = IntegratedUnitTestContext()
-        return IntegratedUnitTestContext.__INSTANCE
-
-    @staticmethod
-    def shutdown():
-        if not IntegratedUnitTestContext.__INSTANCE:
-            raise Error("Must call IntegratedUnitTestContext::startup")
-        IntegratedUnitTestContext.__INSTANCE.cleanup()
-        # WSGI shutdown broken :-(
-        # bug731668
-        #IntegratedUnitTestContext.__INSTANCE = None
diff --git a/nova/tests/integrated/test_login.py b/nova/tests/integrated/test_login.py
index 501f8c919..6b241f240 100644
--- a/nova/tests/integrated/test_login.py
+++ b/nova/tests/integrated/test_login.py
@@ -33,12 +33,12 @@ FLAGS.verbose = True
 class LoginTest(test.TestCase):
     def setUp(self):
         super(LoginTest, self).setUp()
-        context = integrated_helpers.IntegratedUnitTestContext.startup()
-        self.user = context.test_user
+        self.context = integrated_helpers.IntegratedUnitTestContext()
+        self.user = self.context.test_user
         self.api = self.user.openstack_api
 
     def tearDown(self):
-        integrated_helpers.IntegratedUnitTestContext.shutdown()
+        self.context.cleanup()
         super(LoginTest, self).tearDown()
 
     def test_login(self):
diff --git a/nova/tests/objectstore_unittest.py b/nova/tests/objectstore_unittest.py
deleted file mode 100644
index 4e2ac205e..000000000
--- a/nova/tests/objectstore_unittest.py
+++ /dev/null
@@ -1,315 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2010 United States Government as represented by the
-# Administrator of the National Aeronautics and Space Administration.
-# 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.
-
-"""
-Unittets for S3 objectstore clone.
-"""
-
-import boto
-import glob
-import hashlib
-import os
-import shutil
-import tempfile
-
-from boto.s3.connection import S3Connection, OrdinaryCallingFormat
-from twisted.internet import reactor, threads, defer
-from twisted.web import http, server
-
-from nova import context
-from nova import flags
-from nova import objectstore
-from nova import test
-from nova.auth import manager
-from nova.exception import NotEmpty, NotFound
-from nova.objectstore import image
-from nova.objectstore.handler import S3
-
-
-FLAGS = flags.FLAGS
-
-# Create a unique temporary directory. We don't delete after test to
-# allow checking the contents after running tests. Users and/or tools
-# running the tests need to remove the tests directories.
-OSS_TEMPDIR = tempfile.mkdtemp(prefix='test_oss-')
-
-# Create bucket/images path
-os.makedirs(os.path.join(OSS_TEMPDIR, 'images'))
-os.makedirs(os.path.join(OSS_TEMPDIR, 'buckets'))
-
-
-class ObjectStoreTestCase(test.TestCase):
-    """Test objectstore API directly."""
-
-    def setUp(self):
-        """Setup users and projects."""
-        super(ObjectStoreTestCase, self).setUp()
-        self.flags(buckets_path=os.path.join(OSS_TEMPDIR, 'buckets'),
-                   images_path=os.path.join(OSS_TEMPDIR, 'images'),
-                   ca_path=os.path.join(os.path.dirname(__file__), 'CA'))
-
-        self.auth_manager = manager.AuthManager()
-        self.auth_manager.create_user('user1')
-        self.auth_manager.create_user('user2')
-        self.auth_manager.create_user('admin_user', admin=True)
-        self.auth_manager.create_project('proj1', 'user1', 'a proj', ['user1'])
-        self.auth_manager.create_project('proj2', 'user2', 'a proj', ['user2'])
-        self.context = context.RequestContext('user1', 'proj1')
-
-    def tearDown(self):
-        """Tear down users and projects."""
-        self.auth_manager.delete_project('proj1')
-        self.auth_manager.delete_project('proj2')
-        self.auth_manager.delete_user('user1')
-        self.auth_manager.delete_user('user2')
-        self.auth_manager.delete_user('admin_user')
-        super(ObjectStoreTestCase, self).tearDown()
-
-    def test_buckets(self):
-        """Test the bucket API."""
-        objectstore.bucket.Bucket.create('new_bucket', self.context)
-        bucket = objectstore.bucket.Bucket('new_bucket')
-
-        # creator is authorized to use bucket
-        self.assert_(bucket.is_authorized(self.context))
-
-        # another user is not authorized
-        context2 = context.RequestContext('user2', 'proj2')
-        self.assertFalse(bucket.is_authorized(context2))
-
-        # admin is authorized to use bucket
-        admin_context = context.RequestContext('admin_user', None)
-        self.assertTrue(bucket.is_authorized(admin_context))
-
-        # new buckets are empty
-        self.assertTrue(bucket.list_keys()['Contents'] == [])
-
-        # storing keys works
-        bucket['foo'] = "bar"
-
-        self.assertEquals(len(bucket.list_keys()['Contents']), 1)
-
-        self.assertEquals(bucket['foo'].read(), 'bar')
-
-        # md5 of key works
-        self.assertEquals(bucket['foo'].md5, hashlib.md5('bar').hexdigest())
-
-        # deleting non-empty bucket should throw a NotEmpty exception
-        self.assertRaises(NotEmpty, bucket.delete)
-
-        # deleting key
-        del bucket['foo']
-
-        # deleting empty bucket
-        bucket.delete()
-
-        # accessing deleted bucket throws exception
-        self.assertRaises(NotFound, objectstore.bucket.Bucket, 'new_bucket')
-
-    def test_images(self):
-        self.do_test_images('1mb.manifest.xml', True,
-                            'image_bucket1', 'i-testing1')
-
-    def test_images_no_kernel_or_ramdisk(self):
-        self.do_test_images('1mb.no_kernel_or_ramdisk.manifest.xml',
-                            False, 'image_bucket2', 'i-testing2')
-
-    def do_test_images(self, manifest_file, expect_kernel_and_ramdisk,
-                             image_bucket, image_name):
-        "Test the image API."
-
-        # create a bucket for our bundle
-        objectstore.bucket.Bucket.create(image_bucket, self.context)
-        bucket = objectstore.bucket.Bucket(image_bucket)
-
-        # upload an image manifest/parts
-        bundle_path = os.path.join(os.path.dirname(__file__), 'bundle')
-        for path in glob.glob(bundle_path + '/*'):
-            bucket[os.path.basename(path)] = open(path, 'rb').read()
-
-        # register an image
-        image.Image.register_aws_image(image_name,
-                                       '%s/%s' % (image_bucket, manifest_file),
-                                       self.context)
-
-        # verify image
-        my_img = image.Image(image_name)
-        result_image_file = os.path.join(my_img.path, 'image')
-        self.assertEqual(os.stat(result_image_file).st_size, 1048576)
-
-        sha = hashlib.sha1(open(result_image_file).read()).hexdigest()
-        self.assertEqual(sha, '3b71f43ff30f4b15b5cd85dd9e95ebc7e84eb5a3')
-
-        if expect_kernel_and_ramdisk:
-            # Verify the default kernel and ramdisk are set
-            self.assertEqual(my_img.metadata['kernelId'], 'aki-test')
-            self.assertEqual(my_img.metadata['ramdiskId'], 'ari-test')
-        else:
-            # Verify that the default kernel and ramdisk (the one from FLAGS)
-            # doesn't get embedded in the metadata
-            self.assertFalse('kernelId' in my_img.metadata)
-            self.assertFalse('ramdiskId' in my_img.metadata)
-
-        # verify image permissions
-        context2 = context.RequestContext('user2', 'proj2')
-        self.assertFalse(my_img.is_authorized(context2))
-
-        # change user-editable fields
-        my_img.update_user_editable_fields({'display_name': 'my cool image'})
-        self.assertEqual('my cool image', my_img.metadata['displayName'])
-        my_img.update_user_editable_fields({'display_name': ''})
-        self.assert_(not my_img.metadata['displayName'])
-
-
-class TestHTTPChannel(http.HTTPChannel):
-    """Dummy site required for twisted.web"""
-
-    def checkPersistence(self, _, __):  # pylint: disable=C0103
-        """Otherwise we end up with an unclean reactor."""
-        return False
-
-
-class TestSite(server.Site):
-    """Dummy site required for twisted.web"""
-    protocol = TestHTTPChannel
-
-
-class S3APITestCase(test.TestCase):
-    """Test objectstore through S3 API."""
-
-    def setUp(self):
-        """Setup users, projects, and start a test server."""
-        super(S3APITestCase, self).setUp()
-
-        FLAGS.auth_driver = 'nova.auth.ldapdriver.FakeLdapDriver'
-        FLAGS.buckets_path = os.path.join(OSS_TEMPDIR, 'buckets')
-
-        self.auth_manager = manager.AuthManager()
-        self.admin_user = self.auth_manager.create_user('admin', admin=True)
-        self.admin_project = self.auth_manager.create_project('admin',
-                                                              self.admin_user)
-
-        shutil.rmtree(FLAGS.buckets_path)
-        os.mkdir(FLAGS.buckets_path)
-
-        root = S3()
-        self.site = TestSite(root)
-        # pylint: disable=E1101
-        self.listening_port = reactor.listenTCP(0, self.site,
-                                                interface='127.0.0.1')
-        # pylint: enable=E1101
-        self.tcp_port = self.listening_port.getHost().port
-
-        if not boto.config.has_section('Boto'):
-            boto.config.add_section('Boto')
-        boto.config.set('Boto', 'num_retries', '0')
-        self.conn = S3Connection(aws_access_key_id=self.admin_user.access,
-                                 aws_secret_access_key=self.admin_user.secret,
-                                 host='127.0.0.1',
-                                 port=self.tcp_port,
-                                 is_secure=False,
-                                 calling_format=OrdinaryCallingFormat())
-
-        def get_http_connection(host, is_secure):
-            """Get a new S3 connection, don't attempt to reuse connections."""
-            return self.conn.new_http_connection(host, is_secure)
-
-        self.conn.get_http_connection = get_http_connection
-
-    def _ensure_no_buckets(self, buckets):  # pylint: disable=C0111
-        self.assertEquals(len(buckets), 0, "Bucket list was not empty")
-        return True
-
-    def _ensure_one_bucket(self, buckets, name):  # pylint: disable=C0111
-        self.assertEquals(len(buckets), 1,
-                          "Bucket list didn't have exactly one element in it")
-        self.assertEquals(buckets[0].name, name, "Wrong name")
-        return True
-
-    def test_000_list_buckets(self):
-        """Make sure we are starting with no buckets."""
-        deferred = threads.deferToThread(self.conn.get_all_buckets)
-        deferred.addCallback(self._ensure_no_buckets)
-        return deferred
-
-    def test_001_create_and_delete_bucket(self):
-        """Test bucket creation and deletion."""
-        bucket_name = 'testbucket'
-
-        deferred = threads.deferToThread(self.conn.create_bucket, bucket_name)
-        deferred.addCallback(lambda _:
-                             threads.deferToThread(self.conn.get_all_buckets))
-
-        deferred.addCallback(self._ensure_one_bucket, bucket_name)
-
-        deferred.addCallback(lambda _:
-                             threads.deferToThread(self.conn.delete_bucket,
-                                                   bucket_name))
-        deferred.addCallback(lambda _:
-                             threads.deferToThread(self.conn.get_all_buckets))
-        deferred.addCallback(self._ensure_no_buckets)
-        return deferred
-
-    def test_002_create_bucket_and_key_and_delete_key_again(self):
-        """Test key operations on buckets."""
-        bucket_name = 'testbucket'
-        key_name = 'somekey'
-        key_contents = 'somekey'
-
-        deferred = threads.deferToThread(self.conn.create_bucket, bucket_name)
-        deferred.addCallback(lambda b:
-                             threads.deferToThread(b.new_key, key_name))
-        deferred.addCallback(lambda k:
-                             threads.deferToThread(k.set_contents_from_string,
-                                                   key_contents))
-
-        def ensure_key_contents(bucket_name, key_name, contents):
-            """Verify contents for a key in the given bucket."""
-            bucket = self.conn.get_bucket(bucket_name)
-            key = bucket.get_key(key_name)
-            self.assertEquals(key.get_contents_as_string(), contents,
-                              "Bad contents")
-
-        deferred.addCallback(lambda _:
-                             threads.deferToThread(ensure_key_contents,
-                                                   bucket_name, key_name,
-                                                   key_contents))
-
-        def delete_key(bucket_name, key_name):
-            """Delete a key for the given bucket."""
-            bucket = self.conn.get_bucket(bucket_name)
-            key = bucket.get_key(key_name)
-            key.delete()
-
-        deferred.addCallback(lambda _:
-                             threads.deferToThread(delete_key, bucket_name,
-                                                   key_name))
-        deferred.addCallback(lambda _:
-                             threads.deferToThread(self.conn.get_bucket,
-                                                   bucket_name))
-        deferred.addCallback(lambda b: threads.deferToThread(b.get_all_keys))
-        deferred.addCallback(self._ensure_no_buckets)
-        return deferred
-
-    def tearDown(self):
-        """Tear down auth and test server."""
-        self.auth_manager.delete_user('admin')
-        self.auth_manager.delete_project('admin')
-        stop_listening = defer.maybeDeferred(self.listening_port.stopListening)
-        super(S3APITestCase, self).tearDown()
-        return defer.DeferredList([stop_listening])
diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py
index cf8ee7eff..00803d0ad 100644
--- a/nova/tests/test_cloud.py
+++ b/nova/tests/test_cloud.py
@@ -35,31 +35,22 @@ from nova import log as logging
 from nova import rpc
 from nova import service
 from nova import test
+from nova import utils
 from nova.auth import manager
 from nova.compute import power_state
 from nova.api.ec2 import cloud
 from nova.api.ec2 import ec2utils
 from nova.image import local
-from nova.objectstore import image
 
 
 FLAGS = flags.FLAGS
 LOG = logging.getLogger('nova.tests.cloud')
 
-# Temp dirs for working with image attributes through the cloud controller
-# (stole this from objectstore_unittest.py)
-OSS_TEMPDIR = tempfile.mkdtemp(prefix='test_oss-')
-IMAGES_PATH = os.path.join(OSS_TEMPDIR, 'images')
-os.makedirs(IMAGES_PATH)
 
-
-# TODO(termie): these tests are rather fragile, they should at the lest be
-#               wiping database state after each run
 class CloudTestCase(test.TestCase):
     def setUp(self):
         super(CloudTestCase, self).setUp()
-        self.flags(connection_type='fake',
-                   images_path=IMAGES_PATH)
+        self.flags(connection_type='fake')
 
         self.conn = rpc.Connection.instance()
 
@@ -70,6 +61,7 @@ class CloudTestCase(test.TestCase):
         self.compute = self.start_service('compute')
         self.scheduter = self.start_service('scheduler')
         self.network = self.start_service('network')
+        self.image_service = utils.import_object(FLAGS.image_service)
 
         self.manager = manager.AuthManager()
         self.user = self.manager.create_user('admin', 'admin', 'admin', True)
@@ -318,41 +310,6 @@ class CloudTestCase(test.TestCase):
                 LOG.debug(_("Terminating instance %s"), instance_id)
                 rv = self.compute.terminate_instance(instance_id)
 
-    @staticmethod
-    def _fake_set_image_description(ctxt, image_id, description):
-        from nova.objectstore import handler
-
-        class req:
-            pass
-
-        request = req()
-        request.context = ctxt
-        request.args = {'image_id': [image_id],
-                        'description': [description]}
-
-        resource = handler.ImagesResource()
-        resource.render_POST(request)
-
-    def test_user_editable_image_endpoint(self):
-        pathdir = os.path.join(FLAGS.images_path, 'ami-testing')
-        os.mkdir(pathdir)
-        info = {'isPublic': False}
-        with open(os.path.join(pathdir, 'info.json'), 'w') as f:
-            json.dump(info, f)
-        img = image.Image('ami-testing')
-        # self.cloud.set_image_description(self.context, 'ami-testing',
-        #                                  'Foo Img')
-        # NOTE(vish): Above won't work unless we start objectstore or create
-        #             a fake version of api/ec2/images.py conn that can
-        #             call methods directly instead of going through boto.
-        #             for now, just cheat and call the method directly
-        self._fake_set_image_description(self.context, 'ami-testing',
-                                         'Foo Img')
-        self.assertEqual('Foo Img', img.metadata['description'])
-        self._fake_set_image_description(self.context, 'ami-testing', '')
-        self.assertEqual('', img.metadata['description'])
-        shutil.rmtree(pathdir)
-
     def test_update_of_instance_display_fields(self):
         inst = db.instance_create(self.context, {})
         ec2_id = ec2utils.id_to_ec2_id(inst['id'])
diff --git a/nova/tests/test_objectstore.py b/nova/tests/test_objectstore.py
new file mode 100644
index 000000000..c78772f27
--- /dev/null
+++ b/nova/tests/test_objectstore.py
@@ -0,0 +1,148 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2010 United States Government as represented by the
+# Administrator of the National Aeronautics and Space Administration.
+# 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.
+
+"""
+Unittets for S3 objectstore clone.
+"""
+
+import boto
+import glob
+import hashlib
+import os
+import shutil
+import tempfile
+
+from boto import exception as boto_exception
+from boto.s3 import connection as s3
+
+from nova import context
+from nova import exception
+from nova import flags
+from nova import wsgi
+from nova import test
+from nova.auth import manager
+from nova.objectstore import s3server
+
+
+FLAGS = flags.FLAGS
+
+# Create a unique temporary directory. We don't delete after test to
+# allow checking the contents after running tests. Users and/or tools
+# running the tests need to remove the tests directories.
+OSS_TEMPDIR = tempfile.mkdtemp(prefix='test_oss-')
+
+# Create bucket/images path
+os.makedirs(os.path.join(OSS_TEMPDIR, 'images'))
+os.makedirs(os.path.join(OSS_TEMPDIR, 'buckets'))
+
+
+class S3APITestCase(test.TestCase):
+    """Test objectstore through S3 API."""
+
+    def setUp(self):
+        """Setup users, projects, and start a test server."""
+        super(S3APITestCase, self).setUp()
+        self.flags(auth_driver='nova.auth.ldapdriver.FakeLdapDriver',
+                   buckets_path=os.path.join(OSS_TEMPDIR, 'buckets'),
+                   s3_host='127.0.0.1')
+
+        self.auth_manager = manager.AuthManager()
+        self.admin_user = self.auth_manager.create_user('admin', admin=True)
+        self.admin_project = self.auth_manager.create_project('admin',
+                                                              self.admin_user)
+
+        shutil.rmtree(FLAGS.buckets_path)
+        os.mkdir(FLAGS.buckets_path)
+
+        router = s3server.S3Application(FLAGS.buckets_path)
+        server = wsgi.Server()
+        server.start(router, FLAGS.s3_port, host=FLAGS.s3_host)
+
+        if not boto.config.has_section('Boto'):
+            boto.config.add_section('Boto')
+        boto.config.set('Boto', 'num_retries', '0')
+        conn = s3.S3Connection(aws_access_key_id=self.admin_user.access,
+                               aws_secret_access_key=self.admin_user.secret,
+                               host=FLAGS.s3_host,
+                               port=FLAGS.s3_port,
+                               is_secure=False,
+                               calling_format=s3.OrdinaryCallingFormat())
+        self.conn = conn
+
+        def get_http_connection(host, is_secure):
+            """Get a new S3 connection, don't attempt to reuse connections."""
+            return self.conn.new_http_connection(host, is_secure)
+
+        self.conn.get_http_connection = get_http_connection
+
+    def _ensure_no_buckets(self, buckets):  # pylint: disable=C0111
+        self.assertEquals(len(buckets), 0, "Bucket list was not empty")
+        return True
+
+    def _ensure_one_bucket(self, buckets, name):  # pylint: disable=C0111
+        self.assertEquals(len(buckets), 1,
+                          "Bucket list didn't have exactly one element in it")
+        self.assertEquals(buckets[0].name, name, "Wrong name")
+        return True
+
+    def test_000_list_buckets(self):
+        """Make sure we are starting with no buckets."""
+        self._ensure_no_buckets(self.conn.get_all_buckets())
+
+    def test_001_create_and_delete_bucket(self):
+        """Test bucket creation and deletion."""
+        bucket_name = 'testbucket'
+
+        self.conn.create_bucket(bucket_name)
+        self._ensure_one_bucket(self.conn.get_all_buckets(), bucket_name)
+        self.conn.delete_bucket(bucket_name)
+        self._ensure_no_buckets(self.conn.get_all_buckets())
+
+    def test_002_create_bucket_and_key_and_delete_key_again(self):
+        """Test key operations on buckets."""
+        bucket_name = 'testbucket'
+        key_name = 'somekey'
+        key_contents = 'somekey'
+
+        b = self.conn.create_bucket(bucket_name)
+        k = b.new_key(key_name)
+        k.set_contents_from_string(key_contents)
+
+        bucket = self.conn.get_bucket(bucket_name)
+
+        # make sure the contents are correct
+        key = bucket.get_key(key_name)
+        self.assertEquals(key.get_contents_as_string(), key_contents,
+                          "Bad contents")
+
+        # delete the key
+        key.delete()
+
+        self._ensure_no_buckets(bucket.get_all_keys())
+
+    def test_unknown_bucket(self):
+        bucket_name = 'falalala'
+        self.assertRaises(boto_exception.S3ResponseError,
+                          self.conn.get_bucket,
+                          bucket_name)
+
+    def tearDown(self):
+        """Tear down auth and test server."""
+        self.auth_manager.delete_user('admin')
+        self.auth_manager.delete_project('admin')
+        super(S3APITestCase, self).tearDown()