Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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'
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"
)
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)
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)
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))
: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)
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))
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)