在全球 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 资源上使用 moto，但是在运行测试时，似乎没有设置 mock_s3 或者正在运行的测试无法屏蔽全局 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，我们必须确保在声明实际的 bot3 资源之前初始化 moto 模拟。我们如何确保全局资源被 moto 模拟？

Answer 1

我会制作一个固定装置并删除该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)