diff --git a/src/django_watchfiles/__init__.py b/src/django_watchfiles/__init__.py index 7437f2c..a0d49cc 100644 --- a/src/django_watchfiles/__init__.py +++ b/src/django_watchfiles/__init__.py @@ -57,9 +57,7 @@ def __init__(self) -> None: def file_filter(self, change: watchfiles.Change, filename: str) -> bool: path = Path(filename) - # print(f"Path: {path} / {change}") if path in set(self.watched_files(include_globs=False)): - # print("Path in watched files") return True for directory, globs in self.directory_globs.items(): try: @@ -67,12 +65,9 @@ def file_filter(self, change: watchfiles.Change, filename: str) -> bool: except ValueError: pass else: - # print("Path is sub dir") for glob in globs: if fnmatch.fnmatch(str(relative_path), glob): - # print("Path is glob match") return True - # print("file filter", change, path) return False def watched_roots(self, watched_files: list[Path]) -> frozenset[Path]: diff --git a/tests/compat.py b/tests/compat.py new file mode 100644 index 0000000..e934867 --- /dev/null +++ b/tests/compat.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +import sys +from contextlib import AbstractContextManager +from typing import Any +from typing import Callable +from typing import TypeVar + +from django import test + +_T = TypeVar("_T") + +if sys.version_info < (3, 11): + + def _enter_context(cm: Any, addcleanup: Callable[..., None]) -> Any: + # We look up the special methods on the type to match the with + # statement. + cls = type(cm) + try: + enter = cls.__enter__ + exit = cls.__exit__ + except AttributeError: # pragma: no cover + raise TypeError( + f"'{cls.__module__}.{cls.__qualname__}' object does " + f"not support the context manager protocol" + ) from None + result = enter(cm) + addcleanup(exit, cm, None, None, None) + return result + + +class SimpleTestCase(test.SimpleTestCase): + if sys.version_info < (3, 11): + + def enterContext(self, cm: AbstractContextManager[_T]) -> _T: + result: _T = _enter_context(cm, self.addCleanup) + return result diff --git a/tests/test_django_watchfiles.py b/tests/test_django_watchfiles.py index 1c96d72..718a35f 100644 --- a/tests/test_django_watchfiles.py +++ b/tests/test_django_watchfiles.py @@ -1,9 +1,82 @@ from __future__ import annotations +import tempfile +import time +from pathlib import Path -def test_import(): - # TODO: figure out tests +from django.utils import autoreload - import django_watchfiles +from django_watchfiles import MutableWatcher +from django_watchfiles import WatchfilesReloader +from tests.compat import SimpleTestCase - assert django_watchfiles # type: ignore [truthy-bool] + +class MutableWatcherTests(SimpleTestCase): + def setUp(self): + self.watcher = MutableWatcher(lambda *args: True) + self.addCleanup(self.watcher.stop) + + temp_dir = self.enterContext(tempfile.TemporaryDirectory()) + self.temp_path = Path(temp_dir) + + def test_set_roots_unchanged(self): + assert not self.watcher.change_event.is_set() + self.watcher.set_roots(set()) + assert not self.watcher.change_event.is_set() + + def test_set_roots_changed(self): + assert not self.watcher.change_event.is_set() + self.watcher.set_roots({Path("/tmp")}) + assert self.watcher.change_event.is_set() + + def test_stop(self): + assert not self.watcher.stop_event.is_set() + self.watcher.stop() + assert self.watcher.stop_event.is_set() + + def test_iter_no_changes(self): + (self.temp_path / "test.txt").touch() + self.watcher.set_roots({self.temp_path}) + iterator = iter(self.watcher) + # flush initial events + next(iterator) + time.sleep(0.1) # 100ms Rust timeout + + changes = next(iterator) + + assert changes == set() + + def test_iter_yields_changes(self): + (self.temp_path / "test.txt").touch() + self.watcher.set_roots({self.temp_path}) + iterator = iter(self.watcher) + # flush initial events + next(iterator) + + (self.temp_path / "test.txt").touch() + changes = next(iterator) + + assert isinstance(changes, set) + assert len(changes) == 1 + _, path = changes.pop() + assert path == str(self.temp_path.resolve() / "test.txt") + + def test_iter_respects_change_event(self): + (self.temp_path / "test.txt").touch() + self.watcher.set_roots({self.temp_path}) + iterator = iter(self.watcher) + # flush initial events + next(iterator) + + self.watcher.set_roots(set()) + self.watcher.set_roots({self.temp_path}) + changes = next(iterator) + + assert isinstance(changes, set) + assert len(changes) == 0 + + +class ReplacedGetReloaderTests(SimpleTestCase): + def test_replaced_get_reloader(self): + reloader = autoreload.get_reloader() + assert isinstance(reloader, WatchfilesReloader)