feat: centralize platform checks (#1051)

* feat: centralize platform checks

* fixes

* skip macOS pypy

.github/workflows/tests.yml

@@ -77,6 +77,12 @@ jobs:
               emoji: 🐧
               runs-on: [ubuntu-latest]
         exclude:
+          - os:
+              matrix: macos
+            python: "pypy-3.8"
+          - os:
+              matrix: macos
+            python: "pypy-3.9"
           - os:
               matrix: windows
             python: "pypy-3.8"

docs/source/conf.py

@@ -15,11 +15,11 @@
 # If extensions (or modules to document with autodoc) are in another directory,
 # add these directories to sys.path here. If the directory is relative to the
 # documentation root, use os.path.abspath to make it absolute, like shown here.
-TOP_DIR_PATH = os.path.abspath("../../")  # noqa
-SRC_DIR_PATH = os.path.join(TOP_DIR_PATH, "src")  # noqa
-sys.path.insert(0, SRC_DIR_PATH)  # noqa
+TOP_DIR_PATH = os.path.abspath("../../")
+SRC_DIR_PATH = os.path.join(TOP_DIR_PATH, "src")
+sys.path.insert(0, SRC_DIR_PATH)
 
-import watchdog.version  # noqa
+import watchdog.version  # noqa: E402
 
 PROJECT_NAME = "watchdog"
 AUTHOR_NAME = "Yesudeep Mangalapilly and contributors"

src/watchdog/observers/__init__.py

@@ -52,23 +52,22 @@
 
 from __future__ import annotations
 
-import sys
 import warnings
 
-from watchdog.utils import UnsupportedLibc
+from watchdog.utils import UnsupportedLibc, platform
 
 from .api import BaseObserverSubclassCallable
 
 Observer: BaseObserverSubclassCallable
 
 
-if sys.platform.startswith("linux"):
+if platform.is_linux():
     try:
         from .inotify import InotifyObserver as Observer
     except UnsupportedLibc:
         from .polling import PollingObserver as Observer
 
-elif sys.platform.startswith("darwin"):
+elif platform.is_darwin():
     try:
         from .fsevents import FSEventsObserver as Observer
     except Exception:
@@ -81,10 +80,10 @@
 
             warnings.warn("Failed to import fsevents and kqueue. Fall back to polling.")
 
-elif sys.platform in ("dragonfly", "freebsd", "netbsd", "openbsd", "bsd"):
+elif platform.is_bsd():
     from .kqueue import KqueueObserver as Observer
 
-elif sys.platform.startswith("win"):
+elif platform.is_windows():
     try:
         from .read_directory_changes import WindowsApiObserver as Observer
     except Exception:

src/watchdog/observers/read_directory_changes.py

@@ -18,7 +18,6 @@
 
 import os.path
 import platform
-import sys
 import threading
 
 from watchdog.events import (
@@ -34,10 +33,7 @@
     generate_sub_moved_events,
 )
 from watchdog.observers.api import DEFAULT_EMITTER_TIMEOUT, DEFAULT_OBSERVER_TIMEOUT, BaseObserver, EventEmitter
-
-assert sys.platform.startswith("win"), f"{__name__} requires Windows"
-
-from watchdog.observers.winapi import close_directory_handle, get_directory_handle, read_events  # noqa: E402
+from watchdog.observers.winapi import close_directory_handle, get_directory_handle, read_events
 
 # Obsolete constant, it's no more used since v4.0.0.
 WATCHDOG_TRAVERSE_MOVED_DIR_DELAY = 1  # seconds

src/watchdog/observers/winapi.py

@@ -36,13 +36,10 @@
 
 from __future__ import annotations
 
-import sys
+import ctypes.wintypes
 from dataclasses import dataclass
 from functools import reduce
 
-assert sys.platform.startswith("win"), f"{__name__} requires Windows"
-import ctypes.wintypes  # noqa: E402
-
 LPVOID = ctypes.wintypes.LPVOID
 
 # Invalid handle value.

src/watchdog/tricks/__init__.py

@@ -46,12 +46,11 @@
 import os
 import signal
 import subprocess
-import sys
 import threading
 import time
 
 from watchdog.events import EVENT_TYPE_OPENED, FileSystemEvent, PatternMatchingEventHandler
-from watchdog.utils import echo
+from watchdog.utils import echo, platform
 from watchdog.utils.event_debouncer import EventDebouncer
 from watchdog.utils.process_watcher import ProcessWatcher
 
@@ -296,10 +295,10 @@ def _restart_process(self):
         self.restart_count += 1
 
 
-if not sys.platform.startswith("win"):
+if not platform.is_windows():
 
     def kill_process(pid, stop_signal):
-        os.killpg(os.getpgid(pid), stop_signal)
+        os.killpg(os.getpgid(pid), stop_signal)  # type: ignore[attr-defined]
 
 else:
 

src/watchdog/utils/platform.py

@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
 from __future__ import annotations
 
 import sys

src/watchdog/watchmedo.py

@@ -35,7 +35,7 @@
 from typing import TYPE_CHECKING
 
 from watchdog.observers.api import BaseObserverSubclassCallable
-from watchdog.utils import WatchdogShutdown, load_class
+from watchdog.utils import WatchdogShutdown, load_class, platform
 from watchdog.version import VERSION_STRING
 
 logging.basicConfig(level=logging.INFO)
@@ -264,7 +264,7 @@ def tricks_from(args):
         from watchdog.observers.polling import PollingObserver as Observer
     elif args.debug_force_kqueue:
         from watchdog.observers.kqueue import KqueueObserver as Observer
-    elif (not TYPE_CHECKING and args.debug_force_winapi) or (TYPE_CHECKING and sys.platform.startswith("win")):
+    elif (not TYPE_CHECKING and args.debug_force_winapi) or (TYPE_CHECKING and platform.is_windows()):
         from watchdog.observers.read_directory_changes import WindowsApiObserver as Observer
     elif args.debug_force_inotify:
         from watchdog.observers.inotify import InotifyObserver as Observer
@@ -466,7 +466,7 @@ def log(args):
         from watchdog.observers.polling import PollingObserver as Observer
     elif args.debug_force_kqueue:
         from watchdog.observers.kqueue import KqueueObserver as Observer
-    elif (not TYPE_CHECKING and args.debug_force_winapi) or (TYPE_CHECKING and sys.platform.startswith("win")):
+    elif (not TYPE_CHECKING and args.debug_force_winapi) or (TYPE_CHECKING and platform.is_windows()):
         from watchdog.observers.read_directory_changes import WindowsApiObserver as Observer
     elif args.debug_force_inotify:
         from watchdog.observers.inotify import InotifyObserver as Observer

tests/test_0_watchmedo.py

@@ -10,16 +10,15 @@
 # Skip if import PyYAML failed. PyYAML missing possible because
 # watchdog installed without watchmedo. See Installation section
 # in README.rst
-yaml = pytest.importorskip("yaml")  # noqa
+yaml = pytest.importorskip("yaml")
 
+from yaml.constructor import ConstructorError  # noqa: E402
+from yaml.scanner import ScannerError  # noqa: E402
 
-from yaml.constructor import ConstructorError  # noqa
-from yaml.scanner import ScannerError  # noqa
-
-from watchdog import watchmedo  # noqa
-from watchdog.events import FileModifiedEvent, FileOpenedEvent  # noqa
-from watchdog.tricks import AutoRestartTrick, ShellCommandTrick  # noqa
-from watchdog.utils import WatchdogShutdown  # noqa
+from watchdog import watchmedo  # noqa: E402
+from watchdog.events import FileModifiedEvent, FileOpenedEvent  # noqa: E402
+from watchdog.tricks import AutoRestartTrick, ShellCommandTrick  # noqa: E402
+from watchdog.utils import WatchdogShutdown, platform  # noqa: E402
 
 
 def test_load_config_valid(tmpdir):
@@ -149,7 +148,7 @@ def test_auto_restart_on_file_change(tmpdir, capfd):
 
 
 @pytest.mark.xfail(
-    condition=sys.platform.startswith(("win", "darwin")) or sys.implementation.name == "pypy",
+    condition=platform.is_darwin() or platform.is_windows() or sys.implementation.name == "pypy",
     reason="known to be problematic, see #973",
 )
 def test_auto_restart_on_file_change_debounce(tmpdir, capfd):
@@ -181,7 +180,7 @@ def test_auto_restart_on_file_change_debounce(tmpdir, capfd):
         pytest.param(
             False,
             marks=pytest.mark.xfail(
-                condition=sys.platform.startswith(("win", "darwin")),
+                condition=platform.is_darwin() or platform.is_windows(),
                 reason="known to be problematic, see #972",
             ),
         ),

tests/test_fsevents.py

@@ -4,7 +4,7 @@
 
 from watchdog.utils import platform
 
-if not platform.is_darwin():  # noqa
+if not platform.is_darwin():
     pytest.skip("macOS only.", allow_module_level=True)
 
 import logging

tests/test_inotify_buffer.py

@@ -18,8 +18,8 @@
 
 from watchdog.utils import platform
 
-if not platform.is_linux():  # noqa
-    pytest.skip("GNU/Linux only.", allow_module_level=True)  # noqa
+if not platform.is_linux():
+    pytest.skip("GNU/Linux only.", allow_module_level=True)
 
 import os
 import random

tests/test_inotify_c.py

@@ -4,7 +4,7 @@
 
 from watchdog.utils import platform
 
-if not platform.is_linux():  # noqa
+if not platform.is_linux():
     pytest.skip("GNU/Linux only.", allow_module_level=True)
 
 import ctypes

tests/test_observers_winapi.py

@@ -17,25 +17,22 @@
 
 import os
 import os.path
-import sys
 from queue import Empty, Queue
 from time import sleep
 
 import pytest
 
 from watchdog.events import DirCreatedEvent, DirMovedEvent
 from watchdog.observers.api import ObservedWatch
+from watchdog.utils import platform
 
 from .shell import mkdir, mkdtemp, mv, rm
 
 # make pytest aware this is windows only
-if not sys.platform.startswith("win"):
+if not platform.is_windows():
     pytest.skip("Windows only.", allow_module_level=True)
 
-# make mypy aware this is windows only and provide a clear runtime error just in case
-assert sys.platform.startswith("win"), f"{__name__} requires Windows"
-
-from watchdog.observers.read_directory_changes import WindowsApiEmitter  # noqa: E402
+from watchdog.observers.read_directory_changes import WindowsApiEmitter
 
 SLEEP_TIME = 2
 

tests/utils.py

@@ -2,24 +2,23 @@
 
 import dataclasses
 import os
-import sys
 from queue import Empty, Queue
 from typing import List, Optional, Tuple, Type, Union
 
 from watchdog.events import FileSystemEvent
 from watchdog.observers.api import EventEmitter, ObservedWatch
-from watchdog.utils import Protocol
+from watchdog.utils import Protocol, platform
 
 Emitter: Type[EventEmitter]
 
-if sys.platform.startswith("linux"):
+if platform.is_linux():
     from watchdog.observers.inotify import InotifyEmitter as Emitter
     from watchdog.observers.inotify import InotifyFullEmitter
-elif sys.platform.startswith("darwin"):
+elif platform.is_darwin():
     from watchdog.observers.fsevents import FSEventsEmitter as Emitter
-elif sys.platform.startswith("win"):
+elif platform.is_windows():
     from watchdog.observers.read_directory_changes import WindowsApiEmitter as Emitter
-elif sys.platform.startswith(("dragonfly", "freebsd", "netbsd", "openbsd", "bsd")):
+elif platform.is_bsd():
     from watchdog.observers.kqueue import KqueueEmitter as Emitter
 
 
@@ -65,14 +64,14 @@ def start_watching(
         path = self.tmp if path is None else path
 
         emitter: EventEmitter
-        if sys.platform.startswith("linux") and use_full_emitter:
+        if platform.is_linux() and use_full_emitter:
             emitter = InotifyFullEmitter(self.event_queue, ObservedWatch(path, recursive=recursive))
         else:
             emitter = Emitter(self.event_queue, ObservedWatch(path, recursive=recursive))
 
         self.emitters.append(emitter)
 
-        if sys.platform.startswith("darwin"):
+        if platform.is_darwin():
             # TODO: I think this could be better...  .suppress_history should maybe
             #       become a common attribute.
             from watchdog.observers.fsevents import FSEventsEmitter

tox.ini

@@ -37,7 +37,8 @@ usedevelop = true
 deps =
     -r requirements-tests.txt
 commands =
-    mypy
+    # "--platform win32" to not fail on ctypes.windll (it does not affect the overall check on other OSes)
+    mypy --platform win32
 
 [testenv:isort]
 usedevelop = true