python - 单元测试：使用 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)

现在，我想确保我测试了这个功能，但我在这里被阻止了，因为：

我不知道测试这种怪异行为的最佳方法：
- 我是否应该实际创建一个假的本地 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.

我的问题如下：

我是否应该继续创建一个虚假的本地 s3 存储桶？
- 如果是这样，我应该如何让凭证工作？
我应该只模拟子流程调用吗？如何？
- 我很难理解模拟是如何工作的以及它应该如何完成。据我了解，我只是假装调用aws s3 sync并返回一些文件？
是否有另一种我没有想到的单元测试就足够了？
- 毕竟，我只想知道当我在那个桶中传输一个格式良好的s3://bucketurla和 a时，其中包含的文件是否会下载到我的.dirlocal dirs3://bucketurl/dirlocal dir

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

标签： pythonunit-testingamazon-s3mockingsubprocess

python - 单元测试：使用 Python 模拟运行“aws s3 sync”的子进程

问题描述

解决方案

推荐阅读