Covers the file creation events by tests

notifykit/_notifier.py

@@ -33,6 +33,8 @@ def __next__(self) -> List[Event]: ...
 
     async def __anext__(self) -> List[Event]: ...
 
+    def stop(self) -> None: ...
+
 
 class Notifier:
     """

pyproject.toml

@@ -47,4 +47,6 @@ dev = [
     "ruff>=0.12.10",
     "mkdocs-material[imaging]>=9.6.18",
     "mkdocs>=1.6.1",
+    "async-timeout>=5.0.1",
+    "pytest-asyncio>=1.2.0",
 ]

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+asyncio_mode = auto
+xfail_strict = true

tests/conftest.py

@@ -0,0 +1,62 @@
+import asyncio
+from contextlib import asynccontextmanager
+
+import async_timeout
+from typing import AsyncContextManager, AsyncGenerator, Self
+
+from notifykit import NotifierT, Event
+
+
+class EventCollector:
+    def __init__(self, ):
+        self._events: list[Event] = []
+        self._notifier_task: asyncio.Task | None = None
+
+        self._waiters: set[tuple[int, asyncio.Event]] = set()
+
+    @property
+    def events(self) -> list[Event]:
+        return self._events
+
+    @asynccontextmanager
+    async def collect(self, notifier: NotifierT) -> AsyncGenerator[Self, None]:
+        if self._notifier_task is not None:
+            raise RuntimeError("EventCollector is already running.")
+
+        self._notifier_task = asyncio.create_task(self._collect_events(notifier))
+
+        yield self
+
+        notifier.stop()
+
+        if self._notifier_task:
+            self._notifier_task.cancel()
+            try:
+                await self._notifier_task
+            except asyncio.CancelledError:
+                pass
+            self._notifier_task = None
+
+    async def wait_for_events(self, items: int, timeout: float = 2) -> None:
+        if items <= len(self._events):
+            return
+
+        waiter = asyncio.Event()
+        waiter_id = (items, waiter)
+
+        self._waiters.add(waiter_id)
+
+        async with async_timeout.timeout(timeout):
+            await waiter.wait()
+
+        self._waiters.remove(waiter_id)
+
+    async def _collect_events(self, notifier: NotifierT) -> None:
+        async for event in notifier:
+            self._events.extend(event)
+            self._wakeup_waiters()
+
+    def _wakeup_waiters(self) -> None:
+        for items, event in self._waiters:
+            if items <= len(self._events):
+                event.set()

tests/test_events_create.py

@@ -0,0 +1,32 @@
+import asyncio
+import tempfile
+from pathlib import Path
+
+from notifykit import Notifier, NotifierT
+from tests.conftest import EventCollector
+
+
+async def test__events__create_file() -> None:
+    files_to_create = 3
+    expected_events = files_to_create * 3 # each file triggers 3 events: Create, ModifyMetadata, ModifyData
+    tmp_dir = Path(tempfile.mkdtemp())
+
+    await asyncio.sleep(0.1)  # avoid catching directory creation event
+
+    notifier: NotifierT = Notifier(debounce_ms=100, tick_ms=10, debug=False)
+    await notifier.watch([tmp_dir])
+
+    collector = EventCollector()
+
+    async with collector.collect(notifier) as c:
+        expected_paths = []
+
+        for idx in range(files_to_create):
+            file_path = tmp_dir / f'lorem_{idx}.txt'
+            file_path.write_text("new lorem ipsum")
+
+            expected_paths.append(str(file_path))
+
+        await c.wait_for_events(expected_events, timeout=3)
+
+    assert len(collector.events) == expected_events, collector.events