- Mock
- 测试
【免费下载链接】moto
A library that allows you to easily mock out tests based on AWS infrastructure.
Amazon Managed Prometheus(AMP,AWS 的托管 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 = [ r"https?://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_definition(Alert 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(alias="test", clientToken="mytoken")的写法也证实了该参数可被传入但被忽略。
Workspace 生命周期管理:创建、查询、更新与删除
Workspace 是 AMP 的核心资源,对应源码中的 Workspace 模型。模拟实现会在创建时生成以下关键字段:
workspace_id:形如ws-{uuid4}的随机 ID(来自moto.moto_api._internal.mock_random.uuid4);arn:形如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_name="ap-southeast-1") resp = client.create_workspace(alias="my-prom", tags={"env": "dev"}) workspace_id = resp["workspaceId"] assert resp["status"] == {"statusCode": "ACTIVE"} assert resp["alias"] == "my-prom" detail = client.describe_workspace(workspaceId=workspace_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 == alias);maxResults/nextToken:分页参数,默认每页 100 条。
client.update_workspace_alias(alias="renamed", workspaceId=workspace_id) # alias 过滤 only_renamed = client.list_workspaces(alias="renamed")["workspaces"] assert len(only_renamed) == 1删除与异常行为
delete_workspace直接从内部 dict 中弹出该 workspace(pop(workspace_id, None)),随后对该 ID 执行describe_workspace会抛出ResourceNotFoundException。测试 test_amp_workspaces.py 验证了这一点:
client.delete_workspace(workspaceId=workspace_id) with pytest.raises(ClientError) as exc: client.describe_workspace(workspaceId=workspace_id) assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException" assert exc.value.response["Error"]["Message"] == "Workspace not found"该异常由 moto/amp/exceptions.py 中的WorkspaceNotFound抛出,HTTP 状态码为 404,resourceType为AWS::APS::Workspace。
Rule Groups Namespace:Prometheus 规则组的命名空间管理
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_name="eu-west-1") workspace_id = client.create_workspace()["workspaceId"] client.create_rule_groups_namespace( data=b"groups:\n - name: example\n rules: []", name="my-rules", workspaceId=workspace_id, ) ns = client.describe_rule_groups_namespace( name="my-rules", workspaceId=workspace_id )["ruleGroupsNamespace"] assert ns["data"] == b"groups:\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( name="my-rules", workspaceId=workspace_id, data=b"groups: []" ) ns = client.describe_rule_groups_namespace( name="my-rules", workspaceId=workspace_id )["ruleGroupsNamespace"] assert ns["data"] == b"groups: []" client.delete_rule_groups_namespace(name="my-rules", workspaceId=workspace_id)删除后查询同一命名空间会抛出RuleGroupNamespaceNotFound(404,resourceType为AWS::APS::RuleGroupNamespace,见 exceptions.py)。
列表与名称前缀过滤
list_rule_groups_namespaces的name参数是前缀匹配而非精确匹配——源码中通过ns_name.startswith(name)实现(models.py)。测试 test_amp_rulegroupnamespaces.py 演示了典型行为:存在ns0~ns14共 15 个命名空间时,name="ns1"会匹配到ns1、ns10~ns14共 6 个。
Logging Configuration:工作区日志输出配置
日志配置用于将 AMP 的查询与指标日志发送到 CloudWatch Logs,对应 models.py 中的四个方法。模拟实现的数据结构包含logGroupArn、createdAt、status(固定ACTIVE)与workspace字段。
创建与查询
@mock_aws def test_logging(): client = boto3.client("amp", region_name="us-east-2") workspace_id = client.create_workspace()["workspaceId"] resp = client.create_logging_configuration( workspaceId=workspace_id, logGroupArn="arn:aws:logs:us-east-2:123456789012:log-group:/amp/logs" ) assert resp["status"] == {"statusCode": "ACTIVE"} cfg = client.describe_logging_configuration(workspaceId=workspace_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( workspaceId=workspace_id, logGroupArn="arn:aws:logs:us-east-2:123456789012:log-group:/amp/logs-v2" ) cfg = client.describe_logging_configuration(workspaceId=workspace_id)["loggingConfiguration"] assert "modifiedAt" in cfg assert cfg["logGroupArn"].endswith("/amp/logs-v2") client.delete_logging_configuration(workspaceId=workspace_id) assert client.describe_logging_configuration(workspaceId=workspace_id)["loggingConfiguration"] == {}update_logging_configuration会覆盖logGroupArn并追加modifiedAt时间戳;delete_logging_configuration将内部配置置为None,使后续查询重新回到空字典状态。
标签管理:tag_resource / untag_resource / list_tags_for_resource
AMP 模拟的标签能力统一由TaggingService提供(models.py),可作用于 Workspace 与 RuleGroupNamespace 两类资源。标签以 ARN 为键存储,并通过tag_fn回调挂接到每个模型上,因此describe返回的响应中会直接携带tags字段。
@mock_aws def test_tags(): client = boto3.client("amp", region_name="us-east-2") arn = client.create_workspace(alias="test", tags={"env": "prod"})["arn"] client.tag_resource(resourceArn=arn, tags={"team": "platform"}) assert client.list_tags_for_resource(resourceArn=arn)["tags"] == { "env": "prod", "team": "platform" } client.untag_resource(resourceArn=arn, tagKeys=["env"]) assert client.list_tags_for_resource(resourceArn=arn)["tags"] == {"team": "platform"}由于底层是统一的TaggingService,Workspace 与 RuleGroupNamespace 的标签行为完全一致(两组测试文件分别验证了各自的场景)。untag_resource通过untag_resource_using_names按标签名批量删除,可一次传入多个 tagKey。
分页行为:默认页大小与 nextToken
AMP 的两个列表接口(list_workspaces、list_rule_groups_namespaces)都支持 AWS 风格的分页,分页模型定义在 moto/amp/utils.py:
PAGINATION_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 条并附带nextToken; maxResults可缩小或扩大:测试中展示了maxResults=15搭配nextToken逐页取数,以及maxResults=1000一次性取完的场景(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.py):
PrometheusServiceResponse继承BaseResponse,根据请求中携带的 API action 名调用对应的处理函数(如create_workspace、describe_workspace等),处理函数负责从请求 body/路径中解析参数、调用 backend 方法、并把结果序列化为与真实 AWS 一致的 JSON 结构; - 后端模型(moto/amp/models.py):
PrometheusServiceBackend持有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按 account+region 建实例,同一测试中跨 region 创建的 workspace 彼此独立。
运行 amp 相关测试
仓库在 tests/test_amp/ 下提供了三个测试文件,分别覆盖 workspace(test_amp_workspaces.py)、rule group namespaces(test_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
- 测试
【免费下载链接】moto
A library that allows you to easily mock out tests based on AWS infrastructure.
相关推荐
Floci 的 Amazon Managed Service for Prometheus (AMP) 模拟实现:Workspace 与 Rule Groups Namespace 全生命周期指南
Floci 的 Amazon Managed Service for Prometheus AMP 模拟实现:Workspace 与 Rule Groups N
moto 中的 Amazon ECS 模拟实现:API 覆盖清单、资源放置原理与测试实战指南
moto 中的 Amazon ECS 模拟实现:API 覆盖清单、资源放置原理与测试实战指南 本篇以 moto 仓库中的 ECS 服务实现文档 https://
Mock测试moto 中 emr-containers 服务的模拟实现:虚拟集群与作业运行的完整 Mock 指南
moto 中 emr containers 服务的模拟实现:虚拟集群与作业运行的完整 Mock 指南 在基于 AWS EMR on EKS(即 emr cont
Mock测试
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考