mirror of
https://github.com/sstent/FitTrack_GarminSync.git
synced 2026-01-25 16:41:41 +00:00
sstent
ca9d7d9e90
- Ensure code aligns with CentralDB models - Document code alignment with CentralDB models - Remove informal reference documents (data-model.md, DB_API_SPEC.json, GARMINSYNC_SPEC.md) - Run linters and formatters (black, isort, mypy) - Update project configuration files - Add .dockerignore for Docker builds - Perform code formatting and import sorting - Fix type checking issues - Update documentation files - Complete implementation tasks as per spec
174 lines
6.4 KiB
Python
174 lines
6.4 KiB
Python
import uuid
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from backend.src.models.central_db_models import User
|
|
from backend.src.schemas import TokenCreate
|
|
from backend.src.services.auth_service import AuthService
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_service():
|
|
"""Fixture for AuthService with mocked CentralDBService."""
|
|
with patch(
|
|
"backend.src.services.central_db_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()
|
|
mock_central_db_instance.get_token = AsyncMock()
|
|
mock_central_db_instance.create_token = AsyncMock()
|
|
mock_central_db_instance.update_token = AsyncMock()
|
|
service = AuthService()
|
|
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:
|
|
yield mock_login
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_garth_client():
|
|
"""Fixture to mock garth.client attributes."""
|
|
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
|
|
):
|
|
"""Test successful Garmin authentication with a new user."""
|
|
email = "new_user@example.com"
|
|
password = "password123"
|
|
mock_user = User(id=uuid.uuid4(), name=email, email=email)
|
|
|
|
auth_service.central_db.get_user_by_email.return_value = None
|
|
auth_service.central_db.create_user.return_value = mock_user
|
|
auth_service.central_db.get_token.return_value = None
|
|
|
|
result = await auth_service.authenticate_garmin_connect(email, password)
|
|
|
|
mock_garth_login.assert_called_once_with(email, password)
|
|
auth_service.central_db.get_user_by_email.assert_called_once_with(email=email)
|
|
auth_service.central_db.create_user.assert_called_once()
|
|
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),
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authenticate_garmin_connect_existing_user_success(
|
|
auth_service, mock_garth_login, mock_garth_client
|
|
):
|
|
"""Test successful Garmin authentication with an existing user and no existing token."""
|
|
email = "existing_user@example.com"
|
|
password = "password123"
|
|
mock_user = User(id=uuid.uuid4(), name=email, email=email)
|
|
|
|
auth_service.central_db.get_user_by_email.return_value = mock_user
|
|
auth_service.central_db.get_token.return_value = None
|
|
|
|
result = await auth_service.authenticate_garmin_connect(email, password)
|
|
|
|
mock_garth_login.assert_called_once_with(email, password)
|
|
auth_service.central_db.get_user_by_email.assert_called_once_with(email=email)
|
|
auth_service.central_db.create_user.assert_not_called()
|
|
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),
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authenticate_garmin_connect_existing_user_existing_token_success(
|
|
auth_service, mock_garth_login, mock_garth_client
|
|
):
|
|
"""Test successful Garmin authentication with an existing user and existing token."""
|
|
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_existing_token = TokenCreate(
|
|
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
|
|
auth_service.central_db.get_token.return_value = mock_existing_token
|
|
|
|
result = await auth_service.authenticate_garmin_connect(email, password)
|
|
|
|
mock_garth_login.assert_called_once_with(email, password)
|
|
auth_service.central_db.get_user_by_email.assert_called_once_with(email=email)
|
|
auth_service.central_db.create_user.assert_not_called()
|
|
auth_service.central_db.get_token.assert_called_once_with(user_id=mock_user_id)
|
|
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),
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authenticate_garmin_connect_garmin_failure(
|
|
auth_service, mock_garth_login
|
|
):
|
|
"""Test Garmin authentication failure."""
|
|
email = "fail_garmin@example.com"
|
|
password = "password123"
|
|
mock_garth_login.side_effect = Exception("Garmin login failed")
|
|
|
|
result = await auth_service.authenticate_garmin_connect(email, password)
|
|
|
|
mock_garth_login.assert_called_once_with(email, password)
|
|
auth_service.central_db.get_user_by_email.assert_not_called()
|
|
auth_service.central_db.create_user.assert_not_called()
|
|
auth_service.central_db.get_token.assert_not_called()
|
|
auth_service.central_db.create_token.assert_not_called()
|
|
auth_service.central_db.update_token.assert_not_called()
|
|
|
|
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
|
|
):
|
|
"""Test CentralDB user creation failure."""
|
|
email = "fail_user_create@example.com"
|
|
password = "password123"
|
|
|
|
auth_service.central_db.get_user_by_email.return_value = None
|
|
auth_service.central_db.create_user.return_value = None
|
|
|
|
result = await auth_service.authenticate_garmin_connect(email, password)
|
|
|
|
mock_garth_login.assert_called_once_with(email, password)
|
|
auth_service.central_db.get_user_by_email.assert_called_once_with(email=email)
|
|
auth_service.central_db.create_user.assert_called_once()
|
|
auth_service.central_db.get_token.assert_not_called()
|
|
auth_service.central_db.create_token.assert_not_called()
|
|
auth_service.central_db.update_token.assert_not_called()
|
|
|
|
assert result is None
|