fix: polish gorakhargosh#1070

src/watchdog/observers/inotify_c.py

@@ -244,7 +244,7 @@ def close(self) -> None:
                 if self._waiting_to_read:
                     # inotify_rm_watch() should write data to _inotify_fd and wake
                     # the thread, but writing to the kill channel will gaurentee this
-                    os.write(self._kill_w, b'!')
+                    os.write(self._kill_w, b"!")
                 else:
                     self._close_resources()
 
@@ -365,7 +365,7 @@ def _recursive_simulate(src_path: bytes) -> list[InotifyEvent]:
 
         return event_list
 
-    def _close_resources(self):
+    def _close_resources(self) -> None:
         os.close(self._inotify_fd)
         os.close(self._kill_r)
         os.close(self._kill_w)

src/watchdog/utils/bricks.py

@@ -72,7 +72,10 @@ def _init(self, maxsize: int) -> None:
         super()._init(maxsize)
         self._last_item = None
 
-    def put(self, item: Any, block: bool = True, timeout: float | None = None) -> None:
+    def put(self, item: Any, block: bool = True, timeout: float | None = None) -> None:  # noqa: FBT001,FBT002
+        """This method will be used by `eventlet`, when enabled, so we cannot use force proper keyword-only
+        arguments nor touch the signature. Also, the `timeout` argument will be ignored in that case.
+        """
         if self._last_item is None or item != self._last_item:
             super().put(item, block, timeout)
 

tests/isolated/eventlet_observer_stops.py

@@ -1,4 +1,4 @@
-if __name__ == '__main__':
+if __name__ == "__main__":
     import eventlet
 
     eventlet.monkey_patch()
@@ -7,10 +7,11 @@
     import sys
     import tempfile
 
-    from watchdog.observers import Observer
     from watchdog.events import LoggingEventHandler
+    from watchdog.observers import Observer
 
     with tempfile.TemporaryDirectory() as temp_dir:
+
         def run_observer():
             event_handler = LoggingEventHandler()
             observer = Observer()
@@ -20,7 +21,7 @@ def run_observer():
             observer.stop()
 
         def on_alarm(signum, frame):
-            print("Observer.stop() never finished!", file=sys.stderr)
+            print("Observer.stop() never finished!", file=sys.stderr)  # noqa: T201
             sys.exit(1)
 
         signal.signal(signal.SIGALRM, on_alarm)

tests/isolated/eventlet_skip_repeat_queue.py

@@ -1,32 +1,32 @@
-if __name__ == '__main__':
+if __name__ == "__main__":
     import eventlet
 
     eventlet.monkey_patch()
 
     from watchdog.utils.bricks import SkipRepeatsQueue
 
     q = SkipRepeatsQueue(10)
-    q.put('A')
-    q.put('A')
-    q.put('A')
-    q.put('A')
-    q.put('B')
-    q.put('A')
+    q.put("A")
+    q.put("A")
+    q.put("A")
+    q.put("A")
+    q.put("B")
+    q.put("A")
 
     value = q.get()
-    assert value == 'A'
+    assert value == "A"
     q.task_done()
 
     assert q.unfinished_tasks == 2
 
     value = q.get()
-    assert value == 'B'
+    assert value == "B"
     q.task_done()
 
     assert q.unfinished_tasks == 1
 
     value = q.get()
-    assert value == 'A'
+    assert value == "A"
     q.task_done()
 
     assert q.empty()

tests/test_isolated.py

@@ -1,6 +1,7 @@
-import pytest
 import importlib
 
+import pytest
+
 from watchdog.utils import platform
 
 from .utils import run_isolated_test
@@ -10,15 +11,15 @@
 # Current usage ReadDirectoryChangesW on Windows is blocking, though async may be possible
 @pytest.mark.skipif(not platform.is_linux(), reason="Eventlet only supported in Linux")
 def test_observer_stops_in_eventlet():
-    if not importlib.util.find_spec('eventlet'):
+    if not importlib.util.find_spec("eventlet"):
         pytest.skip("eventlet not installed")
 
-    run_isolated_test('eventlet_observer_stops.py')
+    run_isolated_test("eventlet_observer_stops.py")
 
 
 @pytest.mark.skipif(not platform.is_linux(), reason="Eventlet only supported in Linux")
 def test_eventlet_skip_repeat_queue():
-    if not importlib.util.find_spec('eventlet'):
+    if not importlib.util.find_spec("eventlet"):
         pytest.skip("eventlet not installed")
 
-    run_isolated_test('eventlet_skip_repeat_queue.py')
+    run_isolated_test("eventlet_skip_repeat_queue.py")

tests/utils.py

@@ -102,12 +102,12 @@ def close(self) -> None:
 
 
 def run_isolated_test(path):
-    ISOALTED_TEST_PREFIX = os.path.join('tests', 'isolated')
-    path = os.path.abspath(os.path.join(ISOALTED_TEST_PREFIX, path))
+    isolated_test_prefix = os.path.join("tests", "isolated")
+    path = os.path.abspath(os.path.join(isolated_test_prefix, path))
 
-    src_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'src')
+    src_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src")
     new_env = os.environ.copy()
-    new_env['PYTHONPATH'] = os.pathsep.join(sys.path + [src_dir])
+    new_env["PYTHONPATH"] = os.pathsep.join([*sys.path, src_dir])
 
     new_argv = [sys.executable, path]
 
@@ -122,6 +122,6 @@ def run_isolated_test(path):
         p.communicate(timeout=timeout)
     except subprocess.TimeoutExpired:
         p.kill()
-        assert False, 'timed out'
+        raise
 
     assert p.returncode == 0

tox.ini

@@ -33,7 +33,7 @@ extras =
     watchmedo
 commands =
     python -m ruff format docs/source/examples src tests
-    python -m ruff check --fix src docs/source/examples tests
+    python -m ruff check --fix --unsafe-fixes src docs/source/examples tests
 
 [testenv:types]
 usedevelop = true