Skip to main content

Overview

Thorough testing ensures your provider integration is reliable, secure, and handles edge cases properly.

Testing Strategy

Unit Tests

Test individual methods in isolation

Integration Tests

Test OAuth flows end-to-end

Manual Testing

Verify in real environment

Unit Testing

Test individual 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

Test complete 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

Local HTTPS Setup for OAuth Development

Health data providers like WHOOP require HTTPS callback URLs for OAuth authentication due to security requirements.
During local development, your server typically runs on http://localhost, which these providers reject. To test OAuth flows locally, you need to set up a local HTTPS reverse proxy that terminates SSL/TLS and forwards requests to your development server. This is accomplished by configuring your system’s hosts file to map the production callback domain (e.g., data-gray.thetahealth.ai) to 127.0.0.1, then running a reverse proxy on port 443 that handles HTTPS and proxies to your local HTTP server. Several tools can provide this functionality, including web servers with reverse proxy capabilities (nginx, Apache), specialized HTTPS proxy tools, or cloud-based tunneling services like ngrok. The general approach:
  1. Install a local certificate authority tool like mkcert to generate trusted SSL certificates for your domain
  2. Configure the proxy to listen on port 443 with your SSL certificate and forward to your local development port (e.g., localhost:18080)
  3. Update your hosts file to resolve the callback domain to localhost
This setup allows you to use the exact same callback URL in development as in production, making OAuth integration testing seamless without requiring deployment for every code change.

Manual Testing

# Get authorization URL
curl "http://localhost:18080/api/v1/pulse/theta/theta_yourprovider/link?user_id=manual_test_user"
Visit the returned URL and complete authorization.

2. Test Callback

After authorization, verify the callback was successful:
# Check provider status
curl "http://localhost:18080/api/v1/pulse/providers?user_id=manual_test_user"
Expected: Provider shows "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;"

Test Scenarios

Test normal operation:
  • Fresh OAuth connection
  • Data fetch with valid tokens
  • Proper data transformation
  • Successful database storage
Test error scenarios:
  • Invalid OAuth credentials
  • Expired tokens
  • API rate limiting
  • Network failures
  • Malformed API responses
  • Missing required fields
Test edge cases:
  • Empty data responses
  • Very large datasets
  • Duplicate records
  • Time zone conversions
  • Null/missing values

Mocking External APIs

Use mocks for unit tests:
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

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