Fix FileWatcher DELETE event reporting and rearrange tests (#86)

This pull request have 2 different but related main changes:

* The `FileWatcher` was skipping `DELETE` events, except on very weird
cases where the file was deleted and re-created just before the event
was triggered by `awatch()`.

This also extends the integration test to test both situations (delete
events when a file exist and when it doesn't).

Because there is usually a considerable delay between a change in the
filesystem happens and `awatch()` emits an event it is difficult to
decide if we should filter events based on the file existing or not. For
example, if a create event is received but before is triggered the file
was removed, should we emit the create event or not? Or the other way
around, should we emit a delete event if a file was created again before
the event was triggered.

It is probably best to leave this decision to the user, and let them
deal with races.

* `FileWatcher` tests were rearranged and split between unit and
integration tests.

The existing tests were really integration tests, as they are not only
testing `FileWatcher`, but also `Select` and `PeriodicTimer`, which is
confusing when there is a bug in any of the other classes as we might
get test failures here (this actually happened during the development of
#79).
    
Because of this the current tests were moved to
`test/util/test_integration.py`. We also mark the tests with the
`pytest` mark `integration` to make it more clear they are integration
tests, and to enable an easy way to only run (exclude) integration
tests.

New unit tests were added to the current test file, properly mocking
`awatch` to avoid interactions with the OS.

There are many more integration tests posing as unit tests, but this PR
only takes care of `FileWatcher`. With time we should move the rest too.

Author: llucax

RELEASE_NOTES.md

@@ -2,7 +2,7 @@
 
 ## Summary
 
-This release adds support to pass `None` values via channels and revamps the `Timer` class to support custom policies for handling missed ticks and use the loop monotonic clock.
+This release adds support to pass `None` values via channels and revamps the `Timer` class to support custom policies for handling missed ticks and use the loop monotonic clock.  There is also a fix for the `FileWatcher` which includes a change in behavior when reporting changes for deleted files.
 
 ## Upgrading
 
@@ -27,10 +27,32 @@ This release adds support to pass `None` values via channels and revamps the `Ti
 
   They are not **exactly** the same because the `triggered_datetime` in the second case will not be exactly when the timer had triggered, but that shouldn't be relevant, the important part is when your code can actually react to the timer trigger and to know how much drift there was to be able to take corrective actions.
 
-  Also the new `Timer` uses the `asyncio` loop monotonic clock and the old one used the wall clock (`datetime.now()`) to track time. This means that when using `async-solipsism` to test, the new `Timer` will always trigger immediately regarless of the state of the wall clock.
+  Also the new `Timer` uses the `asyncio` loop monotonic clock and the old one used the wall clock (`datetime.now()`) to track time. This means that when using `async-solipsism` to test, the new `Timer` will always trigger immediately regarless of the state of the wall clock.  This also means that we don't need to mock the wall clock with `time-machine` either now.
+
+  With the previous timer one needed to create a separate task to run the timer, because otherwise it would block as it loops until the wall clock was at a specific time. Now the code will run like this:
+
+  ```python
+  timer = Timer.timeout(timedelta(seconds=1.0))
+  asyncio.sleep(0.5)  # Advances the loop monotonic clock by 0.5 seconds immediately
+  await drift = timer.receive()  # Advances the clock by 0.5 immediately too
+  assert drift == approx(timedelta(0))  # Because it could trigger exactly at the tick time
+
+  # Simulates a delay in the timer trigger time
+  asyncio.sleep(1.5)  # Advances the loop monotonic clock by 1.5 seconds immediately
+  await drift = timer.receive()  # The timer should have triggerd 0.5 seconds ago, so it doesn't even sleep
+  assert drift == approx(timedelta(seconds=0.5))  # Now we should observe a drift of 0.5 seconds
+  ```
 
   **Note:** Before replacing this code blindly in all uses of `Timer.timeout()`, please consider using the periodic timer constructor `Timer.periodic()` if you need a timer that triggers reliable on a periodic fashion, as the old `Timer` (and `Timer.timeout()`) accumulates drift, which might not be what you want.
 
+* `FileWatcher` now will emit events even if the file doesn't exist anymore.
+
+  Because the underlying library has a considerable delay in triggering filesystem events, it can happen that, for example, a `CREATE` event is received but at the time of receiving the file doesn't exist anymore (because if was removed just after creation and before the event was triggered).
+
+  Before the `FileWatcher` will only emit events when the file exists, but this doesn't work for `DELETE` events (clearly). Given the nature of this mechanism can lead to races easily, it is better to leave it to the user to decide when these situations happen and just report all events.
+
+  Therefore, you should now check a file receiving an event really exist before trying to operate on it.
+
 ## New Features
 
 * `util.Timer` was replaced by a more generic implementation that allows for customizable policies to handle missed ticks.
@@ -40,3 +62,7 @@ This release adds support to pass `None` values via channels and revamps the `Ti
 ## Bug Fixes
 
 * `util.Select` / `util.Merge` / `util.MergeNamed`: Cancel pending tasks in `__del__` methods only if possible (the loop is not already closed).
+
+* `FileWatcher` will now report `DELETE` events correctly.
+
+  Due to a bug, before this release `DELETE` events were only reported if the file was re-created before the event was triggered.

pyproject.toml

@@ -74,6 +74,9 @@ src_paths = ["src", "examples", "tests"]
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 required_plugins = ["pytest-asyncio", "pytest-mock"]
+markers = [
+  "integration: integration tests (deselect with '-m \"not integration\"')",
+]
 
 [[tool.mypy.overrides]]
 module = ["async_solipsism", "async_solipsism.*"]

src/frequenz/channels/util/_file_watcher.py

@@ -46,16 +46,27 @@ def __init__(
             for path in paths
         ]
         self._awatch = awatch(
-            *self._paths,
-            stop_event=self._stop_event,
-            watch_filter=lambda change, path_str: (
-                change in [event_type.value for event_type in event_types]  # type: ignore
-                and pathlib.Path(path_str).is_file()
-            ),
+            *self._paths, stop_event=self._stop_event, watch_filter=self._filter_events
         )
         self._awatch_stopped_exc: Optional[Exception] = None
         self._changes: Set[FileChange] = set()
 
+    def _filter_events(
+        self,
+        change: Change,
+        path: str,  # pylint: disable=unused-argument
+    ) -> bool:
+        """Filter events based on the event type and path.
+
+        Args:
+            change: The type of change to be notified.
+            path: The path of the file that changed.
+
+        Returns:
+            Whether the event should be notified.
+        """
+        return change in [event_type.value for event_type in self.event_types]
+
     def __del__(self) -> None:
         """Cleanup registered watches.
 

tests/utils/test_file_watcher.py

@@ -3,72 +3,99 @@
 
 """Tests for `channel.FileWatcher`."""
 
-import os
+from __future__ import annotations
+
 import pathlib
-from datetime import timedelta
+from collections.abc import AsyncGenerator, Iterator, Sequence
+from typing import Any
+from unittest import mock
 
-from frequenz.channels.util import FileWatcher, Select, Timer
+import hypothesis
+import hypothesis.strategies as st
+import pytest
+from watchfiles import Change
+from watchfiles.main import FileChange
 
+from frequenz.channels.util import FileWatcher
 
-async def test_file_watcher(tmp_path: pathlib.Path) -> None:
-    """Ensure file watcher is returning paths on file events.
 
-    Args:
-        tmp_path (pathlib.Path): A tmp directory to run the file watcher on.
-            Created by pytest.
-    """
-    filename = tmp_path / "test-file"
-    file_watcher = FileWatcher(paths=[str(tmp_path)])
+class _FakeAwatch:
+    """Fake awatch class to mock the awatch function."""
 
-    number_of_writes = 0
-    expected_number_of_writes = 3
+    def __init__(self, changes: Sequence[FileChange] = ()) -> None:
+        """Create a `_FakeAwatch` instance.
 
-    select = Select(
-        timer=Timer.timeout(timedelta(seconds=0.1)),
-        file_watcher=file_watcher,
-    )
-    while await select.ready():
-        if msg := select.timer:
-            filename.write_text(f"{msg.inner}")
-        elif msg := select.file_watcher:
-            assert msg.inner == filename
-            number_of_writes += 1
-            # After receiving a write 3 times, unsubscribe from the writes channel
-            if number_of_writes == expected_number_of_writes:
-                break
-
-    assert number_of_writes == expected_number_of_writes
-
-
-async def test_file_watcher_change_types(tmp_path: pathlib.Path) -> None:
-    """Ensure file watcher is returning paths only on the DELETE change.
-
-    Args:
-        tmp_path (pathlib.Path): A tmp directory to run the file watcher on.
-            Created by pytest.
-    """
-    filename = tmp_path / "test-file"
-    file_watcher = FileWatcher(
-        paths=[str(tmp_path)], event_types={FileWatcher.EventType.DELETE}
-    )
+        Args:
+            changes: A sequence of file changes to be returned by the fake awatch
+                function.
+        """
+        self.changes: Sequence[FileChange] = changes
+
+    async def fake_awatch(
+        self, *paths: str, **kwargs: Any  # pylint: disable=unused-argument
+    ) -> AsyncGenerator[set[FileChange], None]:
+        """Fake awatch function.
+
+        Args:
+            paths: Paths to watch.
+            kwargs: Keyword arguments to pass to the awatch function.
+        """
+        for change in self.changes:
+            yield {change}
+
+
+@pytest.fixture
+def fake_awatch() -> Iterator[_FakeAwatch]:
+    """Fixture to mock the awatch function."""
+    fake = _FakeAwatch()
+    with mock.patch(
+        "frequenz.channels.util._file_watcher.awatch",
+        autospec=True,
+        side_effect=fake.fake_awatch,
+    ):
+        yield fake
 
-    select = Select(
-        write_timer=Timer.timeout(timedelta(seconds=0.1)),
-        deletion_timer=Timer.timeout(timedelta(seconds=0.25)),
-        watcher=file_watcher,
+
+async def test_file_watcher_receive_updates(
+    fake_awatch: _FakeAwatch,  # pylint: disable=redefined-outer-name
+) -> None:
+    """Test the file watcher receive the expected events."""
+    filename = "test-file"
+    changes = (
+        (Change.added, filename),
+        (Change.deleted, filename),
+        (Change.modified, filename),
     )
-    number_of_deletes = 0
-    number_of_write = 0
-    while await select.ready():
-        if msg := select.write_timer:
-            filename.write_text(f"{msg.inner}")
-            number_of_write += 1
-        elif _ := select.deletion_timer:
-            os.remove(filename)
-        elif _ := select.watcher:
-            number_of_deletes += 1
-            break
-
-    assert number_of_deletes == 1
-    # Can be more because the watcher could take some time to trigger
-    assert number_of_write >= 2
+    fake_awatch.changes = changes
+    file_watcher = FileWatcher(paths=[filename])
+
+    for change in changes:
+        recv_changes = await file_watcher.receive()
+        assert recv_changes == pathlib.Path(change[1])
+
+
+@hypothesis.given(event_types=st.sets(st.sampled_from(FileWatcher.EventType)))
+async def test_file_watcher_filter_events(
+    event_types: set[FileWatcher.EventType],
+) -> None:
+    """Test the file watcher events filtering."""
+    good_path = "good-file"
+
+    # We need to reset the mock explicitly because hypothesis runs all the produced
+    # inputs in the same context.
+    with mock.patch(
+        "frequenz.channels.util._file_watcher.awatch", autospec=True
+    ) as awatch_mock:
+        file_watcher = FileWatcher(paths=[good_path], event_types=event_types)
+
+        filter_events = file_watcher._filter_events  # pylint: disable=protected-access
+
+        assert awatch_mock.mock_calls == [
+            mock.call(
+                pathlib.Path(good_path), stop_event=mock.ANY, watch_filter=filter_events
+            )
+        ]
+        for event_type in FileWatcher.EventType:
+            assert filter_events(event_type.value, good_path) == (
+                event_type in event_types
+            )

tests/utils/test_integration.py

@@ -0,0 +1,109 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Integration tests for the `util` module."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+from datetime import timedelta
+
+import pytest
+
+from frequenz.channels.util import FileWatcher, Select, Timer
+
+
+@pytest.mark.integration
+async def test_file_watcher(tmp_path: pathlib.Path) -> None:
+    """Ensure file watcher is returning paths on file events.
+
+    Args:
+        tmp_path (pathlib.Path): A tmp directory to run the file watcher on.
+            Created by pytest.
+    """
+    filename = tmp_path / "test-file"
+    file_watcher = FileWatcher(paths=[str(tmp_path)])
+
+    number_of_writes = 0
+    expected_number_of_writes = 3
+
+    select = Select(
+        timer=Timer.timeout(timedelta(seconds=0.1)),
+        file_watcher=file_watcher,
+    )
+    while await select.ready():
+        if msg := select.timer:
+            filename.write_text(f"{msg.inner}")
+        elif msg := select.file_watcher:
+            assert msg.inner == filename
+            number_of_writes += 1
+            # After receiving a write 3 times, unsubscribe from the writes channel
+            if number_of_writes == expected_number_of_writes:
+                break
+
+    assert number_of_writes == expected_number_of_writes
+
+
+@pytest.mark.integration
+async def test_file_watcher_deletes(tmp_path: pathlib.Path) -> None:
+    """Ensure file watcher is returning paths only on the DELETE change.
+
+    Also ensures that DELETE events are sent even if the file was recreated and even if
+    the file doesn't exist.
+
+    Args:
+        tmp_path (pathlib.Path): A tmp directory to run the file watcher on.
+            Created by pytest.
+    """
+    filename = tmp_path / "test-file"
+    file_watcher = FileWatcher(
+        paths=[str(tmp_path)], event_types={FileWatcher.EventType.DELETE}
+    )
+
+    select = Select(
+        write_timer=Timer.timeout(timedelta(seconds=0.1)),
+        deletion_timer=Timer.timeout(timedelta(seconds=0.25)),
+        watcher=file_watcher,
+    )
+    number_of_write = 0
+    number_of_deletes = 0
+    number_of_events = 0
+    # We want to write to a file and then removed, and then write again (create it
+    # again) and remove it again and then stop.
+    # Because awatch takes some time to get notified by the OS, we need to stop writing
+    # while a delete was done, to make sure the file is not recreated before the
+    # deletion event arrives.
+    # For the second round of writes and then delete, we allow writing after the delete
+    # was done as an extra test.
+    #
+    # This is an example timeline for this test:
+    #
+    # |-----|--.--|-----|---o-|-----|--.--|-----|--o--|-----|-----|-----|-----|-----|
+    # W     W  D            E W     W  D  W     W  E
+    #
+    # Where:
+    # W: Write
+    # D: Delete
+    # E: FileWatcher Event
+    while await select.ready():
+        if msg := select.write_timer:
+            if number_of_write >= 2 and number_of_events == 0:
+                continue
+            filename.write_text(f"{msg.inner}")
+            number_of_write += 1
+        elif _ := select.deletion_timer:
+            # Avoid removing the file twice
+            if not pathlib.Path(filename).is_file():
+                continue
+            os.remove(filename)
+            number_of_deletes += 1
+        elif _ := select.watcher:
+            number_of_events += 1
+            if number_of_events >= 2:
+                break
+
+    assert number_of_deletes == 2
+    # Can be more because the watcher could take some time to trigger
+    assert number_of_write >= 3
+    assert number_of_events == 2