Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载本篇以 moto 仓库的 DataBrew 服务文档 为主体系统梳理 moto 对 AWS Glue DataBrew 的 Mock 能力边界哪些 API 操作已实现recipes、datasets、rulesets、profile/recipe jobs 及分页哪些尚未覆盖projects、schedules、job runs、tags。读完本文你将掌握如何用mock_aws装饰器在单元测试中完整地模拟 DataBrew 的创建、发布、版本管理与异常分支并能从源码层面理解版本状态机、路由分发与校验逻辑的实现原理。一、服务定位与 API 覆盖清单moto 是一个 AWS 基础设施 Mock 库允许开发者在不访问真实 AWS 的情况下测试基于 boto3 的代码。对于 AWS Glue DataBrew数据剖析与数据转换服务moto 的 服务文档页 以勾选清单形式列出了当前实现的 API 操作。这份清单是评估测试可行性时最直接的依据——在编写依赖 DataBrew 的测试前先核对目标 API 是否在已实现列表中可以避免遇到未实现的 API 调用。已实现的操作Implemented features资源域已实现操作Recipe 食谱create_recipe、describe_recipe、update_recipe、publish_recipe、delete_recipe_version、list_recipes、list_recipe_versionsDataset 数据集create_dataset、describe_dataset、update_dataset、delete_dataset、list_datasetsRuleset 规则集create_ruleset、describe_ruleset、update_ruleset、delete_ruleset、list_rulesetsJob 作业create_profile_job、update_profile_job、create_recipe_job、update_recipe_job、describe_job、delete_job、list_jobs未实现的操作Unimplementedbatch_delete_recipe_version、create_project、create_schedule、delete_project、delete_schedule、describe_job_run、describe_project、describe_schedule、list_job_runs、list_projects、list_schedules、list_tags_for_resource、send_project_session_action、start_job_run、start_project_session、stop_job_run、tag_resource、untag_resource、update_project、update_schedule。从清单可以看出覆盖范围的特点DataBrew 的定义层recipe 的版本管理、dataset 元数据、ruleset、job 定义与更新已被完整 Mock而运行时层project 交互式会话、schedule 调度、job run 的执行与查询以及 tags 资源类接口尚未实现。如果你的被测代码只涉及创建/查询/更新这些定义型资源可以直接使用 moto若涉及start_job_run或 project 会话则需要另行处理。二、请求路由DataBrew 端点如何被识别与分发DataBrew 的 HTTP 端点在 urls.py 中声明。URL 基址匹配databrew.{region}.amazonaws.com形式的域名url_bases [rhttps?://databrew\.(.)\.amazonaws.com] url_paths { {0}/recipeVersions$: DataBrewResponse.dispatch, {0}/recipes$: DataBrewResponse.dispatch, {0}/recipes/(?Precipe_name[^/])$: DataBrewResponse.dispatch, {0}/recipes/(?Precipe_name[^/])/recipeVersion/(?Precipe_version[^/]): DataBrewResponse.dispatch, {0}/recipes/(?Precipe_name[^/])/publishRecipe$: DataBrewResponse.dispatch, {0}/rulesets$: DataBrewResponse.dispatch, {0}/rulesets/(?Pruleset_name[^/])$: DataBrewResponse.dispatch, {0}/datasets$: DataBrewResponse.dispatch, {0}/datasets/(?Pdataset_name[^/])$: DataBrewResponse.dispatch, {0}/jobs$: DataBrewResponse.dispatch, {0}/jobs/(?Pjob_name[^/])$: DataBrewResponse.dispatch, {0}/profileJobs$: DataBrewResponse.dispatch, {0}/recipeJobs$: DataBrewResponse.dispatch, {0}/profileJobs/(?Pjob_name[^/])$: DataBrewResponse.dispatch, {0}/recipeJobs/(?Pjob_name[^/])$: DataBrewResponse.dispatch, }可以注意到几个路由设计细节资源名通过 URL 路径捕获如recipes/(?Precipe_name[^/])用正则捕获 recipe 名称允许含空格响应层再通过unquote解码见 responses.py 的_get_path。REST 风格子路径publishRecipe、recipeVersion/{version}等作为资源子路径而不是独立的资源类型这要求 moto 的 URL 正则必须精确匹配这些后缀。profile job 与 recipe job 共用命名空间profileJobs和recipeJobs各有独立的创建端点但两者写入同一个jobs字典名称冲突时会互相干扰下文详述。所有匹配请求都交给 responses.py 中的DataBrewResponse类由 moto 核心框架根据 boto3 调用的操作名如CreateRecipe分派到同名小写方法如create_recipe。三、Recipe 版本模型理解 moto 对食谱版本的还原models.py 是 DataBrew 后端的核心。DataBrewBackend.__init__用四个OrderedDict分别管理 recipes、rulesets、datasets、jobs见 models.py L58-L63保证列表接口的稳定排序。Recipe 是版本管理最复杂的资源FakeRecipe类models.py L428-L506实现了一套贴近真实 AWS 行为的版本状态机初始工作版本create_recipe时创建版本号为0.1的FakeRecipeVersion作为latest_working此时还没有任何已发布版本latest_published None。发布publish流程FakeRecipe.publishmodels.py L472-L482执行以下操作将当前latest_working设为latest_published并把其版本号向上取整为整数0.1→1.0第二次发布 →2.0同时记录published_date对已发布版本做一次deepcopy作为新的latest_working版本号递增为已发布版本 0.1即1.1、2.1因此发布后工作副本与已发布版本内容相同但版本不同新工作副本的created_time继承上次发布的published_date——测试 test_databrew_recipes.py 中test_publish_recipe用freeze_time验证了这一行为发布后LATEST_WORKING的CreateDate等于发布版本的PublishedDate。描述describe的版本语义describe_recipe在不传RecipeVersion时默认取LATEST_PUBLISHEDmodels.py L170-L197。因此刚创建、从未发布的 recipe 若不带版本号地 describe会抛出ResourceNotFoundException消息为The recipe {name} for version LATEST_PUBLISHED wasnt found.HTTP 404。版本合法性校验version_is_valid要求版本号长度在 1–16 之间且要么是浮点数要么是LATEST_WORKING或LATEST_PUBLISHED之一delete_recipe_version则不允许删除LATEST_PUBLISHED与LATEST_WORKINGValidationException: Recipe version ... is not allowed to be deleted删除最新已发布版本时会向前回退到更早的已发布版本delete_published_versionmodels.py L492-L506。名称长度约束recipe 名称不得超过 255 字符否则返回ValidationExceptionMember must have length less than or equal to 255对应源码中的validate_length静态方法models.py L65-L71。测试中的最小 recipe 示例摘自 test_databrew_recipes.pyclient.create_recipe( Namerecipe_name, Steps[ { Action: { Operation: REMOVE_COMBINED, Parameters: { sourceColumn: FakeColumn, removeSpecialCharacters: true, }, } } ], Tags{env: test, project: moto}, )完整的典型工作流创建 → 发布 → 按版本查询 → 删除已发布版本可参考 test_databrew_recipes.py L480-L515发布两次后list_recipe_versions返回1.0与2.0两个版本删除1.0后LATEST_PUBLISHED指向回退的版本而1.1工作版本仍然可以 describe。四、Dataset格式选项与 ResourceArn 的生成create_dataset接受四类元数据见 models.py L264-L287Format如JSON、CSV、EXCEL与对应的FormatOptionsInput可包含S3InputDefinition、DataCatalogInputDefinition、DatabaseInputDefinition三种输入定义moto 原样保存不做存在性校验PathOptions如LastModifiedDateCondition、FilesLimitMaxFiles/OrderedBy/Order、Parameters等Tags。重复创建同名 dataset 会抛出AlreadyExistsException消息{name} already exists.update_dataset与delete_dataset对不存在的资源抛出ResourceNotFoundException消息为 One or more resources cant be found.。每个 dataset 都带一个自动生成的ResourceArnmodels.py L601-L603farn:{get_partition(self.region_name)}:databrew:{self.region_name}:{self.account_id}:dataset/{self.name}分区前缀通过get_partition按区域推断us-west-1得到aws因此 describe 响应中的 ARN 形如arn:aws:databrew:us-west-1:{account_id}:dataset/{name}——test_databrew_datasets.py 正是用DEFAULT_ACCOUNT_ID断言了这一格式。update_dataset采用部分更新语义仅对非None的字段赋值未传字段保持原值。五、Profile Job 与 Recipe Job共享命名空间与参数校验Job 是另一块实现较深的区域。FakeJob抽象基类models.py L622-L694定义了公共字段Name、DatasetName、EncryptionMode、LogSubscription、MaxCapacity、MaxRetries、RoleArn、Tags与校验逻辑两个子类通过local_attrs声明各自的扩展字段FakeProfileJobjob_type PROFILEoutput_location、configuration、validation_configurationsFakeRecipeJobjob_type RECIPEdatabase_outputs、data_catalog_outputs、outputs、project_name、recipe_reference。as_dict会把子类属性通过camelcase_to_pascal(underscores_to_camelcase(k))转换为 AWS 响应风格的 PascalCase 键并剔除值为None的字段models.py L685-L692。几个关键的实现行为Profile 与 Recipe job 共享名称命名空间两者都写入self.jobs。create_recipe_job时若名称已被 profile job 占用会抛出ConflictException且消息中的类型取自已存在 job 的类型The job {name} profile job already exists.。test_databrew_jobs.py L131-L141 专门验证了这种跨类型冲突。枚举值校验validate方法models.py L648-L658限制EncryptionMode ∈ {SSE-S3, SSE-KMS}、LogSubscription ∈ {ENABLE, DISABLE}非法值抛出带完整约束描述的ValidationExceptionHTTP 400。参数解析responses.py L284-L302 中create_profile_job从请求体解析出 14 个参数含Timeout、JobSample、ValidationConfigurations等传给后端update_profile_job与update_recipe_job共用update_job对非None参数执行setattr式覆盖models.py L384-L395。Job 名称长度上限 240 字符describe_job/delete_job/创建类接口都会先做长度校验超长名称返回 400。ResourceArn与 dataset 类似job 的 ARN 为arn:{partition}:databrew:{region}:{account_id}:job/{name}。list_jobs支持按DatasetName与ProjectName过滤models.py L405-L425。从源码结构看过滤函数对 recipe job 检查project_name属性、对 profile job 用getattr(job, project_name, None)兜底因此ProjectName过滤实际上只对 recipe job 生效——这与测试test_list_jobs_project_name_filter的断言一致。六、Ruleset 与其他资源Ruleset 的 CRUD 实现相对直接创建时校验重名RulesetAlreadyExistsException消息 Ruleset already exists.describe/update/delete 对不存在资源抛RulesetNotFoundException错误码为EntityNotFoundException消息 Ruleset {name} not found.见 exceptions.py L42-L44。注意 ruleset 与其他资源的错误类型并不统一dataset/job 用ResourceNotFoundException404ruleset 用EntityNotFoundException默认 400这与测试 test_databrew_rulesets.py 的断言一致编写断言时需留意。测试中 ruleset 的规则示例test_databrew_rulesets.py L18-L35展示了CheckExpressionSubstitutionMapThreshold的结构Rules[{ Name: Assert values 0, Disabled: False, CheckExpression: :col1 :val1, SubstitutionMap: {:col1: Value, :val1: 0}, Threshold: {Value: 100, Type: GREATER_THAN_OR_EQUAL, Unit: PERCENTAGE}, }]七、分页与异常体系分页list_recipes、list_recipe_versions、list_rulesets、list_datasets、list_jobs五个列表接口都通过paginate装饰器接入 moto 通用分页模型PAGINATION_MODELmodels.py L25-L56统一使用next_token作为分页游标、max_results作为页大小默认 100、以name作为唯一排序键。列表接口因此返回Recipes/Datasets/Rulesets/Jobs数组外加NextTokentest_databrew_recipes.py L91-L98 验证了 10 条记录、MaxResults3时第二页恰好返回 7 条。异常体系exceptions.py全部继承自JsonRESTError按 HTTP 状态码分层异常类状态码触发场景ValidationException400名称超长、版本号非法、枚举值非法AlreadyExistsException/RulesetAlreadyExistsException400dataset/ruleset 重名创建ConflictException409recipe/job 重名创建ResourceNotFoundException404recipe/dataset/job 不存在RulesetNotFoundExceptionEntityNotFoundException400ruleset 不存在八、实战在测试中 Mock DataBrew 的完整示例基于仓库测试的组织方式tests/test_databrew/一个典型的 DataBrew Mock 测试骨架如下import boto3 import pytest from botocore.exceptions import ClientError from moto import mock_aws def _create_databrew_client(): return boto3.client(databrew, region_nameus-west-1) mock_aws def test_recipe_publish_flow(): client _create_databrew_client() response client.create_recipe( Namemy_recipe, Steps[{Action: {Operation: REMOVE_COMBINED, Parameters: {sourceColumn: Col1}}}], ) # 未发布前默认LATEST_PUBLISHED查询会 404 with pytest.raises(ClientError) as exc: client.describe_recipe(Namemy_recipe) assert exc.value.response[Error][Code] ResourceNotFoundException client.publish_recipe(Namemy_recipe, Description1st desc) recipe client.describe_recipe(Namemy_recipe) assert recipe[RecipeVersion] 1.0 assert recipe[Description] 1st desc working client.describe_recipe(Namemy_recipe, RecipeVersionLATEST_WORKING) assert working[RecipeVersion] 1.1 with pytest.raises(ClientError) as exc: client.delete_recipe_version(Namemy_recipe, RecipeVersionLATEST_WORKING) assert exc.value.response[Error][Code] ValidationException运行前置条件安装 moto 与 botocore/boto3版本要求以仓库 requirements.txt 为准测试通过mock_aws装饰器拦截 boto3 对databrew.us-west-1.amazonaws.com的 HTTP 请求并转发给DataBrewBackend处理也可以在 server mode 下运行 moto_server 让真实网络请求打到本地服务。九、参考路径汇总内容仓库路径服务覆盖清单本文主体文档docs/docs/services/databrew.rst后端模型版本状态机、分页、校验moto/databrew/models.py请求解析与响应构造moto/databrew/responses.pyURL 路由moto/databrew/urls.py异常定义moto/databrew/exceptions.pyRecipe 测试版本流与错误分支tests/test_databrew/test_databrew_recipes.pyDataset 测试tests/test_databrew/test_databrew_datasets.pyJob 测试tests/test_databrew/test_databrew_jobs.pyRuleset 测试tests/test_databrew/test_databrew_rulesets.py总结而言moto 当前对 DataBrew 的支持聚焦于定义型APIrecipe 的多版本状态机0.1 工作版、整版发布、0.1 工作副本递增、dataset/ruleset 的完整 CRUD、两类 job 的共享命名空间与参数校验以及统一的 NextToken 分页。在编写依赖 DataBrew 的测试时先对照 docs/docs/services/databrew.rst 的清单确认 API 可用性再参考tests/test_databrew/下的断言风格处理异常分支即可得到与真实 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测试Windows 上手 faster-whisperCUDA 12 加速语音转写实战Windows 上手 faster whisperCUDA 12 加速语音转写实战 faster whisper 用 CTranslate2 重写了 OpenMock测试moto 中 cognito-identity 服务 mock 全解Identity Pool 操作覆盖、实现原理与测试实战moto 中 cognito identity 服务 mock 全解Identity Pool 操作覆盖、实现原理与测试实战 本文基于 moto 仓库中 doMock测试上一篇Element 下拉菜单 Dropdown 组件完全指南触发方式、指令事件与源码级原理剖析下一篇Jupyter AI实战指南三步打造智能编程工作流创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 moto 中 DataBrew 的 Mock 实现：API 覆盖范围、源码结构与测试实战指南
Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载本篇以 moto 仓库的 DataBrew 服务文档 为主体系统梳理 moto 对 AWS Glue DataBrew 的 Mock 能力边界哪些 API 操作已实现recipes、datasets、rulesets、profile/recipe jobs 及分页哪些尚未覆盖projects、schedules、job runs、tags。读完本文你将掌握如何用mock_aws装饰器在单元测试中完整地模拟 DataBrew 的创建、发布、版本管理与异常分支并能从源码层面理解版本状态机、路由分发与校验逻辑的实现原理。一、服务定位与 API 覆盖清单moto 是一个 AWS 基础设施 Mock 库允许开发者在不访问真实 AWS 的情况下测试基于 boto3 的代码。对于 AWS Glue DataBrew数据剖析与数据转换服务moto 的 服务文档页 以勾选清单形式列出了当前实现的 API 操作。这份清单是评估测试可行性时最直接的依据——在编写依赖 DataBrew 的测试前先核对目标 API 是否在已实现列表中可以避免遇到未实现的 API 调用。已实现的操作Implemented features资源域已实现操作Recipe 食谱create_recipe、describe_recipe、update_recipe、publish_recipe、delete_recipe_version、list_recipes、list_recipe_versionsDataset 数据集create_dataset、describe_dataset、update_dataset、delete_dataset、list_datasetsRuleset 规则集create_ruleset、describe_ruleset、update_ruleset、delete_ruleset、list_rulesetsJob 作业create_profile_job、update_profile_job、create_recipe_job、update_recipe_job、describe_job、delete_job、list_jobs未实现的操作Unimplementedbatch_delete_recipe_version、create_project、create_schedule、delete_project、delete_schedule、describe_job_run、describe_project、describe_schedule、list_job_runs、list_projects、list_schedules、list_tags_for_resource、send_project_session_action、start_job_run、start_project_session、stop_job_run、tag_resource、untag_resource、update_project、update_schedule。从清单可以看出覆盖范围的特点DataBrew 的定义层recipe 的版本管理、dataset 元数据、ruleset、job 定义与更新已被完整 Mock而运行时层project 交互式会话、schedule 调度、job run 的执行与查询以及 tags 资源类接口尚未实现。如果你的被测代码只涉及创建/查询/更新这些定义型资源可以直接使用 moto若涉及start_job_run或 project 会话则需要另行处理。二、请求路由DataBrew 端点如何被识别与分发DataBrew 的 HTTP 端点在 urls.py 中声明。URL 基址匹配databrew.{region}.amazonaws.com形式的域名url_bases [rhttps?://databrew\.(.)\.amazonaws.com] url_paths { {0}/recipeVersions$: DataBrewResponse.dispatch, {0}/recipes$: DataBrewResponse.dispatch, {0}/recipes/(?Precipe_name[^/])$: DataBrewResponse.dispatch, {0}/recipes/(?Precipe_name[^/])/recipeVersion/(?Precipe_version[^/]): DataBrewResponse.dispatch, {0}/recipes/(?Precipe_name[^/])/publishRecipe$: DataBrewResponse.dispatch, {0}/rulesets$: DataBrewResponse.dispatch, {0}/rulesets/(?Pruleset_name[^/])$: DataBrewResponse.dispatch, {0}/datasets$: DataBrewResponse.dispatch, {0}/datasets/(?Pdataset_name[^/])$: DataBrewResponse.dispatch, {0}/jobs$: DataBrewResponse.dispatch, {0}/jobs/(?Pjob_name[^/])$: DataBrewResponse.dispatch, {0}/profileJobs$: DataBrewResponse.dispatch, {0}/recipeJobs$: DataBrewResponse.dispatch, {0}/profileJobs/(?Pjob_name[^/])$: DataBrewResponse.dispatch, {0}/recipeJobs/(?Pjob_name[^/])$: DataBrewResponse.dispatch, }可以注意到几个路由设计细节资源名通过 URL 路径捕获如recipes/(?Precipe_name[^/])用正则捕获 recipe 名称允许含空格响应层再通过unquote解码见 responses.py 的_get_path。REST 风格子路径publishRecipe、recipeVersion/{version}等作为资源子路径而不是独立的资源类型这要求 moto 的 URL 正则必须精确匹配这些后缀。profile job 与 recipe job 共用命名空间profileJobs和recipeJobs各有独立的创建端点但两者写入同一个jobs字典名称冲突时会互相干扰下文详述。所有匹配请求都交给 responses.py 中的DataBrewResponse类由 moto 核心框架根据 boto3 调用的操作名如CreateRecipe分派到同名小写方法如create_recipe。三、Recipe 版本模型理解 moto 对食谱版本的还原models.py 是 DataBrew 后端的核心。DataBrewBackend.__init__用四个OrderedDict分别管理 recipes、rulesets、datasets、jobs见 models.py L58-L63保证列表接口的稳定排序。Recipe 是版本管理最复杂的资源FakeRecipe类models.py L428-L506实现了一套贴近真实 AWS 行为的版本状态机初始工作版本create_recipe时创建版本号为0.1的FakeRecipeVersion作为latest_working此时还没有任何已发布版本latest_published None。发布publish流程FakeRecipe.publishmodels.py L472-L482执行以下操作将当前latest_working设为latest_published并把其版本号向上取整为整数0.1→1.0第二次发布 →2.0同时记录published_date对已发布版本做一次deepcopy作为新的latest_working版本号递增为已发布版本 0.1即1.1、2.1因此发布后工作副本与已发布版本内容相同但版本不同新工作副本的created_time继承上次发布的published_date——测试 test_databrew_recipes.py 中test_publish_recipe用freeze_time验证了这一行为发布后LATEST_WORKING的CreateDate等于发布版本的PublishedDate。描述describe的版本语义describe_recipe在不传RecipeVersion时默认取LATEST_PUBLISHEDmodels.py L170-L197。因此刚创建、从未发布的 recipe 若不带版本号地 describe会抛出ResourceNotFoundException消息为The recipe {name} for version LATEST_PUBLISHED wasnt found.HTTP 404。版本合法性校验version_is_valid要求版本号长度在 1–16 之间且要么是浮点数要么是LATEST_WORKING或LATEST_PUBLISHED之一delete_recipe_version则不允许删除LATEST_PUBLISHED与LATEST_WORKINGValidationException: Recipe version ... is not allowed to be deleted删除最新已发布版本时会向前回退到更早的已发布版本delete_published_versionmodels.py L492-L506。名称长度约束recipe 名称不得超过 255 字符否则返回ValidationExceptionMember must have length less than or equal to 255对应源码中的validate_length静态方法models.py L65-L71。测试中的最小 recipe 示例摘自 test_databrew_recipes.pyclient.create_recipe( Namerecipe_name, Steps[ { Action: { Operation: REMOVE_COMBINED, Parameters: { sourceColumn: FakeColumn, removeSpecialCharacters: true, }, } } ], Tags{env: test, project: moto}, )完整的典型工作流创建 → 发布 → 按版本查询 → 删除已发布版本可参考 test_databrew_recipes.py L480-L515发布两次后list_recipe_versions返回1.0与2.0两个版本删除1.0后LATEST_PUBLISHED指向回退的版本而1.1工作版本仍然可以 describe。四、Dataset格式选项与 ResourceArn 的生成create_dataset接受四类元数据见 models.py L264-L287Format如JSON、CSV、EXCEL与对应的FormatOptionsInput可包含S3InputDefinition、DataCatalogInputDefinition、DatabaseInputDefinition三种输入定义moto 原样保存不做存在性校验PathOptions如LastModifiedDateCondition、FilesLimitMaxFiles/OrderedBy/Order、Parameters等Tags。重复创建同名 dataset 会抛出AlreadyExistsException消息{name} already exists.update_dataset与delete_dataset对不存在的资源抛出ResourceNotFoundException消息为 One or more resources cant be found.。每个 dataset 都带一个自动生成的ResourceArnmodels.py L601-L603farn:{get_partition(self.region_name)}:databrew:{self.region_name}:{self.account_id}:dataset/{self.name}分区前缀通过get_partition按区域推断us-west-1得到aws因此 describe 响应中的 ARN 形如arn:aws:databrew:us-west-1:{account_id}:dataset/{name}——test_databrew_datasets.py 正是用DEFAULT_ACCOUNT_ID断言了这一格式。update_dataset采用部分更新语义仅对非None的字段赋值未传字段保持原值。五、Profile Job 与 Recipe Job共享命名空间与参数校验Job 是另一块实现较深的区域。FakeJob抽象基类models.py L622-L694定义了公共字段Name、DatasetName、EncryptionMode、LogSubscription、MaxCapacity、MaxRetries、RoleArn、Tags与校验逻辑两个子类通过local_attrs声明各自的扩展字段FakeProfileJobjob_type PROFILEoutput_location、configuration、validation_configurationsFakeRecipeJobjob_type RECIPEdatabase_outputs、data_catalog_outputs、outputs、project_name、recipe_reference。as_dict会把子类属性通过camelcase_to_pascal(underscores_to_camelcase(k))转换为 AWS 响应风格的 PascalCase 键并剔除值为None的字段models.py L685-L692。几个关键的实现行为Profile 与 Recipe job 共享名称命名空间两者都写入self.jobs。create_recipe_job时若名称已被 profile job 占用会抛出ConflictException且消息中的类型取自已存在 job 的类型The job {name} profile job already exists.。test_databrew_jobs.py L131-L141 专门验证了这种跨类型冲突。枚举值校验validate方法models.py L648-L658限制EncryptionMode ∈ {SSE-S3, SSE-KMS}、LogSubscription ∈ {ENABLE, DISABLE}非法值抛出带完整约束描述的ValidationExceptionHTTP 400。参数解析responses.py L284-L302 中create_profile_job从请求体解析出 14 个参数含Timeout、JobSample、ValidationConfigurations等传给后端update_profile_job与update_recipe_job共用update_job对非None参数执行setattr式覆盖models.py L384-L395。Job 名称长度上限 240 字符describe_job/delete_job/创建类接口都会先做长度校验超长名称返回 400。ResourceArn与 dataset 类似job 的 ARN 为arn:{partition}:databrew:{region}:{account_id}:job/{name}。list_jobs支持按DatasetName与ProjectName过滤models.py L405-L425。从源码结构看过滤函数对 recipe job 检查project_name属性、对 profile job 用getattr(job, project_name, None)兜底因此ProjectName过滤实际上只对 recipe job 生效——这与测试test_list_jobs_project_name_filter的断言一致。六、Ruleset 与其他资源Ruleset 的 CRUD 实现相对直接创建时校验重名RulesetAlreadyExistsException消息 Ruleset already exists.describe/update/delete 对不存在资源抛RulesetNotFoundException错误码为EntityNotFoundException消息 Ruleset {name} not found.见 exceptions.py L42-L44。注意 ruleset 与其他资源的错误类型并不统一dataset/job 用ResourceNotFoundException404ruleset 用EntityNotFoundException默认 400这与测试 test_databrew_rulesets.py 的断言一致编写断言时需留意。测试中 ruleset 的规则示例test_databrew_rulesets.py L18-L35展示了CheckExpressionSubstitutionMapThreshold的结构Rules[{ Name: Assert values 0, Disabled: False, CheckExpression: :col1 :val1, SubstitutionMap: {:col1: Value, :val1: 0}, Threshold: {Value: 100, Type: GREATER_THAN_OR_EQUAL, Unit: PERCENTAGE}, }]七、分页与异常体系分页list_recipes、list_recipe_versions、list_rulesets、list_datasets、list_jobs五个列表接口都通过paginate装饰器接入 moto 通用分页模型PAGINATION_MODELmodels.py L25-L56统一使用next_token作为分页游标、max_results作为页大小默认 100、以name作为唯一排序键。列表接口因此返回Recipes/Datasets/Rulesets/Jobs数组外加NextTokentest_databrew_recipes.py L91-L98 验证了 10 条记录、MaxResults3时第二页恰好返回 7 条。异常体系exceptions.py全部继承自JsonRESTError按 HTTP 状态码分层异常类状态码触发场景ValidationException400名称超长、版本号非法、枚举值非法AlreadyExistsException/RulesetAlreadyExistsException400dataset/ruleset 重名创建ConflictException409recipe/job 重名创建ResourceNotFoundException404recipe/dataset/job 不存在RulesetNotFoundExceptionEntityNotFoundException400ruleset 不存在八、实战在测试中 Mock DataBrew 的完整示例基于仓库测试的组织方式tests/test_databrew/一个典型的 DataBrew Mock 测试骨架如下import boto3 import pytest from botocore.exceptions import ClientError from moto import mock_aws def _create_databrew_client(): return boto3.client(databrew, region_nameus-west-1) mock_aws def test_recipe_publish_flow(): client _create_databrew_client() response client.create_recipe( Namemy_recipe, Steps[{Action: {Operation: REMOVE_COMBINED, Parameters: {sourceColumn: Col1}}}], ) # 未发布前默认LATEST_PUBLISHED查询会 404 with pytest.raises(ClientError) as exc: client.describe_recipe(Namemy_recipe) assert exc.value.response[Error][Code] ResourceNotFoundException client.publish_recipe(Namemy_recipe, Description1st desc) recipe client.describe_recipe(Namemy_recipe) assert recipe[RecipeVersion] 1.0 assert recipe[Description] 1st desc working client.describe_recipe(Namemy_recipe, RecipeVersionLATEST_WORKING) assert working[RecipeVersion] 1.1 with pytest.raises(ClientError) as exc: client.delete_recipe_version(Namemy_recipe, RecipeVersionLATEST_WORKING) assert exc.value.response[Error][Code] ValidationException运行前置条件安装 moto 与 botocore/boto3版本要求以仓库 requirements.txt 为准测试通过mock_aws装饰器拦截 boto3 对databrew.us-west-1.amazonaws.com的 HTTP 请求并转发给DataBrewBackend处理也可以在 server mode 下运行 moto_server 让真实网络请求打到本地服务。九、参考路径汇总内容仓库路径服务覆盖清单本文主体文档docs/docs/services/databrew.rst后端模型版本状态机、分页、校验moto/databrew/models.py请求解析与响应构造moto/databrew/responses.pyURL 路由moto/databrew/urls.py异常定义moto/databrew/exceptions.pyRecipe 测试版本流与错误分支tests/test_databrew/test_databrew_recipes.pyDataset 测试tests/test_databrew/test_databrew_datasets.pyJob 测试tests/test_databrew/test_databrew_jobs.pyRuleset 测试tests/test_databrew/test_databrew_rulesets.py总结而言moto 当前对 DataBrew 的支持聚焦于定义型APIrecipe 的多版本状态机0.1 工作版、整版发布、0.1 工作副本递增、dataset/ruleset 的完整 CRUD、两类 job 的共享命名空间与参数校验以及统一的 NextToken 分页。在编写依赖 DataBrew 的测试时先对照 docs/docs/services/databrew.rst 的清单确认 API 可用性再参考tests/test_databrew/下的断言风格处理异常分支即可得到与真实 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测试Windows 上手 faster-whisperCUDA 12 加速语音转写实战Windows 上手 faster whisperCUDA 12 加速语音转写实战 faster whisper 用 CTranslate2 重写了 OpenMock测试moto 中 cognito-identity 服务 mock 全解Identity Pool 操作覆盖、实现原理与测试实战moto 中 cognito identity 服务 mock 全解Identity Pool 操作覆盖、实现原理与测试实战 本文基于 moto 仓库中 doMock测试上一篇Element 下拉菜单 Dropdown 组件完全指南触发方式、指令事件与源码级原理剖析下一篇Jupyter AI实战指南三步打造智能编程工作流创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考