chore: add coverage

Author: bdraco

.github/workflows/ci.yml

@@ -40,6 +40,7 @@ jobs:
           - "3.10"
           - "3.11"
           - "3.12"
+          - "3.13"
         os:
           - ubuntu-latest
           - windows-latest

pyproject.toml

@@ -160,3 +160,9 @@ match = "main"
 [tool.semantic_release.branches.noop]
 match = "(?!main$)"
 prerelease = true
+
+[tool.uv]
+dev-dependencies = [
+    "pytest-asyncio>=0.25.2",
+    "pytest-cov>=6.0.0",
+]

src/aiousbwatcher/__init__.py

@@ -1,5 +1,5 @@
 __version__ = "0.0.0"
 
-from .impl import AIOUSBWatcher
+from .impl import AIOUSBWatcher, InotifyNotAvailableError
 
-__all__ = ["AIOUSBWatcher"]
+__all__ = ["AIOUSBWatcher", "InotifyNotAvailableError"]

src/aiousbwatcher/impl.py

@@ -7,16 +7,15 @@
 from pathlib import Path
 from typing import Callable
 
-from asyncinotify import Inotify, Mask
-
-_BASE_MASK = (
-    Mask.MOVED_FROM | Mask.MOVED_TO | Mask.CREATE | Mask.DELETE_SELF | Mask.IGNORED
-)
 _PATH = "/dev/bus/usb"
 
 _LOGGER = logging.getLogger(__name__)
 
 
+class InotifyNotAvailableError(Exception):
+    """Raised when inotify is not available on the platform."""
+
+
 def _get_directories_recursive_gen(path: Path) -> Generator[Path, None, None]:
     if path.is_dir():
         yield path
@@ -38,9 +37,7 @@ class AIOUSBWatcher:
     """A watcher for USB devices that uses asyncio."""
 
     def __init__(self) -> None:
-        self._inotify = Inotify()
         self._path = Path(_PATH)
-        self._mask = Mask.CREATE | _BASE_MASK
         self._loop = asyncio.get_running_loop()
         self._task: asyncio.Task[None] | None = None
         self._callbacks: set[Callable[[], None]] = set()
@@ -49,6 +46,12 @@ def async_start(self) -> Callable[[], None]:
         """Start the watcher."""
         if self._task is not None:
             raise RuntimeError("Watcher already started")
+        try:
+            from asyncinotify import Inotify  # noqa
+        except AttributeError as ex:
+            raise InotifyNotAvailableError(
+                "Inotify not available on this platform"
+            ) from ex
         self._task = self._loop.create_task(self._watcher())
         return self._async_stop
 
@@ -66,22 +69,33 @@ def _async_stop(self) -> None:
         self._task = None
 
     async def _watcher(self) -> None:
+        from asyncinotify import Inotify, Mask
+
+        mask = (
+            Mask.CREATE
+            | Mask.MOVED_FROM
+            | Mask.MOVED_TO
+            | Mask.CREATE
+            | Mask.DELETE_SELF
+            | Mask.IGNORED
+        )
+
         with Inotify() as inotify:
             for directory in await _async_get_directories_recursive(
                 self._loop, self._path
             ):
-                inotify.add_watch(directory, self._mask)
+                inotify.add_watch(directory, mask)
 
             async for event in inotify:
                 # Add subdirectories to watch if a new directory is added.
                 if Mask.CREATE in event.mask and event.path is not None:
                     for directory in await _async_get_directories_recursive(
                         self._loop, event.path
                     ):
-                        inotify.add_watch(directory, self._mask)
+                        inotify.add_watch(directory, mask)
 
                 # If there is at least some overlap, assume the user wants this event.
-                if event.mask & self._mask:
+                if event.mask & mask:
                     self._async_call_callbacks()
 
     def _async_unregister_callback(self, callback: Callable[[], None]) -> None:

tests/test_impl.py

@@ -0,0 +1,42 @@
+import asyncio
+from pathlib import Path
+from sys import platform
+from unittest.mock import patch
+
+import pytest
+
+from aiousbwatcher import AIOUSBWatcher, InotifyNotAvailableError
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(platform == "linux", reason="Inotify is available on this platform")
+async def test_aiousbwatcher_not_available() -> None:
+    with pytest.raises(InotifyNotAvailableError):
+        watcher = AIOUSBWatcher()
+        watcher.async_start()
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(
+    platform != "linux", reason="Inotify not available on this platform"
+)
+async def test_aiousbwatcher_callbacks(tmp_path: Path) -> None:
+    called: bool = False
+
+    def callback() -> None:
+        nonlocal called
+        called = True
+
+    with patch("aiousbwatcher.impl._PATH", str(tmp_path)):
+        watcher = AIOUSBWatcher()
+        unregister = watcher.async_register_callback(callback)
+        stop = watcher.async_start()
+        await asyncio.sleep(0.1)
+        assert not called
+        (tmp_path / "test").touch()
+        await asyncio.sleep(0.1)
+        assert called
+        unregister()  # type: ignore[unreachable]
+        stop()
+        await asyncio.sleep(0.1)
+        assert not called