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 is a pure function you can pytest with recorded payloads — no server, no database, no network. The transport can only be checked against a running server.

Note

Start with the transformation tests. 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 dict in your test file trimmed from a real delivery

format_data_v2 pure — no I/O names · values · units

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

Stored records read back per user /api/v1/health-indicators · /api/data

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

Test the transformation with pytest

Tests live beside the code they cover, and bare pytest collects every mirobody/**/test_*.py — so a provider’s tests go into its own package directory:

pip install -e '.[test]'

pytest mirobody/pulse/providers/mirobody_acme -v

Write the module against the payloads your source actually sends. Take a real response or webhook body, trim it to the smallest sample that still exercises the branch you care about, and strip anything personal:

mirobody/pulse/providers/mirobody_acme/test_provider_acme.py

import pytest

from mirobody.pulse.providers.mirobody_acme.provider_acme import AcmeProvider


@pytest.fixture
def provider():
    return AcmeProvider()


@pytest.mark.asyncio
async def test_empty_payload_yields_no_records(provider):
    result = await provider.format_data({"user_id": "u1", "data_type": "sleep", "data": []})
    assert result.healthData == []


@pytest.mark.asyncio
async def test_sleep_payload_maps_to_registered_indicators(provider):
    raw = {
        "user_id": "u1",
        "data_type": "sleep",
        "data": [{"start": "2026-08-01T22:00:00Z", "score": {"sleep_duration": 28800}}],
    }
    result = await provider.format_data(raw)
    types = {record.type for record in result.healthData}
    assert "sleepDurations" in types
    # Hand-calculate the converted value: this is the only assertion that
    # proves a unit conversion is right rather than merely stable.
    assert result.healthData[0].value == 8.0

Four assertions carry most of the value:

Record count

An exact len(result.healthData) per payload catches “the mapping stopped emitting HRV” the moment it happens.

Registered indicator names

Every produced type must be a registered indicator name — see Health Indicators. A typo here stores data nothing can query.

Hand-calculated values

Compute the expected converted value yourself. Unit conversions and epoch-millisecond timestamps are where the bugs live.

Malformed input

An empty payload, a missing data_type, a record the source marks unscored — the contract is an empty healthData, not an exception.

Tip

When you deliberately change the mapping, update the expected values in the same commit as the code. The test 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 18060 and mint a token for a predefined 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:18060

# 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 the 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": "18062",
                        "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:18060 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.

Read back what landed

The write path’s output is readable through the same routes any client uses, as the linked user’s own JWT:

# grouped by indicator, the web client's shape
curl -s "$BASE/api/v1/health-indicators" -H "Authorization: Bearer $JWT"

# row-level and newest-first, the hosted platform's shape — count rows here
# to prove an idempotency replay did not double anything
curl -s "$BASE/api/data?indicator=sleep&limit=50" -H "Authorization: Bearer $JWT"

What the scheduler did, what each webhook stored, and any format_data failure land in the logs — docker compose logs -f mirobody mirobody_worker under Compose. A payload that reproduces a bug there is also a payload ready to be trimmed into your test module.

Coverage checklist

The mapping

One test per data_type your provider handles, each with an exact record count, the required indicator names, and hand-calculated values 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 test 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 GET /api/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

→