968 lines
30 KiB
Python
968 lines
30 KiB
Python
"""Tests for playlist service."""
|
|
|
|
from collections.abc import AsyncGenerator
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from fastapi import HTTPException
|
|
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.services.playlist import PlaylistService
|
|
|
|
|
|
class TestPlaylistService:
|
|
"""Test playlist service operations."""
|
|
|
|
@pytest_asyncio.fixture
|
|
async def playlist_service(
|
|
self,
|
|
test_session: AsyncSession,
|
|
) -> AsyncGenerator[PlaylistService, None]:
|
|
"""Create a playlist service instance."""
|
|
yield PlaylistService(test_session)
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_playlist(
|
|
self,
|
|
test_session: AsyncSession,
|
|
test_user: User,
|
|
) -> Playlist:
|
|
"""Create a test playlist."""
|
|
# Extract user_id from test_user within the fixture
|
|
user_id = test_user.id
|
|
playlist = Playlist(
|
|
user_id=user_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 current_playlist(
|
|
self,
|
|
test_session: AsyncSession,
|
|
test_user: User,
|
|
) -> Playlist:
|
|
"""Create a current playlist."""
|
|
# Extract user_id from test_user within the fixture
|
|
user_id = test_user.id
|
|
playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Current Playlist",
|
|
description="Currently active playlist",
|
|
is_main=False,
|
|
is_current=True,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(playlist)
|
|
return playlist
|
|
|
|
@pytest_asyncio.fixture
|
|
async def main_playlist(
|
|
self,
|
|
test_session: AsyncSession,
|
|
) -> Playlist:
|
|
"""Create a main playlist."""
|
|
playlist = Playlist(
|
|
user_id=None,
|
|
name="Main Playlist",
|
|
description="Main playlist",
|
|
is_main=True,
|
|
is_current=False,
|
|
is_deletable=False,
|
|
)
|
|
test_session.add(playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(playlist)
|
|
return playlist
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_sound(
|
|
self,
|
|
test_session: AsyncSession,
|
|
) -> Sound:
|
|
"""Create a test sound."""
|
|
sound = Sound(
|
|
name="Test Sound",
|
|
filename="test.mp3",
|
|
type="SDB",
|
|
duration=5000,
|
|
size=1024,
|
|
hash="test_hash",
|
|
play_count=10,
|
|
)
|
|
test_session.add(sound)
|
|
await test_session.commit()
|
|
await test_session.refresh(sound)
|
|
return sound
|
|
|
|
@pytest_asyncio.fixture
|
|
async def other_user(
|
|
self,
|
|
test_session: AsyncSession,
|
|
ensure_plans,
|
|
) -> 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("password123"),
|
|
role="user",
|
|
is_active=True,
|
|
plan_id=ensure_plans[0].id,
|
|
credits=100,
|
|
)
|
|
test_session.add(user)
|
|
await test_session.commit()
|
|
await test_session.refresh(user)
|
|
return user
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_playlist_by_id_success(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_playlist: Playlist,
|
|
) -> None:
|
|
"""Test getting playlist by ID successfully."""
|
|
assert test_playlist.id is not None
|
|
|
|
playlist = await playlist_service.get_playlist_by_id(test_playlist.id)
|
|
|
|
assert playlist.id == test_playlist.id
|
|
assert playlist.name == test_playlist.name
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_playlist_by_id_not_found(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
) -> None:
|
|
"""Test getting non-existent playlist."""
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await playlist_service.get_playlist_by_id(99999)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "not found" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_main_playlist_existing(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test getting existing main playlist."""
|
|
# Create main playlist manually
|
|
main_playlist = Playlist(
|
|
user_id=None,
|
|
name="Main Playlist",
|
|
description="Main playlist",
|
|
is_main=True,
|
|
is_deletable=False,
|
|
)
|
|
test_session.add(main_playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(main_playlist)
|
|
|
|
playlist = await playlist_service.get_main_playlist()
|
|
|
|
assert playlist.id == main_playlist.id
|
|
assert playlist.is_main is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_main_playlist_create_if_not_exists(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
) -> None:
|
|
"""Test that service fails if main playlist doesn't exist."""
|
|
# Should raise an HTTPException if no main playlist exists
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await playlist_service.get_main_playlist()
|
|
assert exc_info.value.status_code == 500
|
|
assert "Main playlist not found" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_playlist_success(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
) -> None:
|
|
"""Test creating a new playlist successfully."""
|
|
user_id = test_user.id # Extract user_id while session is available
|
|
playlist = await playlist_service.create_playlist(
|
|
user_id=user_id,
|
|
name="New Playlist",
|
|
description="A new playlist",
|
|
genre="rock",
|
|
)
|
|
|
|
assert playlist.name == "New Playlist"
|
|
assert playlist.description == "A new playlist"
|
|
assert playlist.genre == "rock"
|
|
assert playlist.user_id == user_id
|
|
assert playlist.is_main is False
|
|
assert playlist.is_current is False
|
|
assert playlist.is_deletable is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_playlist_duplicate_name(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test creating playlist with duplicate name fails."""
|
|
# Create test playlist within this test
|
|
user_id = test_user.id
|
|
playlist = Playlist(
|
|
user_id=user_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)
|
|
|
|
# Extract name before async call
|
|
playlist_name = playlist.name
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await playlist_service.create_playlist(
|
|
user_id=user_id,
|
|
name=playlist_name, # Same name as existing playlist
|
|
)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "already exists" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_playlist_as_current(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test creating a playlist as current unsets previous current."""
|
|
# Create current playlist within this test
|
|
user_id = test_user.id
|
|
current_playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Current Playlist",
|
|
description="Currently active playlist",
|
|
is_main=False,
|
|
is_current=True,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(current_playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(current_playlist)
|
|
|
|
# Verify the existing current playlist
|
|
assert current_playlist.is_current is True
|
|
|
|
# Extract ID before async call
|
|
current_playlist_id = current_playlist.id
|
|
|
|
# Create new playlist as current
|
|
new_playlist = await playlist_service.create_playlist(
|
|
user_id=user_id,
|
|
name="New Current Playlist",
|
|
is_current=True,
|
|
)
|
|
|
|
assert new_playlist.is_current is True
|
|
|
|
# Verify the old current playlist is no longer current
|
|
# We need to refresh the old playlist from the database
|
|
old_playlist = await playlist_service.get_playlist_by_id(current_playlist_id)
|
|
assert old_playlist.is_current is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_playlist_success(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test updating a playlist successfully."""
|
|
# Create test playlist within this test
|
|
user_id = test_user.id
|
|
playlist = Playlist(
|
|
user_id=user_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)
|
|
|
|
# Extract IDs before async call
|
|
playlist_id = playlist.id
|
|
|
|
updated_playlist = await playlist_service.update_playlist(
|
|
playlist_id=playlist_id,
|
|
user_id=user_id,
|
|
name="Updated Name",
|
|
description="Updated description",
|
|
genre="jazz",
|
|
)
|
|
|
|
assert updated_playlist.name == "Updated Name"
|
|
assert updated_playlist.description == "Updated description"
|
|
assert updated_playlist.genre == "jazz"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_playlist_set_current(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test setting a playlist as current via update."""
|
|
# Create test playlist within this test
|
|
user_id = test_user.id
|
|
test_playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Test Playlist",
|
|
description="A test playlist",
|
|
genre="test",
|
|
is_main=False,
|
|
is_current=False,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(test_playlist)
|
|
|
|
current_playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Current Playlist",
|
|
description="Currently active playlist",
|
|
is_main=False,
|
|
is_current=True,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(current_playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(test_playlist)
|
|
await test_session.refresh(current_playlist)
|
|
|
|
# Extract IDs before async calls
|
|
test_playlist_id = test_playlist.id
|
|
current_playlist_id = current_playlist.id
|
|
|
|
# Verify initial state
|
|
assert test_playlist.is_current is False
|
|
assert current_playlist.is_current is True
|
|
|
|
# Update playlist to be current
|
|
updated_playlist = await playlist_service.update_playlist(
|
|
playlist_id=test_playlist_id,
|
|
user_id=user_id,
|
|
is_current=True,
|
|
)
|
|
|
|
assert updated_playlist.is_current is True
|
|
|
|
# Verify old current playlist is no longer current
|
|
old_current = await playlist_service.get_playlist_by_id(current_playlist_id)
|
|
assert old_current.is_current is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_playlist_success(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test deleting a playlist successfully."""
|
|
# Create test playlist within this test
|
|
user_id = test_user.id
|
|
playlist = Playlist(
|
|
user_id=user_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)
|
|
|
|
# Extract ID before async call
|
|
playlist_id = playlist.id
|
|
|
|
await playlist_service.delete_playlist(playlist_id, user_id)
|
|
|
|
# Verify playlist is deleted
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await playlist_service.get_playlist_by_id(playlist_id)
|
|
assert exc_info.value.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_current_playlist_sets_main_as_current(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test deleting current playlist sets main as current."""
|
|
# Create main playlist first (required by service)
|
|
main_playlist = Playlist(
|
|
user_id=None,
|
|
name="Main Playlist",
|
|
description="Main playlist",
|
|
is_main=True,
|
|
is_current=False,
|
|
is_deletable=False,
|
|
)
|
|
test_session.add(main_playlist)
|
|
|
|
# Create current playlist within this test
|
|
user_id = test_user.id
|
|
current_playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Current Playlist",
|
|
description="Currently active playlist",
|
|
is_main=False,
|
|
is_current=True,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(current_playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(current_playlist)
|
|
|
|
# Extract ID before async call
|
|
current_playlist_id = current_playlist.id
|
|
|
|
# Delete the current playlist
|
|
await playlist_service.delete_playlist(current_playlist_id, user_id)
|
|
|
|
# Verify main playlist is now fallback current (main playlist doesn't have is_current=True)
|
|
# The service returns main playlist when no current is set
|
|
current = await playlist_service.get_current_playlist()
|
|
assert current.is_main is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_non_deletable_playlist(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test deleting non-deletable playlist fails."""
|
|
# Extract user ID immediately
|
|
user_id = test_user.id
|
|
|
|
# Create non-deletable playlist
|
|
non_deletable = Playlist(
|
|
user_id=user_id,
|
|
name="Non-deletable",
|
|
is_deletable=False,
|
|
)
|
|
test_session.add(non_deletable)
|
|
await test_session.commit()
|
|
await test_session.refresh(non_deletable)
|
|
|
|
# Extract ID before async call
|
|
non_deletable_id = non_deletable.id
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await playlist_service.delete_playlist(non_deletable_id, user_id)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "cannot be deleted" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_sound_to_playlist_success(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test adding sound to playlist successfully."""
|
|
# Create test playlist and sound within this test
|
|
user_id = test_user.id
|
|
playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Test Playlist",
|
|
description="A test playlist",
|
|
genre="test",
|
|
is_main=False,
|
|
is_current=False,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(playlist)
|
|
|
|
sound = Sound(
|
|
name="Test Sound",
|
|
filename="test.mp3",
|
|
type="SDB",
|
|
duration=5000,
|
|
size=1024,
|
|
hash="test_hash",
|
|
play_count=10,
|
|
)
|
|
test_session.add(sound)
|
|
await test_session.commit()
|
|
await test_session.refresh(playlist)
|
|
await test_session.refresh(sound)
|
|
|
|
# Extract IDs before async calls
|
|
playlist_id = playlist.id
|
|
sound_id = sound.id
|
|
|
|
await playlist_service.add_sound_to_playlist(
|
|
playlist_id=playlist_id,
|
|
sound_id=sound_id,
|
|
user_id=user_id,
|
|
)
|
|
|
|
# Verify sound was added
|
|
sounds = await playlist_service.get_playlist_sounds(playlist_id)
|
|
assert len(sounds) == 1
|
|
assert sounds[0].id == sound_id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_sound_to_playlist_already_exists(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test adding sound that's already in playlist fails."""
|
|
# Create test playlist and sound within this test
|
|
user_id = test_user.id
|
|
playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Test Playlist",
|
|
description="A test playlist",
|
|
genre="test",
|
|
is_main=False,
|
|
is_current=False,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(playlist)
|
|
|
|
sound = Sound(
|
|
name="Test Sound",
|
|
filename="test.mp3",
|
|
type="SDB",
|
|
duration=5000,
|
|
size=1024,
|
|
hash="test_hash",
|
|
play_count=10,
|
|
)
|
|
test_session.add(sound)
|
|
await test_session.commit()
|
|
await test_session.refresh(playlist)
|
|
await test_session.refresh(sound)
|
|
|
|
# Extract IDs before async calls
|
|
playlist_id = playlist.id
|
|
sound_id = sound.id
|
|
|
|
# Add sound first time
|
|
await playlist_service.add_sound_to_playlist(
|
|
playlist_id=playlist_id,
|
|
sound_id=sound_id,
|
|
user_id=user_id,
|
|
)
|
|
|
|
# Try to add same sound again
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await playlist_service.add_sound_to_playlist(
|
|
playlist_id=playlist_id,
|
|
sound_id=sound_id,
|
|
user_id=user_id,
|
|
)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "already in this playlist" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_sound_from_playlist_success(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test removing sound from playlist successfully."""
|
|
# Create test playlist and sound within this test
|
|
user_id = test_user.id
|
|
playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Test Playlist",
|
|
description="A test playlist",
|
|
genre="test",
|
|
is_main=False,
|
|
is_current=False,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(playlist)
|
|
|
|
sound = Sound(
|
|
name="Test Sound",
|
|
filename="test.mp3",
|
|
type="SDB",
|
|
duration=5000,
|
|
size=1024,
|
|
hash="test_hash",
|
|
play_count=10,
|
|
)
|
|
test_session.add(sound)
|
|
await test_session.commit()
|
|
await test_session.refresh(playlist)
|
|
await test_session.refresh(sound)
|
|
|
|
# Extract IDs before async calls
|
|
playlist_id = playlist.id
|
|
sound_id = sound.id
|
|
|
|
# Add sound first
|
|
await playlist_service.add_sound_to_playlist(
|
|
playlist_id=playlist_id,
|
|
sound_id=sound_id,
|
|
user_id=user_id,
|
|
)
|
|
|
|
# Remove sound
|
|
await playlist_service.remove_sound_from_playlist(
|
|
playlist_id=playlist_id,
|
|
sound_id=sound_id,
|
|
user_id=user_id,
|
|
)
|
|
|
|
# Verify sound was removed
|
|
sounds = await playlist_service.get_playlist_sounds(playlist_id)
|
|
assert len(sounds) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_sound_not_in_playlist(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test removing sound that's not in playlist fails."""
|
|
# Create test playlist and sound within this test
|
|
user_id = test_user.id
|
|
playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Test Playlist",
|
|
description="A test playlist",
|
|
genre="test",
|
|
is_main=False,
|
|
is_current=False,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(playlist)
|
|
|
|
sound = Sound(
|
|
name="Test Sound",
|
|
filename="test.mp3",
|
|
type="SDB",
|
|
duration=5000,
|
|
size=1024,
|
|
hash="test_hash",
|
|
play_count=10,
|
|
)
|
|
test_session.add(sound)
|
|
await test_session.commit()
|
|
await test_session.refresh(playlist)
|
|
await test_session.refresh(sound)
|
|
|
|
# Extract IDs before async calls
|
|
playlist_id = playlist.id
|
|
sound_id = sound.id
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await playlist_service.remove_sound_from_playlist(
|
|
playlist_id=playlist_id,
|
|
sound_id=sound_id,
|
|
user_id=user_id,
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "not found in this playlist" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_current_playlist(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test setting a playlist as current."""
|
|
# Create test playlists within this test
|
|
user_id = test_user.id
|
|
test_playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Test Playlist",
|
|
description="A test playlist",
|
|
genre="test",
|
|
is_main=False,
|
|
is_current=False,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(test_playlist)
|
|
|
|
current_playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Current Playlist",
|
|
description="Currently active playlist",
|
|
is_main=False,
|
|
is_current=True,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(current_playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(test_playlist)
|
|
await test_session.refresh(current_playlist)
|
|
|
|
# Extract IDs before async calls
|
|
test_playlist_id = test_playlist.id
|
|
current_playlist_id = current_playlist.id
|
|
|
|
# Verify initial state
|
|
assert current_playlist.is_current is True
|
|
assert test_playlist.is_current is False
|
|
|
|
# Set test_playlist as current
|
|
updated_playlist = await playlist_service.set_current_playlist(
|
|
test_playlist_id,
|
|
)
|
|
|
|
assert updated_playlist.is_current is True
|
|
|
|
# Verify old current is no longer current
|
|
old_current = await playlist_service.get_playlist_by_id(current_playlist_id)
|
|
assert old_current.is_current is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unset_current_playlist_sets_main_as_current(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test unsetting current playlist falls back to main playlist."""
|
|
# Create test playlists within this test
|
|
user_id = test_user.id
|
|
current_playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Current Playlist",
|
|
description="Currently active playlist",
|
|
is_main=False,
|
|
is_current=True,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(current_playlist)
|
|
|
|
main_playlist = Playlist(
|
|
user_id=None,
|
|
name="Main Playlist",
|
|
description="Main playlist",
|
|
is_main=True,
|
|
is_current=False,
|
|
is_deletable=False,
|
|
)
|
|
test_session.add(main_playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(current_playlist)
|
|
await test_session.refresh(main_playlist)
|
|
|
|
# Extract IDs before async calls
|
|
current_playlist_id = current_playlist.id
|
|
main_playlist_id = main_playlist.id
|
|
|
|
# Verify initial state
|
|
assert current_playlist.is_current is True
|
|
|
|
# Unset current playlist
|
|
await playlist_service.unset_current_playlist()
|
|
|
|
# Verify get_current_playlist returns main playlist as fallback
|
|
current = await playlist_service.get_current_playlist()
|
|
assert current.id == main_playlist_id
|
|
assert current.is_main is True
|
|
|
|
# Verify old current is no longer current
|
|
old_current = await playlist_service.get_playlist_by_id(current_playlist_id)
|
|
assert old_current.is_current is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_playlist_stats(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test getting playlist statistics."""
|
|
# Create test playlist and sound within this test
|
|
user_id = test_user.id
|
|
playlist = Playlist(
|
|
user_id=user_id,
|
|
name="Test Playlist",
|
|
description="A test playlist",
|
|
genre="test",
|
|
is_main=False,
|
|
is_current=False,
|
|
is_deletable=True,
|
|
)
|
|
test_session.add(playlist)
|
|
|
|
sound = Sound(
|
|
name="Test Sound",
|
|
filename="test.mp3",
|
|
type="SDB",
|
|
duration=5000,
|
|
size=1024,
|
|
hash="test_hash",
|
|
play_count=10,
|
|
)
|
|
test_session.add(sound)
|
|
await test_session.commit()
|
|
await test_session.refresh(playlist)
|
|
await test_session.refresh(sound)
|
|
|
|
# Extract IDs before async calls
|
|
playlist_id = playlist.id
|
|
sound_id = sound.id
|
|
|
|
# Initially empty playlist
|
|
stats = await playlist_service.get_playlist_stats(playlist_id)
|
|
assert stats["sound_count"] == 0
|
|
assert stats["total_duration_ms"] == 0
|
|
assert stats["total_play_count"] == 0
|
|
|
|
# Add sound to playlist
|
|
await playlist_service.add_sound_to_playlist(
|
|
playlist_id=playlist_id,
|
|
sound_id=sound_id,
|
|
user_id=user_id,
|
|
)
|
|
|
|
# Check stats again
|
|
stats = await playlist_service.get_playlist_stats(playlist_id)
|
|
assert stats["sound_count"] == 1
|
|
assert stats["total_duration_ms"] == 5000 # From test_sound fixture
|
|
assert stats["total_play_count"] == 10 # From test_sound fixture
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_sound_to_main_playlist(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test adding sound to main playlist."""
|
|
# Create test sound and main playlist within this test
|
|
user_id = test_user.id
|
|
sound = Sound(
|
|
name="Test Sound",
|
|
filename="test.mp3",
|
|
type="SDB",
|
|
duration=5000,
|
|
size=1024,
|
|
hash="test_hash",
|
|
play_count=10,
|
|
)
|
|
test_session.add(sound)
|
|
|
|
main_playlist = Playlist(
|
|
user_id=None,
|
|
name="Main Playlist",
|
|
description="Main playlist",
|
|
is_main=True,
|
|
is_current=False,
|
|
is_deletable=False,
|
|
)
|
|
test_session.add(main_playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(sound)
|
|
await test_session.refresh(main_playlist)
|
|
|
|
# Extract IDs before async calls
|
|
sound_id = sound.id
|
|
main_playlist_id = main_playlist.id
|
|
|
|
# Add sound to main playlist
|
|
await playlist_service.add_sound_to_main_playlist(sound_id, user_id)
|
|
|
|
# Verify sound was added to main playlist
|
|
sounds = await playlist_service.get_playlist_sounds(main_playlist_id)
|
|
assert len(sounds) == 1
|
|
assert sounds[0].id == sound_id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_sound_to_main_playlist_already_exists(
|
|
self,
|
|
playlist_service: PlaylistService,
|
|
test_user: User,
|
|
test_session: AsyncSession,
|
|
) -> None:
|
|
"""Test adding sound to main playlist when it already exists (should not duplicate)."""
|
|
# Create test sound and main playlist within this test
|
|
user_id = test_user.id
|
|
sound = Sound(
|
|
name="Test Sound",
|
|
filename="test.mp3",
|
|
type="SDB",
|
|
duration=5000,
|
|
size=1024,
|
|
hash="test_hash",
|
|
play_count=10,
|
|
)
|
|
test_session.add(sound)
|
|
|
|
main_playlist = Playlist(
|
|
user_id=None,
|
|
name="Main Playlist",
|
|
description="Main playlist",
|
|
is_main=True,
|
|
is_current=False,
|
|
is_deletable=False,
|
|
)
|
|
test_session.add(main_playlist)
|
|
await test_session.commit()
|
|
await test_session.refresh(sound)
|
|
await test_session.refresh(main_playlist)
|
|
|
|
# Extract IDs before async calls
|
|
sound_id = sound.id
|
|
main_playlist_id = main_playlist.id
|
|
|
|
# Add sound to main playlist twice
|
|
await playlist_service.add_sound_to_main_playlist(sound_id, user_id)
|
|
await playlist_service.add_sound_to_main_playlist(sound_id, user_id)
|
|
|
|
# Verify sound is only added once
|
|
sounds = await playlist_service.get_playlist_sounds(main_playlist_id)
|
|
assert len(sounds) == 1
|
|
assert sounds[0].id == sound_id
|