3ad79a4860
- Add base_url config setting for OAuth callback URLs
- Change OAuth callbacks from relative to absolute URLs
- Add account linking OAuth flow (GET /auth/link/{provider})
- Add unlink endpoint (DELETE /users/me/link/{provider})
- Add AccountLinkingError and service methods for linking
- Add 14 new tests for linking functionality
- Update Phase 2 plan to mark complete (1072 tests passing)
665 lines
22 KiB
Python
665 lines
22 KiB
Python
"""Tests for UserService.
|
|
|
|
Tests the user service CRUD operations and OAuth-based user creation.
|
|
Uses real Postgres via the db_session fixture from conftest.
|
|
"""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from app.db.models import User
|
|
from app.db.models.oauth_account import OAuthLinkedAccount
|
|
from app.schemas.user import OAuthUserInfo, UserCreate, UserUpdate
|
|
from app.services.user_service import AccountLinkingError, user_service
|
|
|
|
# Import db_session fixture from db conftest
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
class TestGetById:
|
|
"""Tests for get_by_id method."""
|
|
|
|
async def test_returns_user_when_found(self, db_session):
|
|
"""Test that get_by_id returns user when it exists.
|
|
|
|
Creates a user and verifies it can be retrieved by ID.
|
|
"""
|
|
# Create user directly
|
|
user = User(
|
|
email="test@example.com",
|
|
display_name="Test User",
|
|
oauth_provider="google",
|
|
oauth_id="123456",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
# Retrieve by ID
|
|
from uuid import UUID
|
|
|
|
user_id = UUID(user.id) if isinstance(user.id, str) else user.id
|
|
result = await user_service.get_by_id(db_session, user_id)
|
|
|
|
assert result is not None
|
|
assert result.email == "test@example.com"
|
|
|
|
async def test_returns_none_when_not_found(self, db_session):
|
|
"""Test that get_by_id returns None for nonexistent users."""
|
|
from uuid import uuid4
|
|
|
|
result = await user_service.get_by_id(db_session, uuid4())
|
|
assert result is None
|
|
|
|
|
|
class TestGetByEmail:
|
|
"""Tests for get_by_email method."""
|
|
|
|
async def test_returns_user_when_found(self, db_session):
|
|
"""Test that get_by_email returns user when it exists."""
|
|
user = User(
|
|
email="findme@example.com",
|
|
display_name="Find Me",
|
|
oauth_provider="discord",
|
|
oauth_id="discord123",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
result = await user_service.get_by_email(db_session, "findme@example.com")
|
|
|
|
assert result is not None
|
|
assert result.display_name == "Find Me"
|
|
|
|
async def test_returns_none_when_not_found(self, db_session):
|
|
"""Test that get_by_email returns None for nonexistent emails."""
|
|
result = await user_service.get_by_email(db_session, "nobody@example.com")
|
|
assert result is None
|
|
|
|
|
|
class TestGetByOAuth:
|
|
"""Tests for get_by_oauth method."""
|
|
|
|
async def test_returns_user_when_found(self, db_session):
|
|
"""Test that get_by_oauth returns user for matching provider+id."""
|
|
user = User(
|
|
email="oauth@example.com",
|
|
display_name="OAuth User",
|
|
oauth_provider="google",
|
|
oauth_id="google-unique-id",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
result = await user_service.get_by_oauth(db_session, "google", "google-unique-id")
|
|
|
|
assert result is not None
|
|
assert result.email == "oauth@example.com"
|
|
|
|
async def test_returns_none_for_wrong_provider(self, db_session):
|
|
"""Test that get_by_oauth returns None if provider doesn't match."""
|
|
user = User(
|
|
email="oauth2@example.com",
|
|
display_name="OAuth User 2",
|
|
oauth_provider="google",
|
|
oauth_id="google-id-2",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
# Same ID, different provider
|
|
result = await user_service.get_by_oauth(db_session, "discord", "google-id-2")
|
|
assert result is None
|
|
|
|
async def test_returns_none_when_not_found(self, db_session):
|
|
"""Test that get_by_oauth returns None for nonexistent OAuth."""
|
|
result = await user_service.get_by_oauth(db_session, "google", "nonexistent")
|
|
assert result is None
|
|
|
|
|
|
class TestCreate:
|
|
"""Tests for create method."""
|
|
|
|
async def test_creates_user_with_all_fields(self, db_session):
|
|
"""Test that create properly persists all user fields."""
|
|
user_data = UserCreate(
|
|
email="new@example.com",
|
|
display_name="New User",
|
|
avatar_url="https://example.com/avatar.jpg",
|
|
oauth_provider="discord",
|
|
oauth_id="discord-new-id",
|
|
)
|
|
|
|
result = await user_service.create(db_session, user_data)
|
|
|
|
assert result.id is not None
|
|
assert result.email == "new@example.com"
|
|
assert result.display_name == "New User"
|
|
assert result.avatar_url == "https://example.com/avatar.jpg"
|
|
assert result.oauth_provider == "discord"
|
|
assert result.oauth_id == "discord-new-id"
|
|
assert result.is_premium is False
|
|
assert result.premium_until is None
|
|
|
|
async def test_creates_user_without_avatar(self, db_session):
|
|
"""Test that create works without optional avatar_url."""
|
|
user_data = UserCreate(
|
|
email="noavatar@example.com",
|
|
display_name="No Avatar",
|
|
oauth_provider="google",
|
|
oauth_id="google-no-avatar",
|
|
)
|
|
|
|
result = await user_service.create(db_session, user_data)
|
|
|
|
assert result.avatar_url is None
|
|
|
|
|
|
class TestCreateFromOAuth:
|
|
"""Tests for create_from_oauth method."""
|
|
|
|
async def test_creates_user_from_oauth_info(self, db_session):
|
|
"""Test that create_from_oauth converts OAuthUserInfo to User."""
|
|
oauth_info = OAuthUserInfo(
|
|
provider="google",
|
|
oauth_id="google-oauth-123",
|
|
email="oauthcreate@example.com",
|
|
name="OAuth Created User",
|
|
avatar_url="https://google.com/avatar.jpg",
|
|
)
|
|
|
|
result = await user_service.create_from_oauth(db_session, oauth_info)
|
|
|
|
assert result.email == "oauthcreate@example.com"
|
|
assert result.display_name == "OAuth Created User"
|
|
assert result.oauth_provider == "google"
|
|
assert result.oauth_id == "google-oauth-123"
|
|
|
|
|
|
class TestGetOrCreateFromOAuth:
|
|
"""Tests for get_or_create_from_oauth method."""
|
|
|
|
async def test_returns_existing_user_by_oauth(self, db_session):
|
|
"""Test that existing user is returned when OAuth matches.
|
|
|
|
Verifies the method returns (user, False) for existing users.
|
|
"""
|
|
# Create existing user
|
|
existing = User(
|
|
email="existing@example.com",
|
|
display_name="Existing",
|
|
oauth_provider="google",
|
|
oauth_id="existing-oauth-id",
|
|
)
|
|
db_session.add(existing)
|
|
await db_session.commit()
|
|
|
|
# Try to get or create with same OAuth
|
|
oauth_info = OAuthUserInfo(
|
|
provider="google",
|
|
oauth_id="existing-oauth-id",
|
|
email="existing@example.com",
|
|
name="Existing",
|
|
)
|
|
|
|
result, created = await user_service.get_or_create_from_oauth(db_session, oauth_info)
|
|
|
|
assert created is False
|
|
assert result.id == existing.id
|
|
|
|
async def test_links_existing_user_by_email(self, db_session):
|
|
"""Test that OAuth is linked when email matches existing user.
|
|
|
|
If a user exists with the same email but different OAuth,
|
|
the new OAuth should be linked to the existing account.
|
|
"""
|
|
# Create user with Google
|
|
existing = User(
|
|
email="link@example.com",
|
|
display_name="Link Me",
|
|
oauth_provider="google",
|
|
oauth_id="google-link-id",
|
|
)
|
|
db_session.add(existing)
|
|
await db_session.commit()
|
|
|
|
# Login with Discord (same email)
|
|
oauth_info = OAuthUserInfo(
|
|
provider="discord",
|
|
oauth_id="discord-link-id",
|
|
email="link@example.com",
|
|
name="Link Me",
|
|
avatar_url="https://discord.com/avatar.jpg",
|
|
)
|
|
|
|
result, created = await user_service.get_or_create_from_oauth(db_session, oauth_info)
|
|
|
|
assert created is False
|
|
assert result.id == existing.id
|
|
# OAuth should be updated to Discord
|
|
assert result.oauth_provider == "discord"
|
|
assert result.oauth_id == "discord-link-id"
|
|
|
|
async def test_creates_new_user_when_not_found(self, db_session):
|
|
"""Test that new user is created when no match exists.
|
|
|
|
Verifies the method returns (user, True) for new users.
|
|
"""
|
|
oauth_info = OAuthUserInfo(
|
|
provider="discord",
|
|
oauth_id="brand-new-id",
|
|
email="brandnew@example.com",
|
|
name="Brand New",
|
|
)
|
|
|
|
result, created = await user_service.get_or_create_from_oauth(db_session, oauth_info)
|
|
|
|
assert created is True
|
|
assert result.email == "brandnew@example.com"
|
|
|
|
|
|
class TestUpdate:
|
|
"""Tests for update method."""
|
|
|
|
async def test_updates_display_name(self, db_session):
|
|
"""Test that update changes display_name when provided."""
|
|
user = User(
|
|
email="update@example.com",
|
|
display_name="Old Name",
|
|
oauth_provider="google",
|
|
oauth_id="update-id",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
update_data = UserUpdate(display_name="New Name")
|
|
result = await user_service.update(db_session, user, update_data)
|
|
|
|
assert result.display_name == "New Name"
|
|
|
|
async def test_updates_avatar_url(self, db_session):
|
|
"""Test that update changes avatar_url when provided."""
|
|
user = User(
|
|
email="avatar@example.com",
|
|
display_name="Avatar User",
|
|
oauth_provider="google",
|
|
oauth_id="avatar-id",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
update_data = UserUpdate(avatar_url="https://new-avatar.com/img.jpg")
|
|
result = await user_service.update(db_session, user, update_data)
|
|
|
|
assert result.avatar_url == "https://new-avatar.com/img.jpg"
|
|
|
|
async def test_ignores_none_values(self, db_session):
|
|
"""Test that update doesn't change fields set to None.
|
|
|
|
Only explicitly provided fields should be updated.
|
|
"""
|
|
user = User(
|
|
email="keep@example.com",
|
|
display_name="Keep Me",
|
|
avatar_url="https://keep.com/avatar.jpg",
|
|
oauth_provider="google",
|
|
oauth_id="keep-id",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
# Update only display_name, leave avatar alone
|
|
update_data = UserUpdate(display_name="Changed")
|
|
result = await user_service.update(db_session, user, update_data)
|
|
|
|
assert result.display_name == "Changed"
|
|
assert result.avatar_url == "https://keep.com/avatar.jpg"
|
|
|
|
|
|
class TestUpdateLastLogin:
|
|
"""Tests for update_last_login method."""
|
|
|
|
async def test_updates_last_login_timestamp(self, db_session):
|
|
"""Test that update_last_login sets current timestamp."""
|
|
user = User(
|
|
email="login@example.com",
|
|
display_name="Login User",
|
|
oauth_provider="google",
|
|
oauth_id="login-id",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
assert user.last_login is None
|
|
|
|
before = datetime.now(UTC)
|
|
result = await user_service.update_last_login(db_session, user)
|
|
after = datetime.now(UTC)
|
|
|
|
assert result.last_login is not None
|
|
# Allow 1 second tolerance
|
|
assert before - timedelta(seconds=1) <= result.last_login <= after + timedelta(seconds=1)
|
|
|
|
|
|
class TestUpdatePremium:
|
|
"""Tests for update_premium method."""
|
|
|
|
async def test_grants_premium(self, db_session):
|
|
"""Test that update_premium sets premium status and expiration."""
|
|
user = User(
|
|
email="premium@example.com",
|
|
display_name="Premium User",
|
|
oauth_provider="google",
|
|
oauth_id="premium-id",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
assert user.is_premium is False
|
|
|
|
expires = datetime.now(UTC) + timedelta(days=30)
|
|
result = await user_service.update_premium(db_session, user, expires)
|
|
|
|
assert result.is_premium is True
|
|
assert result.premium_until == expires
|
|
|
|
async def test_removes_premium(self, db_session):
|
|
"""Test that update_premium with None removes premium status."""
|
|
user = User(
|
|
email="unpremium@example.com",
|
|
display_name="Unpremium User",
|
|
oauth_provider="google",
|
|
oauth_id="unpremium-id",
|
|
is_premium=True,
|
|
premium_until=datetime.now(UTC) + timedelta(days=30),
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
result = await user_service.update_premium(db_session, user, None)
|
|
|
|
assert result.is_premium is False
|
|
assert result.premium_until is None
|
|
|
|
|
|
class TestDelete:
|
|
"""Tests for delete method."""
|
|
|
|
async def test_deletes_user(self, db_session):
|
|
"""Test that delete removes user from database."""
|
|
user = User(
|
|
email="delete@example.com",
|
|
display_name="Delete Me",
|
|
oauth_provider="google",
|
|
oauth_id="delete-id",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
user_id = user.id
|
|
await user_service.delete(db_session, user)
|
|
|
|
# Verify user is gone
|
|
from uuid import UUID
|
|
|
|
result = await user_service.get_by_id(
|
|
db_session, UUID(user_id) if isinstance(user_id, str) else user_id
|
|
)
|
|
assert result is None
|
|
|
|
|
|
class TestGetLinkedAccount:
|
|
"""Tests for get_linked_account method."""
|
|
|
|
async def test_returns_linked_account_when_found(self, db_session):
|
|
"""Test that get_linked_account returns account when it exists.
|
|
|
|
Creates a user with a linked account and verifies it can be retrieved.
|
|
"""
|
|
# Create user
|
|
user = User(
|
|
email="primary@example.com",
|
|
display_name="Primary User",
|
|
oauth_provider="google",
|
|
oauth_id="google-primary",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
|
|
# Create linked account
|
|
linked = OAuthLinkedAccount(
|
|
user_id=user.id,
|
|
provider="discord",
|
|
oauth_id="discord-linked-123",
|
|
email="linked@example.com",
|
|
)
|
|
db_session.add(linked)
|
|
await db_session.commit()
|
|
|
|
# Retrieve linked account
|
|
result = await user_service.get_linked_account(db_session, "discord", "discord-linked-123")
|
|
|
|
assert result is not None
|
|
assert result.provider == "discord"
|
|
assert result.oauth_id == "discord-linked-123"
|
|
|
|
async def test_returns_none_when_not_found(self, db_session):
|
|
"""Test that get_linked_account returns None for nonexistent accounts."""
|
|
result = await user_service.get_linked_account(db_session, "discord", "nonexistent-id")
|
|
assert result is None
|
|
|
|
|
|
class TestLinkOAuthAccount:
|
|
"""Tests for link_oauth_account method."""
|
|
|
|
async def test_links_new_provider(self, db_session):
|
|
"""Test that link_oauth_account successfully links a new provider.
|
|
|
|
Creates a Google user and links Discord to them.
|
|
"""
|
|
# Create user with Google
|
|
user = User(
|
|
email="google-user@example.com",
|
|
display_name="Google User",
|
|
oauth_provider="google",
|
|
oauth_id="google-123",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
# Link Discord
|
|
discord_info = OAuthUserInfo(
|
|
provider="discord",
|
|
oauth_id="discord-456",
|
|
email="discord@example.com",
|
|
name="Discord Name",
|
|
avatar_url="https://discord.com/avatar.png",
|
|
)
|
|
|
|
result = await user_service.link_oauth_account(db_session, user, discord_info)
|
|
|
|
assert result is not None
|
|
assert result.provider == "discord"
|
|
assert result.oauth_id == "discord-456"
|
|
assert result.email == "discord@example.com"
|
|
assert result.display_name == "Discord Name"
|
|
assert str(result.user_id) == str(user.id)
|
|
|
|
async def test_raises_error_if_already_linked_to_same_user(self, db_session):
|
|
"""Test that linking same provider twice raises error.
|
|
|
|
A user cannot have the same provider linked multiple times.
|
|
"""
|
|
user = User(
|
|
email="double-link@example.com",
|
|
display_name="Double Link",
|
|
oauth_provider="google",
|
|
oauth_id="google-double",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
# Link Discord first time
|
|
discord_info = OAuthUserInfo(
|
|
provider="discord",
|
|
oauth_id="discord-first",
|
|
email="first@discord.com",
|
|
name="First",
|
|
)
|
|
await user_service.link_oauth_account(db_session, user, discord_info)
|
|
await db_session.refresh(user)
|
|
|
|
# Try to link same Discord account again
|
|
with pytest.raises(AccountLinkingError) as exc_info:
|
|
await user_service.link_oauth_account(db_session, user, discord_info)
|
|
|
|
assert "already linked to your account" in str(exc_info.value)
|
|
|
|
async def test_raises_error_if_linked_to_another_user(self, db_session):
|
|
"""Test that linking account already linked to another user raises error.
|
|
|
|
The same OAuth provider+ID cannot be linked to multiple users.
|
|
"""
|
|
# Create first user and link Discord
|
|
user1 = User(
|
|
email="user1@example.com",
|
|
display_name="User 1",
|
|
oauth_provider="google",
|
|
oauth_id="google-user1",
|
|
)
|
|
db_session.add(user1)
|
|
await db_session.commit()
|
|
await db_session.refresh(user1)
|
|
|
|
discord_info = OAuthUserInfo(
|
|
provider="discord",
|
|
oauth_id="shared-discord",
|
|
email="shared@discord.com",
|
|
name="Shared",
|
|
)
|
|
await user_service.link_oauth_account(db_session, user1, discord_info)
|
|
|
|
# Create second user
|
|
user2 = User(
|
|
email="user2@example.com",
|
|
display_name="User 2",
|
|
oauth_provider="google",
|
|
oauth_id="google-user2",
|
|
)
|
|
db_session.add(user2)
|
|
await db_session.commit()
|
|
await db_session.refresh(user2)
|
|
|
|
# Try to link same Discord account to second user
|
|
with pytest.raises(AccountLinkingError) as exc_info:
|
|
await user_service.link_oauth_account(db_session, user2, discord_info)
|
|
|
|
assert "already linked to another user" in str(exc_info.value)
|
|
|
|
async def test_raises_error_if_linking_primary_provider(self, db_session):
|
|
"""Test that linking the same provider as primary raises error.
|
|
|
|
User cannot link Google if they already signed up with Google.
|
|
"""
|
|
user = User(
|
|
email="google-primary@example.com",
|
|
display_name="Google Primary",
|
|
oauth_provider="google",
|
|
oauth_id="google-primary-id",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
# Try to link another Google account
|
|
google_info = OAuthUserInfo(
|
|
provider="google",
|
|
oauth_id="google-different-id",
|
|
email="different@gmail.com",
|
|
name="Different",
|
|
)
|
|
|
|
with pytest.raises(AccountLinkingError) as exc_info:
|
|
await user_service.link_oauth_account(db_session, user, google_info)
|
|
|
|
assert "primary login provider" in str(exc_info.value)
|
|
|
|
|
|
class TestUnlinkOAuthAccount:
|
|
"""Tests for unlink_oauth_account method."""
|
|
|
|
async def test_unlinks_linked_account(self, db_session):
|
|
"""Test that unlink_oauth_account removes a linked account.
|
|
|
|
Links Discord then unlinks it successfully.
|
|
"""
|
|
user = User(
|
|
email="unlink@example.com",
|
|
display_name="Unlink User",
|
|
oauth_provider="google",
|
|
oauth_id="google-unlink",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
# Link Discord
|
|
discord_info = OAuthUserInfo(
|
|
provider="discord",
|
|
oauth_id="discord-unlink",
|
|
email="discord@unlink.com",
|
|
name="Discord Unlink",
|
|
)
|
|
await user_service.link_oauth_account(db_session, user, discord_info)
|
|
await db_session.refresh(user)
|
|
|
|
# Verify linked
|
|
assert len(user.linked_accounts) == 1
|
|
|
|
# Unlink
|
|
result = await user_service.unlink_oauth_account(db_session, user, "discord")
|
|
|
|
assert result is True
|
|
|
|
# Verify unlinked
|
|
linked = await user_service.get_linked_account(db_session, "discord", "discord-unlink")
|
|
assert linked is None
|
|
|
|
async def test_returns_false_if_not_linked(self, db_session):
|
|
"""Test that unlink returns False if provider isn't linked."""
|
|
user = User(
|
|
email="not-linked@example.com",
|
|
display_name="Not Linked",
|
|
oauth_provider="google",
|
|
oauth_id="google-notlinked",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
result = await user_service.unlink_oauth_account(db_session, user, "discord")
|
|
|
|
assert result is False
|
|
|
|
async def test_raises_error_if_unlinking_primary(self, db_session):
|
|
"""Test that unlinking primary provider raises error.
|
|
|
|
User cannot unlink their primary OAuth provider.
|
|
"""
|
|
user = User(
|
|
email="primary-unlink@example.com",
|
|
display_name="Primary Unlink",
|
|
oauth_provider="google",
|
|
oauth_id="google-primary-unlink",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
with pytest.raises(AccountLinkingError) as exc_info:
|
|
await user_service.unlink_oauth_account(db_session, user, "google")
|
|
|
|
assert "primary login provider" in str(exc_info.value)
|