python - pytest 如何模拟 s3fs.S3FileSystem 打开文件

Question

我正在尝试模拟调用以打开 S3 存储桶中的文件。我的代码是:

# mymodule.py
import s3fs
#...
def __init__(self):
    self.s3_filesystem = s3fs.S3FileSystem(anon=False, key=s3_key,
                                           secret=s3_secret)
#...
def mymethod(self):
    with self.s3_filesystem.open(filename, 'r') as csv_file:
        file_dataframe = pd.read_csv(csv_file)

我在 pytest 中的模拟是:

import pytest

def test(mocker)                                       
    mocker.patch('mypackage.mymodule.s3fs.S3FileSystem.open',
                 return_value=open)

但是在运行测试时我得到了错误:

>       with self.s3_filesystem.open(filename, 'r') as csv_file:
E       AttributeError: __enter__

知道为什么吗？

Answer 1

免责声明:我回答的所有功劳都归功于 OP Juan Fernando Jaramillo Botero以及他对上面自己的问题的评论。我在这里发布一个答案，因为 OP 目前似乎没有至少 15 个声望点 - 因此，无法回答他自己的问题，如 in this help page 所示。 . 拜托，如果不是这种情况，我鼓励 OP 添加他的答案 - 并接受它 - 因为它对我帮助很大!

---

我遇到了同样的问题，我不得不从 s3fs.S3FileSystem 模拟两个方法:open(作为 OP)，以及 ls，在 pytest 测试套件中。

通过使用装饰器 patch 及其参数 side_effect，所有对 s3fs.S3FileSystem.open 和 s3fs.S3FileSystem 的调用.ls 在测试期间被正确地替换为 open 和 os.listdir，在我的例子中这是预期的效果。

简化情况:

我的模块.py

import s3fs


class Operation():
    def compile_files(self, fs, input_bucket):
        files = fs.ls(input_bucket)

        return files

    def copy_files(self, fs, files, input_bucket, output_bucket):
        for file in files:
            with fs.open(f'{input_bucket}/{file}', 'rb') as source:
                with fs.open(f'{output_bucket}/{file}', 'wb') as destination:
                    destination.write(source.read())

    def transform(self, input_bucket, output_bucket):
        fs = s3fs.S3FileSystem()

        files = self.compile_files(fs, input_bucket)

        self.copy_files(fs, files, input_bucket, output_bucket)

my_module_test.py

import os
from unittest.mock import patch

from my_module import Operation


@patch('s3fs.S3FileSystem.open', side_effect=open)
@patch('s3fs.S3FileSystem.ls', side_effect=os.listdir)
def test_my_module(mock_s3fs_ls, mock_s3fs_open):
    input_bucket = 'test/fixtures/files'
    output_bucket = 'test/fixtures/tmp_output'
    os.mkdir(output_bucket)

    Operation().transform(input_bucket, output_bucket)

    # Proceed to assert everything went as expected...

请注意，测试文件不导入 pytest，但它是完全兼容的，可以在 pytest 测试套件中运行。