Merge pull request #15 from jeffrey82221/feature/add_cp

feature: add copy method

Thank you @jeffrey82221

Author: MarineChap

dropboxdrivefs/core.py

@@ -1,6 +1,4 @@
 import logging
-import os.path
-
 import dropbox.files
 import requests
 from dropbox.exceptions import ApiError
@@ -176,7 +174,56 @@ def _refactor_metadata(self, metadata, detail=True):
                 return {"name": metadata.path_display, "size": None, "type": None}
         else:
             return metadata.path_display
-
+        
+    def copy(self, path1, path2, recursive=True, on_error=None, **kwargs):
+        """ Copy objects from path1 to path2
+        Parameters:
+        ----------
+        path1: a folder or file, or a list of folders or files
+            should we add the metadate with the path
+        path2: a folder or a file
+        recursive: bool
+            whether to copy files in a folder recursively.
+        on_error : "raise", "ignore"
+            If raise, any not-found exceptions will be raised; if ignore any
+            not-found exceptions will cause the path to be skipped; defaults to
+            raise unless recursive is true, where the default is ignore
+        """
+        if on_error is None and recursive:
+            on_error = "ignore"
+        elif on_error is None:
+            on_error = "raise"
+        if isinstance(path1, list):
+            for file_path in path1:
+                if on_error == 'raise':
+                    assert not file_path.endswith('/'), 'multiple file copy should takes files as input'
+                try:
+                    self.copy(file_path, path2)
+                except BaseException as e:
+                    if on_error == 'raise':
+                        raise e
+        else:
+            if path1.endswith('/') and path2.endswith('/'):
+                if on_error == 'raise':
+                    assert recursive, 'recursive should be True for folder copying'
+                if not self.exists(path2[:-1]):
+                    path1 = path1[:-1]
+                    path2 = path2[:-1]
+                else:
+                    try:
+                        self.rm(path2[:-1], recursive=True)
+                    except BaseException as e:
+                        if on_error == 'raise':
+                            raise e
+                    path1 = path1[:-1]
+                    path2 = path2[:-1]
+            elif path2.endswith('/'):
+                path2 = path2 + path1.split('/')[-1]
+            try:
+                self.dbx.files_copy(path1, path2)
+            except BaseException as e:
+                if on_error == 'raise':
+                    raise e
 
 class DropboxDriveFile(AbstractBufferedFile):
     """ fetch_all, fetch_range, and read method are based from the http implementation

test/test.py

@@ -1,11 +1,57 @@
 import unittest
-from unittest.mock import patch
-
+from unittest.mock import patch, call
 import dropbox
 
 import dropboxdrivefs
 from dropboxdrivefs import DropboxDriveFileSystem
 
+import os
+
+# Get the directory containing the current file
+dropboxdrivefs_dir = os.path.dirname(os.path.dirname( os.path.abspath(__file__)))
+
+
+@patch.object(dropbox.Dropbox, "files_copy")
+class TestDropboxCopy(unittest.TestCase):
+
+    def setUp(self) -> None:
+
+        self.fs = DropboxDriveFileSystem('123')
+
+    def test_copy_dir_to_dir_recursive(self, mock_files_copy):
+        self.fs.cp(os.path.join(dropboxdrivefs_dir, "example_file/"), "/target/newdir/", recursive=True)
+        self.assertEqual(mock_files_copy.call_count, 1)
+        mock_files_copy.assert_has_calls([call(os.path.join(dropboxdrivefs_dir, "example_file"), "/target/newdir")])
+    
+
+    def test_copy_dir_to_dir_non_recursive(self, mock_files_copy):
+        self.fs.cp(os.path.join(dropboxdrivefs_dir, "example_file/"), "/target/newdir/", recursive=True)
+        with self.assertRaises(AssertionError) as context:
+            self.fs.cp(os.path.join(dropboxdrivefs_dir, "example_file/"), "/target/newdir/", recursive=False)
+        self.assertEqual(str(context.exception), "recursive should be True for folder copying")
+    
+    def test_copy_file_to_dir(self, mock_files_copy): # 1a. File to existing directory
+        self.fs.cp(os.path.join(dropboxdrivefs_dir, "example_file/test_db/test1.txt"), '/target/')
+        self.assertEqual(mock_files_copy.call_count, 1)
+        mock_files_copy.assert_has_calls([call(os.path.join(dropboxdrivefs_dir, "example_file/test_db/test1.txt"), "/target/test1.txt")])
+    
+
+    def test_copy_file_to_file(self, mock_files_copy):
+        # 1d. File to File in new directory
+        self.fs.cp(os.path.join(dropboxdrivefs_dir, "example_file/test_db/test1.txt"), "/target/newdir/newfile.txt")
+        self.assertEqual(mock_files_copy.call_count, 1)
+        mock_files_copy.assert_has_calls([call(os.path.join(dropboxdrivefs_dir, "example_file/test_db/test1.txt"), "/target/newdir/newfile.txt")])
+    
+
+    def test_copy_multiple_files_to_dir(self, mock_files_copy):
+
+        self.fs.cp([os.path.join(dropboxdrivefs_dir, "example_file/test_db/test1.txt"), 
+                    os.path.join(dropboxdrivefs_dir, "example_file/test_db/test2.txt")], "/target/")
+        self.assertEqual(mock_files_copy.call_count, 2)
+        mock_files_copy.assert_has_calls([call(os.path.join(dropboxdrivefs_dir, "example_file/test_db/test1.txt"), "/target/test1.txt"), 
+                                          call(os.path.join(dropboxdrivefs_dir, "example_file/test_db/test2.txt"), "/target/test2.txt")], 
+                                          any_order=True)
+
 
 @patch.object(dropbox.Dropbox, "files_list_folder")
 class TestDropboxLs(unittest.TestCase):

test/test_copy.py

@@ -0,0 +1,69 @@
+"""
+Testing Copy Method of DropboxDriveFileSystem
+
+Test case reference from https://filesystem-spec.readthedocs.io/en/latest/copying.html
+"""
+import pytest
+from dropboxdrivefs import DropboxDriveFileSystem
+
+@pytest.fixture
+def dropbox_fs():
+    fs = DropboxDriveFileSystem('123')
+    if not fs.exists('/source'):
+        fs.mkdir('/source')
+    if not fs.exists('/source/subdir'):
+        fs.mkdir('/source/subdir')
+    if not fs.exists('/target'):
+        fs.mkdir('/target')
+    with fs.open('/source/subdir/subfile1', 'w') as f:
+        f.write('hello')
+    with fs.open('/source/file1', 'w') as f:
+        f.write('hello')
+    with fs.open('/source/file2', 'w') as f:
+        f.write('hello')
+    return fs
+
+def test_db_cp(dropbox_fs):
+    try:
+        # 1f. Directory to new directory
+        dropbox_fs.cp("/source/subdir/", "/target/newdir/", recursive=True)
+        assert dropbox_fs.exists('/target/newdir/subfile1')
+        dropbox_fs.rm('/target/newdir', recursive=True)
+        # 1e. Directory to existing directory
+        dropbox_fs.cp("/source/subdir/", "/target/", recursive=True)
+        assert dropbox_fs.exists('/target/subfile1')
+        dropbox_fs.rm('/target/subfile1')
+        # 1a. File to existing directory
+        dropbox_fs.cp('/source/subdir/subfile1', '/target/')
+        assert dropbox_fs.exists('/target/subfile1')
+        dropbox_fs.rm('/target/subfile1')
+        # 1b. File to new directory
+        dropbox_fs.cp('/source/subdir/subfile1', '/target/newdir/')
+        assert dropbox_fs.exists('/target/newdir/subfile1')
+        dropbox_fs.rm('/target/newdir', recursive=True)
+        # 1c. File to File in existing directory
+        dropbox_fs.cp("/source/subdir/subfile1", "/target/newfile")
+        assert dropbox_fs.exists('/target/newfile')
+        dropbox_fs.rm('/target/newfile')
+        # 1d. File to File in new directory
+        dropbox_fs.cp("/source/subdir/subfile1", "/target/newdir/newfile")
+        assert dropbox_fs.exists('/target/newdir/newfile')
+        dropbox_fs.rm('/target/newdir', recursive=True)
+        # 2a. List of Files to existing directory 
+        dropbox_fs.cp(["/source/file1", "/source/file2", "/source/subdir/subfile1"], "/target/")
+        assert dropbox_fs.exists('/target/file1')
+        assert dropbox_fs.exists('/target/file2')
+        assert dropbox_fs.exists('/target/subfile1')
+        dropbox_fs.rm('/target/file1')
+        dropbox_fs.rm('/target/file2')
+        dropbox_fs.rm('/target/subfile1') 
+        # 2b. List of Files to new directory
+        dropbox_fs.cp(["/source/file1", "/source/file2", "/source/subdir/subfile1"], "/target/newdir/")
+        assert dropbox_fs.exists('/target/newdir/file1')
+        assert dropbox_fs.exists('/target/newdir/file2')
+        assert dropbox_fs.exists('/target/newdir/subfile1')
+        dropbox_fs.rm('/target/newdir/', recursive=True)
+    finally:
+        # Final Delete
+        dropbox_fs.rm('/source/', recursive=True)
+        dropbox_fs.rm('/target/', recursive=True)