服务的模拟实现与实战指南)
Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载Amazon Managed PrometheusAMPAWS 的托管 Prometheus 服务是云上运行 Prometheus 工作负载的核心托管组件。本文聚焦开源仓库 moto 中amp服务的模拟实现以 docs/docs/services/amp.rst 中列出的功能清单为骨架结合 moto/amp/models.py 等源码与 tests/test_amp/ 下的测试用例全面讲解 Workspace、Rule Groups Namespace、Logging Configuration、标签管理与分页等已实现 API 的用法、参数细节与底层原理。读完本文你将能在本地测试环境中用mock_aws装饰器完整模拟 AMP 工作流的创建、查询、更新与删除操作。AMP 服务在 Moto 中的整体支持范围AMP 服务在 moto 中对应moto/amp模块其网络入口模拟的是 AWS AMP 的真实端点aps.{region}.amazonaws.com这一点可以从 moto/amp/urls.py 的url_bases定义得到确认url_bases [ rhttps?://aps\.(.)\.amazonaws\.com, ]核心后端类为 moto/amp/models.py 中的PrometheusServiceBackend它继承自BaseBackend并通过BackendDict(PrometheusServiceBackend, amp)按账户account 区域region隔离存储因此不同 region 的 workspace 互不干扰这与 moto 其他服务的多区域模型一致。已实现功能清单原文档通过勾选列表标注了每个 API 的实现状态这是判断 moto 中 amp 服务可用能力的最权威依据。已实现[X]的接口如下API 操作说明ClientToken 支持create_workspace创建工作区未实现describe_workspace查询工作区详情—list_workspaces列出工作区支持分页与 alias 过滤—update_workspace_alias更新工作区别名未实现delete_workspace删除工作区未实现create_rule_groups_namespace创建规则组命名空间未实现describe_rule_groups_namespace查询规则组命名空间—list_rule_groups_namespaces列出规则组命名空间支持分页与名称前缀过滤—put_rule_groups_namespace更新规则组命名空间内容未实现delete_rule_groups_namespace删除规则组命名空间未实现create_logging_configuration配置工作区日志输出—describe_logging_configuration查询日志配置—update_logging_configuration更新日志配置—delete_logging_configuration删除日志配置—tag_resource为资源打标签—untag_resource移除资源标签—list_tags_for_resource查询资源标签—同时原文档也明确标注了未实现的操作包括create_alert_manager_definition、delete_alert_manager_definition、describe_alert_manager_definition、put_alert_manager_definitionAlert Manager 定义相关全套、create_scraper、delete_scraper、describe_scraper、list_scrapers、update_scraper等 Scraper 相关接口以及create_anomaly_detector、delete_anomaly_detector、describe_anomaly_detector、list_anomaly_detectors、put_anomaly_detector异常检测器、查询日志配置create/describe/update/delete_query_logging_configuration、资源策略describe/delete/put_resource_policy、工作区配置describe/update_workspace_configuration与get_default_scraper_configuration。在使用这些未实现的 API 时boto3 客户端会抛出服务端错误这一点需要在使用前评估。关于 ClientToken 参数的特别说明原文档在多个已实现 API 下方反复标注The ClientToken-parameter is not yet implemented即create_workspace、create_rule_groups_namespace、delete_rule_groups_namespace、put_rule_groups_namespace、delete_workspace、update_workspace_alias等接口虽然在签名上接受clientToken参数调用时不会报错但该参数在模拟实现中不会产生任何幂等性语义。这一点在 moto/amp/models.py 中可以直接看到——create_workspace的 docstring 即为该句说明方法体内并未读取 clientToken。测试文件 tests/test_amp/test_amp_workspaces.py 中create_workspace(aliastest, clientTokenmytoken)的写法也证实了该参数可被传入但被忽略。Workspace 生命周期管理创建、查询、更新与删除Workspace 是 AMP 的核心资源对应源码中的 Workspace 模型。模拟实现会在创建时生成以下关键字段workspace_id形如ws-{uuid4}的随机 ID来自moto.moto_api._internal.mock_random.uuid4arn形如arn:{partition}:aps:{region}:{account_id}:workspace/{workspace_id}其中 partition 由get_partition(region)计算如aws、aws-cn等prometheusEndpoint形如https://aps-workspaces.{region}.amazonaws.com/workspaces/{workspace_id}/status固定为{statusCode: ACTIVE}createdAt创建时的 unix 时间戳。创建与查询使用mock_aws装饰器即可在本地模拟完整的 AWS 调用链import boto3 from moto import mock_aws mock_aws def test_workspace_lifecycle(): client boto3.client(amp, region_nameap-southeast-1) resp client.create_workspace(aliasmy-prom, tags{env: dev}) workspace_id resp[workspaceId] assert resp[status] {statusCode: ACTIVE} assert resp[alias] my-prom detail client.describe_workspace(workspaceIdworkspace_id)[workspace] assert detail[workspaceId] workspace_id assert prometheusEndpoint in detail assert createdAt in detail注意响应结构的差异create_workspace直接返回 workspace 对象的所有字段顶层展开而describe_workspace返回嵌套的{workspace: {...}}结构与真实 AWS API 保持一致见 moto/amp/responses.py。列出与别名更新list_workspaces支持两个筛选维度alias精确匹配过滤仅返回alias完全相等的 workspace见 models.py 中的w.alias aliasmaxResults/nextToken分页参数默认每页 100 条。client.update_workspace_alias(aliasrenamed, workspaceIdworkspace_id) # alias 过滤 only_renamed client.list_workspaces(aliasrenamed)[workspaces] assert len(only_renamed) 1删除与异常行为delete_workspace直接从内部 dict 中弹出该 workspacepop(workspace_id, None)随后对该 ID 执行describe_workspace会抛出ResourceNotFoundException。测试 test_amp_workspaces.py 验证了这一点client.delete_workspace(workspaceIdworkspace_id) with pytest.raises(ClientError) as exc: client.describe_workspace(workspaceIdworkspace_id) assert exc.value.response[Error][Code] ResourceNotFoundException assert exc.value.response[Error][Message] Workspace not found该异常由 moto/amp/exceptions.py 中的WorkspaceNotFound抛出HTTP 状态码为 404resourceType为AWS::APS::Workspace。Rule Groups NamespacePrometheus 规则组的命名空间管理Rule Groups Namespace 用于承载 Prometheus 的 recording rule 与 alerting rule 配置是 AMP 模拟中最常用的内容型资源。每个命名空间归属于某个 workspace对应 RuleGroupNamespace 模型其 ARN 格式为arn:{partition}:aps:{region}:{account_id}:rulegroupsnamespace/{workspace_id}/{name}命名空间模型内部维护created_at与modified_at两个时间戳put_rule_groups_namespace更新data内容时仅刷新modified_at见 update 方法。创建与描述data参数在 boto3 中为 bytes 类型底层为 YAML 格式的规则组内容mock_aws def test_rulegroups(): client boto3.client(amp, region_nameeu-west-1) workspace_id client.create_workspace()[workspaceId] client.create_rule_groups_namespace( databgroups:\n - name: example\n rules: [], namemy-rules, workspaceIdworkspace_id, ) ns client.describe_rule_groups_namespace( namemy-rules, workspaceIdworkspace_id )[ruleGroupsNamespace] assert ns[data] bgroups:\n - name: example\n rules: [] assert ns[status] {statusCode: ACTIVE} assert createdAt in ns and modifiedAt in ns与 workspace 一样创建/更新操作直接返回命名空间对象字段而describe_rule_groups_namespace返回嵌套的{ruleGroupsNamespace: {...}}结构。更新put与删除client.put_rule_groups_namespace( namemy-rules, workspaceIdworkspace_id, databgroups: [] ) ns client.describe_rule_groups_namespace( namemy-rules, workspaceIdworkspace_id )[ruleGroupsNamespace] assert ns[data] bgroups: [] client.delete_rule_groups_namespace(namemy-rules, workspaceIdworkspace_id)删除后查询同一命名空间会抛出RuleGroupNamespaceNotFound404resourceType为AWS::APS::RuleGroupNamespace见 exceptions.py。列表与名称前缀过滤list_rule_groups_namespaces的name参数是前缀匹配而非精确匹配——源码中通过ns_name.startswith(name)实现models.py。测试 test_amp_rulegroupnamespaces.py 演示了典型行为存在ns0ns14共 15 个命名空间时namens1会匹配到ns1、ns10ns14共 6 个。Logging Configuration工作区日志输出配置日志配置用于将 AMP 的查询与指标日志发送到 CloudWatch Logs对应 models.py 中的四个方法。模拟实现的数据结构包含logGroupArn、createdAt、status固定ACTIVE与workspace字段。创建与查询mock_aws def test_logging(): client boto3.client(amp, region_nameus-east-2) workspace_id client.create_workspace()[workspaceId] resp client.create_logging_configuration( workspaceIdworkspace_id, logGroupArnarn:aws:logs:us-east-2:123456789012:log-group:/amp/logs ) assert resp[status] {statusCode: ACTIVE} cfg client.describe_logging_configuration(workspaceIdworkspace_id)[loggingConfiguration] assert cfg[logGroupArn].endswith(/amp/logs) assert cfg[workspace] workspace_id值得注意的细节在从未创建过日志配置时describe_logging_configuration返回的是空字典{}models.py 中logging_config is None时返回{}而不是报错。测试 test_amp_logging_config.py 专门验证了这一行为。更新与删除client.update_logging_configuration( workspaceIdworkspace_id, logGroupArnarn:aws:logs:us-east-2:123456789012:log-group:/amp/logs-v2 ) cfg client.describe_logging_configuration(workspaceIdworkspace_id)[loggingConfiguration] assert modifiedAt in cfg assert cfg[logGroupArn].endswith(/amp/logs-v2) client.delete_logging_configuration(workspaceIdworkspace_id) assert client.describe_logging_configuration(workspaceIdworkspace_id)[loggingConfiguration] {}update_logging_configuration会覆盖logGroupArn并追加modifiedAt时间戳delete_logging_configuration将内部配置置为None使后续查询重新回到空字典状态。标签管理tag_resource / untag_resource / list_tags_for_resourceAMP 模拟的标签能力统一由TaggingService提供models.py可作用于 Workspace 与 RuleGroupNamespace 两类资源。标签以 ARN 为键存储并通过tag_fn回调挂接到每个模型上因此describe返回的响应中会直接携带tags字段。mock_aws def test_tags(): client boto3.client(amp, region_nameus-east-2) arn client.create_workspace(aliastest, tags{env: prod})[arn] client.tag_resource(resourceArnarn, tags{team: platform}) assert client.list_tags_for_resource(resourceArnarn)[tags] { env: prod, team: platform } client.untag_resource(resourceArnarn, tagKeys[env]) assert client.list_tags_for_resource(resourceArnarn)[tags] {team: platform}由于底层是统一的TaggingServiceWorkspace 与 RuleGroupNamespace 的标签行为完全一致两组测试文件分别验证了各自的场景。untag_resource通过untag_resource_using_names按标签名批量删除可一次传入多个 tagKey。分页行为默认页大小与 nextTokenAMP 的两个列表接口list_workspaces、list_rule_groups_namespaces都支持 AWS 风格的分页分页模型定义在 moto/amp/utils.pyPAGINATION_MODEL { list_workspaces: { input_token: next_token, limit_key: max_results, limit_default: 100, unique_attribute: arn, }, list_rule_groups_namespaces: { input_token: next_token, limit_key: max_results, limit_default: 100, unique_attribute: name, }, }关键行为默认页大小 100不传maxResults时超过 100 条的列表只返回前 100 条并附带nextTokenmaxResults可缩小或扩大测试中展示了maxResults15搭配nextToken逐页取数以及maxResults1000一次性取完的场景test_amp_workspaces.py翻页游标属性workspace 分页以arn为唯一游标属性rulegroupsnamespace 分页以name为唯一游标属性最后一页无nextToken取完所有数据后响应中不再包含nextToken字段test_amp_rulegroupnamespaces.py。分页逻辑由moto.utilities.paginator中的paginate装饰器实现与 moto 其他服务保持一致。请求路由与响应分发机制理解 amp 的模拟实现还需要看清HTTP 请求 → 后端方法的调用链。整个链路由三部分构成URL 路由moto/amp/urls.py将所有aps.{region}.amazonaws.com下的路径统一派发到PrometheusServiceResponse.dispatch包括/workspaces、/workspaces/{id}/alias、/workspaces/{id}/logging、/workspaces/{id}/rulegroupsnamespaces以及/tags/...系列路径响应分发moto/amp/responses.pyPrometheusServiceResponse继承BaseResponse根据请求中携带的 API action 名调用对应的处理函数如create_workspace、describe_workspace等处理函数负责从请求 body/路径中解析参数、调用 backend 方法、并把结果序列化为与真实 AWS 一致的 JSON 结构后端模型moto/amp/models.pyPrometheusServiceBackend持有self.workspaces字典与self.tagger通过amp_backends[current_account][region]获取当前账户与区域对应的后端实例实现数据隔离。路径参数的解析方式值得注意responses.py中大量使用self.path.split(/)配合unquote从 URL 中提取workspace_id、name、resource_arn等参数例如describe_workspace取路径最后一段作为 workspaceId、create_rule_groups_namespace取倒数第二段。使用限制与注意事项总结综合原文档标注与源码实现使用 moto 的 amp 模拟时有以下几点需要提前知悉ClientToken 参数仅占位create_workspace、create_rule_groups_namespace、put_rule_groups_namespace、delete_rule_groups_namespace、delete_workspace、update_workspace_alias均接受但不处理clientToken不会提供幂等去重语义Alert Manager / Scraper / Anomaly Detector 等 API 未实现涉及告警管理、抓取任务、异常检测的业务场景目前无法在 moto 中端到端模拟需要等待后续版本补齐状态字段简化workspace 与命名空间的status固定为{statusCode: ACTIVE}不会模拟CREATING、DELETING、UPDATING等中间状态数据仅存在于内存所有 workspace、命名空间与标签均保存在后端实例的内存字典中进程退出即丢失适合测试场景而非持久化需求区域隔离由于BackendDict按 accountregion 建实例同一测试中跨 region 创建的 workspace 彼此独立。运行 amp 相关测试仓库在 tests/test_amp/ 下提供了三个测试文件分别覆盖 workspacetest_amp_workspaces.py、rule group namespacestest_amp_rulegroupnamespaces.py与日志配置test_amp_logging_config.py。在仓库根目录执行以下命令即可验证 amp 模拟的全部已实现功能pytest tests/test_amp/这些测试既是实现行为的权威佐证也是撰写自定义测试时可直接参考的模板——它们展示了mock_aws装饰器、boto3 客户端的 region 选择如ap-southeast-1、eu-west-1、us-east-2、以及ClientError断言的标准写法。总体而言moto 对 AMP 的模拟已经覆盖了工作区生命周期、规则组命名空间、日志配置与标签管理这四类最核心的日常操作足以支撑大多数基于 AMP 的应用在本地测试环境中的行为验证。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐Floci 的 Amazon Managed Service for Prometheus (AMP) 模拟实现Workspace 与 Rule Groups Namespace 全生命周期指南Floci 的 Amazon Managed Service for Prometheus AMP 模拟实现Workspace 与 Rule Groups Nmoto 中的 Amazon ECS 模拟实现API 覆盖清单、资源放置原理与测试实战指南moto 中的 Amazon ECS 模拟实现API 覆盖清单、资源放置原理与测试实战指南 本篇以 moto 仓库中的 ECS 服务实现文档 https://Mock测试moto 中 emr-containers 服务的模拟实现虚拟集群与作业运行的完整 Mock 指南moto 中 emr containers 服务的模拟实现虚拟集群与作业运行的完整 Mock 指南 在基于 AWS EMR on EKS即 emr contMock测试上一篇CommandLineParser选项组详解互斥选项与分组管理的完整指南下一篇神经网络从零实现Machine Learning with PyTorch and Scikit-Learn深度学习基础创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考