使用moto模拟S3桶时遭遇SSL验证失败问题求助
问题分析
你的代码中使用moto mock AWS服务,但aiobotocore的S3客户端仍尝试连接真实AWS S3域名,触发SSL验证失败。核心原因是moto对异步aiobotocore客户端不会自动替换请求端点,必须显式指定指向moto本地mock服务的地址。
修复方案
创建S3客户端时添加endpoint_url参数,指向moto默认的本地服务地址http://localhost:5000,同时修正代码中两处潜在问题:
put_object缺少Body参数,无法完成对象上传;get_object返回的是字典,需从中取出Body流对象传入AsyncFileIO。
修改后的完整代码:
@moto.mock_aws async def _load_asl_by_type(self, file: File, typ: ContentType) -> AsyncIterable[Tuple[str, Content]]: session = get_session() async with session.create_client( 's3', region_name='us-east-1', aws_secret_access_key='moto', aws_access_key_id='moto', endpoint_url='http://localhost:5000' # 关键:指定moto的mock端点 ) as s3: bucket_name = 'y-bucket' await s3.create_bucket(Bucket=bucket_name) # 补充Body参数,上传实际内容(示例为JSON格式,根据你的需求调整) await s3.put_object( Bucket=bucket_name, Key=file.path, Body=json.dumps({"test_key": "test_value"}).encode() ) try: # 获取对象并取出Body流 obj_response = await s3.get_object(Bucket=bucket_name, Key=file.path) async with AsyncFileIO(io=obj_response['Body']) as f: file.in_progress() collection = json.load(f) for name, data in collection.items(): def _spec_load() -> Any: _logger.info(f"Loading {typ.__qualname__} from {file.path}: {name}") return spec.load(typ, data, key=name, options=self._options) obj = await asyncio.get_running_loop().run_in_executor(None, _spec_load) file.loaded() yield name, obj except Exception as error: _logger.error(f"Failed to load file {file.path}: {error}") file.failed() raise error
额外提示
- 确保moto版本为最新,旧版本对aiobotocore的异步支持可能存在缺陷:
pip install --upgrade moto - 如果你的测试环境无法使用5000端口,可以在启动moto时指定其他端口,并同步修改
endpoint_url中的端口号。
内容的提问来源于stack exchange,提问作者kumaroid
相关产品推荐
相关产品推荐

