通过模拟 S3 存储桶进行单元测试

Question

我是单元测试的新手，我需要对对象存储类执行一些简单的单元测试。

我有一个名为 OSBucket 的类，如下所示：

def __initBucket(self):        
        
        ecs_session = boto3.Session(
        aws_access_key_id="OSKEY",
        aws_secret_access_key="SECRETKEY"
        ) 

        OS_resource = ecs_session.resource('s3', verify=cert, endpoint_url=endpoint)
        self.mybucket = OS_resource.Bucket(OS_BUCKET)
                
        
def get_mybucket(self): 
        
   return self.mybucket


def download_file(self, fileName,filepath):
        
   self.mybucket.download_file(fileName, filepath)

def upload_file(self, filepath,dest_file_name):   
   self.mybucket.upload_file(filepath, '%s%s' % ("/",dest_file_name))

在类的构造函数中调用方法 __initBucket。

我如何开始创建一个单元测试类来测试，例如 download_file 方法？

更新1

        moto_fake.start()
        conn = boto3.resource('s3', aws_access_key_id="fake_id",
        aws_secret_access_key="fake_secret")
        conn.create_bucket(Bucket="OS_BUCKET")    

        os_bucket = OSBucket.OSBucket(thisRun)

        sourcefile = "testingMoto.txt"   
        filePath = os.path.join("/", sourcefile)   
        os_bucket.upload_file(filePath, sourcefile)  

moto_fake.start()在创建对象之前执行的操作os_bucket对我不起作用。

更新2

使用 patch.object 将端点变量更改为 None，使测试通过

Answer 1

第一种方法：使用 python 模拟

您可以使用标准 python 模拟来模拟 s3 存储桶，然后检查您是否使用所需的参数调用方法。

然而，这种方法实际上并不能保证您的实现是正确的，因为您不会连接到 s3。例如，如果您拼错了名称，您可以调用不存在的 boto 函数 - 模拟不会抛出任何异常。

第二种方法：使用 moto

Moto是测试 boto 的最佳实践。它在本地计算机中模拟 boto，在本地创建存储桶，以便您可以进行完整的端到端测试。

我使用 moto 为你的类编写了几个测试（我可能在你的实现中发现了一个错误 - 检查最后一个测试行 - 这些是让它找到文件而不抛出异常所需的参数）

import pathlib
from tempfile import NamedTemporaryFile

import boto3
import moto
import pytest
from botocore.exceptions import ClientError

from os_bucket import OSBucket


@pytest.fixture
def empty_bucket():
    moto_fake = moto.mock_s3()
    try:
        moto_fake.start()
        conn = boto3.resource('s3')
        conn.create_bucket(Bucket="OS_BUCKET")  # or the name of the bucket you use
        yield conn
    finally:
        moto_fake.stop()


def test_download_non_existing_path(empty_bucket):
    os_bucket = OSBucket()
    os_bucket.initBucket()
    with pytest.raises(ClientError) as e:
        os_bucket.download_file("bad_path", "bad_file")
    assert "Not Found" in str(e)


def test_upload_and_download(empty_bucket):
    os_bucket = OSBucket()
    os_bucket.initBucket()
    with NamedTemporaryFile() as tmp:
        tmp.write(b'Hi')
        file_name = pathlib.Path(tmp.name).name

        os_bucket.upload_file(tmp.name, file_name)
        os_bucket.download_file("/" + file_name, file_name)  # this might indicate a bug in the implementation

最后注意事项：

• 我实际上就这个主题做了一场 Pycon 讲座，请在这里观看
• 讲座的源代码可以在这里找到