watchdog Observer filesystem events in tmp_path: working versions (tested 2026-10-15)

watchdog==6.0.0
pytest==9.1.1

import queue
import time
from pathlib import Path

from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer


class EventCollector(FileSystemEventHandler):
    def __init__(self):
        self.queue = queue.Queue()

    def on_any_event(self, event):
        self.queue.put(event)

    def clear(self):
        while True:
            try:
                self.queue.get_nowait()
            except queue.Empty:
                return

    def wait_for(self, predicate, timeout=5.0):
        deadline = time.monotonic() + timeout
        seen = []
        while time.monotonic() < deadline:
            try:
                event = self.queue.get(timeout=0.05)
            except queue.Empty:
                continue
            seen.append(event)
            if predicate(event):
                return event
        raise TimeoutError(f"no event matched, saw {[e.event_type for e in seen]}")


class DirectoryWatcher:
    def __init__(self, path, recursive=True):
        self.path = Path(path)
        self.collector = EventCollector()
        self.observer = Observer()
        self.observer.schedule(self.collector, str(self.path), recursive=recursive)

    def __enter__(self):
        self.observer.start()
        self._wait_ready()
        return self.collector

    def __exit__(self, *exc):
        self.observer.stop()
        self.observer.join(timeout=5)

    def _wait_ready(self, timeout=5.0):
        # Observer.start() returns before the inotify watch exists, and the
        # probe's own create/delete events are delivered asynchronously, so the
        # queue is drained explicitly to keep the next assertions clean.
        deadline = time.monotonic() + timeout
        attempt = 0
        while time.monotonic() < deadline:
            attempt += 1
            name = f".ready-{attempt}"
            probe = self.path / name
            probe.write_text("")
            try:
                self.collector.wait_for(
                    lambda e, name=name: Path(e.src_path).name == name, timeout=0.3
                )
            except TimeoutError:
                probe.unlink()
                continue
            probe.unlink()
            try:
                self.collector.wait_for(
                    lambda e, name=name: Path(e.src_path).name == name
                    and e.event_type == "deleted",
                    timeout=0.5,
                )
            except TimeoutError:
                pass
            self.collector.clear()
            return
        raise TimeoutError("observer did not become ready")

from pathlib import Path

from watcher import DirectoryWatcher


def is_file_event(event_type, path):
    return lambda e: (
        e.event_type == event_type
        and not e.is_directory
        and Path(e.src_path) == path
    )


def test_created_file(tmp_path):
    target = tmp_path / "hello.txt"
    with DirectoryWatcher(tmp_path) as events:
        target.write_text("hi")
        event = events.wait_for(is_file_event("created", target))
    assert Path(event.src_path) == target


def test_modified_file(tmp_path):
    target = tmp_path / "data.txt"
    target.write_text("one")
    with DirectoryWatcher(tmp_path) as events:
        target.write_text("two")
        event = events.wait_for(is_file_event("modified", target))
    assert Path(event.src_path) == target


def test_moved_file_reports_dest(tmp_path):
    src = tmp_path / "a.txt"
    src.write_text("x")
    dst = tmp_path / "b.txt"
    with DirectoryWatcher(tmp_path) as events:
        src.rename(dst)
        event = events.wait_for(
            lambda e: e.event_type == "moved" and Path(e.src_path) == src
        )
    assert Path(event.dest_path) == dst


def test_deleted_file(tmp_path):
    target = tmp_path / "gone.txt"
    target.write_text("x")
    with DirectoryWatcher(tmp_path) as events:
        target.unlink()
        event = events.wait_for(is_file_event("deleted", target))
    assert Path(event.src_path) == target


def test_directory_events_are_flagged(tmp_path):
    sub = tmp_path / "folder"
    with DirectoryWatcher(tmp_path) as events:
        sub.mkdir()
        event = events.wait_for(
            lambda e: e.is_directory
            and e.event_type == "created"
            and Path(e.src_path) == sub
        )
    assert event.is_directory


def test_recursive_watches_existing_subdir(tmp_path):
    sub = tmp_path / "nested"
    sub.mkdir()
    with DirectoryWatcher(tmp_path, recursive=True) as events:
        target = sub / "leaf.txt"
        target.write_text("x")
        event = events.wait_for(is_file_event("created", target))
    assert Path(event.src_path) == target

[pytest]
pythonpath = .