654 lines
20 KiB
Python
654 lines
20 KiB
Python
"""Tests for favorite repository."""
|
|
|
|
from collections.abc import AsyncGenerator
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
from app.models.playlist import Playlist
|
|
from app.models.sound import Sound
|
|
from app.models.user import User
|
|
from app.repositories.favorite import FavoriteRepository
|
|
|
|
|
|
class TestFavoriteRepository:
|
|
"""Test favorite repository operations."""
|
|
|
|
@pytest_asyncio.fixture
|
|
async def favorite_repository(
|
|
self,
|
|
test_session: AsyncSession,
|
|
) -> AsyncGenerator[FavoriteRepository]:
|
|
"""Create a favorite repository instance."""
|
|
yield FavoriteRepository(test_session)
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_sound(
|
|
self,
|
|
test_session: AsyncSession,
|
|
) -> Sound:
|
|
"""Create a test sound."""
|
|
sound = Sound(
|
|
filename="test_sound.mp3",
|
|
name="Test Sound",
|
|
type="SDB",
|
|
duration=5000,
|
|
size=1024000,
|
|
hash="abcdef123456789",
|
|
)
|
|
test_session.add(sound)
|
|
await test_session.commit()
|
|
await test_session.refresh(sound)
|
|
return sound
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_sound2(
|
|
self,
|
|
test_session: AsyncSession,
|
|
) -> Sound:
|
|
"""Create a second test sound."""
|
|
sound = Sound(
|
|
filename="test_sound2.mp3",
|
|
name="Test Sound 2",
|
|
type="TTS",
|
|
duration=3000,
|
|
size=512000,
|
|
hash="xyz789456123abc",
|
|
)
|
|
test_session.add(sound)
|
|
await test_session.commit()
|
|
await test_session.refresh(sound)
|
|
return sound
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_playlist(
|
|
self,
|
|
test_session: AsyncSession,
|
|
) -> Playlist:
|
|
"""Create a test playlist."""
|
|
playlist = Playlist(
|
|
user_id=1, # Use hardcoded ID
|
|
name="Test Playlist",
|
|
description="A test playlist",
|
|
genre="test",
|
|
is_main=False,
|
|
is_current=False,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(playlist)
|
|
return playlist
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_playlist2(
|
|
self,
|
|
test_session: AsyncSession,
|
|
) -> Playlist:
|
|
"""Create a second test playlist."""
|
|
playlist = Playlist(
|
|
user_id=1, # Use hardcoded ID
|
|
name="Test Playlist 2",
|
|
description="Another test playlist",
|
|
genre="test2",
|
|
is_main=False,
|
|
is_current=False,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(playlist)
|
|
return playlist
|
|
|
|
@pytest_asyncio.fixture
|
|
async def other_user(
|
|
self,
|
|
test_session: AsyncSession,
|
|
) -> User:
|
|
"""Create another test user."""
|
|
from app.utils.auth import PasswordUtils
|
|
|
|
user = User(
|
|
email="other@example.com",
|
|
name="Other User",
|
|
password_hash=PasswordUtils.hash_password("otherpassword123"),
|
|
role="user",
|
|
is_active=True,
|
|
plan_id=1, # Use hardcoded plan ID
|
|
credits=100,
|
|
)
|
|
test_session.add(user)
|
|
await test_session.commit()
|
|
await test_session.refresh(user)
|
|
return user
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_sound_favorite(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test creating a sound favorite."""
|
|
# Use hardcoded IDs to avoid session issues
|
|
favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
|
|
favorite = await favorite_repository.create(favorite_data)
|
|
|
|
assert favorite.id is not None
|
|
assert favorite.user_id == 1
|
|
assert favorite.sound_id == 1
|
|
assert favorite.playlist_id is None
|
|
assert favorite.created_at is not None
|
|
assert favorite.updated_at is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_playlist_favorite(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test creating a playlist favorite."""
|
|
favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": None,
|
|
"playlist_id": 1,
|
|
}
|
|
|
|
favorite = await favorite_repository.create(favorite_data)
|
|
|
|
assert favorite.id is not None
|
|
assert favorite.user_id == 1
|
|
assert favorite.sound_id is None
|
|
assert favorite.playlist_id == 1
|
|
assert favorite.created_at is not None
|
|
assert favorite.updated_at is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_id(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test getting a favorite by ID."""
|
|
# Create favorite
|
|
favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
created_favorite = await favorite_repository.create(favorite_data)
|
|
favorite_id = created_favorite.id
|
|
|
|
# Get by ID
|
|
retrieved_favorite = await favorite_repository.get_by_id(favorite_id)
|
|
|
|
assert retrieved_favorite is not None
|
|
assert retrieved_favorite.id == favorite_id
|
|
assert retrieved_favorite.user_id == 1
|
|
assert retrieved_favorite.sound_id == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_id_not_found(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test getting a favorite by non-existent ID."""
|
|
result = await favorite_repository.get_by_id(99999)
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_favorites_empty(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
test_user: User,
|
|
) -> None:
|
|
"""Test getting favorites for user with no favorites."""
|
|
favorites = await favorite_repository.get_user_favorites(test_user.id)
|
|
assert favorites == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_favorites_mixed(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test getting mixed favorites (sounds and playlists)."""
|
|
# Create sound favorite
|
|
sound_favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
await favorite_repository.create(sound_favorite_data)
|
|
|
|
# Create playlist favorite
|
|
playlist_favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": None,
|
|
"playlist_id": 1,
|
|
}
|
|
await favorite_repository.create(playlist_favorite_data)
|
|
|
|
# Get all favorites
|
|
favorites = await favorite_repository.get_user_favorites(1)
|
|
|
|
assert len(favorites) == 2
|
|
|
|
# Should be ordered by created_at desc (most recent first)
|
|
sound_favorite = next((f for f in favorites if f.sound_id is not None), None)
|
|
playlist_favorite = next((f for f in favorites if f.playlist_id is not None), None)
|
|
|
|
assert sound_favorite is not None
|
|
assert playlist_favorite is not None
|
|
assert sound_favorite.sound_id == 1
|
|
assert playlist_favorite.playlist_id == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_favorites_pagination(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test pagination for get user favorites."""
|
|
# Create multiple favorites
|
|
for sound_id in [1, 2]:
|
|
favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": sound_id,
|
|
"playlist_id": None,
|
|
}
|
|
await favorite_repository.create(favorite_data)
|
|
|
|
# Test limit
|
|
favorites = await favorite_repository.get_user_favorites(1, limit=1)
|
|
assert len(favorites) == 1
|
|
|
|
# Test offset
|
|
favorites_page1 = await favorite_repository.get_user_favorites(1, limit=1, offset=0)
|
|
favorites_page2 = await favorite_repository.get_user_favorites(1, limit=1, offset=1)
|
|
|
|
assert len(favorites_page1) == 1
|
|
assert len(favorites_page2) == 1
|
|
assert favorites_page1[0].id != favorites_page2[0].id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_sound_favorites(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test getting only sound favorites."""
|
|
# Create sound favorite
|
|
sound_favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
await favorite_repository.create(sound_favorite_data)
|
|
|
|
# Create playlist favorite
|
|
playlist_favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": None,
|
|
"playlist_id": 1,
|
|
}
|
|
await favorite_repository.create(playlist_favorite_data)
|
|
|
|
# Get only sound favorites
|
|
sound_favorites = await favorite_repository.get_user_sound_favorites(1)
|
|
|
|
assert len(sound_favorites) == 1
|
|
assert sound_favorites[0].sound_id == 1
|
|
assert sound_favorites[0].playlist_id is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_playlist_favorites(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test getting only playlist favorites."""
|
|
# Create sound favorite
|
|
sound_favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
await favorite_repository.create(sound_favorite_data)
|
|
|
|
# Create playlist favorite
|
|
playlist_favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": None,
|
|
"playlist_id": 1,
|
|
}
|
|
await favorite_repository.create(playlist_favorite_data)
|
|
|
|
# Get only playlist favorites
|
|
playlist_favorites = await favorite_repository.get_user_playlist_favorites(1)
|
|
|
|
assert len(playlist_favorites) == 1
|
|
assert playlist_favorites[0].playlist_id == 1
|
|
assert playlist_favorites[0].sound_id is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_user_and_sound_found(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test getting favorite by user and sound when it exists."""
|
|
# Create favorite
|
|
favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
created_favorite = await favorite_repository.create(favorite_data)
|
|
favorite_id = created_favorite.id
|
|
|
|
# Find by user and sound
|
|
found_favorite = await favorite_repository.get_by_user_and_sound(1, 1)
|
|
|
|
assert found_favorite is not None
|
|
assert found_favorite.id == favorite_id
|
|
assert found_favorite.user_id == 1
|
|
assert found_favorite.sound_id == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_user_and_sound_not_found(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test getting favorite by user and sound when it doesn't exist."""
|
|
result = await favorite_repository.get_by_user_and_sound(1, 1)
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_user_and_playlist_found(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test getting favorite by user and playlist when it exists."""
|
|
# Create favorite
|
|
favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": None,
|
|
"playlist_id": 1,
|
|
}
|
|
created_favorite = await favorite_repository.create(favorite_data)
|
|
favorite_id = created_favorite.id
|
|
|
|
# Find by user and playlist
|
|
found_favorite = await favorite_repository.get_by_user_and_playlist(1, 1)
|
|
|
|
assert found_favorite is not None
|
|
assert found_favorite.id == favorite_id
|
|
assert found_favorite.user_id == 1
|
|
assert found_favorite.playlist_id == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_user_and_playlist_not_found(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test getting favorite by user and playlist when it doesn't exist."""
|
|
result = await favorite_repository.get_by_user_and_playlist(1, 1)
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_sound_favorited_true(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test checking if sound is favorited when it is."""
|
|
# Create favorite
|
|
favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
await favorite_repository.create(favorite_data)
|
|
|
|
result = await favorite_repository.is_sound_favorited(1, 1)
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_sound_favorited_false(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test checking if sound is favorited when it's not."""
|
|
result = await favorite_repository.is_sound_favorited(1, 1)
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_playlist_favorited_true(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test checking if playlist is favorited when it is."""
|
|
# Create favorite
|
|
favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": None,
|
|
"playlist_id": 1,
|
|
}
|
|
await favorite_repository.create(favorite_data)
|
|
|
|
result = await favorite_repository.is_playlist_favorited(1, 1)
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_playlist_favorited_false(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test checking if playlist is favorited when it's not."""
|
|
result = await favorite_repository.is_playlist_favorited(1, 1)
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_count_user_favorites(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test counting user favorites."""
|
|
# Initially no favorites
|
|
count = await favorite_repository.count_user_favorites(1)
|
|
assert count == 0
|
|
|
|
# Create sound favorite
|
|
sound_favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
await favorite_repository.create(sound_favorite_data)
|
|
|
|
count = await favorite_repository.count_user_favorites(1)
|
|
assert count == 1
|
|
|
|
# Create playlist favorite
|
|
playlist_favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": None,
|
|
"playlist_id": 1,
|
|
}
|
|
await favorite_repository.create(playlist_favorite_data)
|
|
|
|
count = await favorite_repository.count_user_favorites(1)
|
|
assert count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_count_sound_favorites(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test counting favorites for a sound."""
|
|
# Initially no favorites
|
|
count = await favorite_repository.count_sound_favorites(1)
|
|
assert count == 0
|
|
|
|
# Create favorite by first user
|
|
favorite_data1 = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
await favorite_repository.create(favorite_data1)
|
|
|
|
count = await favorite_repository.count_sound_favorites(1)
|
|
assert count == 1
|
|
|
|
# Create favorite by second user
|
|
favorite_data2 = {
|
|
"user_id": 2,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
await favorite_repository.create(favorite_data2)
|
|
|
|
count = await favorite_repository.count_sound_favorites(1)
|
|
assert count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_count_playlist_favorites(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test counting favorites for a playlist."""
|
|
# Initially no favorites
|
|
count = await favorite_repository.count_playlist_favorites(1)
|
|
assert count == 0
|
|
|
|
# Create favorite by first user
|
|
favorite_data1 = {
|
|
"user_id": 1,
|
|
"sound_id": None,
|
|
"playlist_id": 1,
|
|
}
|
|
await favorite_repository.create(favorite_data1)
|
|
|
|
count = await favorite_repository.count_playlist_favorites(1)
|
|
assert count == 1
|
|
|
|
# Create favorite by second user
|
|
favorite_data2 = {
|
|
"user_id": 2,
|
|
"sound_id": None,
|
|
"playlist_id": 1,
|
|
}
|
|
await favorite_repository.create(favorite_data2)
|
|
|
|
count = await favorite_repository.count_playlist_favorites(1)
|
|
assert count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_favorite(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test updating a favorite."""
|
|
# Create favorite
|
|
favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
created_favorite = await favorite_repository.create(favorite_data)
|
|
favorite_id = created_favorite.id
|
|
original_updated_at = created_favorite.updated_at
|
|
|
|
# Update (note: there's not much to update in a favorite, but we can test the mechanism)
|
|
updated_favorite = await favorite_repository.update(created_favorite, {})
|
|
|
|
assert updated_favorite.id == favorite_id
|
|
assert updated_favorite.updated_at >= original_updated_at
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_favorite(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test deleting a favorite."""
|
|
# Create favorite
|
|
favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
created_favorite = await favorite_repository.create(favorite_data)
|
|
favorite_id = created_favorite.id
|
|
|
|
# Verify it exists
|
|
found_favorite = await favorite_repository.get_by_id(favorite_id)
|
|
assert found_favorite is not None
|
|
|
|
# Delete it
|
|
await favorite_repository.delete(created_favorite)
|
|
|
|
# Verify it's gone
|
|
deleted_favorite = await favorite_repository.get_by_id(favorite_id)
|
|
assert deleted_favorite is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unique_constraints(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test that unique constraints are enforced."""
|
|
# Create sound favorite
|
|
sound_favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
await favorite_repository.create(sound_favorite_data)
|
|
|
|
# Try to create duplicate sound favorite - should raise exception
|
|
from sqlalchemy.exc import IntegrityError
|
|
with pytest.raises(IntegrityError):
|
|
await favorite_repository.create(sound_favorite_data)
|
|
|
|
# Create playlist favorite
|
|
playlist_favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": None,
|
|
"playlist_id": 1,
|
|
}
|
|
await favorite_repository.create(playlist_favorite_data)
|
|
|
|
# Try to create duplicate playlist favorite - should raise exception
|
|
with pytest.raises(IntegrityError):
|
|
await favorite_repository.create(playlist_favorite_data)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_isolation(
|
|
self,
|
|
favorite_repository: FavoriteRepository,
|
|
) -> None:
|
|
"""Test that favorites are properly isolated by user."""
|
|
# Clean up any existing favorites for user 1 first
|
|
existing_favorites = await favorite_repository.get_user_favorites(1)
|
|
for favorite in existing_favorites:
|
|
await favorite_repository.delete(favorite)
|
|
|
|
# Create favorite for user 1
|
|
favorite_data = {
|
|
"user_id": 1,
|
|
"sound_id": 1,
|
|
"playlist_id": None,
|
|
}
|
|
await favorite_repository.create(favorite_data)
|
|
|
|
# Check that user 1 has the favorite
|
|
user1_favorites = await favorite_repository.get_user_favorites(1)
|
|
assert len(user1_favorites) == 1
|
|
|
|
# Check that user 2 doesn't have any favorites
|
|
user2_favorites = await favorite_repository.get_user_favorites(2)
|
|
assert len(user2_favorites) == 0
|
|
|
|
# Check that the sound is favorited by user 1 but not user 2
|
|
assert await favorite_repository.is_sound_favorited(1, 1) is True
|
|
assert await favorite_repository.is_sound_favorited(2, 1) is False
|
|
|