(
self, touch_soon: Callable[[Path], None], reloader: BaseReload, *files: Path
)
| 69 | return reloader |
| 70 | |
| 71 | def _reload_tester( |
| 72 | self, touch_soon: Callable[[Path], None], reloader: BaseReload, *files: Path |
| 73 | ) -> list[Path] | None: |
| 74 | reloader.restart() |
| 75 | if WatchFilesReload is not None and isinstance(reloader, WatchFilesReload): |
| 76 | touch_soon(*files) |
| 77 | # Poll until the touched files are reported, ignoring unrelated churn. |
| 78 | expected = set(files) |
| 79 | deadline = monotonic() + 5 |
| 80 | seen: set[Path] = set() |
| 81 | while monotonic() < deadline: |
| 82 | changes = next(reloader) |
| 83 | if changes: |
| 84 | seen.update(p for p in changes if p in expected) |
| 85 | if seen == expected: |
| 86 | break |
| 87 | return sorted(seen) if seen else None |
| 88 | assert not next(reloader) |
| 89 | sleep(0.1) |
| 90 | for file in files: |
| 91 | file.touch() |
| 92 | return next(reloader) |
| 93 | |
| 94 | @pytest.mark.parametrize("reloader_class", [StatReload, WatchFilesReload]) |
| 95 | def test_reloader_should_initialize(self) -> None: |
no test coverage detected