Update docstrings for FileWatcher class and add unit test coverage (#152

)

.github/workflows/build_tests.yml

@@ -51,5 +51,9 @@ jobs:
       - uses: pypa/[email protected]
         with:
           # Ignore setuptools vulnerability we can't do much about
+          # Ignore requests vulnerability
+          # Ignore Setuptools vulnerability
           ignore-vulns: |
-            GHSA-r9hx-vwmv-q579
+            GHSA-r9hx-vwmv-q579
+            GHSA-j8r2-6x86-q33q
+            PYSEC-2022-43012

ovos_utils/file_utils.py

@@ -316,19 +316,31 @@ def read_translated_file(filename: str, data: dict) -> Optional[List[str]]:
 
 
 class FileWatcher:
-    def __init__(self, files, callback, recursive=False, ignore_creation=False):
+    def __init__(self, files: List[str], callback: callable,
+                 recursive: bool = False, ignore_creation: bool = False):
+        """
+        Initialize a FileWatcher to monitor the specified files for changes
+        @param files: list of paths to monitor for file changes
+        @param callback: function to call on file change with modified file path
+        @param recursive: If true, recursively include directory contents
+        @param ignore_creation: If true, ignore file creation events
+        """
         self.observer = Observer()
         self.handlers = []
         for file_path in files:
             if os.path.isfile(file_path):
                 watch_dir = dirname(file_path)
             else:
                 watch_dir = file_path
-            self.observer.schedule(FileEventHandler(file_path, callback, ignore_creation),
+            self.observer.schedule(FileEventHandler(file_path, callback,
+                                                    ignore_creation),
                                    watch_dir, recursive=recursive)
         self.observer.start()
 
     def shutdown(self):
+        """
+        Remove observer scheduled events and stop the observer.
+        """
         self.observer.unschedule_all()
         self.observer.stop()
 
@@ -339,7 +351,7 @@ def __init__(self, file_path: str, callback: callable,
         """
         Create a handler for file change events
         @param file_path: file_path being watched Unused(?)
-        @param callback: function or method to call on file change
+        @param callback: function to call on file change with modified file path
         @param ignore_creation: if True, only track file modification events
         """
         super().__init__()

test/unittests/test_file_utils.py

@@ -1,5 +1,9 @@
+import shutil
 import unittest
+from os import makedirs
 from os.path import isdir, join, dirname
+from threading import Event
+from time import time
 from unittest.mock import Mock
 
 
@@ -51,8 +55,93 @@ def test_read_translated_file(self):
 
     def test_filewatcher(self):
         from ovos_utils.file_utils import FileWatcher
-        test_file = join(dirname(__file__), "test.watch")
-        # TODO
+
+        test_dir = join(dirname(__file__), "test_watch")
+        test_file = join(test_dir, "test.watch")
+        makedirs(test_dir, exist_ok=True)
+
+        # Test watch directory
+        called = Event()
+        callback = Mock(side_effect=lambda x: called.set())
+        watcher = FileWatcher([test_dir], callback)
+        with open(test_file, 'w+') as f:
+            callback.assert_not_called()
+
+        # Called on file close after creation
+        self.assertTrue(called.wait(3))
+        callback.assert_called_once()
+        called.clear()
+        with open(test_file, 'w+') as f:
+            callback.assert_called_once()
+        # Called again on file close
+        self.assertTrue(called.wait(3))
+        self.assertEqual(callback.call_count, 2)
+
+        # Not called on directory creation
+        callback.reset_mock()
+        called.clear()
+        makedirs(join(test_dir, "new_dir"))
+        self.assertFalse(called.wait(3))
+        callback.assert_not_called()
+
+        # Not called on recursive file creation
+        with open(join(test_dir, "new_dir", "file.txt"), 'w+') as f:
+            callback.assert_not_called()
+        self.assertFalse(called.wait(3))
+        callback.assert_not_called()
+
+        watcher.shutdown()
+
+        # Test recursive watch
+        called = Event()
+        callback = Mock(side_effect=lambda x: called.set())
+        watcher = FileWatcher([test_dir], callback, recursive=True,
+                              ignore_creation=True)
+        # Called on file change
+        with open(join(test_dir, "new_dir", "file.txt"), 'w+') as f:
+            callback.assert_not_called()
+        self.assertTrue(called.wait(3))
+        callback.assert_called_once()
+
+        # Not called on file creation
+        with open(join(test_dir, "new_dir", "new_file.txt"), 'w+') as f:
+            callback.assert_called_once()
+        self.assertTrue(called.wait(3))
+        callback.assert_called_once()
+
+        watcher.shutdown()
+
+        # Test watch single file
+        called.clear()
+        callback = Mock(side_effect=lambda x: called.set())
+        watcher = FileWatcher([test_file], callback)
+        with open(test_file, 'w+') as f:
+            callback.assert_not_called()
+        # Called on file close after change
+        self.assertTrue(called.wait(3))
+        callback.assert_called_once()
+        watcher.shutdown()
+
+        # Test changes on callback
+        contents = None
+        changed = Event()
+
+        def _on_change(fp):
+            nonlocal contents
+            self.assertEqual(fp, test_file)
+            with open(fp, 'r') as f:
+                contents = f.read()
+            changed.set()
+
+        watcher = FileWatcher([test_file], _on_change)
+        now_time = time()
+        with open(test_file, 'w') as f:
+            f.write(f"test {now_time}")
+        self.assertTrue(changed.wait(3))
+        self.assertEqual(contents, f"test {now_time}")
+        watcher.shutdown()
+
+        shutil.rmtree(test_dir)
 
     def test_file_event_handler(self):
         from ovos_utils.file_utils import FileEventHandler
@@ -95,4 +184,4 @@ def test_file_event_handler(self):
         callback.assert_called_once()
         # Second close won't trigger callback
         handler.on_any_event(FileClosedEvent(test_file))
-        callback.assert_called_once()
+        callback.assert_called_once()