使用 boto3 为将文件上传到 S3 的函数进行单元测试

Question

我有这个功能，可以将存档文件上传到 S3 存储桶：

def upload_file_to_s3_bucket(self, bucket, file, key, log):
    if not os.path.exists(file):
        log.error("File '%s' does not exist." % file)
        tools.exit_gracefully(log)
    log.info("Uploading file '%s' to bucket '%s' ..." % (file, bucket))
    try:
        self._s3.upload_file(file, bucket, key)
    except botocore.exceptions.ClientError as e:
        log.error("Unexpected uploading error : %s" % e)
        tools.exit_gracefully(log)
    log.info("Uploading finished.")

我想对其进行单元测试，这是到目前为止我可以写的内容：

class TestUploadFilesToS3(unittest.TestCase):
    """ Tests unitaires upload_file_to_s3_bucket"""


    def setUp(self):
        conf.LOG_FILE = "/tmp/test.log"
        conf.BUCKET_OUTPUT="name.of.the.bucket"
        conf.Conf.get_level_log()
        self.log = logger(conf.LOG_FILE, conf.LEVEL_LOG).logger
        tools.create_workdir(self.log)
        conf.WORKDIR = os.path.join(conf.LOCAL_DIR, "files/output")
        archive = "file_archive.tar.gz"
        archivePath = "/tmp/clients/file_archive.tar.gz"
        _aws = None

    def tearDown(self):
        tools.delete_workdir(self.log)
        os.remove(conf.LOG_FILE)


    def test_upload_file_to_s3_bucket_success(self):
        self._aws.upload_file_to_s3_bucket(conf.BUCKET_OUTPUT, archivePath, archive, self._log)

为了进行单元测试，我不知道应该在测试函数中使用哪个函数 Asserttest_upload_file_to_s3_bucket_success以及应该准确比较什么。例如，我可以测试文件的 URL 是否存在...？有任何想法吗？谢谢

Answer 1

这是我为 s3 上传功能编写的测试片段

    self.ti.uploadTemplate(contentsOfFile) # this is what is being tested
    # also supplied from elsewhere "contentsOfFile" and "nameOfFile"
    # bucket is assumed to be called "cloud-test-cf"

    s3 = boto3.resource('s3')
    mytname = nameOfFile
    obj = s3.Object(bucket_name='cloud-test-cf', key=mytname)
    response = obj.get()
    self.assertEqual(response['ContentLength'], len(contentsOfFile))
    remoteData = response['Body'].read()
    self.assertEqual(remoteData, contentsOfFile)

正如您所看到的，它在使用“uploadTemplate”后立即获取文件。如果您使用相同的区域/可用区设置，那么它应该可以正常工作，请参阅http://docs.aws.amazon.com/AmazonS3/latest/dev/Introduction.html#ConsistencyModel

该测试的unittest方法使用这样的装饰器

@unittest.skipIf(
         not(boto3.session.Config().region_name),
         "aws creds not loaded")
def testuploadTemplate(self):
    #....test code here

此装饰器意味着如果运行了单元测试套件但 AWS 密钥不可用，则会跳过此特定测试