Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add some proper tests #122

Merged
merged 3 commits into from
Oct 28, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 0 additions & 5 deletions src/django_watchfiles/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,22 +57,17 @@ 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:
relative_path = path.relative_to(directory)
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]:
Expand Down
37 changes: 37 additions & 0 deletions tests/compat.py
Original file line number Diff line number Diff line change
@@ -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
81 changes: 77 additions & 4 deletions tests/test_django_watchfiles.py
Original file line number Diff line number Diff line change
@@ -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)
Loading