mirror of
https://github.com/sstent/AICyclingCoach.git
synced 2026-01-26 00:52:07 +00:00
257 lines
11 KiB
Python
257 lines
11 KiB
Python
import pytest
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
from datetime import datetime, timedelta
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, desc
|
|
import garminconnect # Import garminconnect to resolve NameError
|
|
|
|
from backend.app.services.workout_sync import WorkoutSyncService
|
|
from backend.app.services.garmin import GarminConnectService, GarminAuthError, GarminAPIError
|
|
from backend.app.models.workout import Workout
|
|
from backend.app.models.garmin_sync_log import GarminSyncLog, GarminSyncStatus
|
|
|
|
@pytest.fixture
|
|
def mock_garmin_service():
|
|
"""Fixture to provide a mock GarminConnectService."""
|
|
service = MagicMock(spec=GarminConnectService)
|
|
service.authenticate = AsyncMock(return_value=True)
|
|
service.get_activities = AsyncMock(return_value=[])
|
|
service.get_activity_details = AsyncMock(return_value={})
|
|
return service
|
|
|
|
@pytest.fixture
|
|
def mock_db_session():
|
|
"""Fixture to provide an AsyncMock for the database session."""
|
|
mock_db = MagicMock(spec=AsyncSession)
|
|
mock_db.add = MagicMock()
|
|
mock_db.commit = AsyncMock() # Must be AsyncMock as it's awaited
|
|
mock_db.refresh = AsyncMock() # Must be AsyncMock as it's awaited
|
|
|
|
# Configure execute to return a mock result object that can be awaited
|
|
mock_result_proxy = MagicMock()
|
|
mock_result_proxy.scalar_one_or_none = AsyncMock(return_value=None) # scalar_one_or_none is awaited
|
|
mock_db.execute = AsyncMock(return_value=mock_result_proxy) # Must be AsyncMock as it's awaited
|
|
return mock_db
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_successful_sync(mock_db_session, mock_garmin_service):
|
|
"""Test successful sync of new activities"""
|
|
# Configure mock_db_session.execute for activity_exists check
|
|
mock_db_session.execute.return_value.scalar_one_or_none.return_value = None # No duplicates
|
|
|
|
# Patch GarminService to return our mock_garmin_service
|
|
with patch('backend.app.services.workout_sync.GarminService', return_value=mock_garmin_service):
|
|
service = WorkoutSyncService(mock_db_session)
|
|
|
|
# Mock the garmin service methods
|
|
mock_garmin_service.get_activities.return_value = [
|
|
{
|
|
'activityId': '123456',
|
|
'activityType': {'typeKey': 'cycling'},
|
|
'startTimeLocal': '2024-01-15 08:00:00', # Adjusted format
|
|
'duration': 3600,
|
|
'distance': 25000
|
|
}
|
|
]
|
|
|
|
mock_garmin_service.get_activity_details.return_value = {
|
|
'averageHeartRateInBeatsPerMinute': 150,
|
|
'maxHeartRateInBeatsPerMinute': 180,
|
|
'averagePower': 250,
|
|
'totalElevationGain': 500
|
|
}
|
|
|
|
result = await service.sync_recent_activities(days_back=7)
|
|
|
|
assert result == 1
|
|
mock_db_session.add.assert_called()
|
|
mock_db_session.commit.assert_called()
|
|
mock_db_session.refresh.assert_called()
|
|
mock_garmin_service.authenticate.assert_called_once()
|
|
mock_garmin_service.get_activities.assert_called_once()
|
|
mock_garmin_service.get_activity_details.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_activity_handling(mock_db_session, mock_garmin_service):
|
|
"""Test skipping duplicate activities"""
|
|
# Configure mock_db_session.execute for activity_exists check
|
|
mock_db_session.execute.return_value.scalar_one_or_none.return_value = MagicMock(spec=Workout) # Simulate activity exists
|
|
|
|
with patch('backend.app.services.workout_sync.GarminService', return_value=mock_garmin_service):
|
|
service = WorkoutSyncService(mock_db_session)
|
|
|
|
mock_garmin_service.get_activities.return_value = [
|
|
{'activityId': '123456', 'startTimeLocal': '2024-01-15 08:00:00'}
|
|
]
|
|
|
|
result = await service.sync_recent_activities()
|
|
|
|
assert result == 0
|
|
mock_db_session.add.assert_called_once() # Should add the sync log
|
|
mock_db_session.commit.assert_called() # Commit for sync log update
|
|
mock_garmin_service.authenticate.assert_called_once()
|
|
mock_garmin_service.get_activities.assert_called_once()
|
|
mock_garmin_service.get_activity_details.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activity_detail_retry_logic(mock_db_session, mock_garmin_service):
|
|
"""Test retry logic for activity details"""
|
|
mock_db_session.execute.return_value.scalar_one_or_none.return_value = None # No duplicates
|
|
|
|
with patch('backend.app.services.workout_sync.GarminService', return_value=mock_garmin_service):
|
|
service = WorkoutSyncService(mock_db_session)
|
|
|
|
service.garmin_service.get_activities.return_value = [
|
|
{
|
|
'activityId': '123456',
|
|
'activityType': {'typeKey': 'cycling'},
|
|
'startTimeLocal': '2024-01-15 08:00:00',
|
|
'duration': 3600
|
|
}
|
|
]
|
|
|
|
# First call fails, second succeeds
|
|
mock_garmin_service.get_activity_details.side_effect = [
|
|
GarminAPIError("Temporary failure"),
|
|
{'averageHeartRateInBeatsPerMinute': 150, 'maxHeartRateInBeatsPerMinute': 180}
|
|
]
|
|
|
|
# Mock asyncio.sleep to avoid actual delays in tests
|
|
with patch('asyncio.sleep', new_callable=AsyncMock) as mock_sleep:
|
|
result = await service.sync_recent_activities()
|
|
|
|
assert result == 1
|
|
assert mock_garmin_service.get_activity_details.call_count == 2
|
|
mock_sleep.assert_called_once_with(1) # First retry delay
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_error_handling(mock_db_session, mock_garmin_service):
|
|
"""Test authentication error handling"""
|
|
with patch('backend.app.services.workout_sync.GarminService', return_value=mock_garmin_service):
|
|
service = WorkoutSyncService(mock_db_session)
|
|
|
|
# Mock authentication failure. The authenticate method is called before get_activities
|
|
mock_garmin_service.authenticate.side_effect = GarminAuthError("Authentication failed")
|
|
|
|
with pytest.raises(GarminAuthError):
|
|
await service.sync_recent_activities()
|
|
|
|
mock_db_session.add.assert_called_once() # Should add the sync log
|
|
assert mock_db_session.commit.call_count == 2 # Initial commit and commit after updating status
|
|
mock_db_session.refresh.assert_called_once() # Initial refresh
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_api_error_handling(mock_db_session, mock_garmin_service):
|
|
"""Test API error handling"""
|
|
with patch('backend.app.services.workout_sync.GarminService', return_value=mock_garmin_service):
|
|
service = WorkoutSyncService(mock_db_session)
|
|
|
|
mock_garmin_service.get_activities.side_effect = GarminAPIError("API rate limit exceeded")
|
|
|
|
with pytest.raises(GarminAPIError):
|
|
await service.sync_recent_activities()
|
|
|
|
mock_db_session.add.assert_called_once() # Should add the sync log
|
|
assert mock_db_session.commit.call_count == 2 # Initial commit and commit after updating status
|
|
mock_db_session.refresh.assert_called_once() # Initial refresh
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_sync_status(mock_db_session):
|
|
"""Test retrieval of latest sync status"""
|
|
mock_log = GarminSyncLog(
|
|
status=GarminSyncStatus.COMPLETED,
|
|
activities_synced=5,
|
|
last_sync_time=datetime.now()
|
|
)
|
|
|
|
# Mock the database query for a single result
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none = AsyncMock(return_value=mock_log) # scalar_one_or_none is awaited
|
|
mock_db_session.execute.return_value = mock_result # mock_db_session.execute is awaited
|
|
|
|
service = WorkoutSyncService(mock_db_session)
|
|
result = await service.get_latest_sync_status()
|
|
|
|
assert result == mock_log
|
|
mock_db_session.execute.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activity_exists_check(mock_db_session):
|
|
"""Test the activity_exists helper method"""
|
|
# Mock existing activity
|
|
mock_workout = Workout(garmin_activity_id="123456")
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none = AsyncMock(return_value=mock_workout) # scalar_one_or_none is awaited
|
|
mock_db_session.execute.return_value = mock_result # mock_db_session.execute is awaited
|
|
|
|
service = WorkoutSyncService(mock_db_session)
|
|
exists = await service.activity_exists("123456")
|
|
|
|
assert exists is True
|
|
mock_db_session.execute.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activity_does_not_exist(mock_db_session):
|
|
"""Test activity_exists when activity doesn't exist"""
|
|
# Mock no existing activity
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none = AsyncMock(return_value=None) # scalar_one_or_none is awaited
|
|
mock_db_session.execute.return_value = mock_result # mock_db_session.execute is awaited
|
|
|
|
service = WorkoutSyncService(mock_db_session)
|
|
exists = await service.activity_exists("nonexistent")
|
|
|
|
assert exists is False
|
|
mock_db_session.execute.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parse_activity_data(mock_db_session):
|
|
"""Test parsing of Garmin activity data"""
|
|
service = WorkoutSyncService(mock_db_session)
|
|
|
|
activity_data = {
|
|
'activityId': '987654321',
|
|
'activityType': {'typeKey': 'cycling'},
|
|
'startTimeLocal': '2024-01-15 08:30:00', # Adjusted format
|
|
'duration': 7200,
|
|
'distance': 50000,
|
|
'averageHR': 145, # Corrected key
|
|
'maxHR': 175, # Corrected key
|
|
'avgPower': 230, # Corrected key
|
|
'maxPower': 450, # Corrected key
|
|
'averageBikingCadenceInRevPerMinute': 85,
|
|
'elevationGain': 800 # Corrected key
|
|
}
|
|
|
|
result = await service.parse_activity_data(activity_data)
|
|
|
|
assert result['garmin_activity_id'] == '987654321'
|
|
assert result['activity_type'] == 'cycling'
|
|
assert result['duration_seconds'] == 7200
|
|
assert result['distance_m'] == 50000
|
|
assert result['avg_hr'] == 145
|
|
assert result['max_hr'] == 175
|
|
assert result['avg_power'] == 230
|
|
assert result['max_power'] == 450
|
|
assert result['avg_cadence'] == 85
|
|
assert result['elevation_gain_m'] == 800
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_with_network_timeout(mock_db_session, mock_garmin_service):
|
|
"""Test handling of network timeouts during sync"""
|
|
mock_db_session.execute.return_value.scalar_one_or_none.return_value = None # No duplicates
|
|
|
|
with patch('backend.app.services.workout_sync.GarminService', return_value=mock_garmin_service):
|
|
service = WorkoutSyncService(mock_db_session)
|
|
|
|
# Simulate timeout error
|
|
mock_garmin_service.get_activities.side_effect = garminconnect.GarminConnectConnectionError("Request timed out")
|
|
|
|
with pytest.raises(GarminAPIError): # Should raise the GarminAPIError
|
|
await service.sync_recent_activities()
|
|
|
|
mock_db_session.add.assert_called_once() # Should add the sync log
|
|
assert mock_db_session.commit.call_count == 2 # Initial commit and commit after updating status
|
|
mock_db_session.refresh.assert_called_once() # Initial refresh |