"""ResourceFileWatcher steps for UKO Indexer behave tests.""" from __future__ import annotations import shutil import tempfile import threading import time from pathlib import Path from typing import Any from behave import given, then, when # type: ignore[import-untyped] from watchdog.events import ( DirModifiedEvent, FileCreatedEvent, FileDeletedEvent, FileModifiedEvent, FileMovedEvent, ) from cleveragents.application.services.resource_file_watcher import ( FileChangeType, ResourceFileWatcher, ) from cleveragents.infrastructure.events.types import EventType from features.mocks.uko_indexer_mocks import TrackingEventBus __all__: list[str] = [] def _cleanup_watcher(context: Any) -> None: """Stop the file watcher if running (behave cleanup hook).""" watcher = getattr(context, "fw_watcher", None) if watcher is not None and watcher.is_running: watcher.stop() tmpdir = getattr(context, "fw_tmpdir_obj", None) if tmpdir is not None: tmpdir.cleanup() @given("idx a ResourceFileWatcher with default settings") def step_fw_default(context: Any) -> None: context.fw_watcher = ResourceFileWatcher() context.add_cleanup(_cleanup_watcher, context) @given("idx a ResourceFileWatcher with callback tracking and debounce {secs}") def step_fw_tracking(context: Any, secs: str) -> None: context.fw_changes = [] # list of (resource_id, project, change_type) context.fw_callback_event = threading.Event() def _on_change(rid: str, proj: str, ct: FileChangeType) -> None: context.fw_changes.append((rid, proj, ct)) context.fw_callback_event.set() context.fw_watcher = ResourceFileWatcher( on_change=_on_change, debounce_seconds=float(secs), ) context.add_cleanup(_cleanup_watcher, context) @given("idx a ResourceFileWatcher with EventBus tracking and debounce {secs}") def step_fw_eventbus(context: Any, secs: str) -> None: context.fw_event_bus = TrackingEventBus() context.fw_watcher = ResourceFileWatcher( event_bus=context.fw_event_bus, debounce_seconds=float(secs), ) context.add_cleanup(_cleanup_watcher, context) @given("idx a ResourceFileWatcher with a failing callback and debounce {secs}") def step_fw_failing_cb(context: Any, secs: str) -> None: def _boom(rid: str, proj: str, ct: FileChangeType) -> None: _ = rid, proj, ct msg = "intentional test failure" raise RuntimeError(msg) context.fw_watcher = ResourceFileWatcher( on_change=_boom, debounce_seconds=float(secs), ) context.add_cleanup(_cleanup_watcher, context) @given('idx a temporary watched file "{name}" with content "{content}"') def step_fw_temp_file(context: Any, name: str, content: str) -> None: context.fw_tmpdir_obj = tempfile.TemporaryDirectory() tmpdir = Path(context.fw_tmpdir_obj.name) fpath = tmpdir / name fpath.write_text(content) context.fw_temp_file = fpath @given("idx a temporary watched directory") def step_fw_temp_dir(context: Any) -> None: context.fw_tmpdir_obj = tempfile.TemporaryDirectory() context.fw_temp_dir = Path(context.fw_tmpdir_obj.name) @when("idx the file watcher is started") @given("idx the file watcher is started") def step_fw_start(context: Any) -> None: context.fw_watcher.start() @when("idx the file watcher is stopped") def step_fw_stop(context: Any) -> None: context.fw_watcher.stop() @then("idx the file watcher should be running") def step_fw_running(context: Any) -> None: assert context.fw_watcher.is_running, "Watcher should be running" @then("idx the file watcher should not be running") def step_fw_not_running(context: Any) -> None: assert not context.fw_watcher.is_running, "Watcher should not be running" @when('idx the watcher registers the file for resource "{rid}" project "{proj}"') @given('idx the watcher registers the file for resource "{rid}" project "{proj}"') def step_fw_register(context: Any, rid: str, proj: str) -> None: context.fw_watcher.watch( context.fw_temp_file, resource_id=rid, project=proj, ) @when("idx the watcher unregisters the watched file") def step_fw_unregister(context: Any) -> None: context.fw_watcher.unwatch(context.fw_temp_file) @then("idx the watcher should have {n:d} watched paths") def step_fw_count(context: Any, n: int) -> None: actual = len(context.fw_watcher.watched_paths) assert actual == n, f"Expected {n} watched paths, got {actual}" @when("idx a FileModifiedEvent is simulated for the watched file") def step_fw_sim_modified(context: Any) -> None: event = FileModifiedEvent(str(context.fw_temp_file.resolve())) context.fw_watcher._handle_fs_event(event) @when("idx a FileDeletedEvent is simulated for the watched file") def step_fw_sim_deleted(context: Any) -> None: event = FileDeletedEvent(str(context.fw_temp_file.resolve())) context.fw_watcher._handle_fs_event(event) @when("idx 5 FileModifiedEvents are simulated rapidly for the watched file") def step_fw_sim_rapid(context: Any) -> None: for _ in range(5): event = FileModifiedEvent(str(context.fw_temp_file.resolve())) context.fw_watcher._handle_fs_event(event) @when("idx we wait {secs} seconds for debounce") def step_fw_wait(context: Any, secs: str) -> None: time.sleep(float(secs)) @then("idx the on_change callback should fire within {secs:d} seconds") def step_fw_cb_wait(context: Any, secs: int) -> None: fired = context.fw_callback_event.wait(timeout=float(secs)) assert fired, f"Callback did not fire within {secs}s" @then('idx the callback should receive resource "{rid}" project "{proj}" change "{ct}"') def step_fw_cb_check(context: Any, rid: str, proj: str, ct: str) -> None: assert len(context.fw_changes) > 0, "No callback invocations recorded" last = context.fw_changes[-1] assert last[0] == rid, f"resource_id: expected {rid}, got {last[0]}" assert last[1] == proj, f"project: expected {proj}, got {last[1]}" assert last[2] == ct, f"change_type: expected {ct}, got {last[2]}" @then("idx the on_change callback should have fired exactly {n:d} time") @then("idx the on_change callback should have fired exactly {n:d} times") def step_fw_cb_count(context: Any, n: int) -> None: actual = len(context.fw_changes) assert actual == n, f"Expected {n} callback invocations, got {actual}" @then( "idx the EventBus should receive a RESOURCE_MODIFIED event within {secs:d} seconds" ) def step_fw_eb_wait(context: Any, secs: int) -> None: received = context.fw_event_bus.wait(timeout=float(secs)) assert received, f"No EventBus event within {secs}s" assert len(context.fw_event_bus.events) > 0 assert context.fw_event_bus.events[-1].event_type == EventType.RESOURCE_MODIFIED @then('idx the event details should contain resource_id "{rid}"') def step_fw_eb_detail(context: Any, rid: str) -> None: last = context.fw_event_bus.events[-1] assert last.details.get("resource_id") == rid @then('idx watching path "{path}" should raise ValueError containing "{substr}"') def step_fw_watch_bad_path(context: Any, path: str, substr: str) -> None: try: context.fw_watcher.watch( Path(path), resource_id="r", project="p", ) assert False, "Expected ValueError" # noqa: B011 except ValueError as exc: assert substr in str(exc), f"'{substr}' not in '{exc}'" @then( 'idx watching the temporary directory should raise ValueError containing "{substr}"' ) def step_fw_watch_dir(context: Any, substr: str) -> None: try: context.fw_watcher.watch( context.fw_temp_dir, resource_id="r", project="p", ) assert False, "Expected ValueError" # noqa: B011 except ValueError as exc: assert substr in str(exc), f"'{substr}' not in '{exc}'" @then("idx creating ResourceFileWatcher with debounce -1.0 should raise ValueError") def step_fw_bad_debounce(context: Any) -> None: try: ResourceFileWatcher(debounce_seconds=-1.0) assert False, "Expected ValueError" # noqa: B011 except ValueError: pass @when('idx a FileModifiedEvent is simulated for path "{path}"') def step_fw_sim_unwatched(context: Any, path: str) -> None: event = FileModifiedEvent(path) context.fw_watcher._handle_fs_event(event) @when("idx a directory event is simulated for the watched path") def step_fw_sim_dir_event(context: Any) -> None: event = DirModifiedEvent(str(context.fw_temp_file.resolve())) context.fw_watcher._handle_fs_event(event) @then("idx watching with empty resource_id should raise ValueError") def step_fw_watch_empty_rid(context: Any) -> None: try: context.fw_watcher.watch( context.fw_temp_file, resource_id="", project="p", ) assert False, "Expected ValueError" # noqa: B011 except ValueError: pass @then("idx watching with empty project should raise ValueError") def step_fw_watch_empty_project(context: Any) -> None: try: context.fw_watcher.watch( context.fw_temp_file, resource_id="r", project="", ) assert False, "Expected ValueError" # noqa: B011 except ValueError: pass @then("idx starting the watcher again should raise RuntimeError") def step_fw_double_start(context: Any) -> None: try: context.fw_watcher.start() assert False, "Expected RuntimeError" # noqa: B011 except RuntimeError: pass @when("idx I stop the file watcher") @then("idx I stop the file watcher") def step_fw_explicit_stop(context: Any) -> None: context.fw_watcher.stop() @then("idx the debounce_seconds should be {value}") def step_fw_debounce_prop(context: Any, value: str) -> None: assert context.fw_watcher.debounce_seconds == float(value), ( f"Expected {value}, got {context.fw_watcher.debounce_seconds}" ) @when("idx a FileCreatedEvent is simulated for the watched file") def step_fw_sim_created(context: Any) -> None: event = FileCreatedEvent(str(context.fw_temp_file.resolve())) context.fw_watcher._handle_fs_event(event) @when("idx a FileMovedEvent is simulated for the watched file") def step_fw_sim_moved(context: Any) -> None: event = FileMovedEvent( str(context.fw_temp_file.resolve()), str(context.fw_temp_file.resolve()) + ".bak", ) context.fw_watcher._handle_fs_event(event) @given("idx a ResourceFileWatcher with auto_reindex disabled") def step_fw_auto_reindex_disabled(context: Any) -> None: context.fw_watcher = ResourceFileWatcher(auto_reindex=False) context.add_cleanup(_cleanup_watcher, context) @then("idx the auto_reindex property should be false") def step_fw_auto_reindex_prop(context: Any) -> None: assert context.fw_watcher.auto_reindex is False @given("idx a ResourceFileWatcher with a directly-failing callback and debounce {secs}") def step_fw_direct_fail_cb(context: Any, secs: str) -> None: def _boom(rid: str, proj: str, ct: FileChangeType) -> None: _ = rid, proj, ct raise RuntimeError("intentional") context.fw_watcher = ResourceFileWatcher( on_change=_boom, debounce_seconds=float(secs), ) context.add_cleanup(_cleanup_watcher, context) @when("idx _fire_change is called directly with failing callback") def step_fw_fire_change_fail_cb(context: Any) -> None: path = str(context.fw_temp_file.resolve()) context.fw_watcher._fire_change( path, "res-001", "local/test", FileChangeType.MODIFIED, ) @given("idx a ResourceFileWatcher with a failing event bus and debounce {secs}") def step_fw_failing_event_bus(context: Any, secs: str) -> None: class _FailingBus: def emit(self, event: object) -> None: raise RuntimeError("bus boom") def subscribe(self, event_type: object, handler: object) -> None: pass context.fw_watcher = ResourceFileWatcher( event_bus=_FailingBus(), # type: ignore[arg-type] debounce_seconds=float(secs), ) context.add_cleanup(_cleanup_watcher, context) @when("idx _fire_change is called directly with failing event bus") def step_fw_fire_change_fail_bus(context: Any) -> None: path = str(context.fw_temp_file.resolve()) context.fw_watcher._fire_change( path, "res-001", "local/test", FileChangeType.MODIFIED, ) @when("idx _fire_change is called directly after stop") def step_fw_fire_change_after_stop(context: Any) -> None: path = str(context.fw_temp_file.resolve()) context.fw_watcher._fire_change( path, "res-001", "local/test", FileChangeType.MODIFIED, ) @when("idx a _ResourceChangeHandler is used to route events") def step_fw_handler_routes(context: Any) -> None: from cleveragents.application.services.resource_file_watcher import ( _ResourceChangeHandler, ) handler = _ResourceChangeHandler(context.fw_watcher) path = str(context.fw_temp_file.resolve()) handler.on_modified(FileModifiedEvent(path)) handler.on_created(FileCreatedEvent(path)) handler.on_deleted(FileDeletedEvent(path)) handler.on_moved(FileMovedEvent(path, path + ".bak")) @then("idx file watcher should fire both callback and event_bus on change") def step_callback_and_event_bus_combo(context: Any) -> None: td = tempfile.mkdtemp() try: fp = Path(td) / "combo.py" fp.write_text("x=1\n") chs: list[str] = [] ev = threading.Event() uid = "01HQ8ZDRX50000000000000001" bus = TrackingEventBus() def cb(r: str, _p: str, _ct: FileChangeType) -> None: chs.append(r) ev.set() w = ResourceFileWatcher( on_change=cb, event_bus=bus, debounce_seconds=0.01, ) w.start() w.watch(fp, resource_id=uid, project="local/test") w._handle_fs_event(FileModifiedEvent(str(fp.resolve()))) assert ev.wait(timeout=2.0) and chs[-1] == uid assert bus.wait(timeout=2.0) and len(bus.events) >= 1 w.stop() finally: shutil.rmtree(td, ignore_errors=True) @then("idx file watcher should schedule new dir on cross-directory move") def step_fw_cross_dir_move(context: Any) -> None: ev = threading.Event() def cb(rid: str, _p: str, _ct: str) -> None: ev.set() td = tempfile.mkdtemp() try: sub = Path(td) / "sub" sub.mkdir() fp = Path(td) / "orig.py" fp.write_text("x=1\n", encoding="utf-8") dest = sub / "moved.py" w = ResourceFileWatcher(on_change=cb, debounce_seconds=0.05) w.watch(fp, resource_id="01HQ8ZDRX50000000000000077", project="local/t") w.start() w._handle_fs_event(FileMovedEvent(str(fp.resolve()), str(dest.resolve()))) assert ev.wait(timeout=2.0) assert str(sub.resolve()) in w._dir_watches w.stop() finally: shutil.rmtree(td, ignore_errors=True) @then("idx file watcher should handle move then modify sequence") def step_fw_move_then_modify(context: Any) -> None: chs: list[str] = [] ev = threading.Event() uid = "01HQ8ZDRX50000000000000099" def cb(rid: str, _p: str, _ct: str) -> None: chs.append(rid) ev.set() td = tempfile.mkdtemp() try: fp = Path(td) / "src.py" fp.write_text("x=1\n", encoding="utf-8") dp = str((Path(td) / "dst.py").resolve()) w = ResourceFileWatcher(on_change=cb, debounce_seconds=0.05) w.watch(fp, resource_id=uid, project="local/test") w.start() w._handle_fs_event(FileMovedEvent(str(fp.resolve()), dp)) assert ev.wait(timeout=2.0), "Move cb" ev.clear() w._handle_fs_event(FileModifiedEvent(dp)) assert ev.wait(timeout=2.0), "Modify cb" w.stop() assert len(chs) >= 2 and chs[0] == uid and chs[1] == uid finally: shutil.rmtree(td, ignore_errors=True)