test.py 文件源码

def destroy_bucket(cls, connection_data, bucket):
        """Destroys the bucket and its content, just for teardown."""
        exc_num = 0
        try:
            with contextlib.closing(
                    boto.connect_s3(**connection_data)) as conn:
                if isinstance(bucket, basestring):
                    bucket = conn.lookup(bucket)
                    assert isinstance(bucket, s3.bucket.Bucket)
                for obj in bucket.list():
                    try:
                        bucket.delete_key(obj.key)
                        obj.close()
                    except BaseException:
                        LOG.exception("Failed to delete key %s " % obj.key)
                        exc_num += 1
            conn.delete_bucket(bucket)
        except BaseException:
            LOG.exception("Failed to destroy bucket %s " % bucket)
            exc_num += 1
        if exc_num:
            raise exceptions.TearDownException(num=exc_num)