test_s3StorageDriver.py 文件源码

def setUp(self):
        """
        Set up a mock S3 connection, bucket, and key, using moto.
        """
        self.bucket_name = 's3storagesdrivertest'
        conn = boto.connect_s3()
        # We need to create the bucket since this is all in Moto's 'virtual' AWS account
        conn.create_bucket(self.bucket_name)

        self.bucket = conn.get_bucket(self.bucket_name)
        key = self.bucket.new_key('the_dag/the_task/1983-09-05')

        data = 'this is a test.'
        key.set_metadata('Content-Type', 'text/plain')
        key.set_contents_from_string(data)
        key.set_acl('private')

        self.driver = S3StorageDriver('', '', self.bucket_name)