sstent/FitTrack_GarminSync
1
0
Fork 0
mirror of https://github.com/sstent/FitTrack_GarminSync.git synced 2026-01-25 08:35:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Implement single sync job management and progress tracking

Browse Source
This commit is contained in:
sstent
2025-10-11 18:36:19 -07:00
parent 3819e4f5e2
commit 723ca04aa8
51 changed files with 1625 additions and 596 deletions
Download Patch File Download Diff File Expand all files Collapse all files

137
backend/tests/api/test_garmin_auth_api.py
View File

@@ -2,25 +2,30 @@ from datetime import datetime, timedelta
from unittest.mock import AsyncMock, patch
import pytest
from backend.src.main import app
from backend.src.schemas import GarminCredentials
from httpx import AsyncClient
from fastapi.testclient import TestClient
from src.main import app
from src.schemas import GarminCredentials
@pytest.fixture
def mock_garmin_auth_service():
with patch('backend.src.api.garmin_auth.GarminAuthService') as MockGarminAuthService:
with patch("src.api.garmin_auth.GarminAuthService") as MockGarminAuthService:
service_instance = MockGarminAuthService.return_value
yield service_instance
@pytest.fixture
def mock_central_db_service():
with patch('backend.src.api.garmin_auth.CentralDBService') as MockCentralDBService:
with patch("src.api.garmin_auth.CentralDBService") as MockCentralDBService:
service_instance = MockCentralDBService.return_value
yield service_instance
@pytest.mark.asyncio
async def test_garmin_login_success_new_credentials(mock_garmin_auth_service, mock_central_db_service):
async def test_garmin_login_success_new_credentials(
mock_garmin_auth_service, mock_central_db_service
):
username = "test@example.com"
password = "password123"
@@ -29,28 +34,33 @@ async def test_garmin_login_success_new_credentials(mock_garmin_auth_service, mo
garmin_password_plaintext=password,
access_token="mock_access_token",
access_token_secret="mock_access_token_secret",
token_expiration_date=datetime.utcnow() + timedelta(hours=1)
token_expiration_date=datetime.utcnow() + timedelta(hours=1),
)
mock_central_db_service.get_garmin_credentials.return_value = None # No existing credentials
mock_central_db_service.create_garmin_credentials.return_value = AsyncMock() # Simulate successful creation
mock_central_db_service.get_garmin_credentials.return_value = (
None # No existing credentials
)
mock_central_db_service.create_garmin_credentials.return_value = (
AsyncMock()
) # Simulate successful creation
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post(
"/api/garmin/login",
json={
"username": username,
"password": password
}
with TestClient(app=app) as client:
response = client.post(
"/api/garmin/login", json={"username": username, "password": password}
)
assert response.status_code == 200
assert response.json() == {"message": "Garmin account linked successfully."}
mock_garmin_auth_service.initial_login.assert_called_once_with(username, password)
mock_central_db_service.get_garmin_credentials.assert_called_once_with(1) # Assuming user_id 1
mock_central_db_service.get_garmin_credentials.assert_called_once_with(
1
) # Assuming user_id 1
mock_central_db_service.create_garmin_credentials.assert_called_once()
@pytest.mark.asyncio
async def test_garmin_login_success_update_credentials(mock_garmin_auth_service, mock_central_db_service):
async def test_garmin_login_success_update_credentials(
mock_garmin_auth_service, mock_central_db_service
):
username = "test@example.com"
password = "password123"
@@ -59,24 +69,22 @@ async def test_garmin_login_success_update_credentials(mock_garmin_auth_service,
garmin_password_plaintext=password,
access_token="mock_access_token_new",
access_token_secret="mock_access_token_secret_new",
token_expiration_date=datetime.utcnow() + timedelta(hours=1)
token_expiration_date=datetime.utcnow() + timedelta(hours=1),
)
mock_central_db_service.get_garmin_credentials.return_value = GarminCredentials(
garmin_username=username,
garmin_password_plaintext="old_password",
access_token="old_access_token",
access_token_secret="old_access_token_secret",
token_expiration_date=datetime.utcnow() - timedelta(hours=1)
) # Existing credentials
mock_central_db_service.update_garmin_credentials.return_value = AsyncMock() # Simulate successful update
token_expiration_date=datetime.utcnow() - timedelta(hours=1),
) # Existing credentials
mock_central_db_service.update_garmin_credentials.return_value = (
AsyncMock()
) # Simulate successful update
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post(
"/api/garmin/login",
json={
"username": username,
"password": password
}
with TestClient(app=app) as client:
response = client.post(
"/api/garmin/login", json={"username": username, "password": password}
)
assert response.status_code == 200
@@ -85,20 +93,21 @@ async def test_garmin_login_success_update_credentials(mock_garmin_auth_service,
mock_central_db_service.get_garmin_credentials.assert_called_once_with(1)
mock_central_db_service.update_garmin_credentials.assert_called_once()
@pytest.mark.asyncio
async def test_garmin_login_failure_invalid_credentials(mock_garmin_auth_service, mock_central_db_service):
async def test_garmin_login_failure_invalid_credentials(
mock_garmin_auth_service, mock_central_db_service
):
username = "invalid@example.com"
password = "wrongpassword"
mock_garmin_auth_service.initial_login.return_value = None # Simulate failed Garmin login
mock_garmin_auth_service.initial_login.return_value = (
None # Simulate failed Garmin login
)
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post(
"/api/garmin/login",
json={
"username": username,
"password": password
}
with TestClient(app=app) as client:
response = client.post(
"/api/garmin/login", json={"username": username, "password": password}
)
assert response.status_code == 401
@@ -108,8 +117,11 @@ async def test_garmin_login_failure_invalid_credentials(mock_garmin_auth_service
mock_central_db_service.create_garmin_credentials.assert_not_called()
mock_central_db_service.update_garmin_credentials.assert_not_called()
@pytest.mark.asyncio
async def test_garmin_login_failure_central_db_create_error(mock_garmin_auth_service, mock_central_db_service):
async def test_garmin_login_failure_central_db_create_error(
mock_garmin_auth_service, mock_central_db_service
):
username = "test@example.com"
password = "password123"
@@ -118,25 +130,28 @@ async def test_garmin_login_failure_central_db_create_error(mock_garmin_auth_ser
garmin_password_plaintext=password,
access_token="mock_access_token",
access_token_secret="mock_access_token_secret",
token_expiration_date=datetime.utcnow() + timedelta(hours=1)
token_expiration_date=datetime.utcnow() + timedelta(hours=1),
)
mock_central_db_service.get_garmin_credentials.return_value = None
mock_central_db_service.create_garmin_credentials.return_value = None # Simulate CentralDB create failure
mock_central_db_service.create_garmin_credentials.return_value = (
None # Simulate CentralDB create failure
)
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post(
"/api/garmin/login",
json={
"username": username,
"password": password
}
with TestClient(app=app) as client:
response = client.post(
"/api/garmin/login", json={"username": username, "password": password}
)
assert response.status_code == 500
assert response.json() == {"detail": "Failed to store Garmin credentials in CentralDB."}
assert response.json() == {
"detail": "Failed to store Garmin credentials in CentralDB."
}
@pytest.mark.asyncio
async def test_garmin_login_failure_central_db_update_error(mock_garmin_auth_service, mock_central_db_service):
async def test_garmin_login_failure_central_db_update_error(
mock_garmin_auth_service, mock_central_db_service
):
username = "test@example.com"
password = "password123"
@@ -145,25 +160,25 @@ async def test_garmin_login_failure_central_db_update_error(mock_garmin_auth_ser
garmin_password_plaintext=password,
access_token="mock_access_token_new",
access_token_secret="mock_access_token_secret_new",
token_expiration_date=datetime.utcnow() + timedelta(hours=1)
token_expiration_date=datetime.utcnow() + timedelta(hours=1),
)
mock_central_db_service.get_garmin_credentials.return_value = GarminCredentials(
garmin_username=username,
garmin_password_plaintext="old_password",
access_token="old_access_token",
access_token_secret="old_access_token_secret",
token_expiration_date=datetime.utcnow() - timedelta(hours=1)
token_expiration_date=datetime.utcnow() - timedelta(hours=1),
)
mock_central_db_service.update_garmin_credentials.return_value = (
None # Simulate CentralDB update failure
)
mock_central_db_service.update_garmin_credentials.return_value = None # Simulate CentralDB update failure
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post(
"/api/garmin/login",
json={
"username": username,
"password": password
}
with TestClient(app=app) as client:
response = client.post(
"/api/garmin/login", json={"username": username, "password": password}
)
assert response.status_code == 500
assert response.json() == {"detail": "Failed to update Garmin credentials in CentralDB."}
assert response.json() == {
"detail": "Failed to update Garmin credentials in CentralDB."
}

146
backend/tests/api/test_garmin_sync_api.py
View File

@@ -1,80 +1,88 @@
from datetime import date
from unittest.mock import AsyncMock, patch
import pytest
from backend.src.main import app
from backend.src.schemas import User
from fastapi import HTTPException
from httpx import AsyncClient
from backend.src.services.sync_manager import current_sync_job_manager
from fastapi.testclient import TestClient
client = TestClient(app)
@pytest.fixture
def mock_garmin_activity_service():
with patch('backend.src.api.garmin_sync.GarminActivityService') as MockGarminActivityService:
service_instance = MockGarminActivityService.return_value
yield service_instance
def test_get_sync_status():
response = client.get("/api/sync/garmin/sync/status")
assert response.status_code == 200
@pytest.fixture
def mock_get_current_user():
with patch('backend.src.api.garmin_sync.get_current_user') as mock_current_user:
mock_current_user.return_value = User(id=1, name="Test User", email="test@example.com")
yield mock_current_user
@pytest.mark.asyncio
async def test_trigger_garmin_activity_sync_success(mock_garmin_activity_service, mock_get_current_user):
mock_garmin_activity_service.sync_activities_in_background = AsyncMock()
mock_garmin_activity_service.sync_activities_in_background.return_value = None
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post(
"/api/sync/garmin/activities",
json={
"force_resync": False,
"start_date": "2023-01-01",
"end_date": "2023-01-31"
}
)
def test_trigger_activity_sync_success():
response = client.post("/api/sync/garmin/activities", json={})
assert response.status_code == 202
response_json = response.json()
assert "job_id" in response_json
assert "status" in response_json
assert response_json["status"] == "pending"
mock_garmin_activity_service.sync_activities_in_background.assert_called_once()
args, kwargs = mock_garmin_activity_service.sync_activities_in_background.call_args
assert not args[1] # force_resync
assert args[2] == date(2023, 1, 1) # start_date
assert args[3] == date(2023, 1, 31) # end_date
assert response.json() == {
"message": "Activity synchronization initiated successfully."
}
@pytest.mark.asyncio
async def test_trigger_garmin_activity_sync_no_dates(mock_garmin_activity_service, mock_get_current_user):
mock_garmin_activity_service.sync_activities_in_background = AsyncMock()
mock_garmin_activity_service.sync_activities_in_background.return_value = None
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post(
"/api/sync/garmin/activities",
json={}
)
def test_trigger_activity_sync_conflict():
# Manually start a sync to simulate a conflict
current_sync_job_manager._current_job = current_sync_job_manager.start_sync(
"activities"
)
response = client.post("/api/sync/garmin/activities", json={})
assert response.status_code == 409
assert response.json() == {
"detail": "A synchronization is already in progress. Please wait or check status."
}
# Clean up
current_sync_job_manager._current_job = None
def test_trigger_workout_sync_success():
response = client.post(
"/api/sync/garmin/workouts",
json={"workout_id": "a1b2c3d4-e5f6-7890-1234-567890abcdef"},
)
assert response.status_code == 202
response_json = response.json()
assert "job_id" in response_json
assert "status" in response_json
assert response_json["status"] == "pending"
mock_garmin_activity_service.sync_activities_in_background.assert_called_once()
args, kwargs = mock_garmin_activity_service.sync_activities_in_background.call_args
assert not args[1] # force_resync
assert args[2] is None # start_date
assert args[3] is None # end_date
assert response.json() == {
"message": "Workout synchronization initiated successfully."
}
@pytest.mark.asyncio
async def test_trigger_garmin_activity_sync_unauthorized():
with patch('backend.src.api.garmin_sync.get_current_user', side_effect=HTTPException(status_code=401)):
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post(
"/api/sync/garmin/activities",
json={}
)
assert response.status_code == 401
assert response.json() == {"detail": "Not Authenticated"} # Default FastAPI 401 detail
def test_trigger_workout_sync_conflict():
# Manually start a sync to simulate a conflict
current_sync_job_manager._current_job = current_sync_job_manager.start_sync(
"workouts"
)
response = client.post(
"/api/sync/garmin/workouts",
json={"workout_id": "a1b2c3d4-e5f6-7890-1234-567890abcdef"},
)
assert response.status_code == 409
assert response.json() == {
"detail": "A synchronization is already in progress. Please wait or check status."
}
# Clean up
current_sync_job_manager._current_job = None
def test_trigger_health_sync_success():
response = client.post("/api/sync/garmin/health", json={})
assert response.status_code == 202
assert response.json() == {
"message": "Health metrics synchronization initiated successfully."
}
def test_trigger_health_sync_conflict():
# Manually start a sync to simulate a conflict
current_sync_job_manager._current_job = current_sync_job_manager.start_sync(
"health"
)
response = client.post("/api/sync/garmin/health", json={})
assert response.status_code == 409
assert response.json() == {
"detail": "A synchronization is already in progress. Please wait or check status."
}
# Clean up
current_sync_job_manager._current_job = None

79
backend/tests/integration/test_garmin_authentication_flow.py
View File

@@ -1,40 +1,44 @@
from unittest.mock import AsyncMock, patch
import pytest
from backend.src.schemas import GarminCredentials
from backend.src.services.garmin_activity_service import GarminActivityService
from backend.src.services.garmin_health_service import GarminHealthService
from src.schemas import GarminCredentials
from src.services.garmin_activity_service import GarminActivityService
from src.services.garmin_health_service import GarminHealthService
@pytest.fixture
def mock_garmin_auth_service_instance():
with patch(
'backend.src.services.garmin_activity_service.GarminAuthService'
"src.services.garmin_activity_service.GarminAuthService"
) as MockGarminAuthService:
instance = MockGarminAuthService.return_value
yield instance
@pytest.fixture
def mock_central_db_service_instance():
with patch(
'backend.src.services.garmin_activity_service.CentralDBService'
"src.services.garmin_activity_service.CentralDBService"
) as MockCentralDBService:
service_instance = MockCentralDBService.return_value
yield service_instance
@pytest.fixture
def mock_garmin_client_service_instance():
with patch(
'backend.src.services.garmin_activity_service.GarminClientService'
"src.services.garmin_activity_service.GarminClientService"
) as MockGarminClientService:
instance = MockGarminClientService.return_value
yield instance
@pytest.mark.asyncio
async def test_garmin_activity_sync_authentication_flow(
mock_garmin_auth_service_instance,
mock_central_db_service_instance,
mock_garmin_client_service_instance
mock_garmin_client_service_instance,
):
user_id = 1
username = "test@example.com"
@@ -42,10 +46,11 @@ async def test_garmin_activity_sync_authentication_flow(
# Mock GarminCredentials from CentralDB
mock_credentials = GarminCredentials(
garmin_username=username,
garmin_password_plaintext=password
garmin_username=username, garmin_password_plaintext=password
)
mock_central_db_service_instance.get_garmin_credentials.return_value = (
mock_credentials
)
mock_central_db_service_instance.get_garmin_credentials.return_value = mock_credentials
# Mock GarminClientService authentication
mock_garmin_client_service_instance.is_authenticated.return_value = False
@@ -53,31 +58,40 @@ async def test_garmin_activity_sync_authentication_flow(
# Mock GarminClientService.get_client().get_activities
mock_garmin_client_instance = AsyncMock()
mock_garmin_client_service_instance.get_client.return_value = mock_garmin_client_instance
mock_garmin_client_instance.get_activities.return_value = [] # Simulate no activities
mock_garmin_client_service_instance.get_client.return_value = (
mock_garmin_client_instance
)
mock_garmin_client_instance.get_activities.return_value = (
[]
) # Simulate no activities
activity_service = GarminActivityService(
garmin_client_service=mock_garmin_client_service_instance,
activity_download_service=AsyncMock(), # Mock this dependency
garmin_auth_service=mock_garmin_auth_service_instance, # Still needed for init, but methods not called
central_db_service=mock_central_db_service_instance
activity_download_service=AsyncMock(), # Mock this dependency
garmin_auth_service=mock_garmin_auth_service_instance, # Still needed for init, but methods not called
central_db_service=mock_central_db_service_instance,
)
# Call sync_activities_in_background, which will trigger authentication
await activity_service.sync_activities_in_background(job_id="test_job")
await activity_service.sync_activities_in_background(user_id=user_id)
# Assertions
mock_central_db_service_instance.get_garmin_credentials.assert_called_once_with(user_id)
mock_garmin_client_service_instance.update_credentials.assert_called_once_with(username, password)
mock_central_db_service_instance.get_garmin_credentials.assert_called_once_with(
user_id
)
mock_garmin_client_service_instance.update_credentials.assert_called_once_with(
username, password
)
mock_garmin_client_service_instance.is_authenticated.assert_called_once()
mock_garmin_client_service_instance.authenticate.assert_called_once()
mock_garmin_client_instance.get_activities.assert_called_once()
@pytest.mark.asyncio
async def test_garmin_health_sync_authentication_flow(
mock_garmin_auth_service_instance,
mock_central_db_service_instance,
mock_garmin_client_service_instance
mock_garmin_client_service_instance,
):
user_id = 1
username = "test@example.com"
@@ -85,10 +99,11 @@ async def test_garmin_health_sync_authentication_flow(
# Mock GarminCredentials from CentralDB
mock_credentials = GarminCredentials(
garmin_username=username,
garmin_password_plaintext=password
garmin_username=username, garmin_password_plaintext=password
)
mock_central_db_service_instance.get_garmin_credentials.return_value = (
mock_credentials
)
mock_central_db_service_instance.get_garmin_credentials.return_value = mock_credentials
# Mock GarminClientService authentication
mock_garmin_client_service_instance.is_authenticated.return_value = False
@@ -96,21 +111,29 @@ async def test_garmin_health_sync_authentication_flow(
# Mock GarminClientService.get_client().get_daily_summary
mock_garmin_client_instance = AsyncMock()
mock_garmin_client_service_instance.get_client.return_value = mock_garmin_client_instance
mock_garmin_client_instance.get_daily_summary.return_value = [] # Simulate no summaries
mock_garmin_client_service_instance.get_client.return_value = (
mock_garmin_client_instance
)
mock_garmin_client_instance.get_daily_summary.return_value = (
[]
) # Simulate no summaries
health_service = GarminHealthService(
garmin_client_service=mock_garmin_client_service_instance,
central_db_service=mock_central_db_service_instance,
garmin_auth_service=mock_garmin_auth_service_instance # Still needed for init, but methods not called
garmin_auth_service=mock_garmin_auth_service_instance, # Still needed for init, but methods not called
)
# Call sync_health_metrics_in_background, which will trigger authentication
await health_service.sync_health_metrics_in_background(job_id="test_job")
await health_service.sync_health_metrics_in_background(user_id=user_id)
# Assertions
mock_central_db_service_instance.get_garmin_credentials.assert_called_once_with(user_id)
mock_garmin_client_service_instance.update_credentials.assert_called_once_with(username, password)
mock_central_db_service_instance.get_garmin_credentials.assert_called_once_with(
user_id
)
mock_garmin_client_service_instance.update_credentials.assert_called_once_with(
username, password
)
mock_garmin_client_service_instance.is_authenticated.assert_called_once()
mock_garmin_client_service_instance.authenticate.assert_called_once()
mock_garmin_client_instance.get_daily_summary.assert_called_once()

40
backend/tests/unit/test_auth_service.py
View File

@@ -10,7 +10,7 @@ from src.services.auth_service import AuthService
@pytest.fixture
def auth_service():
"""Fixture for AuthService with mocked CentralDBService."""
with patch('src.services.auth_service.CentralDBService') as MockCentralDBService:
with patch("src.services.auth_service.CentralDBService") as MockCentralDBService:
mock_central_db_instance = MockCentralDBService.return_value
mock_central_db_instance.get_user_by_email = AsyncMock()
mock_central_db_instance.create_user = AsyncMock()
@@ -21,21 +21,24 @@ def auth_service():
service.central_db = mock_central_db_instance
yield service
@pytest.fixture
def mock_garth_login():
"""Fixture to mock garth.login."""
with patch('garth.login') as mock_login:
with patch("garth.login") as mock_login:
yield mock_login
@pytest.fixture
def mock_garth_client():
"""Fixture to mock garth.client attributes."""
with patch('garth.client') as mock_client:
with patch("garth.client") as mock_client:
mock_client.oauth2_token = "mock_oauth2_token"
mock_client.refresh_token = "mock_refresh_token"
mock_client.token_expires_at = 1234567890
yield mock_client
@pytest.mark.asyncio
async def test_authenticate_garmin_connect_new_user_success(
auth_service, mock_garth_login, mock_garth_client
@@ -57,7 +60,11 @@ async def test_authenticate_garmin_connect_new_user_success(
auth_service.central_db.create_token.assert_called_once()
auth_service.central_db.update_token.assert_not_called()
assert result == {"message": "Garmin Connect authentication successful", "user_id": str(mock_user.id)}
assert result == {
"message": "Garmin Connect authentication successful",
"user_id": str(mock_user.id),
}
@pytest.mark.asyncio
async def test_authenticate_garmin_connect_existing_user_success(
@@ -79,7 +86,11 @@ async def test_authenticate_garmin_connect_existing_user_success(
auth_service.central_db.create_token.assert_called_once()
auth_service.central_db.update_token.assert_not_called()
assert result == {"message": "Garmin Connect authentication successful", "user_id": str(mock_user.id)}
assert result == {
"message": "Garmin Connect authentication successful",
"user_id": str(mock_user.id),
}
@pytest.mark.asyncio
async def test_authenticate_garmin_connect_existing_user_existing_token_success(
@@ -89,9 +100,12 @@ async def test_authenticate_garmin_connect_existing_user_existing_token_success(
email = "existing_user_token@example.com"
password = "password123"
mock_user = User(id=uuid.uuid4(), name=email, email=email)
mock_user_id = mock_user.id # Capture the generated UUID
mock_user_id = mock_user.id # Capture the generated UUID
mock_existing_token = TokenCreate(
access_token="old_access", refresh_token="old_refresh", expires_at=1111111111, user_id=mock_user_id
access_token="old_access",
refresh_token="old_refresh",
expires_at=1111111111,
user_id=mock_user_id,
)
auth_service.central_db.get_user_by_email.return_value = mock_user
@@ -106,9 +120,16 @@ async def test_authenticate_garmin_connect_existing_user_existing_token_success(
auth_service.central_db.update_token.assert_called_once()
auth_service.central_db.create_token.assert_not_called()
assert result == {"message": "Garmin Connect authentication successful", "user_id": str(mock_user.id)}
assert result == {
"message": "Garmin Connect authentication successful",
"user_id": str(mock_user.id),
}
@pytest.mark.asyncio
async def test_authenticate_garmin_connect_garmin_failure(auth_service, mock_garth_login):
async def test_authenticate_garmin_connect_garmin_failure(
auth_service, mock_garth_login
):
"""Test Garmin authentication failure."""
email = "fail_garmin@example.com"
password = "password123"
@@ -125,6 +146,7 @@ async def test_authenticate_garmin_connect_garmin_failure(auth_service, mock_gar
assert result is None
@pytest.mark.asyncio
async def test_authenticate_garmin_connect_central_db_user_creation_failure(
auth_service, mock_garth_login, mock_garth_client

49
backend/tests/unit/test_garmin_auth_service.py
View File

@@ -2,25 +2,33 @@ from datetime import datetime, timedelta
from unittest.mock import AsyncMock, patch
import pytest
from backend.src.schemas import GarminCredentials
from backend.src.services.garmin_auth_service import GarminAuthService
from src.schemas import GarminCredentials
from src.services.garmin_auth_service import GarminAuthService
@pytest.fixture
def garmin_auth_service():
return GarminAuthService()
@pytest.mark.asyncio
async def test_initial_login_success(garmin_auth_service):
username = "test@example.com"
password = "password123"
with patch('backend.src.services.garmin_auth_service.garth') as mock_garth:
with patch("src.services.garmin_auth_service.garth") as mock_garth:
mock_garth.Client.return_value = AsyncMock()
mock_garth.Client.return_value.login.return_value = None # garth.login doesn't return anything directly
mock_garth.Client.return_value.login.return_value = (
None # garth.login doesn't return anything directly
)
# Mock the attributes that would be set on the client after login
mock_garth.Client.return_value.access_token = f"mock_access_token_for_{username}"
mock_garth.Client.return_value.access_token_secret = f"mock_access_token_secret_for_{username}"
mock_garth.Client.return_value.access_token = (
f"mock_access_token_for_{username}"
)
mock_garth.Client.return_value.access_token_secret = (
f"mock_access_token_secret_for_{username}"
)
mock_garth.Client.return_value.expires_in = 300
credentials = await garmin_auth_service.initial_login(username, password)
@@ -33,19 +41,23 @@ async def test_initial_login_success(garmin_auth_service):
assert isinstance(credentials.token_expiration_date, datetime)
assert credentials.token_expiration_date > datetime.utcnow()
@pytest.mark.asyncio
async def test_initial_login_failure(garmin_auth_service):
username = "invalid@example.com"
password = "wrongpassword"
with patch('backend.src.services.garmin_auth_service.garth') as mock_garth:
with patch("backend.src.services.garmin_auth_service.garth") as mock_garth:
mock_garth.Client.return_value = AsyncMock()
mock_garth.Client.return_value.login.side_effect = Exception("Garmin login failed")
mock_garth.Client.return_value.login.side_effect = Exception(
"Garmin login failed"
)
credentials = await garmin_auth_service.initial_login(username, password)
assert credentials is None
@pytest.mark.asyncio
async def test_refresh_tokens_success(garmin_auth_service):
credentials = GarminCredentials(
@@ -53,14 +65,16 @@ async def test_refresh_tokens_success(garmin_auth_service):
garmin_password_plaintext="password123",
access_token="old_access_token",
access_token_secret="old_access_token_secret",
token_expiration_date=datetime.utcnow() - timedelta(minutes=1) # Expired token
token_expiration_date=datetime.utcnow() - timedelta(minutes=1), # Expired token
)
with patch('backend.src.services.garmin_auth_service.garth') as mock_garth:
with patch("backend.src.services.garmin_auth_service.garth") as mock_garth:
mock_garth.Client.return_value = AsyncMock()
mock_garth.Client.return_value.reauthorize.return_value = None
mock_garth.Client.return_value.access_token = "refreshed_access_token"
mock_garth.Client.return_value.access_token_secret = "refreshed_access_token_secret"
mock_garth.Client.return_value.access_token_secret = (
"refreshed_access_token_secret"
)
mock_garth.Client.return_value.expires_in = 300
refreshed_credentials = await garmin_auth_service.refresh_tokens(credentials)
@@ -68,10 +82,13 @@ async def test_refresh_tokens_success(garmin_auth_service):
assert refreshed_credentials is not None
assert refreshed_credentials.garmin_username == credentials.garmin_username
assert refreshed_credentials.access_token == "refreshed_access_token"
assert refreshed_credentials.access_token_secret == "refreshed_access_token_secret"
assert (
refreshed_credentials.access_token_secret == "refreshed_access_token_secret"
)
assert isinstance(refreshed_credentials.token_expiration_date, datetime)
assert refreshed_credentials.token_expiration_date > datetime.utcnow()
@pytest.mark.asyncio
async def test_refresh_tokens_failure(garmin_auth_service):
credentials = GarminCredentials(
@@ -79,12 +96,14 @@ async def test_refresh_tokens_failure(garmin_auth_service):
garmin_password_plaintext="invalid_password",
access_token="old_access_token",
access_token_secret="old_access_token_secret",
token_expiration_date=datetime.utcnow() - timedelta(minutes=1)
token_expiration_date=datetime.utcnow() - timedelta(minutes=1),
)
with patch('backend.src.services.garmin_auth_service.garth') as mock_garth:
with patch("backend.src.services.garmin_auth_service.garth") as mock_garth:
mock_garth.Client.return_value = AsyncMock()
mock_garth.Client.return_value.reauthorize.side_effect = Exception("Garmin reauthorize failed")
mock_garth.Client.return_value.reauthorize.side_effect = Exception(
"Garmin reauthorize failed"
)
refreshed_credentials = await garmin_auth_service.refresh_tokens(credentials)

16
backend/tests/unit/test_rate_limiter.py
View File

@@ -18,6 +18,7 @@ async def test_rate_limiter_allows_requests_within_limit():
except HTTPException:
pytest.fail("HTTPException raised unexpectedly.")
@pytest.mark.asyncio
async def test_rate_limiter_raises_exception_when_exceeded():
"""Test that the rate limiter raises an HTTPException when the rate limit is exceeded."""
@@ -25,15 +26,20 @@ async def test_rate_limiter_raises_exception_when_exceeded():
mock_request = MagicMock()
# Mock the limiter.test method
with patch.object(rate_limiter.limiter, 'test') as mock_limiter_test:
mock_limiter_test.side_effect = [True, False] # First call returns True, second returns False
with patch.object(rate_limiter.limiter, "test") as mock_limiter_test:
mock_limiter_test.side_effect = [
True,
False,
] # First call returns True, second returns False
await rate_limiter(mock_request) # First call, should pass
await rate_limiter(mock_request) # First call, should pass
with pytest.raises(HTTPException) as exc_info:
await rate_limiter(mock_request) # Second call, should fail
await rate_limiter(mock_request) # Second call, should fail
assert exc_info.value.status_code == 429
mock_limiter_test.assert_called_with(rate_limiter.rate_limit_item, "single_user_system")
mock_limiter_test.assert_called_with(
rate_limiter.rate_limit_item, "single_user_system"
)
assert exc_info.value.status_code == 429

27
backend/tests/unit/test_sync_job.py Normal file
View File

@@ -0,0 +1,27 @@
from datetime import datetime
from backend.src.models.sync_job import SyncJob
def test_sync_job_defaults():
job = SyncJob()
assert job.status == "pending"
assert job.progress == 0.0
assert job.start_time is None
assert job.end_time is None
assert job.error_message is None
assert job.job_type is None
def test_sync_job_with_values():
start_time = datetime.now()
job = SyncJob(
status="in_progress",
progress=0.5,
start_time=start_time,
job_type="activities",
)
assert job.status == "in_progress"
assert job.progress == 0.5
assert job.start_time == start_time
assert job.job_type == "activities"

46
backend/tests/unit/test_sync_manager.py Normal file
View File

@@ -0,0 +1,46 @@
import pytest
from backend.src.services.sync_manager import CurrentSyncJobManager
@pytest.mark.asyncio
async def test_singleton():
manager1 = CurrentSyncJobManager()
manager2 = CurrentSyncJobManager()
assert manager1 is manager2
@pytest.mark.asyncio
async def test_start_sync():
manager = CurrentSyncJobManager()
await manager.start_sync("activities")
status = await manager.get_current_sync_status()
assert status.status == "in_progress"
assert status.job_type == "activities"
@pytest.mark.asyncio
async def test_start_sync_while_active():
manager = CurrentSyncJobManager()
await manager.start_sync("activities")
with pytest.raises(RuntimeError):
await manager.start_sync("workouts")
@pytest.mark.asyncio
async def test_complete_sync():
manager = CurrentSyncJobManager()
await manager.start_sync("activities")
await manager.complete_sync()
status = await manager.get_current_sync_status()
assert status.status == "completed"
assert status.progress == 1.0
@pytest.mark.asyncio
async def test_fail_sync():
manager = CurrentSyncJobManager()
await manager.start_sync("activities")
await manager.fail_sync("Test error")
status = await manager.get_current_sync_status()
assert status.status == "failed"
assert status.error_message == "Test error"

36
backend/tests/unit/test_sync_status_service.py
View File

@@ -1,36 +0,0 @@
import uuid
from datetime import datetime
from unittest.mock import MagicMock
import pytest
from src.jobs import JobStore, SyncJob
from src.services.sync_status_service import SyncStatusService
@pytest.fixture
def mock_job_store():
"""Fixture to create a mock JobStore."""
job_store = MagicMock(spec=JobStore)
job_id = uuid.uuid4()
job = SyncJob(id=str(job_id), status="completed", created_at=datetime.utcnow())
job_store.get_all_jobs.return_value = [job]
job_store.get_job.return_value = job
return job_store
def test_get_sync_jobs_all(mock_job_store):
"""Test retrieving all sync jobs."""
service = SyncStatusService(job_store=mock_job_store)
jobs = service.get_sync_jobs()
assert len(jobs) == 1
mock_job_store.get_all_jobs.assert_called_once()
def test_get_sync_job_by_id(mock_job_store):
"""Test retrieving a single sync job by ID."""
service = SyncStatusService(job_store=mock_job_store)
job_id = mock_job_store.get_job.return_value.id
# The get_sync_jobs implementation filters all jobs, so we need to mock get_all_jobs
mock_job_store.get_all_jobs.return_value = [mock_job_store.get_job.return_value]
jobs = service.get_sync_jobs(job_id=job_id)
assert len(jobs) == 1
assert jobs[0].id == str(job_id)