跳转到主要内容

​
概览

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

​
测试策略

单元测试

隔离测试单个方法

集成测试

端到端测试 OAuth 流程

手动测试

在真实环境中验证

​
单元测试

测试单个 Provider 方法：
tests/test_yourprovider.py
import pytest
from connect.theta.mirobody_yourprovider.provider_yourprovider import ThetaYourProvider

@pytest.mark.asyncio
async def test_format_data():
    """测试数据转换"""
    provider = ThetaYourProvider.factory()
    
    # Mock 来自厂商的原始数据
    raw_data = {
        "steps": 10000,
        "date": "2024-01-15",
        "calories": 450
    }
    
    # 转换数据
    formatted = provider.format_data(raw_data)
    
    # 验证转换结果
    assert len(formatted) == 2
    assert formatted[0].indicator == StandardIndicator.DAILY_STEPS
    assert formatted[0].value == "10000"

@pytest.mark.asyncio
async def test_deduplication():
    """测试重复数据检测"""
    provider = ThetaYourProvider.factory()
    
    # 检查数据是否已处理
    is_duplicate = await provider.is_data_already_processed(
        user_id="test_user",
        msg_id="unique_record_123"
    )
    
    assert is_duplicate == False

​
集成测试

测试完整的 OAuth 流程： 
@pytest.mark.asyncio
async def test_oauth_flow():
    """测试完整的 OAuth 流程"""
    provider = ThetaYourProvider.factory()
    user_id = "test_user_oauth"
    
    # 第 1 步：发起绑定
    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
    
    # 第 2 步：模拟回调
    # (使用 Mock 的 OAuth 响应)
    callback_result = await provider.callback(
        user_id=user_id,
        code="mock_auth_code",
        state=link_result.get("state")
    )
    
    assert callback_result["success"] == True
    
    # 第 3 步：验证 Token 已存储
    tokens = await provider._get_tokens(user_id)
    assert tokens is not None

@pytest.mark.asyncio
async def test_data_pull():
    """测试从 API 获取数据"""
    provider = ThetaYourProvider.factory()
    user_id = "test_user_data"
    
    # 拉取数据
    raw_data = await provider.pull_from_vendor_api(user_id)
    
    # 验证数据结构
    assert isinstance(raw_data, list)
    assert len(raw_data) > 0
    
    # 验证必要字段
    for record in raw_data:
        assert "date" in record or "timestamp" in record

​
本地 HTTPS 设置（用于 OAuth 开发）

由于安全要求，WHOOP 等健康数据提供商要求 OAuth 身份验证必须使用 HTTPS 回调 URL。
在本地开发时，服务器通常运行在 http://localhost，而这些提供商会拒绝该地址。要在本地测试 OAuth 流程，你需要设置一个本地 HTTPS 反向代理，负责终止 SSL/TLS 并将请求转发到你的开发服务器。 这可以通过配置系统的 hosts 文件将生产回调域名（例如 data-gray.thetahealth.ai）映射到 127.0.0.1，然后在 443 端口运行一个处理 HTTPS 并代理到本地 HTTP 服务器的反向代理来实现。 有多种工具可以提供此功能，包括具有反向代理能力的 Web 服务器（nginx、Apache）、专门的 HTTPS 代理工具，或 ngrok 等基于云的隧道服务。 一般步骤：
  1. 安装本地证书授权工具（如 mkcert）为你的域名生成受信任的 SSL 证书
  2. 配置代理监听 443 端口，使用你的 SSL 证书并转发到本地开发端口（例如 localhost:18080）
  3. 更新 hosts 文件，将回调域名解析到 localhost
这种设置允许你在开发环境中使用与生产环境完全相同的回调 URL，从而使 OAuth 集成测试更加顺畅，而无需为每次代码更改都进行部署。

​
手动测试

​
1. 测试 OAuth 绑定

# 获取授权 URL
curl "http://localhost:18080/api/v1/pulse/theta/theta_yourprovider/link?user_id=manual_test_user"
访问返回的 URL 并完成授权。

​
2. 测试回调

授权后，验证回调是否成功： 
# 检查 Provider 状态
curl "http://localhost:18080/api/v1/pulse/providers?user_id=manual_test_user"
预期结果：Provider 显示 "status": "connected"

​
3. 测试数据拉取

# 触发数据同步
curl -X POST "http://localhost:18080/api/v1/pulse/theta/theta_yourprovider/sync?user_id=manual_test_user"

​
4. 验证数据

# 检查数据库中的新记录
psql -d your_db -c "SELECT COUNT(*) FROM theta_ai.health_data_yourprovider WHERE theta_user_id = 'manual_test_user';"

# 检查标准化数据
psql -d your_db -c "SELECT indicator, COUNT(*) FROM theta_ai.th_series_data WHERE user_id = 'manual_test_user' GROUP BY indicator;"

​
测试场景

测试正常操作：
  • 全新的 OAuth 连接
  • 使用有效 Token 获取数据
  • 正确的数据转换
  • 成功存储到数据库
测试错误场景：
  • 无效的 OAuth 凭据
  • Token 过期
  • API 速率限制
  • 网络故障
  • 格式错误的 API 响应
  • 缺少必要字段
测试边界情况：
  • 空的数据响应
  • 超大数据集
  • 重复记录
  • 时区转换
  • Null/缺失值

​
Mock 外部 API

在单元测试中使用 Mock： 
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
@patch('aiohttp.ClientSession.post')
async def test_token_exchange(mock_post):
    """测试使用 Mock API 进行 Token 交换"""
    # Mock API 响应
    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()
    
    # 使用 Mock 响应测试回调
    result = await provider.callback(
        user_id="test_user",
        code="mock_code",
        state="mock_state"
    )
    
    assert result["success"] == True

​
测试数据

创建真实的测试数据： 
# Mock 厂商 API 响应
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
            }
        }
    ]
}

​
持续集成 (CI)

将 Provider 测试添加到 CI 流水线中：
.github/workflows/test-providers.yml
name: Test Providers

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    
    steps:
      - uses: actions/checkout@v3
      
      - name: 设置 Python 环境
        uses: actions/setup-python@v4
        with:
          python-version: '3.12'
      
      - name: 安装依赖
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-asyncio
      
      - name: 运行 Provider 测试
        run: |
          pytest tests/test_*_provider.py -v

​
下一步

Provider 集成

构建自定义 Provider

OAuth 实现

实现 OAuth 流程

测试指南

通用测试实践

示例

查看真实的 Provider 代码
有关完整的测试示例和模式，请参阅代码库中的 Provider 集成指南。