Fixing issue with event retrieval (#28)

* Fixing the issue with not retrieving events after the first batch

* Implemented the notifier mock class

* 🔖 v0.0.7

* Backed up to the previous implementation after some tests

* Fixed the issue with expiration predicate

* Linting

* Reverted back the sig changes

* Linting

---------

Co-authored-by: Roman Glushko <roman.glushko.m@datarobot.com>

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "notifykit_lib"
-version = "0.0.6"
+version = "0.0.7"
 edition = "2021"
 license = "A toolkit for building applications watching filesystem changes"
 homepage = "https://github.com/roma-glushko/notifykit"

notifykit/_notifier.py

@@ -1,6 +1,6 @@
 from os import PathLike
 import anyio
-from typing import Sequence, Protocol, Optional, List
+from typing import Sequence, Protocol, Optional, List, AsyncGenerator
 from notifykit._notifykit_lib import (
     WatcherWrapper,
     AccessEvent,
@@ -29,7 +29,7 @@ def watch(
     ) -> None:
         ...
 
-    def unwatch(self, paths: Sequence[str]) -> None:
+    def unwatch(self, paths: Sequence[PathLike[str]]) -> None:
         ...
 
     def __aiter__(self) -> "Notifier":
@@ -91,7 +91,7 @@ def __next__(self) -> List[Event]:
 
         return events
 
-    async def __anext__(self) -> List[Event]:
+    async def __anext__(self) -> AsyncGenerator[List[Event], None]:
         CancelledError = anyio.get_cancelled_exc_class()
 
         async with anyio.create_task_group() as tg:
@@ -105,6 +105,6 @@ async def __anext__(self) -> List[Event]:
             tg.cancel_scope.cancel()
 
             if events is None:
-                raise StopIteration
+                raise StopAsyncIteration
 
             return events

notifykit/_testing.py

@@ -1,3 +1,55 @@
+from os import PathLike
+from typing import Sequence, List
+
+from notifykit import Event
+
+
 class NotifierMock:
-    # TODO: implement
-    ...
+    """
+    A notifier mock that allows to control filesystems events without actually watching the filesystem
+    """
+
+    def __init__(self, events_batches: List[List[Event]]) -> None:
+        self._watch_paths: List[PathLike[str]] = []
+        self._events_batches = events_batches
+
+    @property
+    def watch_paths(self) -> List[PathLike[str]]:
+        return self._watch_paths
+
+    @property
+    def events_batches(self) -> List[List[Event]]:
+        return self._events_batches
+
+    def add_event_batch(self, events_batch: List[Event]) -> None:
+        self._events_batches.append(events_batch)
+
+    def watch(
+        self,
+        paths: Sequence[PathLike[str]],
+        recursive: bool = True,
+        ignore_permission_errors: bool = False,
+    ) -> None:
+        self._watch_paths.extend(paths)
+
+    def unwatch(self, paths: Sequence[PathLike[str]]) -> None:
+        for path in paths:
+            self._watch_paths.remove(path)
+
+    def __aiter__(self) -> "NotifierMock":
+        return self
+
+    def __iter__(self) -> "NotifierMock":
+        return self
+
+    def __next__(self) -> List[Event]:
+        if not self._events_batches:
+            raise StopIteration
+
+        return self._events_batches.pop(0)
+
+    async def __anext__(self) -> List[Event]:
+        if not self._events_batches:
+            raise StopAsyncIteration
+
+        return self._events_batches.pop(0)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "notifykit"
-version = "0.0.6"
+version = "0.0.7"
 description = "A modern efficient Python toolkit for building applications that need to watch filesystem changes"
 authors = [
     {name = "Roman Glushko", email = "roman.glushko.m@gmail.com"},

src/processor.rs

@@ -441,13 +441,13 @@ impl EventProcessor for BatchProcessor {
         let now = Instant::now();
 
         // Assuming expired items are contiguous and at the beginning of the vector
-        let first_expired_index = self
+        let first_non_expired_index = self
             .events
             .iter()
-            .position(|event| now.saturating_duration_since(event.time) >= self.buffering_time)
+            .position(|event| now.saturating_duration_since(event.time) < self.buffering_time)
             .unwrap_or(self.events.len());
 
-        self.events.drain(0..first_expired_index).collect()
+        self.events.drain(0..first_non_expired_index).collect()
     }
 
     fn get_errors(&mut self) -> Vec<NotifyError> {