moto 对 AWS Comprehend 的模拟实现：功能覆盖矩阵、底层机制与测试实战

Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载moto 项目中的docs/docs/services/comprehend.rst服务文档列出了 moto 对 AWS ComprehendAmazon Comprehend 自然语言处理服务的 API 覆盖清单——哪些操作可以离线 mock、哪些尚未实现、以及各接口的具体限制。本文将基于这份覆盖清单结合 moto/comprehend 的源码实现与 tests/test_comprehend 测试用例完整讲解 moto 如何模拟 Comprehend 的模型训练资源、异步检测作业、同步检测 API 与资源策略帮助你在不触达真实 AWS 的前提下编写可靠的 Comprehend 单元测试。一、服务覆盖总览文档定义的 API 矩阵服务文档以[X]已实现/[ ]未实现形式给出覆盖矩阵。这是使用 moto mock Comprehend 时最重要的参考只应在已实现列表内编写测试断言未实现接口在 moto 中不会响应请求会落入未匹配路径或抛出异常取决于调用方式。1.1 已实现的操作[X]按功能域归纳后当前仓库已实现的操作如下模型与推理资源生命周期操作说明create_document_classifier创建文档分类器ClientRequestToken参数尚未真正生效create_endpoint创建推理端点create_entity_recognizer创建实体识别器create_flywheel创建数据飞轮describe_document_classifier查询分类器属性describe_endpoint查询端点属性describe_entity_recognizer查询识别器属性describe_flywheel查询飞轮属性delete_document_classifier/delete_endpoint/delete_entity_recognizer/delete_flywheel删除对应资源list_document_classifiers按DocumentClassifierName或Status过滤分页未实现list_endpoints按ModelArn或Status过滤分页未实现list_entity_recognizers仅支持RecognizerName过滤Status、SubmitTimeBefore、SubmitTimeAfter过滤未实现list_flywheels仅支持Status过滤分页未实现update_endpoint更新端点源码中仅回显DesiredModelArnstop_training_document_classifier/stop_training_entity_recognizer请求停止训练同步文本检测操作说明detect_key_phrases返回固定canned短语列表detect_pii_entities返回固定 PII 实体列表detect_sentiment返回固定 NEUTRAL 情感结果异步检测作业每类作业均为 start / describe / list 三件套多数支持 stopstart_pii_entities_detection_job、describe_pii_entities_detection_job、stop_pii_entities_detection_job、list_pii_entities_detection_jobsstart_key_phrases_detection_job、describe_key_phrases_detection_job、stop_key_phrases_detection_job、list_key_phrases_detection_jobsstart_sentiment_detection_job、describe_sentiment_detection_job、stop_sentiment_detection_job、list_sentiment_detection_jobsstart_dominant_language_detection_job、describe_dominant_language_detection_job、stop_dominant_language_detection_job、list_dominant_language_detection_jobsstart_entities_detection_job、describe_entities_detection_job、stop_entities_detection_job、list_entities_detection_jobsstart_document_classification_job、describe_document_classification_job、list_document_classification_jobsstart_events_detection_job、describe_events_detection_job、stop_events_detection_job、list_events_detection_jobsstart_targeted_sentiment_detection_job、describe_targeted_sentiment_detection_job、stop_targeted_sentiment_detection_job、list_targeted_sentiment_detection_jobsstart_topics_detection_job、describe_topics_detection_job、list_topics_detection_jobsstart_flywheel_iteration| 返回随机迭代 ID标签与资源策略操作说明list_tags_for_resource/tag_resource/untag_resource通用标签管理put_resource_policy写入资源策略PolicyRevisionId条件更新参数未实现也不校验资源本身是否存在describe_resource_policy查询策略、修订 ID 与时间戳delete_resource_policy删除策略条件删除参数PolicyRevisionId未实现1.2 未实现的操作[ ]文档明确标记为未实现的接口包括全部batch_*系列batch_detect_dominant_language、batch_detect_entities、batch_detect_key_phrases、batch_detect_sentiment、batch_detect_syntax、batch_detect_targeted_sentiment、classify_document、contains_pii_entities、detect_dominant_language、detect_entities、detect_syntax、detect_targeted_sentiment、detect_toxic_content、import_model、update_flywheel以及数据集/飞轮迭代相关接口create_dataset、describe_dataset、list_datasets、describe_flywheel_iteration、list_flywheel_iteration_history和摘要列表接口list_document_classifier_summaries、list_entity_recognizer_summaries。编写测试时应对这些接口保持清醒它们不会被 moto 拦截并给出真实语义响应。二、mock 的入口与请求路由Comprehend 的 mock 由 moto/comprehend/urls.py 声明的 URL 规则接入url_bases [ rhttps?://comprehend\.(.)\.amazonaws\.com, ] url_paths { {0}/$: ComprehendResponse.dispatch, }只要 boto3 客户端指向comprehend.region.amazonaws.com端点请求即被路由到 moto/comprehend/responses.py 中的ComprehendResponse。Comprehend 是 JSON 协议服务所有操作都 POST 到根路径靠请求头X-Amz-Target区分具体 API因此 moto 只需一条根路径规则即可承接全部接口。ComprehendResponse每个方法遵循同一模式json.loads(self.body)解析请求体、取出参数、调用 backend、json.dumps返回响应例如def detect_pii_entities(self) - str: params json.loads(self.body) text params.get(Text) language params.get(LanguageCode) resp self.comprehend_backend.detect_pii_entities(text, language) return json.dumps({Entities: resp})而comprehend_backend属性按account_id region取出独立的后端实例responses.py 第 17-20 行这保证了 moto 的多账号/多区域隔离语义——不同 region 的识别器互不可见。三、后端模型ARN、状态机与存储结构所有状态都保存在 moto/comprehend/models.py 的ComprehendBackend中按区域与账号隔离def __init__(self, region_name: str, account_id: str): super().__init__(region_name, account_id) self.recognizers: dict[str, EntityRecognizer] {} self.tagger TaggingService() self.endpoints: dict[str, Endpoint] {} self.classifiers: dict[str, DocumentClassifier] {} self.flywheels: dict[str, Flywheel] {} self.resource_policies: dict[str, dict[str, Any]] {} self.jobs: dict[str, ComprehendJob] {}文件末尾的comprehend_backends BackendDict(ComprehendBackend, comprehend)是 moto 标准的后端注册方式测试中也可以直接导入它做白盒断言见 tests/test_comprehend/test_comprehend_integration.py 第 70 行。3.1 四类模型资源的 ARN 生成规则各模型类在构造时即生成确定性 ARN测试可以据此做精确断言实体识别器arn:{partition}:comprehend:{region}:{account}:entity-recognizer/{name}若传了VersionName则追加/version/{version_name}models.py 第 88-90 行文档分类器...:document-classifier/{name}/{version_name}推理端点...:endpoint/{endpoint_name}/{model_arn}数据飞轮...:flywheel/{flywheel_name}异步作业...:{job_type_path}-job/{uuid}其中job_type_path由作业类型做驼峰转短横线得到例如DocumentClassification→document-classificationmodels.py 第 262-265 行。3.2 初始状态与可预测的状态机mock 为每类资源设定了固定的初始状态这是写测试断言时必须知道的“事实”资源初始状态停止操作后的状态EntityRecognizerTRAINEDstop_training_entity_recognizer仅在当前状态为TRAINING时置为STOP_REQUESTED由于初始即TRAINED调用后状态保持不变DocumentClassifierTRAININGstop_training_document_classifier将其置为STOP_REQUESTEDEndpointIN_SERVICE无停止语义FlywheelACTIVE无停止语义ComprehendJobSUBMITTEDstop()在SUBMITTED/IN_PROGRESS下置为STOP_REQUESTED这一差异在测试中有直接体现test_comprehend.py 的 test_stop_training_entity_recognizer 断言识别器停止训练后状态仍是TRAINED而 test_stop_training_document_classifier 断言分类器变为STOP_REQUESTED。3.3 通用作业模型 ComprehendJob十种检测作业共用一个ComprehendJob类models.py 第 235-290 行其设计值得学习start_*_job系列方法如start_pii_entities_detection_job只是薄封装统一调用私有方法_start_job(job_type, **kwargs)_start_job从 kwargs 中弹出InputDataConfig、OutputDataConfig、DataAccessRoleArn、JobName、LanguageCode其余参数如Mode、RedactionConfig作为extra_args原样保留并在to_dict()时合并进响应因此 AWS 返回的扩展字段在 mock 中会“原样回显”JobName缺省时自动生成为moto-job-{job_id}LanguageCode对DominantLanguageDetection作业是可选的源码注释明确说明仅当存在时才写入响应字典。_list_jobs提供通用过滤JobName、JobStatus、SubmitTimeBefore、SubmitTimeAftermodels.py 第 723-745 行。注意响应层的_list_jobs_to_dict_resp会剥离内部字段job_type并把SubmitTime/EndTime序列化为 ISO 字符串。一个特殊校验start_events_detection_job要求请求必须包含TargetEventTypes否则抛出InvalidRequestException(The request is missing the required parameter: TargetEventTypes.)models.py 第 839-844 行这是全部 start 作业中唯一的必填参数强校验。四、同步检测 API 的 canned 响应与校验三个同步接口不执行任何 NLP而是返回 models.py 顶部 定义的固定响应CANNED_DETECT_RESPONSE3 条 PII 实体NAME、EMAIL、BANK_ACCOUNT_NUMBER含Score、BeginOffset、EndOffsetCANNED_PHRASES_RESPONSE3 条短语仅Score与偏移量CANNED_SENTIMENT_RESPONSESentiment: NEUTRAL与四档情感分数。真正的价值在于 mock 复现了 AWS 的参数校验exceptions.py 定义了四个异常类异常触发条件错误码DetectPIIValidationException语言代码不在允许集合内ValidationExceptionTextSizeLimitExceededException文本超过长度上限TextSizeLimitExceededExceptionResourceNotFounddescribe/delete 不存在的资源ResourceNotFoundExceptionInvalidRequestException事件检测作业缺少TargetEventTypesInvalidRequestException允许的语言集合定义在 backend 类属性上models.py 第 302-318 行detect_key_phrases_languagesar, hi, ko, zh-TW, ja, zh, de, pt, en, it, fr, es共 12 种detect_pii_entities_languages仅en。文本长度上限各接口不同detect_pii_entities与detect_key_phrases为 100,000 字节detect_sentiment为 5,000 字节models.py 第 396-424 行。注意detect_sentiment的语言校验复用了detect_key_phrases_languages集合。以下示例展示了如何测试这些校验行为摘自 test_comprehend.py 并整合import boto3 import pytest from botocore.exceptions import ClientError from moto import mock_aws mock_aws def test_detect_pii_and_validation(): client boto3.client(comprehend, region_nameeu-west-1) # 合法请求返回固定响应 result client.detect_pii_entities(Textsome text, LanguageCodeen) assert result[Entities][0][Type] NAME # 非法语言es 不在 PII 支持集合 [en] 中 with pytest.raises(ClientError) as exc: client.detect_pii_entities(Textx, LanguageCodees) err exc.value.response[Error] assert err[Code] ValidationException assert Member must satisfy enum value set: [en] in err[Message] # 超长文本超过 100000 字节 with pytest.raises(ClientError) as exc: client.detect_pii_entities(Textx * 100001, LanguageCodeen) assert exc.value.response[Error][Code] TextSizeLimitExceededException五、典型资源生命周期从测试用例到源码5.1 实体识别器的创建、过滤、删除create_entity_recognizer的请求参数对照 responses.py 第 30-54 行 的解析逻辑RecognizerName、VersionName、DataAccessRoleArn、InputDataConfig、LanguageCode、Tags、VolumeKmsKeyId、VpcConfig、ModelKmsKeyId、ModelPolicy。文档注明ClientRequestToken尚未实现即该参数不会被消费。mock_aws def test_entity_recognizer_lifecycle(): client boto3.client(comprehend, region_nameap-southeast-1) resp client.create_entity_recognizer( DataAccessRoleArniam_role_with_20_chars, InputDataConfig{ DataFormat: COMPREHEND_CSV, Documents: {InputFormat: ONE_DOC_PER_LINE, S3Uri: s3://bucket/documents.txt}, EntityTypes: [{Type: ENGINEER}, {Type: MANAGER}], }, LanguageCodeen, RecognizerNamemy-recognizer, VersionNamev1, ) # ARN 中版本段是可选的 assert resp[EntityRecognizerArn] ( arn:aws:comprehend:ap-southeast-1:123456789012:entity-recognizer/my-recognizer/version/v1 ) # 按名称过滤 assert len(client.list_entity_recognizers( Filter{RecognizerName: my-recognizer} )[EntityRecognizerPropertiesList]) 1 # 删除后 describe 抛 ResourceNotFoundException client.delete_entity_recognizer(EntityRecognizerArnresp[EntityRecognizerArn]) with pytest.raises(ClientError) as exc: client.describe_entity_recognizer(EntityRecognizerArnresp[EntityRecognizerArn]) assert exc.value.response[Error][Code] ResourceNotFoundException assert exc.value.response[Error][Message] ( RESOURCE_NOT_FOUND: Could not find specified resource. )delete_entity_recognizer的源码是self.recognizers.pop(entity_recognizer_arn, None)——对已删除的资源幂等不报错这与 AWS 行为并不完全一致但 describe 已删除资源会抛ResourceNotFound测试应以后者为准。5.2 文档分类器、端点与飞轮create_document_classifier额外接收OutputDataConfig、Mode如MULTI_CLASS与ClientRequestTokencreate_endpoint接收EndpointName、ModelArn、DesiredInferenceUnits、FlywheelArn、DataAccessRoleArncreate_flywheel接收TaskConfig、ModelType、DataLakeS3Uri、DataSecurityConfig等参数清单见 responses.py 第 124-204 行。describe 响应的属性键分别为DocumentClassifierProperties、EndpointProperties、FlywheelProperties。list_document_classifiers/list_endpoints/list_flywheels支持过滤但忽略分页源码中next_token、max_results参数被接收但从不使用返回的NextToken恒为Nonemodels.py 第 540-624 行。过滤语义是 elif 链——只评估第一个匹配的条件分类器先看DocumentClassifierName再看Status端点先看ModelArn再看Status飞轮仅支持Status。start_flywheel_iteration校验飞轮存在后返回一个 0–1,000,000 的随机整数作为FlywheelIterationIdupdate_endpoint则仅回显传入的DesiredModelArnmodels.py 第 641-649 行。5.3 异步检测作业作业的标准四步生命周期以 PII 检测为例完整用例见 test_comprehend_jobs.py 第 58-89 行mock_aws def test_pii_job_lifecycle(): client boto3.client(comprehend, region_nameus-east-1) start client.start_pii_entities_detection_job( InputDataConfig{S3Uri: s3://bucket/docs/, InputFormat: ONE_DOC_PER_FILE}, OutputDataConfig{S3Uri: s3://bucket/out/}, DataAccessRoleArnarn:aws:iam::123456789012:role/testing-role, JobNametest-pii-job, LanguageCodeen, ModeONLY_REDACTION, RedactionConfig{MaskCharacter: *, MaskMode: MASK}, ) assert start[JobStatus] SUBMITTED # describe 响应键为 {JobType}JobProperties props client.describe_pii_entities_detection_job( JobIdstart[JobId])[PiiEntitiesDetectionJobProperties] assert props[Mode] ONLY_REDACTION # 额外参数原样回显 # 按名称过滤 jobs client.list_pii_entities_detection_jobs( Filter{JobName: test-pii-job}) assert len(jobs[PiiEntitiesDetectionJobPropertiesList]) 1 # stop 后状态变为 STOP_REQUESTED client.stop_pii_entities_detection_job(JobIdstart[JobId]) assert client.describe_pii_entities_detection_job(JobIdstart[JobId])[ PiiEntitiesDetectionJobProperties][JobStatus] STOP_REQUESTEDdescribe 响应键的命名规则由响应层统一生成f{job_type}JobPropertiesresponses.py 第 356-365 行list 响应键为f{job_type}JobPropertiesList因此十种作业无需各自的响应模板代码。5.4 标签与 Resource Groups Tagging API 联动backend 通过TaggableResourcesMixin暴露iter_tagged_resources把分类器、端点、识别器、飞轮及各类作业统一交给 resourcegroupstaggingapi 服务models.py 第 871-906 行资源类型分别为comprehend:document-classifier、comprehend:endpoint、comprehend:entity-recognizer、comprehend:flywheel和comprehend:{job_type}-job。tests/test_comprehend/test_comprehend_integration.py 验证了跨服务联动用 Comprehend 客户端创建带标签的分类器与作业再用resourcegroupstaggingapi的get_resources按TagFilters或ResourceTypeFilters检索甚至通过 RGT API 完成tag_resources/untag_resources。作业类型过滤器示例tags resource_groups_client.get_resources( ResourceTypeFilters[comprehend:document-classification-job], )[ResourceTagMappingList] assert tags[0][ResourceARN] ( arn:aws:comprehend:ap-southeast-1:123456789012:document-classification-job/ job_id )5.5 资源策略的修订语义put_resource_policy的 mock 语义models.py 第 651-674 行每次调用生成新的 UUID 作为PolicyRevisionIdCreationTime在资源首次写入策略时固定后续更新只刷新LastModifiedTimePolicyRevisionId条件参数被接受但忽略文档已声明put不校验资源是否存在。完整生命周期断言可参考 test_comprehend.py 的 test_resource_policy_lifecycledescribe 不存在的策略抛ResourceNotFoundException两次 put 产生不同修订 ID 而CreationTime保持不变delete 之后再 describe 或再 delete 均抛ResourceNotFoundException。六、已知限制与测试注意事项综合服务文档与源码使用 moto mock Comprehend 时应注意以下边界未实现的接口见 1.2 节清单不在 mock 覆盖范围内尤其是所有batch_*同步批处理 API 与classify_document、contains_pii_entities分页全部未实现list_document_classifiers、list_endpoints、list_flywheels、list_entity_recognizers及各作业 list 的NextToken/MaxResults参数均被忽略NextToken恒返回None过滤能力不均list_entity_recognizers仅支持RecognizerNameStatus、SubmitTimeBefore、SubmitTimeAfter未实现其余 list 接口的过滤键以源码中的 elif 链为准幂等/条件参数未实现create_document_classifier的ClientRequestToken、put/delete_resource_policy的PolicyRevisionId均被忽略同步检测结果是固定值detect_pii_entities、detect_key_phrases、detect_sentiment永远返回同一 canned 数据不要基于输入文本内容做断言但可以断言语言校验与长度上限行为删除是幂等的 popdelete 类接口对不存在的 ARN 不抛错只有 describe 才抛ResourceNotFoundException状态不自动推进分类器创建后停留在TRAINING、作业停留在SUBMITTED没有任何定时器把它们推进到完成态需要“完成态”语义时应显式调用 stop 接口得到STOP_REQUESTED。七、相关文件索引路径作用docs/docs/services/comprehend.rst服务覆盖矩阵本文主线moto/comprehend/models.pyComprehendBackend与五类模型、canned 响应、状态机moto/comprehend/responses.py请求解析与响应序列化moto/comprehend/urls.py端点 URL 匹配规则moto/comprehend/exceptions.py四个异常类定义tests/test_comprehend/test_comprehend.py模型/检测/策略的生命周期与校验测试tests/test_comprehend/test_comprehend_jobs.py十种异步作业的生命周期测试tests/test_comprehend/test_comprehend_integration.py与 Resource Groups Tagging API 的跨服务联动测试掌握以上内容后你可以为 Comprehend 的客户端代码编写三类测试资源生命周期创建/过滤/删除/停止、同步检测的参数校验与固定响应断言、以及标签与资源策略的跨服务联动——全部在mock_aws装饰器下离线完成。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐moto 中的 CloudHSM V2 模拟:API 覆盖范围、后端实现与 mock 测试实战moto 中的 CloudHSM V2 模拟:API 覆盖范围、后端实现与 mock 测试实战 本文基于 moto 仓库的服务文档 cloudhsmv2.rstMock测试moto 对 AWS CodeCommit 的模拟仓库 CRUD 接口实现、响应行为与覆盖边界详解moto 对 AWS CodeCommit 的模拟仓库 CRUD 接口实现、响应行为与覆盖边界详解 在基于 AWS 基础设施构建的 CI/CD 流水线或代码托Mock测试oh-my-codex 功能覆盖矩阵解析OMX 与 oh-my-claudecode 的 95% 功能对齐实现oh my codex 功能覆盖矩阵解析OMX 与 oh my claudecode 的 95% 功能对齐实现 导读 COVERAGE.md https://人工智能AI AgentAgent 编排Agent 工作流CLI开发工具AI 技能上一篇WSABuilds在Windows上运行Android应用的终极解决方案下一篇mona.py栈溢出漏洞利用详解SEH覆盖与异常处理链攻击终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考