watchfiles 1.2.0: watch file system changes using watchfiles.watch and Change enum

# watchfiles watch and Change API

This sample pins `watchfiles` 1.2.0 and verifies filesystem change observation using `watchfiles.watch` and the `watchfiles.Change` enumeration.

The contract verifies:
- `Change` enum members `added` (1), `modified` (2), and `deleted` (3) map to their string representations via `.raw_str()`.
- `watch()` observes real filesystem modifications and yields sets of `(Change, filepath)` tuples.
- `watch()` terminates cleanly when a `threading.Event` `stop_event` is signaled.
- `watch()` supports custom `watch_filter` callables to filter specific files based on path or event type.
- `watch()` supports `yield_on_timeout=True` to yield an empty set when `rust_timeout` triggers without filesystem changes.

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: watch file system changes using watchfiles.watch and Change enum
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/watchfiles@1.2.0
Demonstrate these symbols/APIs:
  - watchfiles.Change
  - watchfiles.watch

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:07409a5c37e89983e3a7322eb49685a3b0e6ecd28043e07e553806faec6b0d3a","contract":["Change enum members added (1), modified (2), and deleted (3) map to raw_str representations 'added', 'modified', and 'deleted'.","watch() detects file creation, content modification, and deletion events in a target directory as (Change, filepath) tuples.","watch() stops iteration when a threading.Event stop_event is set.","watch() accepts a custom callable watch_filter to selectively include or ignore file paths during change detection.","watch() supports yield_on_timeout to yield an empty set when rust_timeout expires without filesystem activity."],"goal":"watch file system changes using watchfiles.watch and Change enum","kind":"HOW","packages":["pkg:pypi/watchfiles@1.2.0"],"schemaVersion":1,"symbols":["watchfiles.Change","watchfiles.watch"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","executionContext":"python","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/watchfiles@1.2.0"],"schemaVersion":1,"subject":"pkg:pypi/watchfiles@1.2.0","symbols":["watchfiles.Change","watchfiles.watch"],"verifierAdapter":"python@1"}

anyio==4.14.2
idna==3.19
typing_extensions==4.16.0
watchfiles==1.2.0

{
  "schemaVersion": 1,
  "goal": "watch file system changes using watchfiles.watch and Change enum",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/watchfiles@1.2.0"
  ],
  "symbols": [
    "watchfiles.Change",
    "watchfiles.watch"
  ]
}

import tempfile
import threading
import time
from pathlib import Path
import watchfiles
from watchfiles import watch, Change

# 1. Verify Change enum values and raw_str representation
assert Change.added == 1
assert Change.modified == 2
assert Change.deleted == 3
assert Change.added.raw_str() == "added"
assert Change.modified.raw_str() == "modified"
assert Change.deleted.raw_str() == "deleted"
assert Change(1) is Change.added
assert Change(2) is Change.modified
assert Change(3) is Change.deleted

# 2. Verify watch() yields file addition, modification, and deletion events
with tempfile.TemporaryDirectory() as tmpdir:
    dir_path = Path(tmpdir)
    target_file = dir_path / "test_file.txt"
    stop_event = threading.Event()

    def generate_fs_events():
        time.sleep(0.05)
        target_file.write_text("initial content")
        time.sleep(0.05)
        target_file.write_text("updated content")
        time.sleep(0.05)
        target_file.unlink()
        time.sleep(0.05)
        stop_event.set()

    worker = threading.Thread(target=generate_fs_events)
    worker.start()

    detected_changes = []
    for changes in watch(tmpdir, stop_event=stop_event, debounce=30, step=10, rust_timeout=50):
        for change_type, path_str in changes:
            if Path(path_str).name == "test_file.txt":
                detected_changes.append(change_type)

    worker.join()

    assert Change.added in detected_changes, f"Expected Change.added in {detected_changes}"
    assert Change.modified in detected_changes, f"Expected Change.modified in {detected_changes}"
    assert Change.deleted in detected_changes, f"Expected Change.deleted in {detected_changes}"

# 3. Verify watch() custom filter selectively captures files
with tempfile.TemporaryDirectory() as tmpdir:
    dir_path = Path(tmpdir)
    matched_file = dir_path / "keep.json"
    ignored_file = dir_path / "ignore.log"
    filter_stop_event = threading.Event()

    def generate_filtered_events():
        time.sleep(0.05)
        matched_file.write_text("{}")
        ignored_file.write_text("log line")
        time.sleep(0.05)
        filter_stop_event.set()

    def filter_func(change: Change, path: str) -> bool:
        return path.endswith(".json")

    worker = threading.Thread(target=generate_filtered_events)
    worker.start()

    filtered_file_names = []
    for changes in watch(tmpdir, watch_filter=filter_func, stop_event=filter_stop_event, debounce=30, step=10, rust_timeout=50):
        for change_type, path_str in changes:
            filtered_file_names.append(Path(path_str).name)

    worker.join()

    assert "keep.json" in filtered_file_names, "Expected keep.json to pass filter"
    assert "ignore.log" not in filtered_file_names, "Expected ignore.log to be filtered out"

# 4. Verify watch() with yield_on_timeout
timeout_stop_event = threading.Event()
with tempfile.TemporaryDirectory() as tmpdir:
    timeout_yield_count = 0
    for changes in watch(tmpdir, stop_event=timeout_stop_event, rust_timeout=50, yield_on_timeout=True):
        timeout_yield_count += 1
        timeout_stop_event.set()
        break
    assert timeout_yield_count == 1, "Expected watch to yield upon timeout when yield_on_timeout=True"

print("CONTRACT PASS: watchfiles.watch and Change verified offline")