Merge pull request #8 from zhuowang10/master

Add recursive watcher

Author: TaylorRichberger

src/asyncinotify/__init__.py

@@ -12,6 +12,7 @@
 from weakref import ReferenceType
 from asyncio import Future
 import select
+from collections import deque
 
 # Python 3.7 suggests get_running_loop for library code
 try:
@@ -418,7 +419,7 @@ def __init__(self,
     @property
     def sync_timeout(self) -> Optional[float]:
         '''The timeout for :meth:`sync_get` and synchronous iteration.
-        
+
         Set this to None to disable and -1 to wait forever.  These options can
         be different depending on the blocking flags selected.
         '''
@@ -649,3 +650,69 @@ def __next__(self) -> Event:
         if event is None:
             raise StopIteration
         return event
+
+class RecursiveWatcher:
+    """
+    watch a folder recursively:
+    add a watch when a folder is created/moved in
+    delete a watch when a folder is deleted/moved out
+    this also works for folders moving within the watched folders because both move_from event and move_to event will be caught
+    """
+    def __init__(self, path, mask) -> None:
+        self._path = path
+        self._mask = mask
+
+    def _get_directories_recursive(self, path):
+        """
+        DFS to iterate all paths within
+        """
+        if not path.is_dir():
+            return
+
+        stack = deque()
+        stack.append(path)
+        while stack:
+            curr_path = stack.pop()
+            yield curr_path
+            for subpath in curr_path.iterdir():
+                if subpath.is_dir():
+                    stack.append(subpath)
+
+    async def watch_recursive(self, inotify=None):
+        create_inotify = inotify is None
+        if create_inotify:
+            inotify = Inotify()
+
+        try:
+            mask = self._mask | Mask.MOVED_FROM | Mask.MOVED_TO | Mask.CREATE | Mask.IGNORED
+            for directory in self._get_directories_recursive(self._path):
+                inotify.add_watch(directory, mask)
+
+            # Things that can throw this off:
+            #
+            # * Doing two changes on a directory or something before the program
+            #   has a time to handle it (this will also throw off a lot of inotify
+            #   code, though)
+            #
+            # * Trying to watch a path that doesn't exist won't automatically
+            #   create it or anything of the sort.
+
+            async for event in inotify:
+                if Mask.ISDIR in event.mask and event.path is not None:
+                    if Mask.CREATE in event.mask or Mask.MOVED_TO in event.mask:
+                        # created new folder or folder moved in, add watches
+                        for directory in self._get_directories_recursive(event.path):
+                            inotify.add_watch(directory, mask)
+                    if Mask.MOVED_FROM in event.mask:
+                        # a folder is moved to another location, remove watch for this folder and subfolders
+                        watches = [watch for watch in inotify._watches.values() if watch.path.is_relative_to(event.path)]
+                        for watch in watches:
+                            inotify.rm_watch(watch)
+
+                    # DELETE event is not watched/handled here because IGNORED event follows deletion, and handled in asyncinotify
+
+                if event.mask & self._mask:
+                    yield event
+        finally:
+            if create_inotify:
+                inotify.close()

test.py

@@ -4,11 +4,14 @@
 # This code is released under the license described in the LICENSE file
 
 import sys
+import os
+import shutil
 
 import unittest
+from unittest import mock
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from asyncinotify import Event, Inotify, Mask
+from asyncinotify import Event, Inotify, Mask, RecursiveWatcher
 
 if sys.version_info >= (3, 9):
     from collections.abc import Sequence
@@ -437,8 +440,270 @@ async def loop(n):
     def test_events(self):
         run(self._actual_test())
 
-        
+class TestRecursiveWatcher(unittest.TestCase):
+
+    @staticmethod
+    def _mock_iterdir(path):
+        name = path.name
+        if name == "tmp":
+            return [Path.joinpath(path, "level1.1"), Path.joinpath(path, "level1.2")]
+        if name == "level1.1":
+            return [Path.joinpath(path, "level2.1"), Path.joinpath(path, "level2.2")]
+        if name == "level2.1":
+            return [Path.joinpath(path, "level3.1")]
+        if name == "level3.1":
+            return [Path.joinpath(path, "level4.1")]
+        # others
+        return []
+
+    @mock.patch.object(Path, "iterdir", autospec=True, side_effect=_mock_iterdir)
+    @mock.patch.object(Path, "is_dir", return_result=True)
+    def test_get_directories_recursive(self, mocked_isdir, mocked_iterdir):
+        """
+        create folder tree as:
+        level1.1
+            -level2.1
+                -level3.1
+                    -level4.1
+            -level2.2
+        level1.2
+        """
+
+        tmpdirname = "/tmp"
+        watcher = RecursiveWatcher(None, None)
+        paths = [path for path in watcher._get_directories_recursive(Path(tmpdirname))]
+        self.assertEqual(len(paths), 7)
+        self.assertEqual(str(paths[0]), tmpdirname)
+        self.assertEqual(str(paths[1]), os.path.join(tmpdirname, "level1.2"))
+        self.assertEqual(str(paths[2]), os.path.join(tmpdirname, "level1.1"))
+        self.assertEqual(str(paths[3]), os.path.join(tmpdirname, "level1.1", "level2.2"))
+        self.assertEqual(str(paths[4]), os.path.join(tmpdirname, "level1.1", "level2.1"))
+        self.assertEqual(str(paths[5]), os.path.join(tmpdirname, "level1.1", "level2.1", "level3.1"))
+        self.assertEqual(str(paths[6]), os.path.join(tmpdirname, "level1.1", "level2.1", "level3.1", "level4.1"))
+
+    def _assert_paths_watched(self, watchers, path_set):
+        watched_path_set = {str(watch.path) for watch in watchers.values()}
+        self.assertSetEqual(watched_path_set, path_set)
+
+    class _FakeWatcher:
+        def __init__(self, path) -> None:
+            self.path = path
+
+    def test_assert_paths_watched(self):
+        # both empty
+        self._assert_paths_watched({}, set())
+
+        # watchers empty
+        with self.assertRaises(AssertionError):
+            self._assert_paths_watched({}, {"/tmp/path1"})
+
+        # path set empty
+        with self.assertRaises(AssertionError):
+            self._assert_paths_watched({
+                "fd1": self._FakeWatcher(Path("/tmp/path1")),
+                "fd2": self._FakeWatcher(Path("/tmp/path2")),
+            }, set())
+
+        # identical sets
+        self._assert_paths_watched({
+            "fd1": self._FakeWatcher(Path("/tmp/path1")),
+            "fd2": self._FakeWatcher(Path("/tmp/path2")),
+        }, {
+            "/tmp/path2",
+            "/tmp/path1"
+        })
+
+        # diff sets
+        with self.assertRaises(AssertionError):
+            self._assert_paths_watched({
+                "fd1": self._FakeWatcher(Path("/tmp/path1")),
+            }, {
+                "/tmp/path2",
+                "/tmp/path1"
+            })
+
+    def _create_file(self, file_path):
+        with open(str(file_path), "w") as f:
+            f.write(file_path)
+
+    async def _read_events(self, inotify, folder, events):
+        watcher = RecursiveWatcher(Path(folder), Mask.CLOSE_WRITE)
+        async for event in watcher.watch_recursive(inotify):
+            # events/watchers are ephemeral, copy data we want
+            events.append((
+                event.path,
+                event.mask,
+            ))
+
+    async def _watch_recursive(self):
+        """
+        test the cases of folder changes:
+        1. create folder
+        2. create cascading folders
+        3. move folder in from un-monitored folder
+        4. move folders out to un-monitored folder
+        5. move folder within monitored folders
+        6. delete folders
+        """
+        with TemporaryDirectory() as tmpdirbasename:
+            events = []
+
+            tmpdirname = os.path.join(tmpdirbasename, "test")
+            os.makedirs(tmpdirname)
+            existing_dir = os.path.join(tmpdirname, "existing_dir")
+            os.makedirs(existing_dir)
+            outside_dir = os.path.join(tmpdirbasename,  "outside")
+            os.makedirs(outside_dir)
+
+            with Inotify() as inotify:
+                watch_task = asyncio.create_task(self._read_events(inotify, tmpdirname, events))
+                await asyncio.sleep(0.3)
+
+                # existing 2 folders are watched
+                self._assert_paths_watched(inotify._watches, {
+                    tmpdirname,
+                    existing_dir,
+                })
+
+                # create file, event
+                file_path = os.path.join(tmpdirname, "f1.txt")
+                self._create_file(file_path)
+                await asyncio.sleep(0.3)
+
+                # still 2 folders watched
+                self._assert_paths_watched(inotify._watches, {
+                    tmpdirname,
+                    existing_dir,
+                })
+
+                # create folder and a file inside, no event because of racing
+                folder_path = os.path.join(tmpdirname, "d1")
+                os.makedirs(folder_path)
+                file_path = os.path.join(folder_path, "f2.txt")
+                self._create_file(file_path)
+                await asyncio.sleep(0.3)
+
+                # one more folder watched
+                self._assert_paths_watched(inotify._watches, {
+                    tmpdirname,
+                    existing_dir,
+                    os.path.join(tmpdirname, "d1"),
+                })
+
+                # create cascade folders
+                folder_path = os.path.join(tmpdirname, "d2", "dd1", "ddd1")
+                os.makedirs(folder_path)
+                await asyncio.sleep(0.3)
+
+                # 3 more folders watched
+                self._assert_paths_watched(inotify._watches, {
+                    tmpdirname,
+                    existing_dir,
+                    os.path.join(tmpdirname, "d1"),
+                    os.path.join(tmpdirname, "d2"),
+                    os.path.join(tmpdirname, "d2", "dd1"),
+                    os.path.join(tmpdirname, "d2", "dd1", "ddd1"),
+                })
+
+                # move in folder from outside
+                move_folder_path = os.path.join(tmpdirname, "d1", "outside")
+                os.rename(outside_dir, move_folder_path)
+                await asyncio.sleep(0.3)
+
+                # one more folder watched
+                self._assert_paths_watched(inotify._watches, {
+                    tmpdirname,
+                    existing_dir,
+                    os.path.join(tmpdirname, "d1"),
+                    os.path.join(tmpdirname, "d2"),
+                    os.path.join(tmpdirname, "d2", "dd1"),
+                    os.path.join(tmpdirname, "d2", "dd1", "ddd1"),
+                    os.path.join(tmpdirname, "d1", "outside"),
+                })
+
+                # create file in watched outside folder, event
+                file_path = os.path.join(tmpdirname, "d1", "outside", "f3.txt")
+                self._create_file(file_path)
+                await asyncio.sleep(0.3)
+
+                # move out folder
+                folder_path = os.path.join(tmpdirname, "d2", "dd1")
+                move_folder_path = os.path.join(tmpdirbasename, "dd1")
+                os.rename(folder_path, move_folder_path)
+                await asyncio.sleep(0.3)
+
+                # 2 folders not watched
+                self._assert_paths_watched(inotify._watches, {
+                    tmpdirname,
+                    existing_dir,
+                    os.path.join(tmpdirname, "d1"),
+                    os.path.join(tmpdirname, "d2"),
+                    os.path.join(tmpdirname, "d1", "outside"),
+                })
+
+                # create file in not watched folder, no event
+                file_path = os.path.join(tmpdirbasename, "dd1", "ddd1", "f4.txt")
+                self._create_file(file_path)
+                await asyncio.sleep(0.3)
+
+                # move folder within
+                folder_path = os.path.join(tmpdirname, "existing_dir")
+                move_folder_path = os.path.join(tmpdirname, "d1", "existing_dir")
+                os.rename(folder_path, move_folder_path)
+                await asyncio.sleep(0.3)
+
+                # folders change
+                self._assert_paths_watched(inotify._watches, {
+                    tmpdirname,
+                    os.path.join(tmpdirname, "d1"),
+                    os.path.join(tmpdirname, "d2"),
+                    os.path.join(tmpdirname, "d1", "outside"),
+                    os.path.join(tmpdirname, "d1", "existing_dir")
+                })
+
+                # create file in moved folder, event
+                file_path = os.path.join(tmpdirname, "d1", "existing_dir", "f5.txt")
+                self._create_file(file_path)
+                await asyncio.sleep(0.3)
+
+                # delete folder
+                folder_path = os.path.join(tmpdirname, "d2")
+                os.removedirs(folder_path)
+                await asyncio.sleep(0.3)
+
+                # one less folder watched
+                self._assert_paths_watched(inotify._watches, {
+                    tmpdirname,
+                    os.path.join(tmpdirname, "d1"),
+                    os.path.join(tmpdirname, "d1", "outside"),
+                    os.path.join(tmpdirname, "d1", "existing_dir")
+                })
+
+                # delete folders
+                shutil.rmtree(os.path.join(tmpdirname, "d1"))
+                await asyncio.sleep(0.3)
+
+                # less folders watched
+                self._assert_paths_watched(inotify._watches, {
+                    tmpdirname,
+                })
+
+                watch_task.cancel()
+                await asyncio.gather(watch_task, return_exceptions=True)
+
+                # verify events
+                self.assertEqual(len(events), 3)
+                self.assertEqual(str(events[0][0]), os.path.join(tmpdirname, "f1.txt"))
+                self.assertTrue(events[0][1] & Mask.CLOSE_WRITE)
+
+                self.assertEqual(str(events[1][0]), os.path.join(tmpdirname, "d1", "outside", "f3.txt"))
+                self.assertTrue(events[1][1] & Mask.CLOSE_WRITE)
+
+                self.assertEqual(str(events[2][0]), os.path.join(tmpdirname, "d1", "existing_dir", "f5.txt"))
+                self.assertTrue(events[2][1] & Mask.CLOSE_WRITE)
+
+    def test_watch_recursive(self):
+        run(self._watch_recursive())
 
 if __name__ == '__main__':
     unittest.main()
-