Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载本文基于 Moto 仓库中 CloudWatch 服务实现文档 展开完整梳理当前版本已实现Implemented与未实现的 CloudWatch API 清单并结合 moto/cloudwatch/models.py、moto/cloudwatch/responses.py 等源码讲解指标数据写入/查询、告警生命周期、仪表盘、Insight 规则与标签等核心功能的模拟机制与验证边界帮助你在编写基于 CloudWatch 的集成测试时准确评估哪些 API 可以在本地安全地 mock 运行。一、CloudWatch 服务在 Moto 中的定位与请求路由Moto 通过 boto3 请求拦截与本地 HTTP 服务两种方式为各 AWS 服务提供模拟。CloudWatch 服务的模拟模块位于 moto/cloudwatch/ 目录包含以下核心文件models.pyCloudWatchBackend状态容器与Alarm、Dashboard、InsightRule、Statistics等数据模型responses.pyCloudWatchResponse负责参数解析与响应序列化urls.py定义被拦截的 URL 模式utils.pyARN 构造工具告警、仪表盘、Insight 规则三类exceptions.pyInvalidParameterValue、ResourceNotFound、ValidationError等异常类metric_data_expression_parser.pyMetric Data 查询中数学表达式的求值逻辑。URL 匹配规则非常明确——从 urls.py 中可以看到url_bases [rhttps?://monitoring\.(.)\.amazonaws.com] url_paths {{0}/$: CloudWatchResponse.dispatch}即所有发往https://monitoring.region.amazonaws.com的 CloudWatch 请求都会被路由到CloudWatchResponse进行分发。CloudWatchResponse在 responses.py 中声明automated_parameter_parsing True由 Moto 核心自动完成请求参数到后端方法签名的映射。二、已实现与未实现的 API 完整清单以下是 实现覆盖文档 中列出的全部 CloudWatch API 及其当前实现状态这是评估你的测试用例能否被完整 mock 的第一依据。已实现的 APIImplementedAPI功能delete_alarms删除告警不存在的名字不会报错delete_dashboards删除一个或多个仪表盘delete_insight_rules删除 Insight 规则托管规则会返回失败项describe_alarms按名称前缀/动作前缀/状态等过滤告警describe_insight_rules列出 Insight 规则disable_alarm_actions禁用告警动作disable_insight_rules禁用 Insight 规则enable_alarm_actions启用告警动作enable_insight_rules启用 Insight 规则get_dashboard获取仪表盘get_metric_data按 MetricStat / 数学表达式 / Insights 表达式查询指标get_metric_statistics按周期聚合查询指标统计值list_dashboards按前缀列出仪表盘list_metrics列出已写入的指标支持分页list_tags_for_resource列出资源标签put_dashboard创建/覆盖仪表盘put_insight_rule创建 Insight 规则put_metric_alarm创建/更新指标告警put_metric_data写入指标数据单值/多值/统计值set_alarm_state手动设置告警状态tag_resource为告警/规则打标签untag_resource移除标签尚未实现的 APIUnimplemented文档中标记为未实现的操作包括associate_dataset_kms_key、delete_alarm_mute_rule、delete_anomaly_detector、delete_metric_stream、describe_alarm_contributors、describe_alarm_history、describe_alarms_for_metric、describe_anomaly_detectors、disassociate_dataset_kms_key、get_alarm_mute_rule、get_dataset、get_insight_rule_report、get_metric_stream、get_metric_widget_image、get_otel_enrichment、list_alarm_mute_rules、list_managed_insight_rules、list_metric_streams、put_alarm_mute_rule、put_anomaly_detector、put_composite_alarm、put_log_alarm、put_managed_insight_rules、put_metric_stream、start_metric_streams、start_otel_enrichment、stop_metric_streams、stop_otel_enrichment。注意describe_alarms_for_metric的响应方法其实已存在于 responses.py 的describe_alarms_for_metric中只是官方覆盖文档尚未将其标记为已完成使用时以实测行为为准。三、指标数据put_metric_data 的三种写入模式与参数校验put_metric_data是最常用来喂数据的 API。从 models.py 的put_metric_data实现看Moto 支持三种互斥的数据形态全部由MetricDatumBase派生模型承载单值Value映射为MetricDatum一个namespace name dimensions组合对应一条数据记录多值Values 可选Countsvalues[i]会被按counts[i]的权重重复展开默认权重为 1。测试 test_cloudwatch.py 中test_put_metric_data_values_and_counts验证了Values[1.0, 10.0], Counts[2, 4]写入后SampleCount 6.0、Sum 42.0聚合统计StatisticValues映射为MetricAggregatedDatum必须同时提供Sum、Maximum、Minimum、SampleCount四个字段缺一即抛InvalidParameterValueMissing required parameter in MetricData[N].StatisticValues。后端还内置了一组与 AWS 对齐的参数校验位于_validate_parameters_put_metric_data方法中单值或任意一个 Values 为 NaN 时抛InvalidParameterValueThe value NaN for parameter MetricData.member.N.Value is invalid.Value与Values同时提供 →InvalidParameterValuemutually exclusiveValues与Counts长度不一致 →InvalidParameterValuemust be of the same sizeValue与StatisticValues同时提供 →InvalidParameterCombination。这些错误码与错误文案与 exceptions.py 中定义的异常类型一一对应测试断言错误消息时可放心复用。一个完整的最小示例如下import boto3 from datetime import datetime, timedelta, timezone from moto import mock_aws mock_aws def test_write_and_query_metric(): cw boto3.client(cloudwatch, region_nameus-east-1) now datetime.now(tztimezone.utc) cw.put_metric_data( Namespacetester, MetricData[ { MetricName: latency, Timestamp: now, Value: 1.5, Unit: Seconds, Dimensions: [{Name: Environment, Value: prod}], } ], ) stats cw.get_metric_statistics( Namespacetester, MetricNamelatency, StartTimenow - timedelta(seconds60), EndTimenow timedelta(seconds60), Period60, Statistics[Average, Sum, SampleCount], ) assert len(stats[Datapoints]) 1四、指标查询get_metric_statistics 与 get_metric_data 的分桶逻辑get_metric_statistics按 Period 切分时间桶models.py 中get_metric_statistics的处理流程是将StartTime/EndTime的毫秒归零StartTime EndTime时抛InvalidParameterValue从全部数据self.metric_data self.aws_metric_data中按 namespace、metric_name、时间区间过滤再依次按Unit和Dimensions收窄用daterange辅助函数从第一条数据的时间戳开始、以period秒为步长切分时间桶每个桶由Statistics类计算Sum、Average、Minimum、Maximum、SampleCount。其中Statistics.get_statistics_for_type只接受这五种统计名区分大小写未请求的统计量返回None。Average的计算为sum / sample_count样本数为 0 时返回None。get_metric_data三类查询的执行顺序get_metric_data是更通用的查询入口responses.py 中要求每个 query 至少包含MetricStat或Expression之一否则抛出带 The parameter MetricDataQueries.member.1.MetricStat is required. 文案的ValidationError与 AWS 行为一致连空行都保留。后端按以下顺序处理三类查询MetricStat 查询按Period切分周期逐桶匹配 namespace/name可选维度排序相等匹配、可选 Unit 过滤计算指定StatScanBy支持TimestampAscending默认与TimestampDescending对结果反转数学表达式查询Expression不以SELECT开头由 metric_data_expression_parser.py 的parse_expression求值——当前实现是把表达式视为对已有查询结果的引用如m1 m2会引用m1的结果再叠加到results中Insights 表达式查询SELECT开头源码注释明确指出 Moto currently does not support Metrics Insights Queries实际行为是返回该周期内所有数据按Sum计算的粗粒度结果不应在测试中依赖其精确语义。StartTime EndTime抛 The parameter EndTime must be greater than StartTime.两者相等抛 The parameter StartTime must not equal parameter EndTime.测试文件 test_cloudwatch.py 中的test_get_metric_data_endtime_sooner_than_starttime、test_get_metric_data_starttime_endtime_equals等用例覆盖了这些边界。另外get_all_metrics会把self.metric_data与aws_metric_data合并返回——后者来自CloudWatchMetricProvider的子类集合即其他服务如 EC2、S3 等 provider注册的开箱即用指标也会出现在 CloudWatch 查询结果中这是 Moto 跨服务模拟联动的一部分。五、告警生命周期put_metric_alarm 到 set_alarm_state创建与校验put_metric_alarm在 models.py 中做了两项前置校验ExtendedStatistic不以p开头时抛InvalidParameterValueThe value ... for parameter ExtendedStatistic is not supported.EvaluateLowSampleCountPercentile非evaluate/ignore时抛ValidationError。Alarm模型创建时初始化状态为state_value OK、state_reason Unchecked: Initial alarm creationARN 由 utils.py 的make_arn_for_alarm生成为arn:{partition}:cloudwatch:{region}:{account_id}:alarm:{name}——分区号随区域自动适配test_cloudwatch_alarms.py 用eu-west-1/aws与cn-north-1/aws-cn两个参数化用例验证了这一点。响应层的 responses.py 中put_metric_alarm还会解析Metrics参数以构造MetricDataQuery列表并通过读取AlarmRule参数复用同一入口承载复合告警的数据结构尽管put_composite_alarm本身未列入已实现清单。过滤查询与状态操作describe_alarms的过滤优先级在 responses.py 中是显式的ActionPrefix→AlarmNamePrefix→AlarmNames→StateValue→ 全量后端对应get_alarms_by_action_prefix检查任一 alarm_action 是否以该前缀开头、get_alarms_by_alarm_name_prefix、get_alarms_by_alarm_names、get_alarms_by_state_value四个方法。响应按rule is None拆分回MetricAlarms与CompositeAlarms两个列表。set_alarm_state的校验链完整复刻了 AWS 语义StateReasonData提供但不是合法 JSON →InvalidFormat告警不存在 →ResourceNotFoundStateValue不在OK/ALARM/INSUFFICIENT_DATA→ValidationError错误文案包含 Member must satisfy enum value set 完整枚举提示。成功时Alarm.update_state会先把旧状态推入history列表类型标记为StateUpdate再更新state_reason、state_value与state_updated_timestamp。enable_alarm_actions/disable_alarm_actions只是翻转actions_enabled布尔值且不存在的告警名不会报错——这与delete_alarms对不存在名字静默成功的行为一致test_cloudwatch_alarms.py 中test_enable_disable_alarm_actions_without_error有对应断言。六、仪表盘put/get/list/delete 与 CloudFormation 集成Dashboard模型models.py实现CloudFormationModel核心字段为name、bodyJSON 字符串、last_modifiedARN 格式为arn:{partition}:cloudwatch::{account_id}:dashboard/{name}注意仪表盘 ARN 不含 region 段。行为要点put_dashboardresponses.py 会先json.loads(body)非法 JSON 抛DashboardInvalidInputErrorcode 为InvalidParameterInput成功则返回空的DashboardValidationMessages列表同名仪表盘会被直接覆盖name 即存储字典的 keylist_dashboards按DashboardNamePrefix前缀过滤get_dashboard不存在时抛ResourceNotFoundDashboard does not existdelete_dashboards至少需要 1 个名字否则抛InvalidParameterValueNeed at least 1 dashboard部分名字不存在时抛ResourceNotFound且消息会列出缺失的名字清单The specified dashboard does not exist. [a, b]。CloudFormation 侧支持AWS::CloudWatch::Dashboard资源类型create_from_cloudformation_json从Properties中取DashboardName缺省时用 UUID与DashboardBody调用put_dashboard更新走先删后建删除调用delete_dashboards。test_cloudwatch_cloudformation.py 验证了建栈后get_dashboard能拿到正确 ARN、删栈后再次获取抛出ResourceNotFound的完整链路。七、Insight 规则put / describe / delete / enable / disableInsightRule模型models.py保存definition、name、state、schema默认{Name: CloudWatchLogRule, Version: 1}与managed_rule标志ARN 为arn:{partition}:cloudwatch:{region}:{account_id}:insight-rule/{rule_name}。操作语义上delete_insight_rules、disable_insight_rules、enable_insight_rules三者结构相同对每个请求的名字若对应规则是managed_rule由put_insight_rule创建的规则恒为False则向Failures列表追加InvalidParameterValue失败项否则执行删除/置DISABLED/置ENABLED。describe_insight_rules在规则数超过MaxResults默认 500时做截断。put_insight_rule同时支持传入Tags标签会挂在规则 ARN 上。八、标签管理仅限告警与 Insight 规则tag_resource/untag_resource/list_tags_for_resource由 models.py 中的方法实现底层使用TaggingService以 ARN 为键存储。关键约束tag_resource会先校验 ARN 是否属于任一告警Currently, the only CloudWatch resources that can be tagged are alarms and Contributor Insights rules——与 AWS 官方限制一致不属于时抛ResourceNotFoundExceptionuntag_resource对未打标签的 ARN 同样抛ResourceNotFoundExceptionput_metric_alarm与put_insight_rule都接受Tags参数创建时即完成打标后端还重写了TaggableResourcesMixin.iter_tagged_resources向 Resource Groups Tagging API 暴露cloudwatch:alarm与cloudwatch:insight-rule两类带标签资源意味着 resourcegroupstaggingapi 服务的 get_resources/get_resources 也能遍历这些 CloudWatch 资源。九、测试验证参考与适用边界围绕本服务仓库提供了完整的回归测试集可按需定位具体行为tests/test_cloudwatch/test_cloudwatch.py指标写入单值/ValuesCounts/StatisticValues、参数互斥校验、get_metric_data的时间框/维度/Unit 过滤与分页tests/test_cloudwatch/test_cloudwatch_alarms.py告警创建含中国区 ARN 分区、删除、动作启停、按指标过滤tests/test_cloudwatch/test_cloudwatch_dashboards.py仪表盘 CRUD 与错误路径tests/test_cloudwatch/test_cloudwatch_insight_rules.pyInsight 规则各操作tests/test_cloudwatch/test_cloudwatch_tags.py标签接口tests/test_cloudwatch/test_cloudwatch_cloudformation.pyCFN 集成tests/test_cloudwatch/test_cloudwatch_expressions.py 与 tests/test_cloudwatch/test_cloudwatch_expression_parser.py数学表达式查询。适用边界提醒基于当前源码告警不会被自动求值——Moto 只保存告警配置并支持手动set_alarm_state不会像真实 CloudWatch 那样根据指标阈值自动翻转状态get_metric_data对SELECT开头的 Insights 查询仅有粗粒度兜底put_composite_alarm、Metric Streams、Anomaly Detectors 等 API 尚未纳入已实现清单。编写测试时建议先对照本文的 API 清单确认所用接口处于已实现一侧再参考上述测试文件复刻真实 boto3 调用方式即可得到可复制、可运行的 CloudWatch 模拟方案。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐Nightingale 只读 API 实战指南通过 Skill Gateway 安全查询告警、主机、仪表盘与指标数据Nightingale 只读 API 实战指南通过 Skill Gateway 安全查询告警、主机、仪表盘与指标数据 本文是 NightingaleN9E后端运维观测告警可观测性人工智能AI Agent基于 AWS SDK for .NET 的 Amazon CloudWatch 监控实战指南指标、仪表盘、告警与异常检测基于 AWS SDK for .NET 的 Amazon CloudWatch 监控实战指南指标、仪表盘、告警与异常检测 本篇指南以仓库 dotnetv3/C示例工程教程后端Satellizer监控告警Prometheus指标与Grafana仪表盘Satellizer监控告警Prometheus指标与Grafana仪表盘 Satellizer作为Token based AngularJS Authent前端应用安全上一篇MERN Starter 项目教程下一篇列表与哈希表JavaScript中的基础数据结构实现创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考