我的项目在进行处理之前需要定期下载相当多的文件。我尝试直接在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)
现在,我想确保我测试了这个函数但我在这里被阻塞了,因为:
- 我不知道测试这种怪异行为的最好方法:
- 我是否应该实际创建一个假的本地s3桶,以便
aws s3 sync
可以击中它? - 我应该模拟子进程调用而不是实际调用我的
download_bucket
函数吗?
- 我是否应该实际创建一个假的本地s3桶,以便
到目前为止,我的尝试是创建一个假桶并将其传递给我的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.
我的问题如下:
- 我应该继续创建一个假的本地s3桶吗?
- 如果是这样,我应该如何获得凭据工作?
- 我应该只是模拟子进程调用和如何?
- 我很难理解嘲笑是如何工作的,以及它应该如何完成。根据我的理解,我只是打电话给
aws s3 sync
并返回一些文件?
- 我很难理解嘲笑是如何工作的,以及它应该如何完成。根据我的理解,我只是打电话给
- 是否有另一种我没有想到的足够的单元测试?
- 毕竟,我只是想知道当我传输一个格式良好的
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的好处,而不是试图嘲笑。