add missing tests for LocalCollection and LocalLibrary methods (#94)

musify/libraries/local/library/library.py

@@ -390,15 +390,15 @@ def log_playlists(self) -> None:
                 f"\33[1;94m{len(playlist):>6} total \33[0m"
             )
 
-    async def save_playlists(self, dry_run: bool = True) -> dict[str, Result]:
+    async def save_playlists(self, dry_run: bool = True) -> dict[LocalPlaylist, Result]:
         """
         For each Playlist in this Library, saves its associate tracks and its settings (if applicable) to file.
 
         :param dry_run: Run function, but do not modify the file on the disk.
         :return: A map of the playlist name to the results of its sync as a :py:class:`Result` object.
         """
-        async def _save_playlist(pl: LocalPlaylist) -> tuple[str, Result]:
-            return pl.name, await pl.save(dry_run=dry_run)
+        async def _save_playlist(pl: LocalPlaylist) -> tuple[LocalPlaylist, Result]:
+            return pl, await pl.save(dry_run=dry_run)
         results = await self.logger.get_asynchronous_iterator(
             map(_save_playlist, self.playlists.values()), desc="Updating playlists", unit="tracks"
         )
@@ -407,7 +407,6 @@ async def _save_playlist(pl: LocalPlaylist) -> tuple[str, Result]:
     ###########################################################################
     ## Backup/restore
     ###########################################################################
-    # TODO: add test for this
     def restore_tracks(
             self, backup: RestoreTracksType, tags: UnitIterable[LocalTrackField] = LocalTrackField.ALL
     ) -> int:

musify/libraries/local/playlist/m3u.py

@@ -132,12 +132,12 @@ async def save(self, dry_run: bool = True, *_, **__) -> SyncResultM3U:
             with open(self.path, "r", encoding="utf-8") as file:  # get list of paths that were saved for results
                 final_paths = {Path(line.rstrip()) for line in file if line.rstrip()}
         else:  # use current list of tracks as a proxy of paths that were saved for results
-            final_paths = {track.path for track in self._tracks}
+            final_paths = set(map(Path, self.path_mapper.unmap_many(self._tracks, check_existence=False)))
 
         return SyncResultM3U(
             start=len(start_paths),
             added=len(final_paths - start_paths),
-            removed=len(start_paths - final_paths),
+            removed=len(start_paths.difference(final_paths)),
             unchanged=len(start_paths.intersection(final_paths)),
             difference=len(final_paths) - len(start_paths),
             final=len(final_paths),

musify/libraries/local/playlist/xautopf.py

@@ -201,16 +201,21 @@ async def save(self, dry_run: bool = True, *_, **__) -> SyncResultXAutoPF:
         if not dry_run:
             self._original = self.tracks.copy()
 
+        def _get_paths_sum(psr: XMLPlaylistParser, key: str) -> int:
+            if not (value := psr.xml_source.get(key)):
+                return 0
+            return sum(1 for p in value.split("|") if p)
+
         return SyncResultXAutoPF(
             start=initial_count,
-            start_included=sum(1 for p in initial.xml_source.get("ExceptionsInclude", "").split("|") if p),
-            start_excluded=sum(1 for p in initial.xml_source.get("Exceptions", "").split("|") if p),
+            start_included=_get_paths_sum(initial, "ExceptionsInclude"),
+            start_excluded=_get_paths_sum(initial, "Exceptions"),
             start_compared=len(initial.xml_source["Conditions"].get("Condition", [])),
             start_limiter=initial.xml_source["Limit"].get("@Enabled", "False") == "True",
             start_sorter=len(initial.xml_source.get("SortBy", initial.xml_source.get("DefinedSort", []))) > 0,
             final=len(self.tracks),
-            final_included=sum(1 for p in parser.xml_source.get("ExceptionsInclude", "").split("|") if p),
-            final_excluded=sum(1 for p in parser.xml_source.get("Exceptions", "").split("|") if p),
+            final_included=_get_paths_sum(parser, "ExceptionsInclude"),
+            final_excluded=_get_paths_sum(parser, "Exceptions"),
             final_compared=len(parser.xml_source["Conditions"].get("Condition", [])),
             final_limiter=parser.xml_source["Limit"].get("@Enabled", "False") == "True",
             final_sorter=len(parser.xml_source.get("SortBy", parser.xml_source.get("DefinedSort", []))) > 0,
@@ -492,7 +497,7 @@ def parse_exception_paths(
             ItemSorter.sort_by_field(original, field=Fields.LAST_PLAYED, reverse=True)
 
             matched_mapped: dict[Path, File] = {
-                item.path: item for item in matcher.comparers(original, reference=original[0])
+                item.path: item for item in matcher.comparers(original, reference=next(iter(original), None))
             } if matcher.comparers.ready else {}
             # noinspection PyProtectedMember
             matched_mapped |= {

musify/libraries/local/track/_tags/writer.py

@@ -47,7 +47,7 @@ def delete_tags(self, tags: UnitIterable[Tags] = (), dry_run: bool = True) -> Sy
         tag_names = set(Tags.to_tags(tags))
         removed = set()
         for tag_name in tag_names:
-            if self._delete_tag(tag_name, dry_run):
+            if self._clear_tag(tag_name, dry_run):
                 removed.update(Tags.from_name(tag_name))
 
         save = not dry_run and len(removed) > 0
@@ -57,12 +57,12 @@ def delete_tags(self, tags: UnitIterable[Tags] = (), dry_run: bool = True) -> Sy
         removed = sorted(removed, key=lambda x: Tags.all().index(x))
         return SyncResultTrack(saved=save, updated={u: 0 for u in removed})
 
-    def _delete_tag(self, tag_name: str, dry_run: bool = True) -> bool:
+    def _clear_tag(self, tag_name: str, dry_run: bool = True) -> bool:
         """
-        Remove a tag by its tag name.
+        Remove a tag by its tag name from the loaded file object in memory.
 
         :param tag_name: Tag name as found in :py:class:`TagMap` to remove.
-        :param dry_run: Run function, but do not modify the file on the disk.
+        :param dry_run: Run function, but do not modify the loaded file in memory.
         :return: True if tag has been remove, False otherwise.
         """
         removed = False
@@ -79,6 +79,18 @@ def _delete_tag(self, tag_name: str, dry_run: bool = True) -> bool:
 
         return removed
 
+    def clear_loaded_images(self) -> bool:
+        """
+        Clear the loaded embedded images for this track.
+        Does not alter the actual file in anyway, only the loaded object in memory.
+        """
+        tag_names = Tags.IMAGES.to_tag()
+        removed = False
+        for tag_name in tag_names:
+            removed = removed or self._clear_tag(tag_name, dry_run=False)
+
+        return removed
+
     def write(
             self,
             source: Track,

musify/libraries/local/track/flac.py

@@ -79,7 +79,7 @@ def _write_images(self, track: LocalTrack, dry_run: bool = True) -> bool:
 
         return updated
 
-    def _delete_tag(self, tag_name: str, dry_run: bool = True) -> bool:
+    def _clear_tag(self, tag_name: str, dry_run: bool = True) -> bool:
         if tag_name == LocalTrackField.IMAGES.name.lower():
             self.file.clear_pictures()
             return True

musify/libraries/local/track/mp3.py

@@ -67,7 +67,7 @@ class _MP3TagWriter(TagWriter[mutagen.mp3.MP3]):
 
     __slots__ = ()
 
-    def _delete_tag(self, tag_name: str, dry_run: bool = True) -> bool:
+    def _clear_tag(self, tag_name: str, dry_run: bool = True) -> bool:
         removed = False
 
         tag_ids = self.tag_map[tag_name]

musify/libraries/local/track/track.py

@@ -444,7 +444,7 @@ def refresh(self) -> None:
             self.has_image = self._reader.check_for_images()
 
         # to reduce memory usage, remove any embedded images from the loaded file
-        self._writer.delete_tags(Tags.IMAGES, dry_run=True)
+        self._writer.clear_loaded_images()
 
         self._loaded = True
 
@@ -490,12 +490,11 @@ def merge(self, track: Track, tags: UnitIterable[TrackField] = TrackField.ALL) -
 
     def extract_images_to_file(self, output_folder: str | Path) -> int:
         """Reload the file, extract and save all embedded images from file. Returns the number of images extracted."""
-        self._reader.file = mutagen.File(self.path)
-        self._writer.file = self._reader.file
+        self._reader.file.load(self._reader.file.filename)
 
         images = self._reader.read_images()
         if images is None:
-            return False
+            return 0
 
         output_folder = Path(output_folder)
         if not output_folder.is_dir():
@@ -511,7 +510,7 @@ def extract_images_to_file(self, output_folder: str | Path) -> int:
             count += 1
 
         # to reduce memory usage, remove any embedded images from the loaded file
-        self._writer.delete_tags(Tags.IMAGES, dry_run=True)
+        self._writer.clear_loaded_images()
 
         return count
 

tests/api/test_authorise.py

@@ -36,7 +36,7 @@ def token(self) -> dict[str, Any]:
         }
 
     @pytest.fixture(params=[path_token])
-    def token_file_path(self, path: str) -> str:
+    def token_file_path(self, path: Path) -> Path:
         """Yield the temporary path for the token JSON file"""
         return path
 
@@ -64,7 +64,7 @@ def test_properties(self, authoriser: APIAuthoriser, token: dict[str, Any]):
         assert authoriser.token == token
         assert authoriser.token_safe != token
 
-    def test_load_token(self, authoriser: APIAuthoriser, token_file_path: str):
+    def test_load_token(self, authoriser: APIAuthoriser, token_file_path: Path):
         # just check it doesn't fail when no path given
         authoriser.token = None
         authoriser.token_file_path = None
@@ -211,7 +211,7 @@ def test_expiry(self, authoriser: APIAuthoriser):
         authoriser.test_expiry = 2000
         assert not authoriser._test_expiry()
 
-    async def test_auth_new_token(self, token: dict[str, Any], token_file_path: str, requests_mock: aioresponses):
+    async def test_auth_new_token(self, token: dict[str, Any], token_file_path: Path, requests_mock: aioresponses):
         authoriser = APIAuthoriser(name="test", auth_args={"url": "http://localhost/auth"}, test_expiry=1000)
 
         response = {"access_token": "valid token", "expires_in": 3000, "refresh_token": "new_refresh"}
@@ -222,7 +222,7 @@ async def test_auth_new_token(self, token: dict[str, Any], token_file_path: str,
         assert authoriser.headers == expected_header
         assert authoriser.token["refresh_token"] == "new_refresh"
 
-    async def test_auth_load_and_token_valid(self, token_file_path: str, requests_mock: aioresponses):
+    async def test_auth_load_and_token_valid(self, token_file_path: Path, requests_mock: aioresponses):
         authoriser = APIAuthoriser(
             name="test",
             test_args={"url": "http://localhost/test"},
@@ -237,7 +237,7 @@ async def test_auth_load_and_token_valid(self, token_file_path: str, requests_mo
         expected_header = {"Authorization": f"Bearer {authoriser.token["access_token"]}"}
         assert authoriser.headers == expected_header
 
-    async def test_auth_force_load_and_token_valid(self, token_file_path: str):
+    async def test_auth_force_load_and_token_valid(self, token_file_path: Path):
         authoriser = APIAuthoriser(
             name="test",
             token={"this token": "is not valid"},
@@ -253,15 +253,15 @@ async def test_auth_force_load_and_token_valid(self, token_file_path: str):
 
         assert authoriser.headers == expected_header | authoriser.header_extra
 
-    async def test_auth_force_new_and_no_args(self, token: dict[str, Any], token_file_path: str):
+    async def test_auth_force_new_and_no_args(self, token: dict[str, Any], token_file_path: Path):
         authoriser = APIAuthoriser(name="test", token=token, token_file_path=token_file_path)
 
         # force new despite being given token and token file path
         with pytest.raises(APIError):
             await authoriser.authorise(force_new=True)
 
     async def test_auth_new_token_and_no_refresh(
-            self, token: dict[str, Any], token_file_path: str, requests_mock: aioresponses
+            self, token: dict[str, Any], token_file_path: Path, requests_mock: aioresponses
     ):
         authoriser = APIAuthoriser(
             name="test",
@@ -276,7 +276,7 @@ async def test_auth_new_token_and_no_refresh(
         assert authoriser.headers == expected_header
 
     async def test_auth_new_token_and_refresh_valid(
-            self, token: dict[str, Any], token_file_path: str, requests_mock: aioresponses
+            self, token: dict[str, Any], token_file_path: Path, requests_mock: aioresponses
     ):
         authoriser = APIAuthoriser(
             name="test",
@@ -295,7 +295,7 @@ async def test_auth_new_token_and_refresh_valid(
         assert authoriser.token["refresh_token"] == "new_refresh"
 
     async def test_auth_new_token_and_refresh_invalid(
-            self, token: dict[str, Any], token_file_path: str, requests_mock: aioresponses
+            self, token: dict[str, Any], token_file_path: Path, requests_mock: aioresponses
     ):
         authoriser = APIAuthoriser(
             name="test",

tests/conftest.py

@@ -351,7 +351,7 @@ def requests_mock():
 
 
 @pytest.fixture
-def path(request: pytest.FixtureRequest | SubRequest, tmp_path: Path) -> str:
+def path(request: pytest.FixtureRequest | SubRequest, tmp_path: Path) -> Path:
     """
     Copy the path of the source file to the test cache for this test and return the cache path.
     Deletes the test folder when test is done.

tests/libraries/local/conftest.py

@@ -1,3 +1,5 @@
+from pathlib import Path
+
 import pytest
 
 from musify.file.path_mapper import PathStemMapper
@@ -21,25 +23,25 @@ def path_mapper() -> PathStemMapper:
 
 
 @pytest.fixture(params=[path_track_flac])
-async def track_flac(path: str, remote_wrangler: RemoteDataWrangler) -> FLAC:
+async def track_flac(path: Path, remote_wrangler: RemoteDataWrangler) -> FLAC:
     """Yields instantiated :py:class:`FLAC` objects for testing"""
     return await FLAC(file=path, remote_wrangler=remote_wrangler)
 
 
 @pytest.fixture(params=[path_track_mp3])
-async def track_mp3(path: str, remote_wrangler: RemoteDataWrangler) -> MP3:
+async def track_mp3(path: Path, remote_wrangler: RemoteDataWrangler) -> MP3:
     """Yields instantiated :py:class:`MP3` objects for testing"""
     return await MP3(file=path, remote_wrangler=remote_wrangler)
 
 
 @pytest.fixture(params=[path_track_m4a])
-async def track_m4a(path: str, remote_wrangler: RemoteDataWrangler) -> M4A:
+async def track_m4a(path: Path, remote_wrangler: RemoteDataWrangler) -> M4A:
     """Yields instantiated :py:class:`M4A` objects for testing"""
     return await M4A(file=path, remote_wrangler=remote_wrangler)
 
 
 @pytest.fixture(params=[path_track_wma])
-async def track_wma(path: str, remote_wrangler: RemoteDataWrangler) -> WMA:
+async def track_wma(path: Path, remote_wrangler: RemoteDataWrangler) -> WMA:
     """Yields instantiated :py:class:`WMA` objects for testing"""
     return await WMA(file=path, remote_wrangler=remote_wrangler)
 

tests/libraries/local/library/test_library.py

@@ -91,6 +91,12 @@ def test_init_relative_paths(self):
             path.stem: path for path in path_playlist_all
         }
 
+    def test_collection_creators(self, library: LocalLibrary):
+        assert len(library.folders) == len(set(track.folder for track in library.tracks))
+        assert len(library.albums) == len(set(track.album for track in library.tracks))
+        assert len(library.artists) == len(set(artist for track in library.tracks for artist in track.artists))
+        assert len(library.genres) == len(set(genre for track in library.tracks for genre in track.genres))
+
     async def test_load(self, path_mapper: PathMapper):
         library = LocalLibrary(
             library_folders=path_track_resources, playlist_folder=path_playlist_resources, path_mapper=path_mapper
@@ -111,9 +117,3 @@ async def test_load(self, path_mapper: PathMapper):
         await library.load()
 
         assert len(library.tracks) == len(library._track_paths) == len(path_track_all) + 2
-
-    def test_collection_creators(self, library: LocalLibrary):
-        assert len(library.folders) == len(set(track.folder for track in library.tracks))
-        assert len(library.albums) == len(set(track.album for track in library.tracks))
-        assert len(library.artists) == len(set(artist for track in library.tracks for artist in track.artists))
-        assert len(library.genres) == len(set(genre for track in library.tracks for genre in track.genres))

tests/libraries/local/library/testers.py

@@ -1,8 +1,61 @@
 from abc import ABCMeta
+from pathlib import Path
+from typing import Any
 
+from musify.libraries.local.library import LocalLibrary
+from musify.libraries.local.playlist.m3u import SyncResultM3U
+from musify.libraries.local.playlist.xautopf import SyncResultXAutoPF
 from tests.libraries.core.collection import LibraryTester
 from tests.libraries.local.track.testers import LocalCollectionTester
 
 
 class LocalLibraryTester(LibraryTester, LocalCollectionTester, metaclass=ABCMeta):
-    pass
+
+    @staticmethod
+    async def test_save_playlists(library: LocalLibrary):
+        playlists = [pl for pl in library.playlists.values() if len(pl) > 0]
+        for playlist in playlists:
+            playlist.pop()
+
+        results = await library.save_playlists(dry_run=True)
+
+        for pl, result in results.items():
+            print(pl.name, result)
+        for pl in playlists:
+            print(pl.name, len(pl))
+
+        assert len(results) == len(library.playlists)
+        for pl, result in results.items():
+            if pl not in playlists:
+                continue
+
+            if isinstance(result, SyncResultM3U):
+                assert result.removed == 1
+            elif isinstance(result, SyncResultXAutoPF):
+                assert result.start - result.final == 1
+
+    @staticmethod
+    def test_restore_tracks(library: LocalLibrary):
+        new_title = "brand new title"
+        new_artist = "brand new artist"
+
+        for track in library:
+            assert track.title != "brand new title"
+            assert track.artist != new_artist
+
+        backup: list[dict[str, Any]] = library.json()["tracks"]
+        for track in backup:
+            track["title"] = new_title
+
+        library.restore_tracks(backup)
+        for track in library:
+            assert track.title == "brand new title"
+            assert track.artist != new_artist
+
+        for track in backup:
+            track["artist"] = new_artist
+
+        library.restore_tracks({Path(track["path"]): track for track in backup})
+        for track in library:
+            assert track.title == "brand new title"
+            assert track.artist == new_artist