fix(monitor): skip trash folders only below the watched folder (#501 review)

A source named .Trash-1000 (or another trash name) was ignored by the
image monitor while every walker still listed it; the monitor now tests
the folders below the longest watched folder holding the new file.

TESTING.md

@@ -2479,9 +2479,8 @@ files to the system trash. **No code path unlinks or overwrites a media file.**
   `$RECYCLE.BIN` (any case), as none descends `thumbnails`: the workspace and legacy
   listings (`_walk_media`, `_settled_entry`), the metadata scan and duplicate find
   (`_collect_media_files`), the legacy auto-tag listing (`_collect_scan_files`) and the
-  delete review (`trash-folder`). The image monitor tests every folder of the new file's
-  absolute path, so it also ignores outputs written into a source that itself lies
-  inside such a folder. Tests: `tests/test_media_folders.py`,
+  delete review (`trash-folder`), and the image monitor, which tests the folders below the
+  watched folder holding the new file (a source named like a trash folder is still watched). Tests: `tests/test_media_folders.py`,
   `tests/test_image_removal_gaps.py`.
 - **Trash backends** (`utils/trash.py`, no dependency): Linux and BSD implement the
   freedesktop Trash specification (home trash on the same mount, else the mount's

tests/test_image_removal_gaps.py

@@ -239,6 +239,16 @@ class TestWalkersSkipTrashFolders(_GapCase):
                     monitor.is_image_file(os.path.join(base, folder, "files", "a.png"))
                 )
 
+    def test_the_image_monitor_watches_a_root_named_like_a_trash_folder(self):
+        monitor = ImageGenerationHandler.__new__(ImageGenerationHandler)
+        monitor.supported_extensions = (".png",)
+        root = os.path.join(str(self.output_dir), ".Trash-1000")
+        monitor.watched_roots = (root,)
+        self.assertTrue(monitor.is_image_file(os.path.join(root, "sub", "a.png")))
+        self.assertFalse(
+            monitor.is_image_file(os.path.join(root, "sub", ".Trashes", "a.png"))
+        )
+
     async def test_the_workspace_listing(self):
         self.seed()
         self.assertEqual(await self.listed(), ["kept.png"])

utils/image_monitor.py

@@ -147,6 +147,10 @@ class ImageGenerationHandler(FileSystemEventHandler):
 
         # Settle check: the file size must be unchanged across two reads
         self.settle_interval = SETTLE_INTERVAL_SECONDS
+        # The folders this handler is scheduled on (#501): the folder skip only
+        # tests the folders below them, so a source named like a trash folder
+        # is still watched.
+        self.watched_roots = ()
         self.settle_attempts = SETTLE_ATTEMPTS
 
         # Images scheduled but not yet processed: path -> (prompt snapshot, due time).
@@ -433,11 +437,28 @@ class ImageGenerationHandler(FileSystemEventHandler):
         """
         # Previews are derivatives, and a trash folder inside an output
         # folder holds thrown-away files (#501): neither is a new output.
-        folders = filepath.replace("\\", "/").split("/")[:-1]
-        if in_skipped_folder(folders):
+        if in_skipped_folder(self._folders_below_root(filepath)):
             return False
         return filepath.lower().endswith(self.supported_extensions)
 
+    def _folders_below_root(self, filepath: str) -> list:
+        """The folder names of *filepath* below the longest watched root.
+
+        Without a watched root holding it (a handler used on its own), every
+        folder of the path counts.
+        """
+        text = os.path.abspath(filepath)
+        holding = [
+            root
+            for root in getattr(self, "watched_roots", ())
+            if os.path.normcase(text).startswith(
+                os.path.normcase(os.path.join(root, ""))
+            )
+        ]
+        if holding:
+            text = text[len(os.path.join(max(holding, key=len), "")) :]
+        return text.replace("\\", "/").split("/")[:-1]
+
     def process_new_image(self, image_path: str, prompt_snapshot=None):
         """Process a newly created image file for gallery integration.
 
@@ -1238,6 +1259,7 @@ class ImageMonitor:
         handler.review_recorder = self.review_recorder
         handler.thumbnail_queue = self.thumbnail_queue
         handler.output_listener = self.output_listener
+        handler.watched_roots = tuple(os.path.abspath(d) for d in valid_directories)
         observer = Observer()
         scheduled = []
         for output_dir in valid_directories: