| 1 |
"""Loop-level coverage for watch_for_changes and default stream wrapper.""" |
| 2 |
|
| 3 |
from __future__ import annotations |
| 4 |
|
| 5 |
from collections.abc import Iterator |
| 6 |
from pathlib import Path |
| 7 |
from types import SimpleNamespace |
| 8 |
from unittest.mock import patch |
| 9 |
|
| 10 |
import pytest |
| 11 |
|
| 12 |
from dlm.watch.errors import WatchSetupError |
| 13 |
from dlm.watch.watcher import _default_event_stream, watch_for_changes |
| 14 |
|
| 15 |
|
| 16 |
def test_default_event_stream_wraps_watchfiles_iterator(tmp_path: Path) -> None: |
| 17 |
seen: list[tuple[str, object | None]] = [] |
| 18 |
expected = [{("modified", str(tmp_path / "doc.dlm"))}] |
| 19 |
|
| 20 |
def fake_watch( |
| 21 |
path: str, *, stop_event: object | None = None |
| 22 |
) -> Iterator[set[tuple[object, str]]]: |
| 23 |
seen.append((path, stop_event)) |
| 24 |
yield from expected |
| 25 |
|
| 26 |
with patch.dict("sys.modules", {"watchfiles": SimpleNamespace(watch=fake_watch)}): |
| 27 |
batches = list(_default_event_stream(tmp_path / "doc.dlm", stop_event="stop")) |
| 28 |
|
| 29 |
assert batches == expected |
| 30 |
assert seen == [(str(tmp_path), "stop")] |
| 31 |
|
| 32 |
|
| 33 |
def test_watch_for_changes_requires_existing_file(tmp_path: Path) -> None: |
| 34 |
with pytest.raises(WatchSetupError, match="does not exist"): |
| 35 |
watch_for_changes(tmp_path / "missing.dlm", lambda: None) |
| 36 |
|
| 37 |
|
| 38 |
def test_watch_for_changes_invokes_callback_for_matching_batches(tmp_path: Path) -> None: |
| 39 |
target = tmp_path / "doc.dlm" |
| 40 |
target.write_text("x") |
| 41 |
seen: list[str] = [] |
| 42 |
|
| 43 |
def event_stream( |
| 44 |
_path: Path, *, stop_event: object | None = None |
| 45 |
) -> Iterator[set[tuple[object, str]]]: |
| 46 |
assert stop_event == "stop" |
| 47 |
yield {("modified", str(tmp_path / "other.dlm"))} |
| 48 |
yield {("modified", str(target))} |
| 49 |
yield {("added", str(target))} |
| 50 |
|
| 51 |
watch_for_changes( |
| 52 |
target, lambda: seen.append("changed"), stop_event="stop", event_stream=event_stream |
| 53 |
) |
| 54 |
|
| 55 |
assert seen == ["changed", "changed"] |