Provider Testing

A provider is two things bolted together: a transformation (format_data_v2) and a transport (OAuth, webhooks, scheduled pulls). They fail differently, so they are verified differently. The transformation has an offline suite that replays recorded payloads and diffs the output against stored snapshots. The transport can only be checked against a running server.

Note

Start with the gate tests. They need no server, no database and no network, so they are the loop you can run on every edit — and a mapping bug found there is a five-second fix instead of a puzzle in a webhook log.

Two verification loops

Recorded payload a JSON fixture gate_tests/fixtures/

format_data_v2 pure — no I/O rules · values · snapshot

Live routes link, callback, webhook, pull /api/v1/pulse/*

Stored records read back per user /api/v1/manage/pulse/*

The offline loop covers the pure transformation; the live loop covers everything around it.

The gate test suite

mirobody/pulse/gate_tests/ is an acceptance suite for format_data() and format_data_v2(). It imports your provider class dynamically, mocks away its database and config dependencies, feeds it a recorded payload, and checks the resulting StandardPulseData.

mirobody/pulse/gate_tests/

format_tester.py the test framework

test_format_data.py the pytest entry point (parametrized)

conftest.py adds the --update-snapshots option

pytest.ini

fixtures/ one directory per source

apple_health/ 3 fixtures

theta_garmin/ 8 fixtures

theta_oura/ 5 fixtures

theta_whoop/ 5 fixtures

Install the test extra once, then run from the repository root:

pip install -e ".[test]"

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

Every fixture becomes one parametrized case named after its test_id, so a failure points straight at a file. The suite’s own pytest.ini sets asyncio_mode = auto, which is why the async cases run without extra flags.

Fixture anatomy

A fixture is one JSON file. Three keys are required — test_id, provider_class, input — and a file missing any of them is logged as a warning and silently skipped, so a typo shows up as a case that quietly stopped existing rather than as a failure.

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": { }
  }
}

What each key does (the recorded snapshot is elided above — it mirrors the entire output):

Key	Effect
provider_class	Dotted path, so it has to be importable from wherever pytest runs. Every shipped fixture uses a packaged path under mirobody.pulse.providers.…. The class is instantiated directly — the provider loader is not involved.
context	Present → the runner builds a FormatDataContext and calls format_data_v2. Absent → it calls legacy format_data(input).
input	The payload, exactly as your source would send it.
patch_targets	A string patches that name with MagicMock; an object {target, return_value} patches it to return that value. This is how the database and safe_read_cfg are neutralised during __init__.
init_kwargs	Constructor arguments; the literal "__mock__" becomes a MagicMock(). Apple Health’s fixtures use {"platform": "__mock__"}.
mock_context	Method name → return value, applied to the instance as an AsyncMock. Use it for any coroutine you don’t want to run.
expected	The assertions — see below.

Because the patching is fully data-driven, the runner contains no per-provider branching: your provider is testable the moment you list which of its constructor dependencies to stub.

Case validation

expected drives three independent layers, and each one is skipped when its key is absent — so a fixture can be as loose or as strict as you need.

Rule assertions

health_data_count must match len(healthData) exactly, and every name in required_indicators must appear as some record’s type. This is the layer that catches “the mapping stopped emitting HRV”.

Value checks

value_checks is a list of {index, field, expected}, compared against healthData[index][field]. Hand-calculate these: they are the only layer that proves a unit conversion is right rather than merely stable.

Snapshot comparison

snapshot is diffed against the whole output recursively, reporting up to 20 differences with their paths. Fields that change every run are stripped from both sides first: requestId, timestamp, start_time, end_time, processing_duration_ms, success_rate from metaInfo and processingInfo, and timestamp from each healthData record.

Setting "success": false inverts the whole thing: the case passes if format_data raises, or if it returns with an empty healthData. That is the shape for asserting that a malformed payload is refused rather than half-ingested.

Adding a case

Capture a real payload

Take an actual response or webhook body from your source. If it is already stored, check_format (below) will print it back for you. Trim it to the smallest sample that still exercises the branch you care about, and strip anything personal.

Write the fixture with no snapshot

Create mirobody/pulse/gate_tests/fixtures/<slug>/<data_type>.json with test_id, provider_class, context, patch_targets, input, and expected.snapshot set to 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 }
}

Record the snapshot

Run with the flag conftest.py adds. The runner calls your provider, normalises the output, and writes it back into the fixture file:

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

Read the snapshot before trusting it

This is the step people skip. --update-snapshots records whatever your code did, bugs included. Open the diff and check the indicator names, the units, and the timestamps against the source’s documentation.

Tighten it, then re-run clean

Fill in health_data_count, required_indicators, and a few hand-calculated value_checks, then run without the flag and confirm green:

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

Tip

When you deliberately change format_data_v2, re-record with --update-snapshots and commit the fixture diff in the same commit as the code. The diff then documents exactly what the change did to real payloads, providing a lightweight review of the mapping change.

Verifying against a running server

Transport is the other half. Start the server on 18080 and mint a token for a demo account — an address listed in EMAIL_PREDEFINE_CODES is accepted by the validator without any mail being sent, so you can skip /email/login and call /email/verify directly:

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": "exp1@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

Assert on behaviour, not just on 200s: a bad credential should come back as code: 400 with your own ValueError message; the same webhook sent twice with the same Svix-Id should be recognised by is_data_already_processed; and a link for a slug the loader never registered answers with Provider … not found in theta platform rather than a crash.

Caution

Most sources reject a plain http://localhost:18080 redirect URI, so a real consent flow can’t complete against a bare local server. Terminate TLS in front of it — a local reverse proxy with a trusted certificate, or a tunnel — and register that exact HTTPS URL as <SLUG>_REDIRECT_URL. A tunnel also gives you an inbound address for real webhook deliveries.

Replay a stored payload

The management routes are the closest thing to a debugger for a provider, because they read back what you actually stored and re-run your formatter over it. They are gated by a shared key rather than a JWT: verify_manage_key reads a sk query parameter and compares it with the backend_server_sk config value. That key is not in the shipped config.yaml, so add it to your config.<ENV>.yaml as BACKEND_SERVER_SK first — without it every management route answers 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 is the one to reach for first. It returns original_data and formatted_data side by side, plus the resolved theta_user_id, external_user_id and msg_id — and when your formatter raises, it reports success: false with the exception message instead of swallowing it. A payload that reproduces a bug there is also a payload ready to be saved as a fixture.

Coverage checklist

The mapping

One fixture per data_type your provider handles, each with health_data_count, required_indicators, and hand-calculated value_checks for anything that gets converted. Unit conversions and epoch-millisecond timestamps are where the bugs live.

Malformed and empty input

An empty payload, a missing data_type, a record the source marks unscored. The contract is that these produce an empty healthData rather than an exception, and a fixture with "success": false pins the behaviour down.

The factory

With the credential keys unset, create_provider must return None and the slug must be absent from GET /api/v1/pulse/providers. With them set, it must be present. This is a two-line check that catches a whole class of “works on my machine”.

Token lifecycle

For OAuth sources: complete a real consent flow, then confirm the stored expires_at and force a refresh by pulling after expiry. get_valid_access_token refreshes when under five minutes remain and returns None when no refresh token is stored — the second case must surface as “reconnect required”, not as an empty pull.

Idempotency

Deliver the same payload twice. The second one should be skipped by is_data_already_processed, and the record count read back from user-health-data must not double.

Next steps

Building a Provider

The class you are testing

→

Development Setup

The venv, the containers, and the wider test suite

→

Contributing

What a pull request needs before review

→

Provider Overview

The sources that already ship

→