在全局boto资源上使用moto

Question

我的my_module代码是这样的：

import boto3

S3_RESOURCE = boto3.resource('s3')

def some_func():
    local_file='local_file.txt'
    S3_RESOURCE.Object(bucket, key).download_file(Filename=local_file)

我正在尝试用moto测试这个方法，如下所示：

from moto import mock_s3
import unittest
import os

@mock_s3
class TestingS3(unittest.TestCase):

    def setUp(self):
        os.environ["AWS_DEFAULT_REGION"] = "us-east-1"
        os.environ["AWS_ACCESS_KEY_ID"] = "foobar_key"
        os.environ["AWS_SECRET_ACCESS_KEY"] = "foobar_secret"
        os.environ["AWS_SECURITY_TOKEN"] = 'testing'
        os.environ['AWS_SESSION_TOKEN'] = 'testing'
        self.setup_s3()
 
    def setup_s3(self):
        conn = boto3.client('s3')
        conn.create_bucket('my-bucket')
        conn.put_object(Bucket='my-bucket', Key='my-key', Body='data')

    def test_some_func(self):
        import my_module
        local_file = some_func('my-bucket','my-key')
        expected_file_path = 'local_file.txt'
        assert expected_file_path == local_file
        assert os.path.isfile(expected_file_path)

我正尝试在全局s3 boto资源上使用mock_s3，但是在运行测试时，似乎没有设置moto，或者正在运行的测试无法屏蔽全局s3资源。我得到以下错误

conn.create_bucket('my-bucket')
 if http.status_code >= 300:
             error_code = parsed_response.get("Error", {}).get("Code")
             error_class = self.exceptions.from_code(error_code)
 >           raise error_class(parsed_response, operation_name)
 E           botocore.exceptions.ClientError: An error occurred (ExpiredToken) when calling the CreateBucket operation: The provided token has expired.

但是，如果我在some_func()中本地初始化S3_RESOURCE，它就能正常工作。我知道为了使用moto，我们必须确保moto mock在声明实际的bot3资源之前被初始化。我们如何确保全局资源被moto模仿？

Answer 1

我会制作一个fixture并删除set_up s3方法

@pytest.yield_fixture
def s3():
    with mock_s3():
        s3 = boto3.client("s3")
        yield s3

然后在测试中

@mock_s3
def test_some_func(self, s3):
    import my_module
    CLIENT = s3
    bucket_name = "my-bucket"
    CLIENT.create_bucket(
    Bucket=bucket_name, CreateBucketConfiguration=bucket_config)
    CLIENT.put_object(Bucket='my-bucket', Key='my-key', Body='data')
    local_file = some_func('my-bucket','my-key')
    expected_file_path = 'local_file.txt'
    assert expected_file_path == local_file
    assert os.path.isfile(expected_file_path)

Answer 2

所提供的解决方案对我不起作用，我最终所做的是替换fixture部分上的那些资源

@pytest.fixture(autouse=True)
@mock_s3
def setup_module():
    my_module.S3_RESOURCE = boto3.resource("s3")