Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载导读SageMaker Runtime 是 AWS 推理服务中负责向已部署端点发起调用同步InvokeEndpoint、异步InvokeEndpointAsync的运行时组件。在 moto 中sagemaker-runtime后端为这两类调用提供了可编程的 Mock 能力默认情况下返回静态数据开发者可以借助 moto 专属的/moto-api/static/sagemaker/...接口预先配置预期结果队列从而精确控制每次调用的响应内容、失败注入与返回头。读完本文你将掌握该模块的全部已实现能力、结果队列的入队与出队规则、同步/异步调用在底层源码中的真实行为以及如何在单元测试中按场景编排这些结果。本文以 docs/docs/services/sagemaker-runtime.rst 为骨架结合 moto/sagemakerruntime/models.py、moto/sagemakerruntime/responses.py 与 tests/test_sagemakerruntime/test_sagemakerruntime.py 展开源码级讲解。一、已实现功能总览根据官方服务文档见 docs/docs/services/sagemaker-runtime.rstmoto 对sagemaker-runtime的支持情况如下操作状态说明invoke_endpoint✅ 已实现默认返回静态数据可通过结果队列定制invoke_endpoint_async✅ 已实现默认返回静态数据可通过异步结果队列定制支持失败注入与 S3 输出落盘invoke_endpoint_with_response_stream❌ 未实现流式响应即 SageMaker 的流式推理暂不支持也就是说moto 目前覆盖了推理调用中最常用的同步与异步两条路径凡是请求了invoke_endpoint_with_response_stream的代码在 Mock 环境下会得到未实现的响应测试设计时需要避开该操作或自行做条件分支。二、默认行为静态数据2.1 同步调用invoke_endpoint在不做任何额外配置时invoke_endpoint会返回一组写死的静态数据。其默认值定义在 moto/sagemakerruntime/models.py 的SageMakerRuntimeBackend.invoke_endpoint方法中self.results[endpoint_name][unique_repr] ( body, content_type, invoked_production_variant, custom_attributes, )返回值是一个四元组四个元素会被 moto/sagemakerruntime/responses.py 的SageMakerRuntimeResponse.invoke_endpoint映射到 Boto3 客户端返回结构的不同字段四元组元素默认值映射到的返回字段bodybody响应的Body可读流content_typecontent_type响应头Content-Typeinvoked_production_variantinvoked_production_variant响应头x-Amzn-Invoked-Production-Variantcustom_attributescustom_attributes返回结构中的CustomAttributes字段这一点在 tests/test_sagemakerruntime/test_sagemakerruntime.py 的test_invoke_endpoint__default_results中得到验证即使调用时传入了Acceptsth、TargetModeltm返回的Body仍是bbodyCustomAttributes仍是custom_attributes。从源码结构可以推断同一次 Mock 会话内、相同请求特征的调用会命中缓存invoke_endpoint内部以endpoint_name为键维护一个嵌套字典self.results请求的唯一标识unique_repr则由请求头中所有以x-amzn-sagemaker开头的字段、Accept头以及请求体Body共同计算并 base64 编码而来见 moto/sagemakerruntime/responses.py。因此相同 EndpointName 相同请求头 相同 Body的重复调用会复用第一次的结果。2.2 异步调用invoke_endpoint_async异步调用的默认行为同样是静态数据但其返回结构完全不同。真实 AWS 的异步推理会将结果写入用户指定的 S3 位置moto 模拟了这一流程默认返回的数据是json.dumps({default: response})并会在当前账号/分区下自动创建一个名为sagemaker-output-{uuid}的 S3 Bucket将结果以response.json写入见 moto/sagemakerruntime/models.py。底层实现位于SageMakerRuntimeBackend.invoke_endpoint_asyncmoto/sagemakerruntime/models.py它最终返回(output_location, failure_location)两个 S3 路径其中output_locations3://sagemaker-output-{uuid}/response.jsonfailure_locations3://sagemaker-output-{uuid}/failure.json仅当本次结果被标记为失败时数据才会写入该文件。对应的响应层 moto/sagemakerruntime/responses.py 会把这些位置放入响应头X-Amzn-SageMaker-OutputLocationX-Amzn-SageMaker-FailureLocationInferenceId若请求未携带X-Amzn-SageMaker-Inference-Id头则由random.uuid4()生成这与真实 AWS 的异步推理语义保持一致客户端拿到OutputLocation后再到 S3 中读取推理结果。三、用 moto-api 结果队列定制响应3.1 队列机制的核心思想无论是同步还是异步调用moto 都提供了一套预期结果队列机制通过向 moto 专属的 HTTP 接口 POST 一个 JSON 负载将一系列结果按顺序压入后端队列之后每次新的推理请求会从队列头部弹出一个结果。该机制的关键规则官方文档与源码一致是相同请求 → 相同结果后续使用相同请求特征的调用同步场景看请求头/Body异步场景看InputLocation会返回与第一次完全相同的结果而不会再次消费队列不同请求 → 队列下一个结果其他使用不同特征的新请求会取走队列中的下一条队列为空 → 回退静态数据队列耗尽后新请求返回默认静态数据。从源码看这一规则在 moto/sagemakerruntime/models.py同步与第 107-144 行异步中以同样的模式实现先查缓存字典命中即返回未命中再从队列pop(0)否则生成默认值最后把结果写回缓存。3.2 配置同步结果队列endpoint-results向/moto-api/static/sagemaker/endpoint-results发起 POST 即可配置同步调用invoke_endpoint的结果队列。官方文档给出的示例负载如下原文见 docs/docs/services/sagemaker-runtime.rstexpected_results { account_id: 123456789012, # 默认账号可省略 region: us-east-1, # 默认区域可省略 results: [ { Body: first body, ContentType: text/xml, InvokedProductionVariant: prod, CustomAttributes: my_attr, }, # 可按需添加更多结果 ], } requests.post( http://motoapi.amazonaws.com/moto-api/static/sagemaker/endpoint-results, jsonexpected_results, ) client boto3.client(sagemaker-runtime, region_nameus-east-1) details client.invoke_endpoint(EndpointNameasdf, Bodyqwer)每个结果条目支持的字段与对应关系如下字段是否必填对应行为Body是返回给客户端的推理响应体ContentType否写入响应头Content-TypeInvokedProductionVariant否写入响应头x-Amzn-Invoked-Production-VariantCustomAttributes否写入返回结构中的CustomAttributes字段account_id与region缺省时分别使用默认账号与us-east-1这一点在 moto-api 的解析代码 moto/moto_api/_internal/responses.py 中可以看到account_id body.get(account_id, DEFAULT_ACCOUNT_ID)、region body.get(region, us-east-1)随后逐条把结果追加到sagemakerruntime_backends[account_id][region].results_queue见 moto/moto_api/_internal/models.py。也就是说队列是按账号 区域隔离的多账号/多区域测试时需各自配置。3.3 配置异步结果队列async-endpoint-results异步调用的结果队列通过/moto-api/static/sagemaker/async-endpoint-results配置每个条目额外支持is_failure字段来模拟推理失败。官方文档示例expected_results { account_id: 123456789012, # 默认账号可省略 region: us-east-1, # 默认区域可省略 results: [ { data: json.dumps({first: output}), }, { is_failure: True, data: second inference failed, }, # 可按需添加更多结果 ], } requests.post( http://motoapi.amazonaws.com/moto-api/static/sagemaker/async-endpoint-results, jsonexpected_results, ) client boto3.client(sagemaker-runtime, region_nameus-east-1) details client.invoke_endpoint_async(EndpointNameasdf, InputLocationqwer)字段语义字段是否必填对应行为data是写入 S3 对象的内容建议用json.dumps序列化is_failure否默认为False为True时数据写入failure.json并返回对应的FailureLocation队列消费时is_failureTrue的结果会走失败分支把data原样写入failure.json成功结果写入response.json。测试 tests/test_sagemakerruntime/test_sagemakerruntime.py 的test_invoke_endpoint_async验证了这条完整链路配置两个结果 → 第一次调用拿到OutputLocation→ 从 S3 读回内容为{first: output}→ 换一个InputLocation触发第二条结果is_failureTrue→ 从FailureLocation读回second failure同时InferenceId与请求携带的保持一致。3.4 向后兼容异步队列为空时回退同步队列这是异步调用特有的规则官方文档明确说明如果异步队列为空将使用已配置的同步队列for backward compatibility。底层实现见 moto/sagemakerruntime/models.py当async_results_queue为空但results_queue非空时会从同步队列弹出条目并把四元组包装成 JSON 写入 S3elif self.results_queue: # Backward compatibility is_failure False body, _type, variant, attrs self.results_queue.pop(0) data json.dumps( { Body: body, ContentType: _type, InvokedProductionVariant: variant, CustomAttributes: attrs, } )对应测试test_invoke_endpoint_async_should_read_sync_queue_if_async_not_configuredtests/test_sagemakerruntime/test_sagemakerruntime.py验证了该回退路径只配置同步队列、不配置异步队列时异步调用依然能读到同步队列里的Body内容。四、路由与请求协议URL 是如何被匹配的理解路由有助于在 Server Mode / Proxy Mode 下调试问题。sagemaker-runtime 的 URL 规则定义在 moto/sagemakerruntime/urls.pyurl_bases [ rhttps?://runtime\.sagemaker\.(.)\.amazonaws\.com, ] url_paths { {0}/endpoints/(?Pname[^/])/async-invocations$: response.dispatch, {0}/endpoints/(?Pname[^/])/invocations$: response.dispatch, }即同步调用InvokeEndpoint→POST https://runtime.sagemaker.{region}.amazonaws.com/endpoints/{EndpointName}/invocations异步调用InvokeEndpointAsync→POST https://runtime.sagemaker.{region}.amazonaws.com/endpoints/{EndpointName}/async-invocations。异步响应层正是从路径中解析端点名的endpoint_name self.path.split(/)[2]见 moto/sagemakerruntime/responses.py并从请求头读取X-Amzn-SageMaker-InputLocation与X-Amzn-SageMaker-Inference-Id。因此测试中给InputLocation传任意字符串都是允许的——它只作为请求特征参与缓存判定并不校验其是否为真实 S3 路径。此外moto-api 的静态配置接口注册在 moto/moto_api/_internal/urls.py{0}/moto-api/static/sagemaker/endpoint-results: response_instance.set_sagemaker_result, {0}/moto-api/static/sagemaker/async-endpoint-results: response_instance.set_sagemaker_async_result,这解释了为什么在默认非 Server Mode环境下配置队列要访问http://motoapi.amazonaws.com/moto-api/...moto 会把对motoapi.amazonaws.com域名的请求拦截并交给 moto-api 后端处理。而在 Server ModeTEST_SERVER_MODE下测试代码则使用localhost:5000作为 base URL——这一差异在测试文件中通过settings.TEST_SERVER_MODE做了分支见 tests/test_sagemakerruntime/test_sagemakerruntime.py。五、完整可运行的测试示例综合以上机制下面是一段覆盖同步 异步 失败注入 S3 落盘校验的完整示例逻辑参照 tests/test_sagemakerruntime/test_sagemakerruntime.pyimport json import boto3 import requests from moto import mock_aws, settings mock_aws def test_sagemaker_runtime_queue(): client boto3.client(sagemaker-runtime, region_nameus-east-1) base_url localhost:5000 if settings.TEST_SERVER_MODE else motoapi.amazonaws.com # 1. 配置同步结果队列两条结果 requests.post( fhttp://{base_url}/moto-api/static/sagemaker/endpoint-results, json{ results: [ {Body: first body, ContentType: text/xml}, {Body: second body}, ] }, ) # 相同请求 - 返回队列第一条并缓存 r1 client.invoke_endpoint(EndpointNameasdf, Bodyqwer) assert r1[Body].read() bfirst body r1_again client.invoke_endpoint(EndpointNameasdf, Bodyqwer) assert r1_again[Body].read() bfirst body # 不同请求换了 Accept/TargetModel- 取队列第二条 r2 client.invoke_endpoint(EndpointNameasdf, Bodyqwer, Acceptsth) assert r2[Body].read() bsecond body # 2. 配置异步结果队列一个成功、一个失败 requests.post( fhttp://{base_url}/moto-api/static/sagemaker/async-endpoint-results, json{ results: [ {data: json.dumps({first: output})}, {is_failure: True, data: second inference failed}, ] }, ) async_r1 client.invoke_endpoint_async(EndpointNameasdf, InputLocationqwer) s3 boto3.client(s3, us-east-1) # 从 OutputLocation 读回成功结果 output_bucket, output_key async_r1[OutputLocation].replace(s3://, ).split(/, 1) out s3.get_object(Bucketoutput_bucket, Keyoutput_key)[Body].read().decode(utf-8) assert json.loads(out) {first: output} # 不同 InputLocation - 触发失败结果 async_r2 client.invoke_endpoint_async( EndpointNameasdf, InputLocationasf, InferenceIdsth ) assert async_r2[InferenceId] sth failure_bucket, failure_key async_r2[FailureLocation].replace(s3://, ).split(/, 1) fail s3.get_object(Bucketfailure_bucket, Keyfailure_key)[Body].read().decode(utf-8) assert fail second inference failed使用要点小结使用mock_aws装饰器即可启用本模块与 moto 其他服务一致结果队列按(account_id, region)隔离多环境测试需分别配置队列消费是惰性的只有出现新的请求特征时才会弹出下一条重复请求永远命中缓存这在幂等测试中非常有用异步结果最终落在由 moto 自动创建的 S3 Bucketsagemaker-output-{uuid}中因此可以直接用boto3.client(s3)断言落盘内容包括失败场景invoke_endpoint_with_response_stream尚未实现涉及流式推理的用例需绕开。六、扩展阅读服务实现核心moto/sagemakerruntime/models.pySageMakerRuntimeBackend及两条队列的实现响应层与头映射moto/sagemakerruntime/responses.py路由规则moto/sagemakerruntime/urls.pymoto-api 静态配置接口的路由与解析moto/moto_api/_internal/urls.py、moto/moto_api/_internal/responses.py、moto/moto_api/_internal/models.py行为验证测试tests/test_sagemakerruntime/test_sagemakerruntime.py官方服务文档docs/docs/services/sagemaker-runtime.rst赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐Moto 中 RDS Data APIrds-data的 Mock 实现与可预测查询结果配置实战Moto 中 RDS Data APIrds data的 Mock 实现与可预测查询结果配置实战 导读 本文聚焦开源项目 Moto 对 AWS RDS DaMock测试Moto 中 AWS Cost Explorerce服务模拟指南Cost Category 与 get_cost_and_usage 结果队列实战Moto 中 AWS Cost Explorerce服务模拟指南Cost Category 与 get_cost_and_usage 结果队列实战 本文以Mock测试Moto 中 Redshift Data APIredshift-data的 Mock 实现execute_statement 与静态结果集的完整实战指南Moto 中 Redshift Data APIredshift data的 Mock 实现execute_statement 与静态结果集的完整实战指南Mock测试上一篇RxSwift中文文档架构模式对比MVVM vs RxFeedback vs ReactorKit下一篇Stremio-web视频转码服务适配不同设备的媒体处理终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
moto 中 sagemaker-runtime 的 Mock 实践：invoke_endpoint 与 invoke_endpoint_async 的结果队列配置指南
Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载导读SageMaker Runtime 是 AWS 推理服务中负责向已部署端点发起调用同步InvokeEndpoint、异步InvokeEndpointAsync的运行时组件。在 moto 中sagemaker-runtime后端为这两类调用提供了可编程的 Mock 能力默认情况下返回静态数据开发者可以借助 moto 专属的/moto-api/static/sagemaker/...接口预先配置预期结果队列从而精确控制每次调用的响应内容、失败注入与返回头。读完本文你将掌握该模块的全部已实现能力、结果队列的入队与出队规则、同步/异步调用在底层源码中的真实行为以及如何在单元测试中按场景编排这些结果。本文以 docs/docs/services/sagemaker-runtime.rst 为骨架结合 moto/sagemakerruntime/models.py、moto/sagemakerruntime/responses.py 与 tests/test_sagemakerruntime/test_sagemakerruntime.py 展开源码级讲解。一、已实现功能总览根据官方服务文档见 docs/docs/services/sagemaker-runtime.rstmoto 对sagemaker-runtime的支持情况如下操作状态说明invoke_endpoint✅ 已实现默认返回静态数据可通过结果队列定制invoke_endpoint_async✅ 已实现默认返回静态数据可通过异步结果队列定制支持失败注入与 S3 输出落盘invoke_endpoint_with_response_stream❌ 未实现流式响应即 SageMaker 的流式推理暂不支持也就是说moto 目前覆盖了推理调用中最常用的同步与异步两条路径凡是请求了invoke_endpoint_with_response_stream的代码在 Mock 环境下会得到未实现的响应测试设计时需要避开该操作或自行做条件分支。二、默认行为静态数据2.1 同步调用invoke_endpoint在不做任何额外配置时invoke_endpoint会返回一组写死的静态数据。其默认值定义在 moto/sagemakerruntime/models.py 的SageMakerRuntimeBackend.invoke_endpoint方法中self.results[endpoint_name][unique_repr] ( body, content_type, invoked_production_variant, custom_attributes, )返回值是一个四元组四个元素会被 moto/sagemakerruntime/responses.py 的SageMakerRuntimeResponse.invoke_endpoint映射到 Boto3 客户端返回结构的不同字段四元组元素默认值映射到的返回字段bodybody响应的Body可读流content_typecontent_type响应头Content-Typeinvoked_production_variantinvoked_production_variant响应头x-Amzn-Invoked-Production-Variantcustom_attributescustom_attributes返回结构中的CustomAttributes字段这一点在 tests/test_sagemakerruntime/test_sagemakerruntime.py 的test_invoke_endpoint__default_results中得到验证即使调用时传入了Acceptsth、TargetModeltm返回的Body仍是bbodyCustomAttributes仍是custom_attributes。从源码结构可以推断同一次 Mock 会话内、相同请求特征的调用会命中缓存invoke_endpoint内部以endpoint_name为键维护一个嵌套字典self.results请求的唯一标识unique_repr则由请求头中所有以x-amzn-sagemaker开头的字段、Accept头以及请求体Body共同计算并 base64 编码而来见 moto/sagemakerruntime/responses.py。因此相同 EndpointName 相同请求头 相同 Body的重复调用会复用第一次的结果。2.2 异步调用invoke_endpoint_async异步调用的默认行为同样是静态数据但其返回结构完全不同。真实 AWS 的异步推理会将结果写入用户指定的 S3 位置moto 模拟了这一流程默认返回的数据是json.dumps({default: response})并会在当前账号/分区下自动创建一个名为sagemaker-output-{uuid}的 S3 Bucket将结果以response.json写入见 moto/sagemakerruntime/models.py。底层实现位于SageMakerRuntimeBackend.invoke_endpoint_asyncmoto/sagemakerruntime/models.py它最终返回(output_location, failure_location)两个 S3 路径其中output_locations3://sagemaker-output-{uuid}/response.jsonfailure_locations3://sagemaker-output-{uuid}/failure.json仅当本次结果被标记为失败时数据才会写入该文件。对应的响应层 moto/sagemakerruntime/responses.py 会把这些位置放入响应头X-Amzn-SageMaker-OutputLocationX-Amzn-SageMaker-FailureLocationInferenceId若请求未携带X-Amzn-SageMaker-Inference-Id头则由random.uuid4()生成这与真实 AWS 的异步推理语义保持一致客户端拿到OutputLocation后再到 S3 中读取推理结果。三、用 moto-api 结果队列定制响应3.1 队列机制的核心思想无论是同步还是异步调用moto 都提供了一套预期结果队列机制通过向 moto 专属的 HTTP 接口 POST 一个 JSON 负载将一系列结果按顺序压入后端队列之后每次新的推理请求会从队列头部弹出一个结果。该机制的关键规则官方文档与源码一致是相同请求 → 相同结果后续使用相同请求特征的调用同步场景看请求头/Body异步场景看InputLocation会返回与第一次完全相同的结果而不会再次消费队列不同请求 → 队列下一个结果其他使用不同特征的新请求会取走队列中的下一条队列为空 → 回退静态数据队列耗尽后新请求返回默认静态数据。从源码看这一规则在 moto/sagemakerruntime/models.py同步与第 107-144 行异步中以同样的模式实现先查缓存字典命中即返回未命中再从队列pop(0)否则生成默认值最后把结果写回缓存。3.2 配置同步结果队列endpoint-results向/moto-api/static/sagemaker/endpoint-results发起 POST 即可配置同步调用invoke_endpoint的结果队列。官方文档给出的示例负载如下原文见 docs/docs/services/sagemaker-runtime.rstexpected_results { account_id: 123456789012, # 默认账号可省略 region: us-east-1, # 默认区域可省略 results: [ { Body: first body, ContentType: text/xml, InvokedProductionVariant: prod, CustomAttributes: my_attr, }, # 可按需添加更多结果 ], } requests.post( http://motoapi.amazonaws.com/moto-api/static/sagemaker/endpoint-results, jsonexpected_results, ) client boto3.client(sagemaker-runtime, region_nameus-east-1) details client.invoke_endpoint(EndpointNameasdf, Bodyqwer)每个结果条目支持的字段与对应关系如下字段是否必填对应行为Body是返回给客户端的推理响应体ContentType否写入响应头Content-TypeInvokedProductionVariant否写入响应头x-Amzn-Invoked-Production-VariantCustomAttributes否写入返回结构中的CustomAttributes字段account_id与region缺省时分别使用默认账号与us-east-1这一点在 moto-api 的解析代码 moto/moto_api/_internal/responses.py 中可以看到account_id body.get(account_id, DEFAULT_ACCOUNT_ID)、region body.get(region, us-east-1)随后逐条把结果追加到sagemakerruntime_backends[account_id][region].results_queue见 moto/moto_api/_internal/models.py。也就是说队列是按账号 区域隔离的多账号/多区域测试时需各自配置。3.3 配置异步结果队列async-endpoint-results异步调用的结果队列通过/moto-api/static/sagemaker/async-endpoint-results配置每个条目额外支持is_failure字段来模拟推理失败。官方文档示例expected_results { account_id: 123456789012, # 默认账号可省略 region: us-east-1, # 默认区域可省略 results: [ { data: json.dumps({first: output}), }, { is_failure: True, data: second inference failed, }, # 可按需添加更多结果 ], } requests.post( http://motoapi.amazonaws.com/moto-api/static/sagemaker/async-endpoint-results, jsonexpected_results, ) client boto3.client(sagemaker-runtime, region_nameus-east-1) details client.invoke_endpoint_async(EndpointNameasdf, InputLocationqwer)字段语义字段是否必填对应行为data是写入 S3 对象的内容建议用json.dumps序列化is_failure否默认为False为True时数据写入failure.json并返回对应的FailureLocation队列消费时is_failureTrue的结果会走失败分支把data原样写入failure.json成功结果写入response.json。测试 tests/test_sagemakerruntime/test_sagemakerruntime.py 的test_invoke_endpoint_async验证了这条完整链路配置两个结果 → 第一次调用拿到OutputLocation→ 从 S3 读回内容为{first: output}→ 换一个InputLocation触发第二条结果is_failureTrue→ 从FailureLocation读回second failure同时InferenceId与请求携带的保持一致。3.4 向后兼容异步队列为空时回退同步队列这是异步调用特有的规则官方文档明确说明如果异步队列为空将使用已配置的同步队列for backward compatibility。底层实现见 moto/sagemakerruntime/models.py当async_results_queue为空但results_queue非空时会从同步队列弹出条目并把四元组包装成 JSON 写入 S3elif self.results_queue: # Backward compatibility is_failure False body, _type, variant, attrs self.results_queue.pop(0) data json.dumps( { Body: body, ContentType: _type, InvokedProductionVariant: variant, CustomAttributes: attrs, } )对应测试test_invoke_endpoint_async_should_read_sync_queue_if_async_not_configuredtests/test_sagemakerruntime/test_sagemakerruntime.py验证了该回退路径只配置同步队列、不配置异步队列时异步调用依然能读到同步队列里的Body内容。四、路由与请求协议URL 是如何被匹配的理解路由有助于在 Server Mode / Proxy Mode 下调试问题。sagemaker-runtime 的 URL 规则定义在 moto/sagemakerruntime/urls.pyurl_bases [ rhttps?://runtime\.sagemaker\.(.)\.amazonaws\.com, ] url_paths { {0}/endpoints/(?Pname[^/])/async-invocations$: response.dispatch, {0}/endpoints/(?Pname[^/])/invocations$: response.dispatch, }即同步调用InvokeEndpoint→POST https://runtime.sagemaker.{region}.amazonaws.com/endpoints/{EndpointName}/invocations异步调用InvokeEndpointAsync→POST https://runtime.sagemaker.{region}.amazonaws.com/endpoints/{EndpointName}/async-invocations。异步响应层正是从路径中解析端点名的endpoint_name self.path.split(/)[2]见 moto/sagemakerruntime/responses.py并从请求头读取X-Amzn-SageMaker-InputLocation与X-Amzn-SageMaker-Inference-Id。因此测试中给InputLocation传任意字符串都是允许的——它只作为请求特征参与缓存判定并不校验其是否为真实 S3 路径。此外moto-api 的静态配置接口注册在 moto/moto_api/_internal/urls.py{0}/moto-api/static/sagemaker/endpoint-results: response_instance.set_sagemaker_result, {0}/moto-api/static/sagemaker/async-endpoint-results: response_instance.set_sagemaker_async_result,这解释了为什么在默认非 Server Mode环境下配置队列要访问http://motoapi.amazonaws.com/moto-api/...moto 会把对motoapi.amazonaws.com域名的请求拦截并交给 moto-api 后端处理。而在 Server ModeTEST_SERVER_MODE下测试代码则使用localhost:5000作为 base URL——这一差异在测试文件中通过settings.TEST_SERVER_MODE做了分支见 tests/test_sagemakerruntime/test_sagemakerruntime.py。五、完整可运行的测试示例综合以上机制下面是一段覆盖同步 异步 失败注入 S3 落盘校验的完整示例逻辑参照 tests/test_sagemakerruntime/test_sagemakerruntime.pyimport json import boto3 import requests from moto import mock_aws, settings mock_aws def test_sagemaker_runtime_queue(): client boto3.client(sagemaker-runtime, region_nameus-east-1) base_url localhost:5000 if settings.TEST_SERVER_MODE else motoapi.amazonaws.com # 1. 配置同步结果队列两条结果 requests.post( fhttp://{base_url}/moto-api/static/sagemaker/endpoint-results, json{ results: [ {Body: first body, ContentType: text/xml}, {Body: second body}, ] }, ) # 相同请求 - 返回队列第一条并缓存 r1 client.invoke_endpoint(EndpointNameasdf, Bodyqwer) assert r1[Body].read() bfirst body r1_again client.invoke_endpoint(EndpointNameasdf, Bodyqwer) assert r1_again[Body].read() bfirst body # 不同请求换了 Accept/TargetModel- 取队列第二条 r2 client.invoke_endpoint(EndpointNameasdf, Bodyqwer, Acceptsth) assert r2[Body].read() bsecond body # 2. 配置异步结果队列一个成功、一个失败 requests.post( fhttp://{base_url}/moto-api/static/sagemaker/async-endpoint-results, json{ results: [ {data: json.dumps({first: output})}, {is_failure: True, data: second inference failed}, ] }, ) async_r1 client.invoke_endpoint_async(EndpointNameasdf, InputLocationqwer) s3 boto3.client(s3, us-east-1) # 从 OutputLocation 读回成功结果 output_bucket, output_key async_r1[OutputLocation].replace(s3://, ).split(/, 1) out s3.get_object(Bucketoutput_bucket, Keyoutput_key)[Body].read().decode(utf-8) assert json.loads(out) {first: output} # 不同 InputLocation - 触发失败结果 async_r2 client.invoke_endpoint_async( EndpointNameasdf, InputLocationasf, InferenceIdsth ) assert async_r2[InferenceId] sth failure_bucket, failure_key async_r2[FailureLocation].replace(s3://, ).split(/, 1) fail s3.get_object(Bucketfailure_bucket, Keyfailure_key)[Body].read().decode(utf-8) assert fail second inference failed使用要点小结使用mock_aws装饰器即可启用本模块与 moto 其他服务一致结果队列按(account_id, region)隔离多环境测试需分别配置队列消费是惰性的只有出现新的请求特征时才会弹出下一条重复请求永远命中缓存这在幂等测试中非常有用异步结果最终落在由 moto 自动创建的 S3 Bucketsagemaker-output-{uuid}中因此可以直接用boto3.client(s3)断言落盘内容包括失败场景invoke_endpoint_with_response_stream尚未实现涉及流式推理的用例需绕开。六、扩展阅读服务实现核心moto/sagemakerruntime/models.pySageMakerRuntimeBackend及两条队列的实现响应层与头映射moto/sagemakerruntime/responses.py路由规则moto/sagemakerruntime/urls.pymoto-api 静态配置接口的路由与解析moto/moto_api/_internal/urls.py、moto/moto_api/_internal/responses.py、moto/moto_api/_internal/models.py行为验证测试tests/test_sagemakerruntime/test_sagemakerruntime.py官方服务文档docs/docs/services/sagemaker-runtime.rst赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐Moto 中 RDS Data APIrds-data的 Mock 实现与可预测查询结果配置实战Moto 中 RDS Data APIrds data的 Mock 实现与可预测查询结果配置实战 导读 本文聚焦开源项目 Moto 对 AWS RDS DaMock测试Moto 中 AWS Cost Explorerce服务模拟指南Cost Category 与 get_cost_and_usage 结果队列实战Moto 中 AWS Cost Explorerce服务模拟指南Cost Category 与 get_cost_and_usage 结果队列实战 本文以Mock测试Moto 中 Redshift Data APIredshift-data的 Mock 实现execute_statement 与静态结果集的完整实战指南Moto 中 Redshift Data APIredshift data的 Mock 实现execute_statement 与静态结果集的完整实战指南Mock测试上一篇RxSwift中文文档架构模式对比MVVM vs RxFeedback vs ReactorKit下一篇Stremio-web视频转码服务适配不同设备的媒体处理终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
