Provider 测试

一个 provider 是两件事拧在一起：一次变换（format_data_v2）和一套传输（OAuth、webhook、定时拉取）。它们坏的方式不同，验的方式也就不同。变换有一套离线套件，把录制下来的载荷重放一遍、再拿输出跟存档快照做 diff。传输只能对着一个运行中的服务来验。

注意

先运行 gate 测试。它不需要服务、不需要数据库、也不需要联网，所以它是你每改一次都能运行的那条回路：在那里发现的映射 bug 是五秒钟的修改，而不是 webhook 日志里的一道谜题。

两条验证回路

录制的载荷 一个 JSON 夹具 gate_tests/fixtures/

format_data_v2 纯函数 —— 不做 I/O 规则 · 取值 · 快照

线上路由 连接、回调、webhook、拉取 /api/v1/pulse/*

入库记录 按用户读回来 /api/v1/manage/pulse/*

离线回路覆盖纯变换；线上回路覆盖它周围的一切。

Gate 测试套件

mirobody/pulse/gate_tests/ 是给 format_data() 和 format_data_v2() 用的验收套件。它动态导入你的 provider 类，把它的数据库和配置依赖 mock 掉，喂进一条录制好的载荷，再检查得到的 StandardPulseData。

mirobody/pulse/gate_tests/

format_tester.py 测试框架本身

test_format_data.py pytest 入口（参数化）

conftest.py 提供 --update-snapshots 选项

pytest.ini

fixtures/ 每个来源一个目录

apple_health/ 3 份 fixture

theta_garmin/ 8 份 fixture

theta_oura/ 5 份 fixture

theta_whoop/ 5 份 fixture

先装一次 test extra，然后在仓库根目录运行：

pip install -e ".[test]"

python -m pytest mirobody/pulse/gate_tests/test_format_data.py -v

每个夹具变成一个以它的 test_id 命名的参数化用例，所以一次失败直接指到某个文件。套件自带的 pytest.ini 设了 asyncio_mode = auto，这就是那些异步用例不用额外加参数就能运行的原因。

夹具结构

一个夹具就是一个 JSON 文件。三个键是必需的：test_id、provider_class、input，缺任何一个的文件只会被记一条 warning 然后静默跳过，所以一个拼写错误的表现是某个用例悄悄不存在了，而不是一次失败。

mirobody/pulse/gate_tests/fixtures/theta_whoop/body_measurements.json

{
  "test_id": "theta_whoop_body_measurements_001",
  "description": "Whoop body_measurements data",
  "provider_class": "mirobody.pulse.providers.mirobody_whoop.provider_whoop.WhoopProvider",
  "platform": "theta",
  "mock_context": {},
  "patch_targets": [
    "mirobody.pulse.providers.platform.base.ProviderDatabaseService",
    "mirobody.pulse.providers.platform.base.PlatformUserService",
    { "target": "mirobody.utils.config.safe_read_cfg", "return_value": "" }
  ],
  "context": {
    "theta_user_id": "test_user_005",
    "user_timezone": "Asia/Shanghai",
    "msg_id": "test_msg_0009"
  },
  "input": {
    "data": [
      { "height_meter": 1.7, "max_heart_rate": 182, "weight_kilogram": 70.0 }
    ],
    "msg_id": "test_msg_0009",
    "user_id": "test_user_005",
    "data_type": "body_measurements",
    "timestamp": 1768574416892
  },
  "expected": {
    "success": true,
    "health_data_count": 3,
    "required_indicators": ["bodyMasss", "heights", "maxHeartRateProfile"],
    "value_checks": [
      { "index": 0, "field": "type", "expected": "heights" },
      { "index": 0, "field": "value", "expected": 1.7 },
      { "index": 0, "field": "unit", "expected": "m" }
    ],
    "snapshot": { }
  }
}

每个键的作用（上面那份录好的 snapshot 被省略了：它就是整个输出的镜像）：

键	作用
provider_class	点分路径，所以它必须能从 pytest 运行的位置被导入到。随仓库发布的每个夹具用的都是 mirobody.pulse.providers.… 下的包内路径。类是被直接实例化的：provider 加载器不参与。
context	有它 → 运行器构造一个 FormatDataContext 并调 format_data_v2。没有 → 调旧的 format_data(input)。
input	载荷，就按你的数据源实际发出来的样子。
patch_targets	字符串把那个名字 patch 成 MagicMock；对象 {target, return_value} 把它 patch 成返回该值。这就是 __init__ 期间数据库和 safe_read_cfg 被中和掉的方式。
init_kwargs	构造函数参数；字面量 "__mock__" 会变成一个 MagicMock()。Apple Health 的夹具用的是 {"platform": "__mock__"}。
mock_context	方法名 → 返回值，以 AsyncMock 的形式标注在实例上。任何你不想真运行的协程都用它。
expected	断言（见下节）。

因为打桩是完全数据驱动的，运行器里没有任何按 provider 分的分支：只要你列清楚要把构造依赖里的哪几个换掉，你的 provider 就立刻可测。

用例的判定方式

expected 驱动三个互相独立的层，每一层在自己的键缺席时被跳过，所以一个夹具可以想多松就多松、想多严就多严。

规则断言

health_data_count 必须跟 len(healthData) 完全相等，required_indicators 里的每个名字必须作为某条记录的 type 出现。这一层抓的是「映射不再吐 HRV 了」。

取值检查

value_checks 是一串 {index, field, expected}，拿去跟 healthData[index][field] 比。这些要手算出来：它是唯一能证明一次单位换算是对的、而不仅仅是稳定的的那一层。

快照比对

snapshot 会跟整个输出做递归 diff，最多报出 20 处差异及其路径。每次运行都会变的字段先从两边被剥掉：metaInfo 和 processingInfo 里的 requestId、timestamp、start_time、end_time、processing_duration_ms、success_rate，以及每条 healthData 记录里的 timestamp。

把 "success": false 一设，整件事就反过来：format_data 抛异常，或者返回了一个空的 healthData，用例才算通过。这就是「一条畸形载荷应当被拒、而不是被半吞下去」的断言写法。

新增一个用例

抓一条真实载荷

拿一条你数据源真实的响应或 webhook 请求体。如果它已经入库了，下面的 check_format 会把它打印回来给你。把它裁到还能触发你关心的那条分支的最小样本，并把任何个人信息去掉。

先写夹具、不写快照

建 mirobody/pulse/gate_tests/fixtures/<slug>/<data_type>.json，写上 test_id、provider_class、context、patch_targets、input，并把 expected.snapshot 设成 null：

mirobody/pulse/gate_tests/fixtures/theta_acme/sleep.json

{
  "test_id": "theta_acme_sleep_001",
  "description": "Acme sleep payload",
  "provider_class": "mirobody.pulse.providers.mirobody_acme.provider_acme.AcmeProvider",
  "platform": "theta",
  "mock_context": {},
  "patch_targets": [
    "mirobody.pulse.providers.platform.base.ProviderDatabaseService",
    "mirobody.pulse.providers.platform.base.PlatformUserService",
    { "target": "mirobody.utils.config.safe_read_cfg", "return_value": "" }
  ],
  "context": {
    "theta_user_id": "test_user_acme_001",
    "user_timezone": "Asia/Shanghai",
    "msg_id": "test_msg_acme_sleep"
  },
  "input": { "data_type": "sleep", "data": [] },
  "expected": { "success": true, "snapshot": null }
}

录一份快照

带上 conftest.py 加的那个参数运行一次。运行器会调你的 provider、把输出归一化，再写回夹具文件：

python -m pytest mirobody/pulse/gate_tests/test_format_data.py --update-snapshots -v

信它之前先把快照读一遍

这一步是最容易被跳过的。--update-snapshots 记录的是代码当时的实际行为，其中的 bug 也会一起被记录下来。打开 diff，对着数据源的文档核一遍指标名、单位和时间戳。

收紧，然后干净地重运行

补上 health_data_count、required_indicators，以及若干手工计算的 value_checks，然后不带该参数运行一次，确认全部通过：

python -m pytest mirobody/pulse/gate_tests/test_format_data.py -v

提示

当你有意改动 format_data_v2 时，用 --update-snapshots 重录一次，并把夹具的 diff 跟代码放进同一个 commit。那个 diff 于是记录了这次改动对真实载荷做了什么，可以作为映射改动的一次轻量评审。

对运行中的服务验证

传输是另一半。把服务起在 18080，给一个演示账号取一个令牌；列在 EMAIL_PREDEFINE_CODES 里的地址会被校验器直接接受、不会真发信，所以你可以跳过 /email/login 直接调 /email/verify：

BASE=http://localhost:18080

# 1. provider list — no token needed, and the fastest check that your class loaded
curl -s "$BASE/api/v1/pulse/providers"

# 2. get a JWT for a predefined demo account
curl -s -X POST "$BASE/email/verify" \
  -H 'Content-Type: application/json' \
  -d '{"email": "caregiver@mirobody.ai", "code": "111111"}'
# → { "success": true, "code": 0, "data": { "access_token": "...", ... } }

# 3. link — PASSWORD / CUSTOMIZED providers connect in this one call
curl -s -X POST "$BASE/api/v1/pulse/user/providers/link" \
  -H "Authorization: Bearer $JWT" -H 'Content-Type: application/json' \
  -d '{"provider_slug": "theta_pgsql", "platform": "theta",
       "auth_type": "customized",
       "connect_info": {"host": "localhost", "port": "18082",
                        "database": "mirobody", "username": "...", "password": "..."}}'

# 4. link — an OAuth provider answers with a URL to open in a browser
curl -s -X POST "$BASE/api/v1/pulse/user/providers/link" \
  -H "Authorization: Bearer $JWT" -H 'Content-Type: application/json' \
  -d '{"provider_slug": "theta_whoop", "platform": "theta", "auth_type": "oauth2"}'
# → { "data": { "link_web_url": "https://..." } }

# 5. webhook — replay a captured delivery through the real ingest path
curl -s -X POST "$BASE/api/v1/pulse/providers/theta_acme/webhook" \
  -H 'Content-Type: application/json' -H 'Svix-Id: replay-001' \
  --data-binary @captured_payload.json

要断言行为，别只断言 200：一个错的凭据应该以 code: 400 带着你自己那句 ValueError 消息回来；同一条 webhook 用同一个 Svix-Id 发两次，第二次应该被 is_data_already_processed 认出来；而对一个加载器从未注册过的 slug 调 link，回的应该是 Provider … not found in theta platform 而不是一次崩溃。

警告

多数数据源不接受 http://localhost:18080 这样的裸 HTTP 回调地址，所以一次真实的授权流程没法对着一个裸的本地服务走完。在它前面终止 TLS（用一个带受信证书的本地反向代理，或者一条隧道），然后把那个确切的 HTTPS URL 登记成 <SLUG>_REDIRECT_URL。隧道同时还给了你一个能接真实 webhook 投递的入站地址。

重放一条已入库的负载

管理路由是最接近 provider 调试器的工具，因为它把真正入库的原始负载读回来、再拿你的格式化器重运行一遍。它们由一个共享密钥而不是 JWT 把守：verify_manage_key 读的是一个 sk 查询参数，并拿它跟 backend_server_sk 这个配置值比。这个键不在随仓库发布的 config.yaml 里，所以先在你的 config.<ENV>.yaml 里以 BACKEND_SERVER_SK 的形式加上它；没有它，每个管理路由都回 500。

SK=your-management-key

# what has arrived for one provider (paginated, from its own storage table)
curl -s "$BASE/api/v1/manage/pulse/providers/webhooks?provider=theta_acme&page=1&page_size=20&sk=$SK"

# re-run format_data_v2 over stored record #123 and see both sides
curl -s "$BASE/api/v1/manage/pulse/providers/check_format?id=123&provider=theta_acme&sk=$SK"

# force a scheduled pull now, ignoring the interval and the distributed lock
curl -s -X POST "$BASE/api/v1/manage/theta/pull/trigger?sk=$SK" \
  -H 'Content-Type: application/json' \
  -d '{"provider_slug": "theta_acme", "force": true}'

# and the scheduler's view of every registered task
curl -s "$BASE/api/v1/manage/theta/pull/status?sk=$SK"

# finally, read the normalised records back (max 7 days per call)
curl -s "$BASE/api/v1/manage/pulse/user-health-data?user_id=505&start_date=2026-08-01&end_date=2026-08-07&sk=$SK"

check_format 是最该先伸手去拿的那个。它把 original_data 和 formatted_data 并排返回，再加上解析出来的 theta_user_id、external_user_id 和 msg_id，而当你的格式化器抛异常时，它报的是 success: false 加异常消息，不会把它吞掉。一条能在那里复现出 bug 的载荷，同时也就是一条可以直接存成夹具的载荷。

覆盖范围清单

映射本身

你的 provider 处理的每一个 data_type 一个夹具，每个都带上 health_data_count、required_indicators，以及针对任何被换算过的量手算出来的 value_checks。单位换算和毫秒级时间戳就是 bug 住的地方。

畸形与空输入

一条空载荷、一条缺 data_type 的、一条被数据源标为未评分的记录。契约是这些应当产出一个空的 healthData 而不是抛异常，而一个带 "success": false 的夹具把这个行为钉死。

工厂方法

在凭据键没设的情况下，create_provider 必须返回 None，并且 slug 必须不在 GET /api/v1/pulse/providers 里。设上之后必须在。这是一个两行的检查，却挡掉了一整类「在我机器上是好的」。

令牌生命周期

对 OAuth 数据源：走通一次真实授权，然后确认存下来的 expires_at，再靠过期后拉一次来强制触发刷新。get_valid_access_token 在剩余不足五分钟时刷新、在没存 refresh token 时返回 None：第二种情况必须表现为「需要重新连接」，而不是一次空拉取。

幂等性

把同一条载荷投两次。第二次应该被 is_data_already_processed 跳过，而从 user-health-data 读回来的记录数不能翻倍。

下一步

开发一个 Provider

你正在测的那个类

→

开发环境搭建

虚拟环境、容器，以及更外层的测试套件

→

贡献指南

一个 PR 在评审前需要什么

→

Provider 概览

已经随仓库发布的那些数据源

→