From 0e7b93aa993a7c3d5136551e20725fcfd1b83a8b Mon Sep 17 00:00:00 2001 From: Travis Abendshien <46939827+CyanVoxel@users.noreply.github.com> Date: Thu, 10 Sep 2026 22:19:47 -0700 Subject: [PATCH] fix: fix edge cases and rework tests based on docs case table --- src/tagstudio/core/library/sync.py | 223 ++++-- src/tagstudio/qt/qt_driver.py | 6 + src/tagstudio/resources/translations/en.json | 1 + tests/core/library/test_sync.py | 682 ++++++++++++++++--- 4 files changed, 730 insertions(+), 182 deletions(-) diff --git a/src/tagstudio/core/library/sync.py b/src/tagstudio/core/library/sync.py index 6172741e..b3d1520d 100644 --- a/src/tagstudio/core/library/sync.py +++ b/src/tagstudio/core/library/sync.py @@ -2,7 +2,7 @@ # SPDX-License-Identifier: MIT -from collections.abc import Iterator +from collections.abc import Callable, Hashable, Iterator from dataclasses import dataclass, field from datetime import datetime as dt from pathlib import Path @@ -130,6 +130,11 @@ class LibrarySyncEngine: if unlinked_ids: self.unlinked_entries = self.library.get_entries(list(unlinked_ids)) + if self.unlinked_entries: + yield -1, -1 # Signals the UI that repair work is starting + + # Any (rare) duplicate entries are merged first, then the normal relinking process + self._merge_duplicate_path_entries(case_sensitive, cache) self._auto_relink_matched_entries(case_sensitive, cache) yield count, len(self.new_paths) @@ -220,7 +225,8 @@ class LibrarySyncEngine: def find_relink_candidates(self, entry: Entry) -> list[Path]: """Try to find files in the library directory matching an unlinked entry's filename. - Comparisons are made using NFD normalization and the assumed filesystem's case sensitivity. + A file already in the path cache can only be a candidate if its normalized path matches + this entry's own normalized path, such as with an NFC/NFD or case only duplicate. """ case_sensitive = self._get_case_sensitivity() target_key = norm_path(Path(entry.path.name), case_sensitive=case_sensitive) @@ -232,6 +238,15 @@ class LibrarySyncEngine: else: matches = self._glob_for_filename(entry.path.name, case_sensitive) + entry_key = norm_path(entry.path, case_sensitive=case_sensitive) + cache = self.library.get_or_build_path_cache() + filtered: list[Path] = [] + for path in matches: + path_key = norm_path(path, case_sensitive=case_sensitive) + if path_key == entry_key or cache.get(path_key) is None: + filtered.append(path) + matches = filtered + logger.info("[Sync] Relink candidates", entry=entry.path.as_posix(), matches=matches) return matches @@ -253,6 +268,66 @@ class LibrarySyncEngine: return self.library.merge_entries(source, target) return self.library.update_entry_path(entry.id, new_path) + def _relink_unique_matches( + self, + paths: list[Path], + cache: dict[Path, int], + case_sensitive: bool, + key_of_entry: Callable[[Entry], Hashable], + key_of_path: Callable[[Path], Hashable | None], + log_message: str, + ) -> list[Path]: + """Relink entries to paths that share a unique key. + + Args: + paths (list[Path]): Candidate filepaths to try matching against unlinked entries. + cache (dict[Path, int]): Path cache, forwarded to `_apply_relink()`. + case_sensitive (bool): Filesystem case sensitivity, forwarded to `_apply_relink()`. + key_of_entry (Callable[[Entry], Hashable]): Computes a grouping key from an entry. + key_of_path (Callable[[Path], Hashable | None]): Computes a grouping key from a path, + or None if that path can't be evaluated by this key at all (e.g. missing stats). + log_message (str): Logged on each successful relink in this pass. + + Returns: + list[Path]: Remaining unmatched paths. + """ + if not paths or not self.unlinked_entries: + return paths # Nothing to match against + + by_key: dict[Hashable, list[Entry]] = {} # Key -> entries sharing it + for entry in self.unlinked_entries: + by_key.setdefault(key_of_entry(entry), []).append(entry) + + paths_by_key: dict[Hashable, list[Path]] = {} # Key -> candidate paths sharing it + unmatched: list[Path] = [] + for path in paths: + key = key_of_path(path) + if key is None: # Not evaluable by this key (e.g. missing stats) + unmatched.append(path) + continue + paths_by_key.setdefault(key, []).append(path) + + relinked: list[Entry] = [] + for key, candidate_paths in paths_by_key.items(): # Check each key for a unique pairing + candidates = by_key.get(key, []) + if len(candidates) != 1 or len(candidate_paths) != 1: # Ambiguous case + unmatched.extend(candidate_paths) + continue + + entry, new_path = candidates[0], candidate_paths[0] # The only pair sharing this key + if not self._apply_relink(entry, new_path, cache, case_sensitive): # DB write failed + unmatched.append(new_path) + continue + + logger.info(log_message, old_path=entry.path.as_posix(), new_path=new_path.as_posix()) + relinked.append(entry) + + for entry in relinked: + self.unlinked_entries.remove(entry) + self.relinked_entries.extend(relinked) + + return unmatched + def relink_unlinked_entries(self) -> Iterator[int]: """Attempt to fix unlinked entries by finding a single matching file in the library.""" self.manual_relink_count = 0 @@ -280,37 +355,57 @@ class LibrarySyncEngine: for entry in matched: self.unlinked_entries.remove(entry) + def _merge_duplicate_path_entries(self, case_sensitive: bool, cache: dict[Path, int]) -> None: + """Merge any entry whose stored path collides with another entry's onto that entry. + + Entries here have already a full path + filename match. + """ + duplicate_ids = set(self.library.duplicate_path_entry_ids or []) + if not duplicate_ids: + return + + merged: list[Entry] = [] + for entry in self.unlinked_entries: + if self.cancelled: + break + if entry.id in duplicate_ids and self._apply_relink( + entry, entry.path, cache, case_sensitive + ): + merged.append(entry) + + for entry in merged: + self.unlinked_entries.remove(entry) + self.relinked_entries.extend(merged) + def _auto_relink_matched_entries(self, case_sensitive: bool, cache: dict[Path, int]) -> None: """Attempt to automatically relink unlinked entries under available conditions. Auto-relink applies to: - Files with the same filename but different paths - - Handles moves, moves + changes + - Handles moves, moves + changes (Case #7) - Files with different names and/or paths but the same date_modified and file_size - - Handles moves, renames + moves + - Handles moves, renames + moves (Cases #3, #11) + - Files with only a matching filename, as a last resort when nothing else matches + - Handles moves + changes, and entries with no stored stats (Case #6) Auto-relink DOES NOT apply to: - - Renames + Moves + Changes - - Deletions - - Ambiguous (more than one) matches + - Renames + Changes (Cases #2, #10) + - Deletions (Case #1) + - Ambiguous matches (Cases #4, #5, #8, #9, #12, #13) + + *(Cases are listed in the Library documentation)* """ if not self.new_paths or not self.unlinked_entries: return library_dir = unwrap(self.library.library_dir) + self.relinked_entries = [] - # Pass 1: Filename + metadata - by_name_and_stat: dict[tuple[Path, float | None, int | None], list[Entry]] = {} - for entry in self.unlinked_entries: - name_key = norm_path(Path(entry.path.name), case_sensitive=case_sensitive) - by_name_and_stat.setdefault( - (name_key, entry.date_modified, entry.file_size), [] - ).append(entry) + def name_key(path: Path) -> Path: + return norm_path(Path(path.name), case_sensitive=case_sensitive) - relinked: list[Entry] = [] - remaining_new: list[Path] = [] + # Stat every new path once, every pass below reuses this stats_by_path: dict[Path, tuple[float | None, int | None]] = {} - for new_path in self.new_paths: try: file_stat = (library_dir / new_path).stat() @@ -320,69 +415,53 @@ class LibrarySyncEngine: path=new_path, error=e, ) - remaining_new.append(new_path) continue + stats_by_path[new_path] = (get_date_modified(file_stat), get_file_size(file_stat)) - mtime = get_date_modified(file_stat) - size = get_file_size(file_stat) - stats_by_path[new_path] = (mtime, size) + def name_and_stat_key(path: Path) -> tuple[Path, float | None, int | None] | None: + stat = stats_by_path.get(path) + return None if stat is None else (name_key(path), *stat) - name_key = norm_path(Path(new_path.name), case_sensitive=case_sensitive) - key = (name_key, mtime, size) - candidates = by_name_and_stat.get(key, []) - if len(candidates) != 1: - remaining_new.append(new_path) - continue + # Pass 1: Filename + metadata (Case #7) + remaining = self._relink_unique_matches( + self.new_paths, + cache, + case_sensitive, + key_of_entry=lambda e: (name_key(e.path), e.date_modified, e.file_size), + key_of_path=name_and_stat_key, + log_message="[Sync] Automatically relinked moved file", + ) - entry = candidates[0] - if not self._apply_relink(entry, new_path, cache, case_sensitive): - remaining_new.append(new_path) - continue - by_name_and_stat[key] = [] # Don't match a second new_path here + if self.cancelled: + self.new_paths = remaining + return - logger.info( - "[Sync] Automatically relinked moved file", - old_path=entry.path.as_posix(), - new_path=new_path.as_posix(), - ) - relinked.append(entry) + # Pass 2: Different filename checking for same metadata (Cases #3, #11) + remaining = self._relink_unique_matches( + remaining, + cache, + case_sensitive, + key_of_entry=lambda e: (e.date_modified, e.file_size), + key_of_path=stats_by_path.get, + log_message="[Sync] Automatically relinked renamed file (matched by size + mdate)", + ) - for entry in relinked: - self.unlinked_entries.remove(entry) + if self.cancelled: + self.new_paths = remaining + return - # Pass 2: Different filename checking for same metadata - by_stat_only: dict[tuple[float | None, int | None], list[Entry]] = {} - for entry in self.unlinked_entries: - by_stat_only.setdefault((entry.date_modified, entry.file_size), []).append(entry) + # Pass 3: Filename only, for anything that wasn't matched before + # (Case #6, and entries with no stored stats) + remaining = self._relink_unique_matches( + remaining, + cache, + case_sensitive, + key_of_entry=lambda e: name_key(e.path), + key_of_path=name_key, + log_message="[Sync] Automatically relinked file by filename (no metadata found)", + ) - still_remaining: list[Path] = [] - for new_path in remaining_new: - stat_key = stats_by_path.get(new_path) - if stat_key is None: - still_remaining.append(new_path) - continue - - candidates = by_stat_only.get(stat_key, []) - if len(candidates) != 1: - still_remaining.append(new_path) - continue - - entry = candidates[0] - if not self._apply_relink(entry, new_path, cache, case_sensitive): - still_remaining.append(new_path) - continue - by_stat_only[stat_key] = [] - - logger.info( - "[Sync] Automatically relinked renamed file (matched by size/date only)", - old_path=entry.path.as_posix(), - new_path=new_path.as_posix(), - ) - relinked.append(entry) - self.unlinked_entries.remove(entry) - - self.new_paths = still_remaining - self.relinked_entries = relinked + self.new_paths = remaining def remove_unlinked_entries(self) -> None: """Remove unlinked entries from the Library.""" diff --git a/src/tagstudio/qt/qt_driver.py b/src/tagstudio/qt/qt_driver.py index d38dfb18..e468c2e0 100644 --- a/src/tagstudio/qt/qt_driver.py +++ b/src/tagstudio/qt/qt_driver.py @@ -1120,6 +1120,12 @@ class QtDriver(DriverMixin, QObject): if engine.cancelled: return searched_count, found_count = progress + if searched_count < 0: + # Scan finished, duplicate entry merging/relinking is running before the next yield + self.main_window.banner.show_progress( + Translations["library.sync.repairing"], phase="repairing" + ) + return self.main_window.banner.show_progress( Translations.format( "library.sync.scanning", diff --git a/src/tagstudio/resources/translations/en.json b/src/tagstudio/resources/translations/en.json index cd8915dc..24f183d4 100644 --- a/src/tagstudio/resources/translations/en.json +++ b/src/tagstudio/resources/translations/en.json @@ -265,6 +265,7 @@ "library.sync.relinked_banner.singular": "{count} File Automatically Relinked", "library.sync.relinked_suffix": "({count} Automatically Relinked)", "library.sync.remaining_unlinked_suffix": "({count} Still Unlinked)", + "library.sync.repairing": "Repairing Entries…", "library.sync.scanning": "Discovering Files… ({found_count} New Out of {searched_count})", "library.sync.unlinked_banner.plural": "{count} Unlinked Entries Found", "library.sync.unlinked_banner.singular": "{count} Unlinked Entry Found", diff --git a/tests/core/library/test_sync.py b/tests/core/library/test_sync.py index b9b071fe..fc6746f9 100644 --- a/tests/core/library/test_sync.py +++ b/tests/core/library/test_sync.py @@ -9,14 +9,17 @@ from pathlib import Path from tempfile import TemporaryDirectory import pytest +from sqlalchemy import text +from sqlalchemy.orm import Session from tagstudio.core.constants import IGNORE_NAME, TS_FOLDER_NAME -from tagstudio.core.library.alchemy.enums import BrowsingState from tagstudio.core.library.alchemy.library import Library from tagstudio.core.library.alchemy.models import Entry from tagstudio.core.library.sync import LibrarySyncEngine from tagstudio.core.utils.types import unwrap +# NOTE: Case numbers are described in the Library documentation. + @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) def test_sync_new_files(library: Library): @@ -51,29 +54,6 @@ def test_sync_multi_byte_filenames(library: Library): assert Path("umlaute äöü.txt") in engine.new_paths -@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) -def test_sync_unlinked_entries(library: Library): - """An unlinked entry with one matching file must be found by `relink_unlinked_entries()`.""" - library_dir = unwrap(library.library_dir) - engine = LibrarySyncEngine(library=library) - - # Touch the file "bar.md" but in the wrong location, to simulate a moved file - (library_dir / "bar.md").touch() - - # Neither library entry ("foo.txt", "one/two/bar.md") exists on disk, so both are unlinked - list(engine.sync_dir(library_dir, force_internal_scanner=True)) - assert engine.unlinked_entries_count == 2 - - # Relinking bar.md should match and relink to the entry that was at "one/two/bar.md" - list(engine.relink_unlinked_entries()) - assert engine.manual_relink_count == 1 - assert engine.unlinked_entries_count == 1 - - results = library.search_library(BrowsingState.from_path("bar.md"), page_size=500) - entries = library.get_entries(results.ids) - assert entries[0].path == Path("bar.md") - - @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) def test_sync_nfd_nfc_false_positive(library: Library): """A file reappearing in a different Unicode form must not look like a new/duplicate entry.""" @@ -98,27 +78,6 @@ def test_sync_nfd_nfc_false_positive(library: Library): assert engine.unlinked_entries_count == 2 -@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) -def test_sync_case_sensitivity_aware_relink(library: Library): - """`find_relink_candidates()` must respect the library's case-sensitivity setting.""" - library_dir = unwrap(library.library_dir) - engine = LibrarySyncEngine(library=library) - - (library_dir / "Other").mkdir() - (library_dir / "Other" / "name.txt").touch() - library.add_entries([Entry(path=Path("Folder/Name.txt"), fields=[])]) - - list(engine.sync_dir(library_dir, force_internal_scanner=True)) - unlinked_entry = next(e for e in engine.unlinked_entries if e.path == Path("Folder/Name.txt")) - - library.is_case_sensitive_fs = True - assert engine.find_relink_candidates(unlinked_entry) == [] - - library.is_case_sensitive_fs = False - engine._filename_to_path_map = None # Rebuild the index with the new case sensitivity - assert engine.find_relink_candidates(unlinked_entry) == [Path("Other/name.txt")] - - @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) def test_sync_cache_pruned_on_remove(library: Library): """Removing an unlinked entry must prune the path cache.""" @@ -149,11 +108,22 @@ def test_sync_cache_pruned_on_remove(library: Library): @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) -def test_sync_duplicate_case_collision_treated_as_unlinked(library: Library): - """A duplicate entry displaced by a case-insensitive collision must be treated as unlinked.""" +def test_sync_signals_repair_phase_when_entries_are_unlinked(library: Library): + """`sync_dir()` must yield a (-1, -1) signal before repair work when entries are unlinked.""" library_dir = unwrap(library.library_dir) engine = LibrarySyncEngine(library=library) - library.is_case_sensitive_fs = False # Force a collision using a case difference + + # The baseline entries never existed on disk, so they're always unlinked here + progress = list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert (-1, -1) in progress + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_duplicate_case_collision_merged(library: Library): + """A duplicate entry displaced by a case-insensitive collision must be merged automatically.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + library.is_case_sensitive_fs = False # Force a case-insensitive collision (library_dir / "Dupe").mkdir() (library_dir / "Dupe" / "photo.jpg").touch() @@ -172,103 +142,235 @@ def test_sync_duplicate_case_collision_treated_as_unlinked(library: Library): original_id = entry_b_id if dupe_id == entry_a_id else entry_a_id assert cache.get(path_key) == original_id - # The dupe ID should be marked as unlinked, and the original ID should not + entries_before = library.entries_count list(engine.sync_dir(library_dir, force_internal_scanner=True)) - unlinked_ids = {e.id for e in engine.unlinked_entries} - assert dupe_id in unlinked_ids - assert original_id not in unlinked_ids - - # Removing the duplicate entry shouldn't remove the original from the cache - engine.unlinked_entries = [e for e in engine.unlinked_entries if e.id == dupe_id] - engine.remove_unlinked_entries() + assert engine.relinked_entries_count == 1 + assert engine.relinked_entries[0].id == dupe_id + assert dupe_id not in {e.id for e in engine.unlinked_entries} + # A merge deletes the duplicate outright, rather than just reassigning its path + assert library.entries_count == entries_before - 1 assert cache.get(path_key) == original_id assert library.duplicate_path_entry_ids == [] @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) -def test_sync_auto_relink_moved_file(library: Library): - """A moved file (same filename + stats, different path) must auto-relink to its entry.""" +def test_sync_unlinked_entries(library: Library): + """`relink_unlinked_entries()` must relink an entry to its one matching, unclaimed file.""" library_dir = unwrap(library.library_dir) engine = LibrarySyncEngine(library=library) - (library_dir / "moveme.txt").touch() + (library_dir / "sub").mkdir() + (library_dir / "sub" / "found.txt").touch() + library.add_entries([Entry(path=Path("found.txt"), fields=[])]) + entry_id = library.get_entry_id_from_path(Path("found.txt")) + assert entry_id >= 0 + # Skip sync_dir() to avoid a scan, and manually create unlinked entries to test + engine.unlinked_entries = [unwrap(library.get_entry_full(entry_id))] + + list(engine.relink_unlinked_entries()) + assert engine.manual_relink_count == 1 + assert engine.unlinked_entries == [] + assert library.get_entry_id_from_path(Path("sub/found.txt")) == entry_id + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_relink_ignores_files_already_linked_elsewhere(library: Library): + """`find_relink_candidates()` must not offer a file linked elsewhere at a different path.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "kept").mkdir() + (library_dir / "kept" / "target.txt").touch() list(engine.sync_dir(library_dir, force_internal_scanner=True)) list(engine.save_new_entries()) - original_id = library.get_entry_id_from_path(Path("moveme.txt")) - assert original_id >= 0 + kept_id = library.get_entry_id_from_path(Path("kept/target.txt")) + assert kept_id >= 0 - (library_dir / "sub").mkdir() - (library_dir / "moveme.txt").rename(library_dir / "sub" / "moveme.txt") + # A different, unrelated entry that happens to share a filename with the kept file above + library.add_entries([Entry(path=Path("old/target.txt"), fields=[])]) + orphan_id = library.get_entry_id_from_path(Path("old/target.txt")) + # Skip sync_dir() to avoid a scan, and manually create unlinked entries to test + engine.unlinked_entries = [unwrap(library.get_entry_full(orphan_id))] + list(engine.relink_unlinked_entries()) + assert engine.manual_relink_count == 0 + assert len(engine.unlinked_entries) == 1 + # The kept entry must be untouched, no merge should have occurred + assert library.get_entry_id_from_path(Path("kept/target.txt")) == kept_id + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_relink_merges_nfc_nfd_duplicate(library: Library): + """`relink_unlinked_entries()` must merge a duplicate entry differing only by Unicode form. + + PathType always normalizes to NFD, so this test uses raw SQL to simulate legacy data + from before that rule existed. + """ + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + nfc_name = unicodedata.normalize("NFC", "SKÅL.txt") + nfd_name = unicodedata.normalize("NFD", "SKÅL.txt") + assert nfc_name != nfd_name + + # The real file lands in NFD form, as most filesystems store it regardless of input form + (library_dir / nfd_name).touch() + entry_a_id, entry_b_id = library.add_entries( + [ + Entry(path=Path(nfd_name), fields=[]), + Entry(path=Path("placeholder.txt"), fields=[]), + ] + ) + with Session(library.engine) as session: + session.execute( + text("UPDATE entries SET path = :path WHERE id = :id"), + {"path": nfc_name, "id": entry_b_id}, + ) + session.commit() + library.path_cache = None # Force a rebuild to pick up the raw SQL change + + library.get_or_build_path_cache() + assert library.duplicate_path_entry_ids is not None + assert len(library.duplicate_path_entry_ids) == 1 + dupe_id = library.duplicate_path_entry_ids[0] + kept_id = entry_b_id if dupe_id == entry_a_id else entry_a_id + # Skip sync_dir() to isolate manual relink + engine.unlinked_entries = [unwrap(library.get_entry_full(dupe_id))] + + entries_before = library.entries_count + list(engine.relink_unlinked_entries()) + assert engine.manual_relink_count == 1 + assert dupe_id not in {e.id for e in engine.unlinked_entries} + # A merge deletes the duplicate outright, rather than just reassigning its path + assert library.entries_count == entries_before - 1 + assert unwrap(library.get_entry_full(kept_id)).id == kept_id + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_merges_nfc_nfd_duplicate(library: Library): + """A duplicate entry differing only by Unicode form must be merged automatically. + + PathType always normalizes to NFD, so this test uses raw SQL to simulate legacy data + from before that rule existed. + """ + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + nfc_name = unicodedata.normalize("NFC", "SKÅL.txt") + nfd_name = unicodedata.normalize("NFD", "SKÅL.txt") + assert nfc_name != nfd_name + + # The real file lands in NFD form, as most filesystems store it regardless of input form + (library_dir / nfd_name).touch() + entry_a_id, entry_b_id = library.add_entries( + [ + Entry(path=Path(nfd_name), fields=[]), + Entry(path=Path("placeholder.txt"), fields=[]), + ] + ) + with Session(library.engine) as session: + session.execute( + text("UPDATE entries SET path = :path WHERE id = :id"), + {"path": nfc_name, "id": entry_b_id}, + ) + session.commit() + library.path_cache = None # Force a rebuild to pick up the raw SQL change + + library.get_or_build_path_cache() + assert library.duplicate_path_entry_ids is not None + assert len(library.duplicate_path_entry_ids) == 1 + dupe_id = library.duplicate_path_entry_ids[0] + kept_id = entry_b_id if dupe_id == entry_a_id else entry_a_id + + entries_before = library.entries_count list(engine.sync_dir(library_dir, force_internal_scanner=True)) - assert engine.new_paths == [] assert engine.relinked_entries_count == 1 - assert engine.relinked_entries[0].id == original_id - assert Path("sub/moveme.txt") not in {e.path for e in engine.unlinked_entries} - assert library.get_entry_id_from_path(Path("sub/moveme.txt")) == original_id - # The original path should no longer be associated with an entry - assert library.get_entry_id_from_path(Path("moveme.txt")) == -1 + assert engine.relinked_entries[0].id == dupe_id + assert dupe_id not in {e.id for e in engine.unlinked_entries} + assert library.entries_count == entries_before - 1 + assert unwrap(library.get_entry_full(kept_id)).id == kept_id @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) -def test_sync_auto_relink_moved_file_ambiguous(library: Library): - """Two equally-matching candidates (same filename + stats) must not auto-relink. - - NOTE: This may change in the future as capability expands. - """ +def test_sync_case_sensitivity_aware_relink(library: Library): + """`find_relink_candidates()` must respect the library's case-sensitivity setting.""" library_dir = unwrap(library.library_dir) engine = LibrarySyncEngine(library=library) - (library_dir / "a").mkdir() - (library_dir / "b").mkdir() - (library_dir / "a" / "dupe.txt").touch() - (library_dir / "b" / "dupe.txt").touch() - st = (library_dir / "a" / "dupe.txt").stat() - os.utime(library_dir / "b" / "dupe.txt", (st.st_atime, st.st_mtime)) + (library_dir / "Other").mkdir() + (library_dir / "Other" / "name.txt").touch() + library.add_entries([Entry(path=Path("Folder/Name.txt"), fields=[])]) + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + unlinked_entry = next(e for e in engine.unlinked_entries if e.path == Path("Folder/Name.txt")) + + library.is_case_sensitive_fs = True + assert engine.find_relink_candidates(unlinked_entry) == [] + + library.is_case_sensitive_fs = False + engine._filename_to_path_map = None # Rebuild the index with the new case sensitivity + assert engine.find_relink_candidates(unlinked_entry) == [Path("Other/name.txt")] + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_modified_file_stays_linked(library: Library): + """[Case #0] Modifying a file's content in place must not create an unlinked entry.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "stable.txt").touch() list(engine.sync_dir(library_dir, force_internal_scanner=True)) list(engine.save_new_entries()) - (library_dir / "a" / "dupe.txt").unlink() - (library_dir / "b" / "dupe.txt").unlink() - (library_dir / "c").mkdir() - (library_dir / "c" / "dupe.txt").touch() - os.utime(library_dir / "c" / "dupe.txt", (st.st_atime, st.st_mtime)) + (library_dir / "stable.txt").write_text("modified content") + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert Path("stable.txt") not in {e.path for e in engine.unlinked_entries} + assert engine.new_paths == [] + assert len(engine.paths_to_restat) == 1 + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_deleted_file(library: Library): + """[Case #1] A deleted file with no candidates anywhere must not auto-relink.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "gone.txt").touch() + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + + (library_dir / "gone.txt").unlink() list(engine.sync_dir(library_dir, force_internal_scanner=True)) assert engine.relinked_entries_count == 0 - assert Path("c/dupe.txt") in engine.new_paths - unlinked_paths = {e.path for e in engine.unlinked_entries} - assert Path("a/dupe.txt") in unlinked_paths - assert Path("b/dupe.txt") in unlinked_paths + assert engine.new_paths == [] + assert Path("gone.txt") in {e.path for e in engine.unlinked_entries} @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) -def test_sync_auto_relink_moved_file_stat_mismatch(library: Library): - """A file with (same filename, different stats) must not auto-relink. - - NOTE: This is a limitation of the current auto-relinking system, NOT a design principle. - """ +def test_sync_auto_relink_moved_and_renamed_file_modified(library: Library): + """[Case #2] A moved, renamed, and modified file must not auto-relink.""" library_dir = unwrap(library.library_dir) engine = LibrarySyncEngine(library=library) - (library_dir / "notmoved.txt").touch() + (library_dir / "original_name.txt").touch() list(engine.sync_dir(library_dir, force_internal_scanner=True)) list(engine.save_new_entries()) - (library_dir / "notmoved.txt").unlink() + (library_dir / "original_name.txt").unlink() (library_dir / "sub").mkdir() - (library_dir / "sub" / "notmoved.txt").touch() + (library_dir / "sub" / "renamed.txt").write_text("different content, different size") list(engine.sync_dir(library_dir, force_internal_scanner=True)) assert engine.relinked_entries_count == 0 - assert Path("sub/notmoved.txt") in engine.new_paths - assert Path("notmoved.txt") in {e.path for e in engine.unlinked_entries} + assert Path("sub/renamed.txt") in engine.new_paths + assert Path("original_name.txt") in {e.path for e in engine.unlinked_entries} @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) -def test_sync_auto_relink_renamed_file(library: Library): - """A renamed file (different filename, same stats) must auto-relink to its entry.""" +def test_sync_auto_relink_moved_and_renamed_file(library: Library): + """[Case #3] A single file that is both moved and renamed must still auto-relink.""" library_dir = unwrap(library.library_dir) engine = LibrarySyncEngine(library=library) @@ -278,27 +380,52 @@ def test_sync_auto_relink_renamed_file(library: Library): original_id = library.get_entry_id_from_path(Path("original_name.txt")) assert original_id >= 0 - (library_dir / "original_name.txt").rename(library_dir / "renamed.txt") + (library_dir / "sub").mkdir() + (library_dir / "original_name.txt").rename(library_dir / "sub" / "renamed.txt") list(engine.sync_dir(library_dir, force_internal_scanner=True)) assert engine.new_paths == [] assert engine.relinked_entries_count == 1 assert engine.relinked_entries[0].id == original_id - assert Path("renamed.txt") not in {e.path for e in engine.unlinked_entries} - assert library.get_entry_id_from_path(Path("renamed.txt")) == original_id - # The original path should no longer be associated with an entry + assert library.get_entry_id_from_path(Path("sub/renamed.txt")) == original_id assert library.get_entry_id_from_path(Path("original_name.txt")) == -1 @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) -def test_sync_auto_relink_renamed_file_ambiguous(library: Library): - """Two equally-matching candidates (different filenames, same stats) must not auto-relink. +def test_sync_auto_relink_single_entry_multiple_moved_and_renamed_files(library: Library): + """[Case #4] A single entry must not auto-relink to 2+ equally-matching moved+renamed files. - NOTE: This may be automated in the future, but for now differs to the user's judgement. + NOTE: This may change in the future as capability expands. """ library_dir = unwrap(library.library_dir) engine = LibrarySyncEngine(library=library) + (library_dir / "original_name.txt").touch() + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + original_stat = (library_dir / "original_name.txt").stat() + + (library_dir / "original_name.txt").unlink() + (library_dir / "a").mkdir() + (library_dir / "b").mkdir() + (library_dir / "a" / "renamed1.txt").touch() + (library_dir / "b" / "renamed2.txt").touch() + os.utime(library_dir / "a" / "renamed1.txt", (original_stat.st_atime, original_stat.st_mtime)) + os.utime(library_dir / "b" / "renamed2.txt", (original_stat.st_atime, original_stat.st_mtime)) + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.relinked_entries_count == 0 + assert Path("a/renamed1.txt") in engine.new_paths + assert Path("b/renamed2.txt") in engine.new_paths + assert Path("original_name.txt") in {e.path for e in engine.unlinked_entries} + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_multiple_entries_one_moved_and_renamed_file(library: Library): + """[Case #5] Two or more entries must not auto-relink to one matching moved+renamed file.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + (library_dir / "a").mkdir() (library_dir / "b").mkdir() (library_dir / "a" / "one.txt").touch() @@ -324,8 +451,239 @@ def test_sync_auto_relink_renamed_file_ambiguous(library: Library): @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) -def test_sync_auto_relink_moved_and_renamed_together(library: Library): - """A moved file and a renamed file in the same sync must each auto-relink via their own pass.""" +def test_sync_auto_relink_moved_file_modified(library: Library): + """[Case #6] A moved + modified file must still auto-relink by its unique filename.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "notmoved.txt").touch() + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + original_id = library.get_entry_id_from_path(Path("notmoved.txt")) + assert original_id >= 0 + + (library_dir / "notmoved.txt").unlink() + (library_dir / "sub").mkdir() + (library_dir / "sub" / "notmoved.txt").write_text("different content, different size") + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.new_paths == [] + assert engine.relinked_entries_count == 1 + assert engine.relinked_entries[0].id == original_id + assert library.get_entry_id_from_path(Path("sub/notmoved.txt")) == original_id + # The original path should no longer be associated with an entry + assert library.get_entry_id_from_path(Path("notmoved.txt")) == -1 + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_moved_file(library: Library): + """[Case #7] A moved file (same filename + stats, different path) must auto-relink.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "moveme.txt").touch() + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + original_id = library.get_entry_id_from_path(Path("moveme.txt")) + assert original_id >= 0 + + (library_dir / "sub").mkdir() + (library_dir / "moveme.txt").rename(library_dir / "sub" / "moveme.txt") + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.new_paths == [] + assert engine.relinked_entries_count == 1 + assert engine.relinked_entries[0].id == original_id + assert Path("sub/moveme.txt") not in {e.path for e in engine.unlinked_entries} + assert library.get_entry_id_from_path(Path("sub/moveme.txt")) == original_id + # The original path should no longer be associated with an entry + assert library.get_entry_id_from_path(Path("moveme.txt")) == -1 + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_moved_files_same_names(library: Library): + """[Case #7] A file candidate with same name + different stats must not block the real match.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "photo.jpg").touch() + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + original_id = library.get_entry_id_from_path(Path("photo.jpg")) + assert original_id >= 0 + original_stat = (library_dir / "photo.jpg").stat() + + (library_dir / "a").mkdir() + (library_dir / "b").mkdir() + (library_dir / "photo.jpg").rename(library_dir / "a" / "photo.jpg") + (library_dir / "b" / "photo.jpg").touch() + # Ensure a different mtime than "a/photo.jpg" + os.utime( + library_dir / "b" / "photo.jpg", (original_stat.st_atime, original_stat.st_mtime + 100) + ) + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.relinked_entries_count == 1 + assert engine.relinked_entries[0].id == original_id + assert library.get_entry_id_from_path(Path("a/photo.jpg")) == original_id + assert Path("b/photo.jpg") in engine.new_paths + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_single_entry_multiple_moved_files(library: Library): + """[Case #8] A single entry must not auto-relink to multiple equally-matching file candidates. + + NOTE: This may change in the future as capability expands. + """ + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "photo.jpg").touch() + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + assert library.get_entry_id_from_path(Path("photo.jpg")) >= 0 + original_stat = (library_dir / "photo.jpg").stat() + + (library_dir / "a").mkdir() + (library_dir / "b").mkdir() + (library_dir / "photo.jpg").rename(library_dir / "a" / "photo.jpg") + (library_dir / "b" / "photo.jpg").touch() + os.utime(library_dir / "b" / "photo.jpg", (original_stat.st_atime, original_stat.st_mtime)) + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.relinked_entries_count == 0 + assert Path("a/photo.jpg") in engine.new_paths + assert Path("b/photo.jpg") in engine.new_paths + assert Path("photo.jpg") in {e.path for e in engine.unlinked_entries} + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_multiple_entries_one_moved_file(library: Library): + """[Case #9] Two or more entries must not auto-relink to one matching file candidate.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "a").mkdir() + (library_dir / "b").mkdir() + (library_dir / "a" / "dupe.txt").touch() + (library_dir / "b" / "dupe.txt").touch() + st = (library_dir / "a" / "dupe.txt").stat() + os.utime(library_dir / "b" / "dupe.txt", (st.st_atime, st.st_mtime)) + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + + (library_dir / "a" / "dupe.txt").unlink() + (library_dir / "b" / "dupe.txt").unlink() + (library_dir / "c").mkdir() + (library_dir / "c" / "dupe.txt").touch() + os.utime(library_dir / "c" / "dupe.txt", (st.st_atime, st.st_mtime)) + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.relinked_entries_count == 0 + assert Path("c/dupe.txt") in engine.new_paths + unlinked_paths = {e.path for e in engine.unlinked_entries} + assert Path("a/dupe.txt") in unlinked_paths + assert Path("b/dupe.txt") in unlinked_paths + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_renamed_file_modified(library: Library): + """[Case #10] A renamed file with different stats must not auto-relink.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "original_name.txt").touch() + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + + (library_dir / "original_name.txt").unlink() + (library_dir / "renamed_and_modified.txt").write_text("different content, different size") + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.relinked_entries_count == 0 + assert Path("renamed_and_modified.txt") in engine.new_paths + assert Path("original_name.txt") in {e.path for e in engine.unlinked_entries} + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_renamed_file(library: Library): + """[Case #11] A renamed file (different filename, same stats) must auto-relink to its entry.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "original_name.txt").touch() + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + original_id = library.get_entry_id_from_path(Path("original_name.txt")) + assert original_id >= 0 + + (library_dir / "original_name.txt").rename(library_dir / "renamed.txt") + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.new_paths == [] + assert engine.relinked_entries_count == 1 + assert engine.relinked_entries[0].id == original_id + assert Path("renamed.txt") not in {e.path for e in engine.unlinked_entries} + assert library.get_entry_id_from_path(Path("renamed.txt")) == original_id + # The original path should no longer be associated with an entry + assert library.get_entry_id_from_path(Path("original_name.txt")) == -1 + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_single_entry_multiple_renamed_files(library: Library): + """[Case #12] A single entry must not auto-relink to 2+ matched renamed files.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "original_name.txt").touch() + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + original_stat = (library_dir / "original_name.txt").stat() + + (library_dir / "original_name.txt").unlink() + (library_dir / "renamed1.txt").touch() + (library_dir / "renamed2.txt").touch() + os.utime(library_dir / "renamed1.txt", (original_stat.st_atime, original_stat.st_mtime)) + os.utime(library_dir / "renamed2.txt", (original_stat.st_atime, original_stat.st_mtime)) + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.relinked_entries_count == 0 + assert Path("renamed1.txt") in engine.new_paths + assert Path("renamed2.txt") in engine.new_paths + assert Path("original_name.txt") in {e.path for e in engine.unlinked_entries} + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_multiple_entries_one_renamed_file(library: Library): + """[Case #13] Two or more entries must not auto-relink to one matching renamed file.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "one.txt").touch() + (library_dir / "two.txt").touch() + st = (library_dir / "one.txt").stat() + os.utime(library_dir / "two.txt", (st.st_atime, st.st_mtime)) + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + + (library_dir / "one.txt").unlink() + (library_dir / "two.txt").unlink() + (library_dir / "renamed.txt").touch() + os.utime(library_dir / "renamed.txt", (st.st_atime, st.st_mtime)) + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.relinked_entries_count == 0 + assert Path("renamed.txt") in engine.new_paths + unlinked_paths = {e.path for e in engine.unlinked_entries} + assert Path("one.txt") in unlinked_paths + assert Path("two.txt") in unlinked_paths + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_separate_moved_and_renamed_files(library: Library): + """[Cases #7, #11] Two separate files, one moved and one renamed, must each auto-relink.""" library_dir = unwrap(library.library_dir) engine = LibrarySyncEngine(library=library) @@ -350,6 +708,47 @@ def test_sync_auto_relink_moved_and_renamed_together(library: Library): assert library.get_entry_id_from_path(Path("rename_new.txt")) == renamed_id +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_matches_entry_without_stored_stats_by_filename(library: Library): + """An entry with no stored stats must auto-relink to a single matching filename.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + library.add_entries([Entry(path=Path("legacy.txt"), fields=[])]) + original_id = library.get_entry_id_from_path(Path("legacy.txt")) + assert original_id >= 0 + + (library_dir / "sub").mkdir() + (library_dir / "sub" / "legacy.txt").touch() + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.new_paths == [] + assert engine.relinked_entries_count == 1 + assert engine.relinked_entries[0].id == original_id + assert Path("legacy.txt") not in {e.path for e in engine.unlinked_entries} + assert library.get_entry_id_from_path(Path("sub/legacy.txt")) == original_id + # The original path should no longer be associated with an entry + assert library.get_entry_id_from_path(Path("legacy.txt")) == -1 + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_auto_relink_skips_ambiguous_entry_without_stored_stats(library: Library): + """An entry with no stored stats must not auto-relink to 2+ files with the same name.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + library.add_entries([Entry(path=Path("legacy.txt"), fields=[])]) + + (library_dir / "a").mkdir() + (library_dir / "b").mkdir() + (library_dir / "a" / "legacy.txt").touch() + (library_dir / "b" / "legacy.txt").touch() + + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + assert engine.relinked_entries_count == 0 + assert Path("a/legacy.txt") in engine.new_paths + assert Path("b/legacy.txt") in engine.new_paths + assert Path("legacy.txt") in {e.path for e in engine.unlinked_entries} + + @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) def test_sync_cancel_skips_unlinked_finalization(library: Library): """A cancelled `sync_dir()` call must not mark any entries as unlinked.""" @@ -368,6 +767,69 @@ def test_sync_cancel_skips_unlinked_finalization(library: Library): assert engine.new_paths == [] +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_cancel_skips_duplicate_merge(library: Library): + """Cancelling right after the scan must stop duplicate merging before it starts.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + library.is_case_sensitive_fs = False # Force a collision using a case difference + + (library_dir / "Dupe").mkdir() + (library_dir / "Dupe" / "photo.jpg").touch() + library.add_entries( + [ + Entry(path=Path("Dupe/photo.jpg"), fields=[]), + Entry(path=Path("Dupe/PHOTO.JPG"), fields=[]), + ] + ) + library.get_or_build_path_cache() + assert library.duplicate_path_entry_ids is not None + assert len(library.duplicate_path_entry_ids) == 1 + dupe_id = library.duplicate_path_entry_ids[0] + + # Cancel right as the repair phase signal comes through, before merging actually runs + generator = engine.sync_dir(library_dir, force_internal_scanner=True) + for progress in generator: + if progress == (-1, -1): + engine.cancelled = True + break + list(generator) + + assert engine.relinked_entries_count == 0 + assert dupe_id in {e.id for e in engine.unlinked_entries} + + +@pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) +def test_sync_cancel_skips_remaining_relink_passes(library: Library): + """Cancelling right after the scan must let Pass 1 finish, but stop before Pass 2.""" + library_dir = unwrap(library.library_dir) + engine = LibrarySyncEngine(library=library) + + (library_dir / "moveme.txt").touch() + (library_dir / "original_name.txt").touch() + list(engine.sync_dir(library_dir, force_internal_scanner=True)) + list(engine.save_new_entries()) + moved_id = library.get_entry_id_from_path(Path("moveme.txt")) + renamed_id = library.get_entry_id_from_path(Path("original_name.txt")) + + (library_dir / "sub").mkdir() + (library_dir / "moveme.txt").rename(library_dir / "sub" / "moveme.txt") # Pass 1 match + (library_dir / "original_name.txt").rename(library_dir / "renamed.txt") # Pass 2 match + + # Cancel right as the repair phase signal comes through, before relinking actually runs + generator = engine.sync_dir(library_dir, force_internal_scanner=True) + for progress in generator: + if progress == (-1, -1): + engine.cancelled = True + break + list(generator) + + relinked_ids = {e.id for e in engine.relinked_entries} + assert moved_id in relinked_ids + assert renamed_id not in relinked_ids + assert Path("renamed.txt") in engine.new_paths + + @pytest.mark.parametrize("library", [TemporaryDirectory()], indirect=True) def test_sync_cancel_skips_save_new_entries(library: Library): """A cancelled `save_new_entries()` call must not save any new entries."""