Provider 测试

概览

充分测试能确保你的 Provider 集成可靠、安全，并能正确处理各种边界情况。

测试策略

Unit Tests

隔离测试单个 method

Integration Tests

端到端测试 OAuth flows

Manual Testing

在真实环境中验证

Unit Testing

测试单个 Provider methods：

tests/test_yourprovider.py

import pytest
from connect.theta.mirobody_yourprovider.provider_yourprovider import ThetaYourProvider

@pytest.mark.asyncio
async def test_format_data():
    """Test data transformation"""
    provider = ThetaYourProvider.factory()
    
    # Mock raw data from vendor
    raw_data = {
        "steps": 10000,
        "date": "2024-01-15",
        "calories": 450
    }
    
    # Transform data
    formatted = provider.format_data(raw_data)
    
    # Verify transformation
    assert len(formatted) == 2
    assert formatted[0].indicator == StandardIndicator.DAILY_STEPS
    assert formatted[0].value == "10000"

@pytest.mark.asyncio
async def test_deduplication():
    """Test duplicate detection"""
    provider = ThetaYourProvider.factory()
    
    # Check if data is already processed
    is_duplicate = await provider.is_data_already_processed(
        user_id="test_user",
        msg_id="unique_record_123"
    )
    
    assert is_duplicate == False

Integration Testing

测试完整 OAuth flows：

@pytest.mark.asyncio
async def test_oauth_flow():
    """Test complete OAuth flow"""
    provider = ThetaYourProvider.factory()
    user_id = "test_user_oauth"
    
    # Step 1: Initiate link
    link_result = await provider.link(user_id)
    assert "link_web_url" in link_result
    assert "state" in link_result or "oauth_token" in link_result
    
    # Step 2: Simulate callback
    # (Use mock OAuth response)
    callback_result = await provider.callback(
        user_id=user_id,
        code="mock_auth_code",
        state=link_result.get("state")
    )
    
    assert callback_result["success"] == True
    
    # Step 3: Verify tokens stored
    tokens = await provider._get_tokens(user_id)
    assert tokens is not None

@pytest.mark.asyncio
async def test_data_pull():
    """Test data fetching from API"""
    provider = ThetaYourProvider.factory()
    user_id = "test_user_data"
    
    # Pull data
    raw_data = await provider.pull_from_vendor_api(user_id)
    
    # Verify data structure
    assert isinstance(raw_data, list)
    assert len(raw_data) > 0
    
    # Verify required fields
    for record in raw_data:
        assert "date" in record or "timestamp" in record

本地 HTTPS（用于 OAuth 开发）

Health data providers like WHOOP require HTTPS callback URLs for OAuth authentication due to security requirements.

本地开发时，服务通常运行在 http://localhost，而这些 Providers 会拒绝非 HTTPS 的 callback URL。要在本地测试 OAuth flows，你需要配置一个本地 HTTPS reverse proxy：负责终止 SSL/TLS，并将请求转发到本地开发服务。一般做法是：修改系统 hosts 文件，把生产 callback 域名（例如 data-gray.thetahealth.ai）指向 127.0.0.1，然后在本地 443 端口运行 reverse proxy 来处理 HTTPS，并代理到本地 HTTP 服务。实现方式很多，例如带反向代理能力的 Web server（nginx、Apache）、专用 HTTPS proxy 工具，或 ngrok 这类云端隧道服务。 The general approach:

安装本地证书工具（如 mkcert）为域名生成可信的 SSL 证书
配置 proxy 监听 443 端口并使用该证书，然后转发到本地开发端口（例如 localhost:18080）
更新 hosts 文件，让 callback 域名解析到 localhost

这样你就能在开发环境中使用与生产环境完全一致的 callback URL，从而无需每次改代码都部署一次，也能顺畅测试 OAuth 集成。

手动测试

1. Test OAuth Link

# Get authorization URL
curl "http://localhost:18080/api/v1/pulse/theta/theta_yourprovider/link?user_id=manual_test_user"

访问返回的 URL 并完成授权。

2. Test Callback

授权后，验证 callback 是否成功：

# Check provider status
curl "http://localhost:18080/api/v1/pulse/providers?user_id=manual_test_user"

期望：Provider 显示 "status": "connected"

3. Test Data Pull

# Trigger data sync
curl -X POST "http://localhost:18080/api/v1/pulse/theta/theta_yourprovider/sync?user_id=manual_test_user"

4. Verify Data

# Check database for new records
psql -d your_db -c "SELECT COUNT(*) FROM theta_ai.health_data_yourprovider WHERE theta_user_id = 'manual_test_user';"

# Check standardized data
psql -d your_db -c "SELECT indicator, COUNT(*) FROM theta_ai.th_series_data WHERE user_id = 'manual_test_user' GROUP BY indicator;"

测试场景

Happy Path

测试正常流程：

Fresh OAuth connection
Data fetch with valid tokens
Proper data transformation
Successful database storage

Error Handling

测试错误场景：

Invalid OAuth credentials
Expired tokens
API rate limiting
Network failures
Malformed API responses
Missing required fields

Edge Cases

测试边界情况：

Empty data responses
Very large datasets
Duplicate records
Time zone conversions
Null/missing values

Mock 外部 APIs

在 unit tests 中使用 mocks：

from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
@patch('aiohttp.ClientSession.post')
async def test_token_exchange(mock_post):
    """Test token exchange with mocked API"""
    # Mock API response
    mock_response = AsyncMock()
    mock_response.json.return_value = {
        "access_token": "mock_token_123",
        "refresh_token": "mock_refresh_456",
        "expires_in": 3600
    }
    mock_post.return_value.__aenter__.return_value = mock_response
    
    provider = ThetaYourProvider.factory()
    
    # Test callback with mocked response
    result = await provider.callback(
        user_id="test_user",
        code="mock_code",
        state="mock_state"
    )
    
    assert result["success"] == True

Test Data

Create realistic test data:

# Mock vendor API response
MOCK_GARMIN_DAILY_DATA = {
    "summaries": [
        {
            "calendarDate": "2024-01-15",
            "steps": 10543,
            "distanceInMeters": 8234,
            "activeKilocalories": 450,
            "bmrKilocalories": 1650,
            "floorsClimbed": 12,
            "activeTimeInSeconds": 5400
        }
    ]
}

MOCK_WHOOP_SLEEP_DATA = {
    "records": [
        {
            "id": "sleep_123",
            "score": {
                "total": 85,
                "efficiency": 92
            },
            "duration_millis": 27000000,
            "stages": {
                "deep_millis": 7200000,
                "light_millis": 14400000,
                "rem_millis": 5400000
            }
        }
    ]
}

Continuous Integration

Add provider tests to CI pipeline:

.github/workflows/test-providers.yml

name: Test Providers

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.12'
      
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-asyncio
      
      - name: Run provider tests
        run: |
          pytest tests/test_*_provider.py -v

Next Steps

Provider Integration

Build a custom provider

OAuth Implementation

Implement OAuth flows

Testing Guide

General testing practices

Examples

See real provider code

For complete testing examples and patterns, see the Provider Integration Guide in the repository.

开发者指南

Providers 开发

部署

概览

测试策略

Unit Tests

Integration Tests

Manual Testing

Unit Testing

Integration Testing

本地 HTTPS（用于 OAuth 开发）

手动测试

1. Test OAuth Link

2. Test Callback

3. Test Data Pull

4. Verify Data

测试场景

Mock 外部 APIs

Test Data

Continuous Integration

Next Steps

Provider Integration

OAuth Implementation

Testing Guide

Examples

开发者指南

Providers 开发

部署

​概览

​测试策略

Unit Tests

Integration Tests

Manual Testing

​Unit Testing

​Integration Testing

​本地 HTTPS（用于 OAuth 开发）

​手动测试

​1. Test OAuth Link

​2. Test Callback

​3. Test Data Pull

​4. Verify Data

​测试场景

​Mock 外部 APIs

​Test Data

​Continuous Integration

​Next Steps

Provider Integration

OAuth Implementation

Testing Guide

Examples

概览

测试策略

Unit Testing

Integration Testing

本地 HTTPS（用于 OAuth 开发）

手动测试

1. Test OAuth Link

2. Test Callback

3. Test Data Pull

4. Verify Data

测试场景

Mock 外部 APIs

Test Data

Continuous Integration

Next Steps