2

我正在尝试模拟在 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__

知道为什么吗？

4

1 回答 1

0

Disclaimer: All credits to my answer are due to the OP Juan Fernando Jaramillo Botero and his comment to his own question, above. I am posting an answer here, because it appears that the OP does not have at least 15 reputation points at the moment - and thus, can't answer his own question, as indicated in this help page. Please, if this is not the case, I encourage the OP to add his answer - and accept it - because it helped me a lot!

I was having the same problem, where I had to mock two methods from s3fs.S3FileSystem: open (as OP), and also ls, in a pytest suite of tests.

By using the decorator patch, and its argument side_effect, all calls for s3fs.S3FileSystem.open and s3fs.S3FileSystem.ls during tests are properly replaced by open and os.listdir, respectively, which in my case was the desired effect.

Simplified situation:

my_module.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...

Note that the test file does not import pytest, but it is fully compatible and can be run within a pytest test suite.

于 2021-07-04T16:32:14.447 回答