"""Step definitions for resource_file_watcher_coverage.feature. These steps target specific uncovered lines in resource_file_watcher.py: - Lines 168-183: watch() OSError fallback to polling / re-raise - Lines 255-264: start() OSError fallback to PollingObserver - Lines 295-304: _switch_to_polling_observer_locked() - Lines 325-329: stop() observer join timeout warning - _fire_change with dest_path in event details """ from __future__ import annotations import contextlib import tempfile from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch from behave import given, then, when from cleveragents.application.services.resource_file_watcher import ( FileChangeType, ResourceFileWatcher, ) from cleveragents.infrastructure.events.models import DomainEvent from cleveragents.infrastructure.events.types import EventType __all__: list[str] = [] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _cleanup_watcher(context: Any) -> None: """Stop watcher and clean up temp directory.""" watcher = getattr(context, "rfw_watcher", None) if watcher is not None: # Force-reset running flag to avoid errors during cleanup watcher._running = False watcher._observer = None tmpdir = getattr(context, "rfw_tmpdir_obj", None) if tmpdir is not None: tmpdir.cleanup() # Restore any patches for patcher in getattr(context, "rfw_patches", []): with contextlib.suppress(RuntimeError): patcher.stop() def _make_mock_observer( *, schedule_raises: bool = False, alive_after_join: bool = False ) -> MagicMock: """Create a mock observer with configurable behaviour.""" mock_obs = MagicMock() mock_obs.daemon = True if schedule_raises: mock_obs.schedule.side_effect = OSError("Simulated inotify limit") if alive_after_join: mock_obs.is_alive.return_value = True else: mock_obs.is_alive.return_value = False return mock_obs class _TrackingEventBus: """Minimal event bus that records emitted events.""" def __init__(self) -> None: self.events: list[DomainEvent] = [] def emit(self, event: DomainEvent) -> None: self.events.append(event) def subscribe(self, event_type: EventType, handler: Any) -> None: pass # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("the resource_file_watcher module is imported") def step_rfw_module_imported(context: Any) -> None: """Ensure the module is importable.""" assert ResourceFileWatcher is not None assert FileChangeType is not None context.rfw_patches = [] # --------------------------------------------------------------------------- # start() OSError fallback to PollingObserver (lines 255-264) # --------------------------------------------------------------------------- @given("a ResourceFileWatcher with default auto_reindex enabled") def step_rfw_default_watcher(context: Any) -> None: context.rfw_watcher = ResourceFileWatcher(auto_reindex=True, debounce_seconds=0.01) context.add_cleanup(_cleanup_watcher, context) @given("the native Observer start method is patched to raise OSError") def step_rfw_patch_observer_start(context: Any) -> None: """Patch Observer so its start() raises OSError, and PollingObserver works.""" mock_native = MagicMock() mock_native.return_value = _make_mock_observer() mock_native.return_value.start.side_effect = OSError("No inotify watches available") mock_native.return_value.daemon = True mock_polling = MagicMock() mock_polling_instance = _make_mock_observer() mock_polling_instance.start.side_effect = None # polling works fine mock_polling.return_value = mock_polling_instance p1 = patch( "cleveragents.application.services.resource_file_watcher.Observer", mock_native, ) p2 = patch( "cleveragents.application.services.resource_file_watcher.PollingObserver", mock_polling, ) p1.start() p2.start() context.rfw_patches.extend([p1, p2]) context.rfw_mock_polling_instance = mock_polling_instance @when("the watcher start method is called") def step_rfw_start(context: Any) -> None: context.rfw_watcher.start() @then("the watcher should be running") def step_rfw_is_running(context: Any) -> None: assert context.rfw_watcher.is_running, "Expected watcher to be running" @then("the watcher should be using polling observer") def step_rfw_using_polling(context: Any) -> None: assert context.rfw_watcher._using_polling, "Expected watcher to be using polling" # --------------------------------------------------------------------------- # watch() schedule OSError fallback (lines 168-180, 295-304) # --------------------------------------------------------------------------- @given("a ResourceFileWatcher that is running with a mock native observer") def step_rfw_running_with_native_observer(context: Any) -> None: """Create a watcher and set it to running state with a mock observer.""" context.rfw_watcher = ResourceFileWatcher(debounce_seconds=0.01) context.add_cleanup(_cleanup_watcher, context) context.rfw_mock_observer = _make_mock_observer() context.rfw_watcher._observer = context.rfw_mock_observer context.rfw_watcher._running = True context.rfw_watcher._using_polling = False @given("a temporary file exists for watching") def step_rfw_temp_file(context: Any) -> None: context.rfw_tmpdir_obj = tempfile.TemporaryDirectory() tmpdir = Path(context.rfw_tmpdir_obj.name) fpath = tmpdir / "testfile.py" fpath.write_text("# test content\n") context.rfw_temp_file = fpath @given("the mock observer schedule method raises OSError") def step_rfw_mock_schedule_raises(context: Any) -> None: context.rfw_mock_observer.schedule.side_effect = OSError("Simulated inotify limit") # Patch PollingObserver constructor to return a working mock mock_polling_cls = MagicMock() mock_polling_instance = _make_mock_observer() mock_polling_instance.schedule.return_value = MagicMock() # watch handle mock_polling_cls.return_value = mock_polling_instance context.rfw_mock_polling_instance = mock_polling_instance p = patch( "cleveragents.application.services.resource_file_watcher.PollingObserver", mock_polling_cls, ) p.start() context.rfw_patches.append(p) @when('I watch the temporary file with resource "{rid}" project "{proj}"') def step_rfw_watch_file(context: Any, rid: str, proj: str) -> None: context.rfw_watcher.watch( context.rfw_temp_file, resource_id=rid, project=proj, ) @then("the watcher should have switched to polling observer") def step_rfw_switched_to_polling(context: Any) -> None: assert context.rfw_watcher._using_polling, ( "Expected watcher to have switched to polling observer" ) @then("the temporary file should be in the watched paths") def step_rfw_temp_file_in_watched(context: Any) -> None: resolved = str(context.rfw_temp_file.resolve()) assert resolved in context.rfw_watcher._watched_paths, ( f"Expected {resolved} in watched paths" ) # --------------------------------------------------------------------------- # watch() schedule OSError re-raise when already polling (line 183) # --------------------------------------------------------------------------- @given("a ResourceFileWatcher that is running with a mock polling observer") def step_rfw_running_with_polling_observer(context: Any) -> None: """Create a watcher set to running state with polling already active.""" context.rfw_watcher = ResourceFileWatcher(debounce_seconds=0.01) context.add_cleanup(_cleanup_watcher, context) context.rfw_mock_observer = _make_mock_observer(schedule_raises=True) context.rfw_watcher._observer = context.rfw_mock_observer context.rfw_watcher._running = True context.rfw_watcher._using_polling = True # already polling @when( 'I watch the temporary file expecting OSError with resource "{rid}" project "{proj}"' ) def step_rfw_watch_file_expect_error(context: Any, rid: str, proj: str) -> None: context.rfw_watch_error = None try: context.rfw_watcher.watch( context.rfw_temp_file, resource_id=rid, project=proj, ) except OSError as exc: context.rfw_watch_error = exc @then("the watch call should have raised OSError") def step_rfw_watch_raised_oserror(context: Any) -> None: assert context.rfw_watch_error is not None, "Expected OSError to be raised" assert isinstance(context.rfw_watch_error, OSError) # --------------------------------------------------------------------------- # stop() with observer that stays alive (lines 325-329) # --------------------------------------------------------------------------- @given("a ResourceFileWatcher with a mock observer that stays alive after join") def step_rfw_mock_observer_stays_alive(context: Any) -> None: context.rfw_watcher = ResourceFileWatcher(debounce_seconds=0.01) context.add_cleanup(_cleanup_watcher, context) mock_obs = _make_mock_observer(alive_after_join=True) context.rfw_watcher._observer = mock_obs context.rfw_watcher._running = True context.rfw_watcher._using_polling = False context.rfw_mock_alive_observer = mock_obs @when("the watcher is stopped") def step_rfw_stop(context: Any) -> None: context.rfw_watcher.stop() @then("the watcher should not be running after stop") def step_rfw_not_running_after_stop(context: Any) -> None: assert not context.rfw_watcher.is_running, "Expected watcher to not be running" @then("the observer join timeout warning path should have been exercised") def step_rfw_observer_join_timeout_exercised(context: Any) -> None: """Verify the observer's is_alive was checked (which triggers the warning log).""" context.rfw_mock_alive_observer.stop.assert_called_once() context.rfw_mock_alive_observer.join.assert_called_once() context.rfw_mock_alive_observer.is_alive.assert_called_once() # --------------------------------------------------------------------------- # _switch_to_polling_observer_locked (lines 295-304) # --------------------------------------------------------------------------- @given("a ResourceFileWatcher with a mock native observer and watched paths") def step_rfw_watcher_with_native_and_paths(context: Any) -> None: context.rfw_watcher = ResourceFileWatcher(debounce_seconds=0.01) context.add_cleanup(_cleanup_watcher, context) context.rfw_old_observer = _make_mock_observer() context.rfw_watcher._observer = context.rfw_old_observer context.rfw_watcher._running = True context.rfw_watcher._using_polling = False @when("_switch_to_polling_observer_locked is called directly") def step_rfw_call_switch_to_polling(context: Any) -> None: # Need to add a watched path so _build_observer schedules something resolved = str(context.rfw_temp_file.resolve()) context.rfw_watcher._watched_paths[resolved] = ("res-switch", "proj-switch") # Patch PollingObserver to return a mock mock_polling_cls = MagicMock() mock_polling_instance = _make_mock_observer() mock_polling_instance.schedule.return_value = MagicMock() mock_polling_cls.return_value = mock_polling_instance context.rfw_new_polling_instance = mock_polling_instance p = patch( "cleveragents.application.services.resource_file_watcher.PollingObserver", mock_polling_cls, ) p.start() context.rfw_patches.append(p) with context.rfw_watcher._lock: context.rfw_watcher._switch_to_polling_observer_locked() @then("the old observer should have been stopped") def step_rfw_old_observer_stopped(context: Any) -> None: context.rfw_old_observer.stop.assert_called_once() context.rfw_old_observer.join.assert_called_once() @then("the watcher should now have a polling observer") def step_rfw_now_has_polling_observer(context: Any) -> None: assert context.rfw_watcher._using_polling, "Expected _using_polling to be True" assert context.rfw_watcher._observer is context.rfw_new_polling_instance # --------------------------------------------------------------------------- # start() with pre-registered paths + polling fallback (lines 255-264) # --------------------------------------------------------------------------- @given("a ResourceFileWatcher with a pre-registered watched path") def step_rfw_watcher_with_preregistered_path(context: Any) -> None: context.rfw_tmpdir_obj = tempfile.TemporaryDirectory() tmpdir = Path(context.rfw_tmpdir_obj.name) fpath = tmpdir / "preregistered.py" fpath.write_text("# pre-registered\n") context.rfw_temp_file = fpath context.rfw_watcher = ResourceFileWatcher(auto_reindex=True, debounce_seconds=0.01) context.add_cleanup(_cleanup_watcher, context) # Pre-register the path by directly injecting into _watched_paths # (Normally watch() requires file to exist and watcher to be started, # but _build_observer reads _watched_paths to set up directory watches.) resolved = str(fpath.resolve()) context.rfw_watcher._watched_paths[resolved] = ("res-pre", "proj-pre") context.rfw_preregistered_parent = str(fpath.resolve().parent) @then("the directory watches should include the pre-registered path parent") def step_rfw_dir_watches_include_parent(context: Any) -> None: parent = context.rfw_preregistered_parent assert parent in context.rfw_watcher._dir_watches, ( f"Expected {parent} in dir_watches, got {list(context.rfw_watcher._dir_watches.keys())}" ) # --------------------------------------------------------------------------- # _fire_change with dest_path in event details # --------------------------------------------------------------------------- @given("a ResourceFileWatcher with a tracking EventBus and debounce 0") def step_rfw_with_tracking_event_bus(context: Any) -> None: context.rfw_event_bus = _TrackingEventBus() context.rfw_watcher = ResourceFileWatcher( event_bus=context.rfw_event_bus, debounce_seconds=0.0, ) context.add_cleanup(_cleanup_watcher, context) @given("the watcher is marked as running") def step_rfw_mark_running(context: Any) -> None: context.rfw_watcher._running = True @when("_fire_change is invoked with a dest_path argument") def step_rfw_fire_change_with_dest_path(context: Any) -> None: context.rfw_watcher._fire_change( "/tmp/src_file.py", "res-dest", "proj-dest", FileChangeType.MOVED, dest_path="/tmp/dest_file.py", ) @then("the emitted event details should contain the dest_path") def step_rfw_event_has_dest_path(context: Any) -> None: assert len(context.rfw_event_bus.events) == 1, ( f"Expected 1 event, got {len(context.rfw_event_bus.events)}" ) details = context.rfw_event_bus.events[0].details assert details.get("dest_path") == "/tmp/dest_file.py", ( f"Expected dest_path='/tmp/dest_file.py' in details, got {details}" ) assert details.get("change_type") == "moved" # --------------------------------------------------------------------------- # _fire_change exits early when not running (line 459) # --------------------------------------------------------------------------- @given("a ResourceFileWatcher with a tracking callback") def step_rfw_with_tracking_callback(context: Any) -> None: context.rfw_callback_calls = [] def _on_change(rid: str, proj: str, ct: FileChangeType) -> None: context.rfw_callback_calls.append((rid, proj, ct)) context.rfw_watcher = ResourceFileWatcher( on_change=_on_change, debounce_seconds=0.0, ) context.add_cleanup(_cleanup_watcher, context) @given("the watcher is not running") def step_rfw_not_running(context: Any) -> None: context.rfw_watcher._running = False @when("_fire_change is invoked directly") def step_rfw_fire_change_not_running(context: Any) -> None: context.rfw_watcher._fire_change( "/tmp/some_file.py", "res-early", "proj-early", FileChangeType.MODIFIED, ) @then("the callback should not have been called") def step_rfw_callback_not_called(context: Any) -> None: assert len(context.rfw_callback_calls) == 0, ( f"Expected no callback calls, got {len(context.rfw_callback_calls)}" )