|
10 | 10 | from typing import Any |
11 | 11 | from unittest import mock |
12 | 12 |
|
| 13 | +import hypothesis |
| 14 | +import hypothesis.strategies as st |
13 | 15 | import pytest |
14 | 16 | from watchfiles import Change |
15 | 17 | from watchfiles.main import FileChange |
@@ -70,3 +72,30 @@ async def test_file_watcher_receive_updates( |
70 | 72 | for change in changes: |
71 | 73 | recv_changes = await file_watcher.receive() |
72 | 74 | assert recv_changes == pathlib.Path(change[1]) |
| 75 | + |
| 76 | + |
| 77 | +@hypothesis.given(event_types=st.sets(st.sampled_from(FileWatcher.EventType))) |
| 78 | +async def test_file_watcher_filter_events( |
| 79 | + event_types: set[FileWatcher.EventType], |
| 80 | +) -> None: |
| 81 | + """Test the file watcher events filtering.""" |
| 82 | + good_path = "good-file" |
| 83 | + |
| 84 | + # We need to reset the mock explicitly because hypothesis runs all the produced |
| 85 | + # inputs in the same context. |
| 86 | + with mock.patch( |
| 87 | + "frequenz.channels.util._file_watcher.awatch", autospec=True |
| 88 | + ) as awatch_mock: |
| 89 | + file_watcher = FileWatcher(paths=[good_path], event_types=event_types) |
| 90 | + |
| 91 | + filter_events = file_watcher._filter_events # pylint: disable=protected-access |
| 92 | + |
| 93 | + assert awatch_mock.mock_calls == [ |
| 94 | + mock.call( |
| 95 | + pathlib.Path(good_path), stop_event=mock.ANY, watch_filter=filter_events |
| 96 | + ) |
| 97 | + ] |
| 98 | + for event_type in FileWatcher.EventType: |
| 99 | + assert filter_events(event_type.value, good_path) == ( |
| 100 | + event_type in event_types |
| 101 | + ) |
0 commit comments