From b84e6ea1117ef7e0f276d928c35e06a2b33e7459 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Thu, 22 Oct 2026 17:37:00 -0700 Subject: [PATCH] fix(review): refresh the recorded spelling when a file is seen under a new one A case-only rename on Windows or macOS keeps size and mtime_ns, so the rescan skipped the file and the item kept the old spelling. Its target was then formatted from a name the directory no longer lists, the capture refused it, and the item stayed pending with no way to apply it. The skip path and a duplicate insert now record the current spelling for every item at the locator and move the review index revision; item revisions do not change, so an open decision stays valid. Tested under case-folding semantics on every platform. --- database/review.py | 31 ++++++++++- tests/test_review.py | 118 ++++++++++++++++++++++++++++++++++++++++ utils/review_capture.py | 11 +++- 3 files changed, 156 insertions(+), 4 deletions(-) diff --git a/database/review.py b/database/review.py index db7df29..46b6b31 100644 --- a/database/review.py +++ b/database/review.py @@ -50,6 +50,13 @@ _BUMP_REVIEW_INDEX = ( f"UPDATE {METADATA_TABLE} SET " # nosec B608 "review_index_revision=review_index_revision+1 WHERE singleton_id=1" ) +# A later spelling of the same locator (a case-only rename on a case-folding +# filesystem) replaces the recorded one, so targets keep formatting. +_REFRESH_SPELLING = ( + f"UPDATE {ITEMS_TABLE} SET spelled_path=? " # nosec B608 + f"WHERE spelled_path<>? AND byte_id IN (SELECT b.id FROM {BYTES_TABLE} b " + f"JOIN {FILES_TABLE} f ON f.id=b.file_id WHERE f.locator=?)" +) _REVIEW_INDEX = ( f"SELECT review_index_revision FROM {METADATA_TABLE} " # nosec B608 "WHERE singleton_id=1" @@ -276,10 +283,28 @@ class ReviewOperations: ), ) inserted = cursor.rowcount > 0 - if inserted: + refreshed = _refresh_spelling(conn, locator, spelled_path) + if inserted or refreshed: conn.execute(_BUMP_REVIEW_INDEX) return inserted + def refresh_review_spelling(self, locator: str, spelled_path: str) -> bool: + """Record ``spelled_path`` as the spelling of every item at ``locator``. + + The rescan's skip calls it for a file it does not copy again: a + case-only rename keeps size and ``mtime_ns``, and a target formatted + from the old spelling would be refused. True when a spelling changed + (the review index revision moves; item revisions do not, so an open + decision stays valid). + """ + _validate_locator(locator) + _validate_spelling(spelled_path, locator) + with self._pair_transaction() as conn: + refreshed = _refresh_spelling(conn, locator, spelled_path) + if refreshed: + conn.execute(_BUMP_REVIEW_INDEX) + return refreshed + def review_index_revision(self) -> int: """The queue's revision (moved by every queue change).""" row = self.model.get_connection().execute(_REVIEW_INDEX).fetchone() @@ -388,6 +413,10 @@ class ReviewOperations: return found +def _refresh_spelling(conn: sqlite3.Connection, locator: str, spelled: str) -> bool: + return conn.execute(_REFRESH_SPELLING, (spelled, spelled, locator)).rowcount > 0 + + def _queue_row(row, columns) -> dict: item = dict(zip(columns, tuple(row))) item["limited"] = bool(item["limited"]) diff --git a/tests/test_review.py b/tests/test_review.py index 2ad873c..cf1d277 100644 --- a/tests/test_review.py +++ b/tests/test_review.py @@ -11,6 +11,7 @@ import asyncio import hashlib import json import logging +import contextlib import os import sqlite3 import unittest @@ -774,3 +775,120 @@ class TestMonitorRecording(ReviewTestCase): if __name__ == "__main__": unittest.main() + + +class _FoldingPath: + """``os.path`` of a case-folding filesystem (Windows, macOS APFS).""" + + def __getattr__(self, name): + return getattr(os.path, name) + + @staticmethod + def normcase(value): + return os.path.normcase(value).lower() + + +class _FoldingOs: + path = _FoldingPath() + + def __getattr__(self, name): + return getattr(os, name) + + +def _case_folding(): + """Fold case where locators are formed: recording, the store, the database.""" + import database.review as review_module + import utils.media_capture as capture_module + + stack = contextlib.ExitStack() + for module in (review_module, capture_module, review_capture): + stack.enter_context(patch.object(module, "os", _FoldingOs())) + return stack + + +class TestRecordedSpelling(ReviewTestCase): + """A case-only rename refreshes the recorded spelling (I5).""" + + def _spelled(self): + conn = self.api.db.model.get_connection() + return [ + row[0] + for row in conn.execute("SELECT spelled_path FROM media_review_items") + ] + + async def test_case_only_rename_refreshes_the_spelling_on_the_skip_path(self): + self._require_module() + self._roots() + upper = self.output_dir / "Two.PNG" + write_png(upper, prompt=ambiguous_graph("cat", "dog")) + with _case_folding(): + self.assertEqual(self.api.record_review_file(str(upper)), "recorded") + (before,) = self._items() + revision = self._review_revision() + lower = self.output_dir / "two.png" + os.rename(upper, lower) # keeps size and mtime_ns + self.assertEqual(self.api.record_review_file(str(lower)), "duplicate") + (after,) = self._items() + self.assertEqual(self._spelled(), [str(lower)]) + self.assertEqual(self._review_revision(), revision + 1) + # The item itself is unchanged: an open decision stays valid. + self.assertEqual( + {k: v for k, v in after.items() if k != "spelled_path"}, + {k: v for k, v in before.items() if k != "spelled_path"}, + ) + from py.api.review import format_review_target + + real_root = os.path.realpath(str(self.output_dir)) + self.assertEqual( + format_review_target( + after["spelled_path"], + after["locator"], + [(0, real_root)], + pathmod=_FoldingPath(), + ), + (0, "two.png"), + ) + # The new spelling opens; the recorded one would be refused. + capture = self.api._captures.capture( + self.output_dir, "two.png", epoch="e", scope="s" + ) + self.assertEqual(capture.locator, after["locator"]) + self.api._captures.release(capture.token) + # Seeing the same spelling again writes nothing. + self.assertEqual(self.api.record_review_file(str(lower)), "duplicate") + self.assertEqual(self._review_revision(), revision + 1) + + async def test_duplicate_insert_refreshes_the_spelling(self): + self._require_module() + db = self.api.db + root = os.path.realpath(str(self.output_dir)) + upper, lower = os.path.join(root, "Two.PNG"), os.path.join(root, "two.png") + facts = dict( + sha256="a" * 64, + size=10, + mtime_ns=1, + resolver_version=1, + evidence={"reason": "", "candidates": []}, + candidate_count=0, + limited=True, + ) + with _case_folding(): + locator = _FoldingPath.normcase(upper) + epoch = db.get_execution_epoch() + self.assertTrue( + db.record_review_item( + locator=locator, spelled_path=upper, epoch=epoch, **facts + ) + ) + revision = self._review_revision() + self.assertFalse( + db.record_review_item( + locator=locator, spelled_path=lower, epoch=epoch, **facts + ) + ) + self.assertEqual(self._spelled(), [lower]) + self.assertEqual(self._review_revision(), revision + 1) + self.assertFalse(db.refresh_review_spelling(locator, lower)) + self.assertEqual(self._review_revision(), revision + 1) + with self.assertRaises(ValueError): + db.refresh_review_spelling(locator, os.path.join(root, "other.png")) diff --git a/utils/review_capture.py b/utils/review_capture.py index b476c81..90d0ee0 100644 --- a/utils/review_capture.py +++ b/utils/review_capture.py @@ -113,17 +113,22 @@ def _queued_spelling(db, root, relative: str) -> str | None: with the listed relative path), so it folds to the capture's locator. A file whose size and ``mtime_ns`` equal the verified fingerprint of bytes an item already cites is not copied or hashed again (maintainer ruling: - the rescan stays cheap and the file stays in the default scope). A stat - failure leaves the decision to the capture. + the rescan stays cheap and the file stays in the default scope); its + items take the current spelling. A stat failure leaves the decision to + the capture. """ spelled = os.path.join(os.path.realpath(os.fspath(root)), *relative.split("/")) try: facts = os.lstat(spelled) except OSError: return spelled + locator = os.path.normcase(spelled) if db.review_fingerprint_queued( - os.path.normcase(spelled), facts.st_size, facts.st_mtime_ns, RESOLVER_VERSION + locator, facts.st_size, facts.st_mtime_ns, RESOLVER_VERSION ): + # Not copied again, but a case-only rename keeps the stat: record + # the current spelling so the item's target still opens. + db.refresh_review_spelling(locator, spelled) return None return spelled