如何用S3FileSystem、Pytest和Moto访问模拟S3存储桶

使用Pytest、Moto（4.1.6）和s3fs（0.4.2）编写S3交互功能的单元测试时，已成功创建模拟存储桶并上传data文件夹内的所有文件，但通过s3fs.core.S3FileSystem访问存储桶时遭遇权限拒绝错误，且系统错误地将本地路径的一部分识别为存储桶名称。

相关代码（conftest.py） #

#!/usr/bin/env python3

from moto import mock_s3
from pathlib import Path

import boto3
import os
import pytest
import s3fs


@pytest.fixture(scope="session")
def test_data_folder():
    return os.path.join(os.path.dirname(__file__), "data")


@pytest.fixture(scope="session")
@mock_s3
def s3_filesystem(test_data_folder):
    connection = boto3.client("s3", region_name="us-east-1")
    connection.create_bucket(Bucket="bucket")

    for path in Path(test_data_folder).rglob("*"):
        if path.is_file():
            with open(path, "rb") as parquet:
               data = parquet.read()
               connection.put_object(Bucket="bucket", Key=str(path), Body=data)

    bucket = boto3.resource("s3").Bucket("bucket")
    for object in bucket.objects.all():
        print(object.key)

    filesystem = s3fs.S3FileSystem(anon=True)
    filesystem.ls(test_data_folder)
    return filesystem

错误现象 #

1. 打印显示存储桶内存在类似/Users/campos/repos/project/tests/data/20221027/transactions_test_20221027.parquet的文件。
2. 执行filesystem.ls(test_data_folder)报错：PermissionError: All access to this object has been disabled。
3. 深层错误信息显示：

api_params = {'Bucket': 'Users', 'Delimiter': '/', 'EncodingType': 'url', 'Prefix': 'ykb595/repos/gfctr-card-lob-pyexetl/tests/data/'}

...

botocore.errorfactory.NoSuchBucket: An error occurred (NoSuchBucket) when calling the ListObjectsV2 operation: The specified bucket does not exist

问题原因 #

1. 路径格式错误：s3fs访问S3资源的正确格式是存储桶名/对象路径，但你直接传入了本地绝对路径test_data_folder，导致s3fs误将路径的第一个分段（如Users）识别为存储桶名称，而非你创建的bucket桶。
2. 对象Key不规范：上传文件时使用了本地绝对路径作为S3对象的Key，这不仅不符合S3的使用习惯，还会导致后续访问路径混乱。

解决方案 #

1. 修改上传文件的Key为相对路径 #

将上传时的Key改为相对于test_data_folder的路径，避免本地绝对路径干扰：

# 替换原put_object的Key参数
connection.put_object(Bucket="bucket", Key=str(path.relative_to(test_data_folder)), Body=data)

2. 调整s3fs的访问路径 #

使用s3fs访问时，需指定正确的存储桶路径格式：

# 替换原filesystem.ls调用
filesystem.ls("bucket/")  # 列出桶内所有对象
# 或者列出特定子路径
filesystem.ls(f"bucket/{Path(test_data_folder).name}/")

3. 确保s3fs连接到Moto的模拟端点 #

s3fs默认可能尝试连接真实AWS服务，需显式指定Moto的本地端点：

filesystem = s3fs.S3FileSystem(
    anon=True,
    client_kwargs={
        'endpoint_url': 'http://localhost:5000',  # Moto默认端口
        'region_name': 'us-east-1'
    }
)

修改后的完整fixture代码 #

@pytest.fixture(scope="session")
@mock_s3
def s3_filesystem(test_data_folder):
    connection = boto3.client("s3", region_name="us-east-1")
    connection.create_bucket(Bucket="bucket")

    test_data_path = Path(test_data_folder)
    for path in test_data_path.rglob("*"):
        if path.is_file():
            relative_key = path.relative_to(test_data_path)
            with open(path, "rb") as parquet:
               data = parquet.read()
               connection.put_object(Bucket="bucket", Key=str(relative_key), Body=data)

    bucket = boto3.resource("s3").Bucket("bucket")
    for object in bucket.objects.all():
        print(object.key)

    filesystem = s3fs.S3FileSystem(
        anon=True,
        client_kwargs={
            'endpoint_url': 'http://localhost:5000',
            'region_name': 'us-east-1'
        }
    )
    # 正确访问桶内路径
    print(filesystem.ls("bucket/"))
    return filesystem

内容的提问来源于stack exchange，提问作者A Campos