How to use the watchdog.events.FileMovedEvent function in watchdog

To help you get started, we’ve selected a few watchdog examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github CenterForOpenScience / osf-sync / tests / sync / test_consolidated_event_handler.py View on Github external
def test_rename_file(self, sha_mock):
        sha_mock.return_value = '1234'
        project = self.PROJECT_STRUCTURE[0]
        file_path = self.root_dir.join(
            project['files'][0]['children'][0]['rel_path'].lstrip(os.path.sep)
        )
        os.rename(str(file_path), 'foobar.baz')
        self.sync_worker.flushed.wait()
        assert len(self.sync_worker._event_cache.events) == 1, 'exactly one event captured'
        assert isinstance(self.sync_worker._event_cache.events[0], FileMovedEvent) is True, 'the one captured event is a FileMovedEvent'
github brickgao / specchio / tests / test_handlers.py View on Github external
def test_on_moved(self, _mv, _os):
        _mv.return_value = True
        _os.path.abspath.side_effect = ["/a/1.py", "/a/2.py"]
        _os.path.join.side_effect = ["/b/a/1.py", "/b/a/2.py"]
        _event = FileMovedEvent(src_path="/a/1.py", dest_path="/a/2.py")
        self.handler.on_moved(_event)
        _mv.assert_called_once_with(
            dst_ssh=self.handler.dst_ssh,
            src_path="/b/a/1.py",
            dst_path="/b/a/2.py"
        )
github brickgao / specchio / tests / test_handlers.py View on Github external
def test_on_moved_all_ignore(self, _rsync, _mv, _create_folder, _rm, _os):
        _mv.return_value = True
        _create_folder.return_value = True
        _rm.return_value = True
        _rsync.return_value = True
        _os.path.abspath.side_effect = ["/a/test.py", "/a/test.py"]
        _os.path.join.side_effect = ["/b/a/test.py", "/b/a/test.py"]
        _event = FileMovedEvent(src_path="/a/test.py", dest_path="/a/test.py")
        self.handler.on_moved(_event)
        self.assertEqual(_mv.call_count, 0)
        self.assertEqual(_rm.call_count, 0)
        self.assertEqual(_create_folder.call_count, 0)
        self.assertEqual(_rsync.call_count, 0)
github lektor / lektor / lektor / watcher.py View on Github external
def on_any_event(self, event):
        if not isinstance(event, DirModifiedEvent):
            path = event.dest_path if isinstance(event, FileMovedEvent) else event.src_path
            item = (time.time(), event.event_type, path)
            if self.queue is not None:
                self.queue.put(item)
            else:
                self.callback(*item)
github gorakhargosh / watchdog / watchdog / observers / fsevents_observer.py View on Github external
def _dispatch_events_for_path(self, handler, recursive, path):
            with self._lock:
                diff = self._get_directory_snapshot_diff(path, recursive)
                if diff:
                    for path in diff.files_deleted:
                        handler.dispatch(FileDeletedEvent(path, handler))
                    for path in diff.files_modified:
                        handler.dispatch(FileModifiedEvent(path, handler))
                    for path in diff.files_created:
                        handler.dispatch(FileCreatedEvent(path, handler))
                    for path, dest_path in diff.files_moved.items():
                        handler.dispatch(FileMovedEvent(path, dest_path, handler))

                    for path in diff.dirs_modified:
                        handler.dispatch(DirModifiedEvent(path, handler))
                    for path in diff.dirs_deleted:
                        handler.dispatch(DirDeletedEvent(path, handler))
                    for path in diff.dirs_created:
                        handler.dispatch(DirCreatedEvent(path, handler))
                    for path, dest_path in diff.dirs_moved.items():
                        handler.dispatch(DirMovedEvent(path, dest_path, handler))
github ppalex7 / SourcePawnCompletions / watchdog / events.py View on Github external
:func:`os.walk`.
  :returns:
      An iterable of file system events of type :class:`DirMovedEvent` and
      :class:`FileMovedEvent`.
  """
  src_dir_path = absolute_path(src_dir_path) if src_dir_path else None
  dest_dir_path = absolute_path(dest_dir_path)
  for root, directories, filenames in _walker(dest_dir_path):
    for directory in directories:
      full_path = os.path.join(root, directory)
      renamed_path = full_path.replace(dest_dir_path, src_dir_path) if src_dir_path else None
      yield DirMovedEvent(renamed_path, full_path)
    for filename in filenames:
      full_path = os.path.join(root, filename)
      renamed_path = full_path.replace(dest_dir_path, src_dir_path) if src_dir_path else None
      yield FileMovedEvent(renamed_path, full_path)
github gorakhargosh / watchdog / src / watchdog / observers / polling.py View on Github external
self.queue_event(DirDeletedEvent(self.watch.path))
                self.stop()
                return

            events = DirectorySnapshotDiff(self._snapshot, new_snapshot)
            self._snapshot = new_snapshot

            # Files.
            for src_path in events.files_deleted:
                self.queue_event(FileDeletedEvent(src_path))
            for src_path in events.files_modified:
                self.queue_event(FileModifiedEvent(src_path))
            for src_path in events.files_created:
                self.queue_event(FileCreatedEvent(src_path))
            for src_path, dest_path in events.files_moved:
                self.queue_event(FileMovedEvent(src_path, dest_path))

            # Directories.
            for src_path in events.dirs_deleted:
                self.queue_event(DirDeletedEvent(src_path))
            for src_path in events.dirs_modified:
                self.queue_event(DirModifiedEvent(src_path))
            for src_path in events.dirs_created:
                self.queue_event(DirCreatedEvent(src_path))
            for src_path, dest_path in events.dirs_moved:
                self.queue_event(DirMovedEvent(src_path, dest_path))
github gorakhargosh / watchdog / watchdog / observers / kqueue_observer.py View on Github external
ref_stat_info = ref_dir_snapshot.stat_info(path_renamed)
                except KeyError:
                    # Caught a temporary file most probably.
                    # So fire a created+deleted event sequence.
                    self.event_queue.put(
                        FileCreatedEvent(path_renamed,
                                         handlers=self.handlers))
                    self.event_queue.put(
                        FileDeletedEvent(path_renamed,
                                         handlers=self.handlers))
                    continue

                try:
                    path = new_dir_snapshot.path_for_inode(ref_stat_info.st_ino)
                    self.event_queue.put(
                        FileMovedEvent(path_renamed,
                                       dest_path=path,
                                       handlers=self.handlers))
                    self._unregister_path(path_renamed)
                    self._register_path(path, is_directory=False)
                except KeyError:
                    # We could not find the new name.
                    self.event_queue.put(
                        FileDeletedEvent(path_renamed,
                                         handlers=self.handlers))
                    self._unregister_path(path_renamed)