mirror of
https://github.com/sstent/AICyclingCoach.git
synced 2026-01-26 00:52:07 +00:00
237 lines
12 KiB
Python
237 lines
12 KiB
Python
import os
|
|
from unittest.mock import MagicMock, patch, call
|
|
import pytest
|
|
from backend.app.services.garmin import GarminConnectService, GarminAuthError, GarminAPIError
|
|
from backend.app.models.garmin_sync_log import GarminSyncStatus
|
|
from datetime import datetime, timedelta
|
|
from garminconnect import Garmin, GarminConnectAuthenticationError, GarminConnectTooManyRequestsError, GarminConnectConnectionError
|
|
|
|
@pytest.fixture
|
|
def mock_env_vars():
|
|
with patch.dict(os.environ, {"GARMIN_USERNAME": "test_user", "GARMIN_PASSWORD": "test_password"}):
|
|
yield
|
|
|
|
def _create_garmin_client_mock():
|
|
mock_client_instance = MagicMock(spec=Garmin)
|
|
mock_client_instance.get_activities_by_date = MagicMock(return_value=[])
|
|
mock_client_instance.get_activity_details = MagicMock(return_value={})
|
|
|
|
# Mock garth.dump for saving session
|
|
mock_garth = MagicMock()
|
|
mock_garth.dump = MagicMock()
|
|
mock_client_instance.garth = mock_garth
|
|
|
|
return mock_client_instance
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_garmin_authentication_success(db_session, mock_env_vars):
|
|
"""Test successful Garmin Connect authentication"""
|
|
mock_client = _create_garmin_client_mock()
|
|
mock_client.login.side_effect = [FileNotFoundError(), None] # Session load fails, fresh login succeeds
|
|
mock_client.garth.dump.return_value = None # Ensure dump also returns None
|
|
|
|
# Mock Path constructor and its instance methods
|
|
mock_path_instance = MagicMock()
|
|
mock_path_instance.exists.return_value = True
|
|
mock_path_instance.__str__.return_value = "data/sessions"
|
|
mock_path_instance.mkdir.return_value = None
|
|
|
|
with patch('backend.app.services.garmin.Garmin', return_value=mock_client) as mock_garmin_class, \
|
|
patch('backend.app.services.garmin.Path', new_callable=MagicMock) as mock_path_constructor:
|
|
mock_path_constructor.return_value = mock_path_instance # Configure what Path(...) returns
|
|
service = GarminConnectService(db_session)
|
|
|
|
result = await service.authenticate()
|
|
assert result is True
|
|
|
|
mock_garmin_class.assert_called_once() # Garmin() constructor called once
|
|
mock_path_constructor.assert_called_once_with("data/sessions") # Path constructor called
|
|
mock_path_instance.mkdir.assert_called_once_with(parents=True, exist_ok=True) # mkdir called
|
|
|
|
# Verify calls to login on the mock_client instance
|
|
assert mock_client.login.call_count == 2
|
|
mock_client.login.assert_has_calls([
|
|
call(str(service.session_dir)), # First call: session load
|
|
call(os.getenv("GARMIN_USERNAME"), os.getenv("GARMIN_PASSWORD")) # Second call: fresh login
|
|
])
|
|
mock_client.garth.dump.assert_called_once_with(str(service.session_dir))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_garmin_authentication_failure(db_session, mock_env_vars):
|
|
"""Test authentication failure handling"""
|
|
mock_client = _create_garmin_client_mock()
|
|
mock_client.login.side_effect = [FileNotFoundError(), GarminConnectAuthenticationError("Invalid credentials")] # Session load fails, fresh login fails
|
|
|
|
mock_path_instance = MagicMock()
|
|
mock_path_instance.exists.return_value = True
|
|
mock_path_instance.__str__.return_value = "data/sessions"
|
|
mock_path_instance.mkdir.return_value = None
|
|
|
|
with patch('backend.app.services.garmin.Garmin', return_value=mock_client) as mock_garmin_class, \
|
|
patch('backend.app.services.garmin.Path', new_callable=MagicMock) as mock_path_constructor:
|
|
mock_path_constructor.return_value = mock_path_instance
|
|
service = GarminConnectService(db_session)
|
|
|
|
with pytest.raises(GarminAuthError):
|
|
await service.authenticate()
|
|
|
|
mock_garmin_class.assert_called_once() # Garmin() constructor called once
|
|
mock_path_constructor.assert_called_once_with("data/sessions")
|
|
mock_path_instance.mkdir.assert_called_once_with(parents=True, exist_ok=True)
|
|
|
|
# Verify calls to login on the mock_client instance
|
|
assert mock_client.login.call_count == 2
|
|
mock_client.login.assert_has_calls([
|
|
call(str(service.session_dir)), # First call: session load
|
|
call(os.getenv("GARMIN_USERNAME"), os.getenv("GARMIN_PASSWORD")) # Second call: fresh login
|
|
])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_garmin_authentication_load_session_success(db_session, mock_env_vars):
|
|
"""Test successful loading of existing Garmin session"""
|
|
mock_client = _create_garmin_client_mock()
|
|
mock_client.login.return_value = None # Session load succeeds (only one call expected)
|
|
|
|
mock_path_instance = MagicMock()
|
|
mock_path_instance.exists.return_value = True
|
|
mock_path_instance.__str__.return_value = "data/sessions"
|
|
mock_path_instance.mkdir.return_value = None
|
|
|
|
with patch('backend.app.services.garmin.Garmin', return_value=mock_client) as mock_garmin_class, \
|
|
patch('backend.app.services.garmin.Path', new_callable=MagicMock) as mock_path_constructor:
|
|
mock_path_constructor.return_value = mock_path_instance
|
|
service = GarminConnectService(db_session)
|
|
result = await service.authenticate()
|
|
assert result is True
|
|
mock_garmin_class.assert_called_once_with() # Ensure Garmin() is called once without args
|
|
mock_path_constructor.assert_called_once_with("data/sessions")
|
|
mock_path_instance.mkdir.assert_called_once_with(parents=True, exist_ok=True)
|
|
mock_client.login.assert_called_once_with(str(service.session_dir)) # Only session login should be called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_garmin_authentication_missing_credentials(db_session):
|
|
"""Test authentication failure when credentials are missing"""
|
|
mock_path_instance = MagicMock()
|
|
mock_path_instance.exists.return_value = True
|
|
mock_path_instance.__str__.return_value = "data/sessions"
|
|
mock_path_instance.mkdir.return_value = None
|
|
|
|
with patch.dict(os.environ, {"GARMIN_USERNAME": "", "GARMIN_PASSWORD": ""}), \
|
|
patch('backend.app.services.garmin.Path', new_callable=MagicMock) as mock_path_constructor:
|
|
mock_path_constructor.return_value = mock_path_instance
|
|
with pytest.raises(GarminAuthError, match="Garmin username or password not configured."):
|
|
service = GarminConnectService(db_session)
|
|
mock_path_constructor.assert_called_once_with("data/sessions")
|
|
mock_path_instance.mkdir.assert_called_once_with(parents=True, exist_ok=True)
|
|
await service.authenticate()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_activity_sync(db_session, mock_env_vars):
|
|
"""Test successful activity synchronization"""
|
|
mock_client = _create_garmin_client_mock()
|
|
mock_client.get_activities_by_date.return_value = [
|
|
{"activityId": 123, "startTimeLocal": "2024-01-01 08:00:00"}
|
|
]
|
|
mock_path_instance = MagicMock()
|
|
mock_path_instance.exists.return_value = True
|
|
mock_path_instance.__str__.return_value = "data/sessions"
|
|
mock_path_instance.mkdir.return_value = None
|
|
|
|
with patch('backend.app.services.garmin.Garmin', return_value=mock_client) as mock_garmin_class, \
|
|
patch('backend.app.services.garmin.Path', new_callable=MagicMock) as mock_path_constructor, \
|
|
patch('backend.app.services.garmin.GarminConnectService.authenticate', return_value=True) as mock_authenticate:
|
|
mock_path_constructor.return_value = mock_path_instance
|
|
service = GarminConnectService(db_session)
|
|
|
|
activities = await service.get_activities(datetime(2024, 1, 1), datetime(2024, 1, 1))
|
|
assert len(activities) == 1
|
|
assert activities[0]["activityId"] == 123
|
|
mock_authenticate.assert_called_once() # Ensure authenticate was called
|
|
mock_client.get_activities_by_date.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rate_limiting_handling(db_session, mock_env_vars):
|
|
"""Test API rate limit error handling"""
|
|
mock_client = _create_garmin_client_mock()
|
|
mock_client.get_activities_by_date.side_effect = GarminConnectTooManyRequestsError("Rate limit exceeded")
|
|
mock_path_instance = MagicMock()
|
|
mock_path_instance.exists.return_value = True
|
|
mock_path_instance.__str__.return_value = "data/sessions"
|
|
mock_path_instance.mkdir.return_value = None
|
|
|
|
with patch('backend.app.services.garmin.Garmin', return_value=mock_client) as mock_garmin_class, \
|
|
patch('backend.app.services.garmin.Path', new_callable=MagicMock) as mock_path_constructor, \
|
|
patch('backend.app.services.garmin.GarminConnectService.authenticate', return_value=True) as mock_authenticate:
|
|
mock_path_constructor.return_value = mock_path_instance
|
|
service = GarminConnectService(db_session)
|
|
|
|
with pytest.raises(GarminAPIError):
|
|
await service.get_activities(datetime(2024, 1, 1), datetime(2024, 1, 1))
|
|
mock_authenticate.assert_called_once() # Ensure authenticate was called
|
|
mock_client.get_activities_by_date.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_activity_details_success(db_session, mock_env_vars):
|
|
"""Test successful retrieval of activity details."""
|
|
mock_client = _create_garmin_client_mock()
|
|
mock_client.get_activity_details.return_value = {"activityId": 123, "details": "data"}
|
|
mock_path_instance = MagicMock()
|
|
mock_path_instance.exists.return_value = True
|
|
mock_path_instance.__str__.return_value = "data/sessions"
|
|
mock_path_instance.mkdir.return_value = None
|
|
|
|
with patch('backend.app.services.garmin.Garmin', return_value=mock_client) as mock_garmin_class, \
|
|
patch('backend.app.services.garmin.Path', new_callable=MagicMock) as mock_path_constructor, \
|
|
patch('backend.app.services.garmin.GarminConnectService.authenticate', return_value=True) as mock_authenticate:
|
|
mock_path_constructor.return_value = mock_path_instance
|
|
service = GarminConnectService(db_session)
|
|
|
|
details = await service.get_activity_details("123")
|
|
assert details["activityId"] == 123
|
|
mock_authenticate.assert_called_once() # Ensure authenticate was called
|
|
mock_client.get_activity_details.assert_called_once_with("123")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_activity_details_failure(db_session, mock_env_vars):
|
|
"""Test failure in retrieving activity details."""
|
|
mock_client = _create_garmin_client_mock()
|
|
mock_client.get_activity_details.side_effect = GarminConnectConnectionError("Activity not found")
|
|
mock_path_instance = MagicMock()
|
|
mock_path_instance.exists.return_value = True
|
|
mock_path_instance.__str__.return_value = "data/sessions"
|
|
mock_path_instance.mkdir.return_value = None
|
|
|
|
with patch('backend.app.services.garmin.Garmin', return_value=mock_client) as mock_garmin_class, \
|
|
patch('backend.app.services.garmin.Path', new_callable=MagicMock) as mock_path_constructor, \
|
|
patch('backend.app.services.garmin.GarminConnectService.authenticate', return_value=True) as mock_authenticate:
|
|
mock_path_constructor.return_value = mock_path_instance
|
|
service = GarminConnectService(db_session)
|
|
|
|
with pytest.raises(GarminAPIError, match="Failed to fetch activity details"):
|
|
await service.get_activity_details("123")
|
|
mock_authenticate.assert_called_once() # Ensure authenticate was called
|
|
mock_client.get_activity_details.assert_called_once_with("123")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_authenticated(db_session):
|
|
"""Test is_authenticated method"""
|
|
service = GarminConnectService(db_session)
|
|
assert service.client is None
|
|
assert service.username is None
|
|
assert service.password is None
|
|
# After authentication, client should be set and username/password should be present
|
|
service.client = MagicMock(spec=Garmin)
|
|
service.username = "testuser"
|
|
service.password = "testpass"
|
|
assert service.client is not None
|
|
assert service.username is not None
|
|
assert service.password is not None |