Fix a silent fail if path is a file #1034 (Windows only).

src/watchdog/observers/read_directory_changes.py

@@ -44,9 +44,14 @@ def __init__(
         super().__init__(event_queue, watch, timeout=timeout, event_filter=event_filter)
         self._lock = threading.Lock()
         self._whandle: HANDLE | None = None
+        self._watched_files: dict[str, str] = {}
 
     def on_thread_start(self) -> None:
-        self._whandle = get_directory_handle(self.watch.path)
+        watch_path = self.watch.path
+        if os.path.isfile(watch_path):
+            watch_path, basename = os.path.split(watch_path)
+            self._watched_files[self.watch.path] = basename
+        self._whandle = get_directory_handle(watch_path)
 
     if platform.python_implementation() == "PyPy":
 
@@ -71,7 +76,13 @@ def queue_events(self, timeout: float) -> None:
         with self._lock:
             last_renamed_src_path = ""
             for winapi_event in winapi_events:
-                src_path = os.path.join(self.watch.path, winapi_event.src_path)
+                try:
+                    basename = self._watched_files[self.watch.path]  # Is a file?
+                    if basename != winapi_event.src_path:
+                        continue
+                    src_path = self.watch.path
+                except KeyError:
+                    src_path = os.path.join(self.watch.path, winapi_event.src_path)
 
                 if winapi_event.is_renamed_old:
                     last_renamed_src_path = src_path

tests/test_single_file_watched.py

@@ -0,0 +1,172 @@
+"""Test cases when a single file is subscribed."""
+
+import time
+from os import mkdir
+
+from .utils import P, StartWatching, TestEventQueue
+
+SLEEP = 0.1
+
+
+def test_selftest(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
+    """Check if all events are catched in SLEEP time."""
+
+    emitter = start_watching(path=p())  # Pretty sure this should work
+
+    with open(p("file1.bak"), "w") as fp:
+        fp.write("test1")
+
+    with open(p("file2.bak"), "w") as fp:
+        fp.write("test2")
+
+    time.sleep(SLEEP)
+
+    found_files = set()
+    try:
+        while event_queue.qsize():
+            event, _ = event_queue.get()
+            if event.is_directory:  # Not catched on Windows
+                assert event.src_path == p()
+            else:
+                found_files.add(event.src_path)
+    finally:
+        emitter.stop()
+
+    assert len(found_files) == 2, "Number of expected files differ. Increase sleep time."
+
+
+def test_file_access(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
+    """Check if file fires events."""
+
+    file1 = "file1.bak"
+    tmpfile = p(file1)
+
+    with open(tmpfile, "w") as fp:
+        fp.write("init1")
+
+    emitter = start_watching(path=tmpfile)
+
+    # This is what we want to see
+    with open(tmpfile, "w") as fp:
+        fp.write("test1")
+
+    time.sleep(SLEEP)
+
+    try:
+        while event_queue.qsize():
+            event, _ = event_queue.get()
+            if event.is_directory:
+                assert event.src_path == p()
+                continue
+            assert event.src_path.endswith(file1)
+            break
+        else:
+            raise AssertionError  # No event catched
+    finally:
+        emitter.stop()
+
+
+def test_file_access_multiple(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
+    """Check if file fires events multiple times."""
+
+    file1 = "file1.bak"
+    tmpfile = p(file1)
+
+    with open(tmpfile, "w") as fp:
+        fp.write("init1")
+
+    emitter = start_watching(path=tmpfile)
+
+    try:
+        for _i in range(5):
+            # This is what we want to see multiple times
+            with open(tmpfile, "w") as fp:
+                fp.write("test1")
+
+            time.sleep(SLEEP)
+
+            while event_queue.qsize():
+                event, _ = event_queue.get()
+                if event.is_directory:
+                    assert event.src_path == p()
+                    continue
+                assert event.src_path.endswith(file1)
+                break
+            else:
+                raise AssertionError  # No event catched
+
+    finally:
+        emitter.stop()
+
+
+def test_file_access_other_file(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
+    """Check if other files doesn't fires events."""
+
+    file1 = "file1.bak"
+    tmpfile = p(file1)
+
+    with open(tmpfile, "w") as fp:
+        fp.write("init1")
+
+    emitter = start_watching(path=tmpfile)
+
+    # Don't wanted
+    with open(p("file2.bak"), "w") as fp:
+        fp.write("test2")
+
+    # but this
+    with open(tmpfile, "w") as fp:
+        fp.write("test1")
+
+    time.sleep(SLEEP)
+
+    found_files = set()
+    try:
+        while event_queue.qsize():
+            event, _ = event_queue.get()
+            if event.is_directory:
+                assert event.src_path == p()
+            else:
+                found_files.add(event.src_path)
+                assert event.src_path.endswith(file1)
+    finally:
+        emitter.stop()
+
+    assert len(found_files) == 1, "Number of expected files differ. Wrong file catched."
+
+
+def test_create_folder(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
+    """Check if creation of a directory and inside files doesn't fires events."""
+
+    file1 = "file1.bak"
+    tmpfile = p(file1)
+
+    with open(tmpfile, "w") as fp:
+        fp.write("init1")
+
+    emitter = start_watching(path=tmpfile)
+
+    # Don't wanted
+    mkdir(p("myfolder"))
+    with open(p("myfolder/file2.bak"), "w") as fp:
+        fp.write("test2")
+
+    # but this
+    with open(tmpfile, "w") as fp:
+        fp.write("test1")
+
+    time.sleep(SLEEP)
+
+    found_files = set()
+    try:
+        while event_queue.qsize():
+            event, _ = event_queue.get()
+            if event.is_directory:
+                assert event.src_path == p()
+            else:
+                found_files.add(event.src_path)
+                assert event.src_path.endswith(file1)
+    finally:
+        emitter.stop()
+
+    assert len(found_files) == 1, "Number of expected files differ. Wrong file catched."