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.
This commit is contained in:
Vito Sansevero
+30
-1
@@ -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"])
|
||||
|
||||
@@ -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"))
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user