diff --git a/app/services/extraction.py b/app/services/extraction.py index fe96b37..53488ce 100644 --- a/app/services/extraction.py +++ b/app/services/extraction.py @@ -508,7 +508,9 @@ class ExtractionService: async def _add_to_main_playlist(self, sound_id: int, user_id: int) -> None: """Add the sound to the user's main playlist.""" try: - await self.playlist_service.add_sound_to_main_playlist(sound_id, user_id) + await self.playlist_service._add_sound_to_main_playlist_internal( # noqa: SLF001 + sound_id, user_id, + ) logger.info( "Added sound %d to main playlist for user %d", sound_id, diff --git a/app/services/playlist.py b/app/services/playlist.py index 338cac3..17ffbdc 100644 --- a/app/services/playlist.py +++ b/app/services/playlist.py @@ -51,6 +51,16 @@ class PlaylistService: self.playlist_repo = PlaylistRepository(session) self.sound_repo = SoundRepository(session) + async def _is_main_playlist(self, playlist_id: int) -> bool: + """Check if the given playlist is the main playlist.""" + try: + playlist = await self.playlist_repo.get_by_id(playlist_id) + except Exception: + logger.exception("Failed to check if playlist is main: %s", playlist_id) + return False + else: + return playlist is not None and playlist.is_main + async def get_playlist_by_id(self, playlist_id: int) -> Playlist: """Get a playlist by ID.""" playlist = await self.playlist_repo.get_by_id(playlist_id) @@ -145,6 +155,13 @@ class PlaylistService: is_current: bool | None = None, ) -> Playlist: """Update a playlist.""" + # Check if this is the main playlist + if await self._is_main_playlist(playlist_id): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="The main playlist cannot be edited", + ) + playlist = await self.get_playlist_by_id(playlist_id) update_data: dict[str, Any] = {} @@ -186,6 +203,13 @@ class PlaylistService: async def delete_playlist(self, playlist_id: int, user_id: int) -> None: """Delete a playlist.""" + # Check if this is the main playlist + if await self._is_main_playlist(playlist_id): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="The main playlist cannot be deleted", + ) + playlist = await self.get_playlist_by_id(playlist_id) if not playlist.is_deletable: @@ -247,6 +271,13 @@ class PlaylistService: position: int | None = None, ) -> None: """Add a sound to a playlist.""" + # Check if this is the main playlist + if await self._is_main_playlist(playlist_id): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Sounds cannot be added to the main playlist", + ) + # Verify playlist exists await self.get_playlist_by_id(playlist_id) @@ -295,6 +326,13 @@ class PlaylistService: user_id: int, ) -> None: """Remove a sound from a playlist.""" + # Check if this is the main playlist + if await self._is_main_playlist(playlist_id): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Sounds cannot be removed from the main playlist", + ) + # Verify playlist exists await self.get_playlist_by_id(playlist_id) @@ -377,8 +415,23 @@ class PlaylistService: "total_play_count": total_plays, } - async def add_sound_to_main_playlist(self, sound_id: int, user_id: int) -> None: + async def add_sound_to_main_playlist( + self, sound_id: int, user_id: int, # noqa: ARG002 + ) -> None: """Add a sound to the global main playlist.""" + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Sounds cannot be added to the main playlist", + ) + + async def _add_sound_to_main_playlist_internal( + self, sound_id: int, user_id: int, + ) -> None: + """Add sound to main playlist bypassing restrictions. + + This method is intended for internal system use only (e.g., extraction service). + It bypasses the main playlist modification restrictions. + """ main_playlist = await self.get_main_playlist() if main_playlist.id is None: @@ -395,7 +448,7 @@ class PlaylistService: ): await self.playlist_repo.add_sound_to_playlist(main_playlist_id, sound_id) logger.info( - "Added sound %s to main playlist for user %s", + "Added sound %s to main playlist for user %s (internal)", sound_id, user_id, ) diff --git a/tests/api/v1/test_playlist_endpoints.py b/tests/api/v1/test_playlist_endpoints.py index db558ea..fa6bc7b 100644 --- a/tests/api/v1/test_playlist_endpoints.py +++ b/tests/api/v1/test_playlist_endpoints.py @@ -463,7 +463,7 @@ class TestPlaylistEndpoints: f"/api/v1/playlists/{main_playlist_id}", ) - assert response.status_code == 400 + assert response.status_code == 403 assert "cannot be deleted" in response.json()["detail"] @pytest.mark.asyncio diff --git a/tests/services/test_playlist.py b/tests/services/test_playlist.py index 07d4303..5dc547f 100644 --- a/tests/services/test_playlist.py +++ b/tests/services/test_playlist.py @@ -874,13 +874,13 @@ class TestPlaylistService: assert stats["total_play_count"] == 10 # From test_sound fixture @pytest.mark.asyncio - async def test_add_sound_to_main_playlist( + async def test_add_sound_to_main_playlist_forbidden( self, playlist_service: PlaylistService, test_user: User, test_session: AsyncSession, ) -> None: - """Test adding sound to main playlist.""" + """Test that adding sound to main playlist is forbidden.""" # Create test sound and main playlist within this test user_id = test_user.id sound = Sound( @@ -911,57 +911,102 @@ class TestPlaylistService: 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) + # Try to add sound to main playlist (should fail) + with pytest.raises(HTTPException) as exc_info: + await playlist_service.add_sound_to_main_playlist(sound_id, user_id) + + assert exc_info.value.status_code == 403 + assert "cannot be added to the main playlist" in exc_info.value.detail + + @pytest.mark.asyncio + async def test_add_sound_to_main_playlist_blocked( + self, + playlist_service: PlaylistService, + test_user: User, + test_session: AsyncSession, + ) -> None: + """Test that adding sound to main playlist is blocked (should raise 403).""" + # 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 + + # Try to add sound to main playlist (should fail both times) + with pytest.raises(HTTPException) as exc_info: + await playlist_service.add_sound_to_main_playlist(sound_id, user_id) + assert exc_info.value.status_code == 403 + + with pytest.raises(HTTPException) as exc_info: + await playlist_service.add_sound_to_main_playlist(sound_id, user_id) + assert exc_info.value.status_code == 403 + + @pytest.mark.asyncio + async def test_internal_add_sound_to_main_playlist_allowed( + self, + playlist_service: PlaylistService, + test_user: User, + test_session: AsyncSession, + ) -> None: + """Test that internal method can add sound to main playlist (bypasses restrictions).""" + # Create test sound and main playlist within this test + user_id = test_user.id + sound = Sound( + name="Test Sound Internal", + filename="test_internal.mp3", + type="EXT", + duration=6000, + size=2048, + hash="test_hash_internal", + play_count=0, + ) + test_session.add(sound) + + main_playlist = Playlist( + user_id=None, + name="Main Playlist Internal", + description="Main playlist for internal test", + 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 + + # Use internal method to add sound to main playlist (should work) + await playlist_service._add_sound_to_main_playlist_internal(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