单元测试:模拟使用 Python 运行"aws s3 sync"的子进程



我的项目在进行处理之前需要定期下载相当多的文件。我尝试直接在Python中编码,但考虑到桶中的数据量,它非常慢。

我决定使用运行aws-cli的子进程,因为boto3仍然没有同步功能。我知道使用aws-cli的子进程是不理想的,但它确实是有用的,并且开箱操作非常好。

aws-cli的好处之一是我可以看到stdout的进展,这是我使用以下代码获得的:

def download_bucket(bucket_url, dir_name, dest):
"""Download all the files from a bucket into a directory."""
path = Path(dest) / dir_name
bucket_dest = str(os.path.join(bucket_url, dir_name))
with subprocess.Popen(["aws", "s3", "sync", bucket_dest, path], stdout=subprocess.PIPE, bufsize=1, universal_newlines=True) as p:
for b in p.stdout:
print(b, end='')
if p.returncode != 0:
raise subprocess.CalledProcessError(p.returncode, p.args)

现在,我想确保我测试了这个函数但我在这里被阻塞了,因为:

  1. 我不知道测试这种怪异行为的最好方法:
    • 我是否应该实际创建一个假的本地s3桶,以便aws s3 sync可以击中它?
    • 我应该模拟子进程调用而不是实际调用我的download_bucket函数吗?

到目前为止,我的尝试是创建一个假桶并将其传递给我的download_bucket函数。这样,我认为aws s3 sync仍然会工作,尽管在本地:

def test_download_s3(tmpdir):
tmpdir.join(f'frankendir').ensure()
with mock_s3():
conn = boto3.resource('s3', region_name='us-east-1')
conn.create_bucket(Bucket='cool-bucket.us-east-1.dev.000000000000')
s3 = boto3.client('s3', region_name="us-east-1")
s3.put_object(Bucket='cool-bucket.us-east-1.dev.000000000000', Key='frankendir', Body='has no files')
body = conn.Object('cool-bucket.us-east-1.dev.000000000000', 'frankendir').get()[
'Body'].read().decode("utf-8")
download_bucket('s3://cool-bucket.us-east-1.dev.000000000000', 'frankendir', tmpdir)
#assert tmpdir.join('frankendir').join('has not files').exists()
assert body == 'has no files'

但我得到以下错误fatal error: An error occurred (InvalidAccessKeyId) when calling the ListObjects operation: The AWS Access Key Id you provided does not exist in our records.

我的问题如下:

  1. 我应该继续创建一个假的本地s3桶吗?
    • 如果是这样,我应该如何获得凭据工作?
  2. 我应该只是模拟子进程调用和如何?
    • 我很难理解嘲笑是如何工作的,以及它应该如何完成。根据我的理解,我只是打电话给aws s3 sync并返回一些文件?
  3. 是否有另一种我没有想到的足够的单元测试?
    • 毕竟,我只是想知道当我传输一个格式良好的s3://bucketurl,一个dir和一个local dir时,s3://bucketurl/dir中包含的文件是否下载到我的local dir

谢谢你的帮助,我希望我不是到处都是。

更好的方法是在伪造/测试s3时使用moto。您可以查看他们的文档或查看我所做的测试代码示例:https://github.com/pksol/pycon-go-beyond-mocks/blob/main/testrong3_fake.py.

如果你有几分钟的时间,你可以看看这个简短的视频,我解释了使用moto的好处,而不是试图嘲笑。

相关内容

  • 没有找到相关文章

最新更新