tests: Add follow symlink tests (gorakhargosh#1087)

Co-authored-by: Corentin <[email protected]>

Author: Corentin-pro

src/watchdog/observers/fsevents.py

@@ -337,5 +337,6 @@ def schedule(
         if isinstance(path, str):
             path = unicodedata.normalize("NFC", path)
 
-        return super().schedule(event_handler, path, recursive=recursive, follow_symlink=follow_symlink,
-                                event_filter=event_filter)
+        return super().schedule(
+            event_handler, path, recursive=recursive, follow_symlink=follow_symlink, event_filter=event_filter
+        )

tests/shell.py

@@ -42,6 +42,10 @@ def mkdir(path, *, parents=False):
         os.mkdir(path)
 
 
+def symlink(source, destination, *, target_is_directory: bool = False):
+    os.symlink(source, destination, target_is_directory=target_is_directory)
+
+
 def rm(path, *, recursive=False):
     """Deletes files or directories."""
     if os.path.isdir(path):

tests/test_inotify_buffer.py

@@ -13,7 +13,7 @@
 
 from watchdog.observers.inotify_buffer import InotifyBuffer
 
-from .shell import mkdir, mount_tmpfs, mv, rm, touch, unmount
+from .shell import mkdir, mount_tmpfs, mv, rm, symlink, touch, unmount
 
 
 def wait_for_move_event(read_event):
@@ -65,6 +65,21 @@ def test_move_internal(p):
     inotify.close()
 
 
+@pytest.mark.timeout(5)
+def test_move_internal_symlink_followed(p):
+    mkdir(p("dir", "dir1"), parents=True)
+    mkdir(p("dir", "dir2"))
+    touch(p("dir", "dir1", "a"))
+    symlink(p("dir"), p("symdir"), target_is_directory=True)
+
+    inotify = InotifyBuffer(p("symdir").encode(), recursive=True, follow_symlink=True)
+    mv(p("dir", "dir1", "a"), p("dir", "dir2", "b"))
+    frm, to = wait_for_move_event(inotify.read_event)
+    assert frm.src_path == p("symdir", "dir1", "a").encode()
+    assert to.src_path == p("symdir", "dir2", "b").encode()
+    inotify.close()
+
+
 @pytest.mark.timeout(10)
 def test_move_internal_batch(p):
     n = 100
@@ -102,6 +117,21 @@ def test_delete_watched_directory(p):
     inotify.close()
 
 
+@pytest.mark.timeout(5)
+def test_delete_watched_directory_symlink_followed(p):
+    mkdir(p("dir", "dir2"), parents=True)
+    symlink(p("dir"), p("symdir"), target_is_directory=True)
+
+    inotify = InotifyBuffer(p("symdir").encode(), follow_symlink=True)
+    rm(p("dir", "dir2"), recursive=True)
+
+    # Wait for the event to be picked up
+    inotify.read_event()
+
+    # Ensure InotifyBuffer shuts down cleanly without raising an exception
+    inotify.close()
+
+
 @pytest.mark.timeout(5)
 @pytest.mark.skipif("GITHUB_REF" not in os.environ, reason="sudo password prompt")
 def test_unmount_watched_directory_filesystem(p):