Reworking and refactoring

tests/core/test_provider.py

@@ -276,6 +276,15 @@ async def test_passes_on_rename(self, provider1):
             conflict='replace',
         )
 
+    @pytest.mark.asyncio
+    async def test_copy_will_self_overwrite(self, provider1):
+        src_path = await provider1.validate_path('/source/path')
+        dest_path = await provider1.validate_path('/destination/')
+        provider1.will_self_overwrite = utils.MockCoroutine()
+
+        with pytest.raises(exceptions.OverwriteSelfError):
+            await provider1.copy(provider1, src_path, dest_path)
+
     @pytest.mark.asyncio
     async def test_checks_can_intra_copy(self, provider1):
         provider1.can_intra_copy = mock.Mock(return_value=False)
@@ -393,6 +402,15 @@ async def test_passes_on_rename(self, provider1):
             conflict='replace',
         )
 
+    @pytest.mark.asyncio
+    async def test_move_will_self_overwrite(self, provider1):
+        src_path = await provider1.validate_path('/source/path')
+        dest_path = await provider1.validate_path('/destination/')
+        provider1.will_self_overwrite = utils.MockCoroutine()
+
+        with pytest.raises(exceptions.OverwriteSelfError):
+            await provider1.move(provider1, src_path, dest_path)
+
     @pytest.mark.asyncio
     async def test_checks_can_intra_move(self, provider1):
         provider1.can_intra_move = mock.Mock(return_value=False)

tests/providers/box/test_provider.py

@@ -165,7 +165,9 @@ async def test_validate_path(self, provider, root_provider_fixtures):
         provider.folder = '0'
         folder_id = '0'
 
-        good_url = provider.build_url('folders', folder_id, 'items', fields='id,name,type', limit=1000)
+        good_url = provider.build_url('folders', folder_id, 'items',
+                                      fields='id,name,type', limit=1000)
+
         aiohttpretty.register_json_uri('GET', good_url,
                                        body=root_provider_fixtures['revalidate_metadata'],
                                        status=200)
@@ -295,7 +297,7 @@ async def test_upload_checksum_mismatch(self, provider, root_provider_fixtures,
         aiohttpretty.register_json_uri('POST', upload_url, status=201,
                                        body=root_provider_fixtures['checksum_mismatch_metadata'])
 
-        with pytest.raises(exceptions.UploadChecksumMismatchError) as exc:
+        with pytest.raises(exceptions.UploadChecksumMismatchError):
             await provider.upload(file_stream, path)
 
         assert aiohttpretty.has_call(method='POST', uri=upload_url)
@@ -360,8 +362,11 @@ async def test_delete_root(self, provider, root_provider_fixtures):
         url = provider.build_url('folders', root_path.identifier, 'items',
                                  fields='id,name,size,modified_at,etag,total_count',
                                  offset=(0), limit=1000)
-        aiohttpretty.register_json_uri('GET', url,
-                                       body=root_provider_fixtures['one_entry_folder_list_metadata'])
+        aiohttpretty.register_json_uri(
+            'GET',
+            url,
+            body=root_provider_fixtures['one_entry_folder_list_metadata']
+        )
 
         url = provider.build_url('files', item['id'], fields='id,name,path_collection')
         delete_url = provider.build_url('files', path.identifier)
@@ -568,6 +573,7 @@ async def test_get_revisions_free_account(self, provider, root_provider_fixtures
 
 
 class TestIntraCopy:
+
     @pytest.mark.asyncio
     @pytest.mark.aiohttpretty
     async def test_intra_copy_file(self, provider, root_provider_fixtures):
@@ -583,16 +589,6 @@ async def test_intra_copy_file(self, provider, root_provider_fixtures):
 
         assert result == expected
 
-    @pytest.mark.asyncio
-    @pytest.mark.aiohttpretty
-    async def test_intra_copy_overwrite_error(self, provider, root_provider_fixtures):
-        item = root_provider_fixtures['file_metadata']['entries'][0]
-        src_path = WaterButlerPath('/name.txt', _ids=(provider, item['id']))
-        with pytest.raises(exceptions.IntraCopyError) as e:
-            await provider.intra_copy(provider, src_path, src_path)
-
-        assert e.value.code == 409
-
     @pytest.mark.asyncio
     @pytest.mark.aiohttpretty
     async def test_intra_copy_file_replace(self, provider, root_provider_fixtures):
@@ -630,7 +626,9 @@ async def test_intra_copy_folder(self, provider, intra_fixtures, root_provider_f
         expected_folder = BoxFolderMetadata(item, dest_path)
         expected_folder._children = []
         for child_item in list_metadata['entries']:
-            child_path = dest_path.child(child_item['name'], folder=(child_item['type'] == 'folder'))
+            child_path = dest_path.child(child_item['name'],
+                                         folder=(child_item['type'] == 'folder'))
+
             serialized_child = provider._serialize_item(child_item, child_path)
             expected_folder._children.append(serialized_child)
         expected = (expected_folder, True)
@@ -641,7 +639,10 @@ async def test_intra_copy_folder(self, provider, intra_fixtures, root_provider_f
 
     @pytest.mark.asyncio
     @pytest.mark.aiohttpretty
-    async def test_intra_copy_folder_replace(self, provider, intra_fixtures, root_provider_fixtures):
+    async def test_intra_copy_folder_replace(self,
+                                             provider,
+                                             intra_fixtures,
+                                             root_provider_fixtures):
         item = intra_fixtures['intra_folder_metadata']
         list_metadata = root_provider_fixtures['folder_list_metadata']
 
@@ -661,7 +662,9 @@ async def test_intra_copy_folder_replace(self, provider, intra_fixtures, root_pr
         expected_folder = BoxFolderMetadata(item, dest_path)
         expected_folder._children = []
         for child_item in list_metadata['entries']:
-            child_path = dest_path.child(child_item['name'], folder=(child_item['type'] == 'folder'))
+            child_path = dest_path.child(child_item['name'],
+                                         folder=(child_item['type'] == 'folder'))
+
             serialized_child = provider._serialize_item(child_item, child_path)
             expected_folder._children.append(serialized_child)
         expected = (expected_folder, False)
@@ -689,22 +692,13 @@ async def test_intra_move_file(self, provider, root_provider_fixtures):
 
         assert result == expected
 
-    @pytest.mark.asyncio
-    @pytest.mark.aiohttpretty
-    async def test_intra_move_overwrite_error(self, provider, root_provider_fixtures):
-        item = root_provider_fixtures['file_metadata']['entries'][0]
-        src_path = WaterButlerPath('/name.txt', _ids=(provider, item['id']))
-        with pytest.raises(exceptions.IntraCopyError) as e:
-            await provider.intra_move(provider, src_path, src_path)
-
-        assert e.value.code == 409
-
     @pytest.mark.asyncio
     @pytest.mark.aiohttpretty
     async def test_intra_move_file_replace(self, provider, root_provider_fixtures):
         item = root_provider_fixtures['file_metadata']['entries'][0]
         src_path = WaterButlerPath('/name.txt', _ids=(provider, item['id']))
-        dest_path = WaterButlerPath('/charmander/name.txt', _ids=(provider, item['id'], 'YgzZejrj834j'))
+        dest_path = WaterButlerPath('/charmander/name.txt',
+                                    _ids=(provider, item['id'], 'YgzZejrj834j'))
 
         file_url = provider.build_url('files', src_path.identifier)
         delete_url = provider.build_url('files', dest_path.identifier)
@@ -736,7 +730,10 @@ async def test_intra_move_folder(self, provider, intra_fixtures, root_provider_f
         expected_folder = BoxFolderMetadata(item, dest_path)
         expected_folder._children = []
         for child_item in list_metadata['entries']:
-            child_path = dest_path.child(child_item['name'], folder=(child_item['type'] == 'folder'))
+            child_path = dest_path.child(
+                child_item['name'],
+                folder=(child_item['type'] == 'folder')
+            )
             serialized_child = provider._serialize_item(child_item, child_path)
             expected_folder._children.append(serialized_child)
         expected = (expected_folder, True)
@@ -747,7 +744,10 @@ async def test_intra_move_folder(self, provider, intra_fixtures, root_provider_f
 
     @pytest.mark.asyncio
     @pytest.mark.aiohttpretty
-    async def test_intra_move_folder_replace(self, provider, intra_fixtures, root_provider_fixtures):
+    async def test_intra_move_folder_replace(self,
+                                             provider,
+                                             intra_fixtures,
+                                             root_provider_fixtures):
         item = intra_fixtures['intra_folder_metadata']
         list_metadata = root_provider_fixtures['folder_list_metadata']
 
@@ -767,7 +767,9 @@ async def test_intra_move_folder_replace(self, provider, intra_fixtures, root_pr
         expected_folder = BoxFolderMetadata(item, dest_path)
         expected_folder._children = []
         for child_item in list_metadata['entries']:
-            child_path = dest_path.child(child_item['name'], folder=(child_item['type'] == 'folder'))
+            child_path = dest_path.child(child_item['name'],
+                                         folder=(child_item['type'] == 'folder'))
+
             serialized_child = provider._serialize_item(child_item, child_path)
             expected_folder._children.append(serialized_child)
         expected = (expected_folder, False)
@@ -851,25 +853,29 @@ async def test_returns_metadata(self, provider, root_provider_fixtures):
 
 class TestOperations:
 
-    @pytest.mark.asyncio
-    @pytest.mark.aiohttpretty
-    async def test_can_duplicate_names(self, provider):
+    def test_will_self_overwrite(self, provider, other_provider):
+        src_path = WaterButlerPath('/50 shades of nope.txt',
+                                    _ids=(provider.folder, '12231'))
+        dest_path = WaterButlerPath('/50 shades of nope2223.txt',
+                                    _ids=(provider.folder, '2342sdfsd'))
+
+        result = provider.will_self_overwrite(other_provider, src_path, dest_path)
+        assert result is False
+
+        result = provider.will_self_overwrite(other_provider, src_path, src_path)
+        assert result is True
+
+    def test_can_duplicate_names(self, provider):
         assert provider.can_duplicate_names() is False
 
-    @pytest.mark.asyncio
-    @pytest.mark.aiohttpretty
-    async def test_shares_storage_root(self, provider, other_provider):
+    def test_shares_storage_root(self, provider, other_provider):
         assert provider.shares_storage_root(other_provider) is False
         assert provider.shares_storage_root(provider) is True
 
-    @pytest.mark.asyncio
-    @pytest.mark.aiohttpretty
-    async def test_can_intra_move(self, provider, other_provider):
+    def test_can_intra_move(self, provider, other_provider):
         assert provider.can_intra_move(other_provider) is False
         assert provider.can_intra_move(provider) is True
 
-    @pytest.mark.asyncio
-    @pytest.mark.aiohttpretty
-    async def test_can_intra_copy(self, provider, other_provider):
+    def test_can_intra_copy(self, provider, other_provider):
         assert provider.can_intra_copy(other_provider) is False
         assert provider.can_intra_copy(provider) is True

tests/providers/dropbox/test_provider.py

@@ -290,8 +290,12 @@ async def test_folder_with_subdirectory_metadata(self, provider, root_provider_f
         path = await provider.validate_path('/')
         url = provider.build_url('files', 'list_folder')
         data = {'path': path.full_path}
-        aiohttpretty.register_json_uri('POST', url, data=data,
-                                       body=root_provider_fixtures['folder_with_subdirectory_metadata'])
+        aiohttpretty.register_json_uri(
+            'POST',
+            url,
+            data=data,
+            body=root_provider_fixtures['folder_with_subdirectory_metadata']
+        )
         result = await provider.metadata(path)
 
         assert isinstance(result, list)
@@ -308,8 +312,12 @@ async def test_folder_with_hasmore_metadata(self, provider, root_provider_fixtur
         data = {'path': path.full_path}
         aiohttpretty.register_json_uri('POST', url, data=data,
                                        body=root_provider_fixtures['folder_with_hasmore_metadata'])
-        aiohttpretty.register_json_uri('POST', url + '/continue', data=data,
-                                       body=root_provider_fixtures['folder_with_subdirectory_metadata'])
+        aiohttpretty.register_json_uri(
+            'POST',
+            url + '/continue',
+            data=data,
+            body=root_provider_fixtures['folder_with_subdirectory_metadata']
+        )
 
         result = await provider.metadata(path)
 
@@ -544,7 +552,9 @@ async def test_intra_copy_replace_file(self, provider, root_provider_fixtures, e
                 {
                     'headers': {'Content-Type': 'application/json'},
                     'data': data,
-                    'body': json.dumps(error_fixtures['rename_conflict_folder_metadata']).encode('utf-8'),
+                    'body': json.dumps(
+                        error_fixtures['rename_conflict_folder_metadata']
+                    ).encode('utf-8'),
                     'status': 409
                 },
                 {
@@ -574,8 +584,12 @@ async def test_intra_copy_file_different_provider(self, provider, other_provider
 
         url1 = provider.build_url('files', 'copy_reference', 'save')
         data1 = {'copy_reference': 'test', 'path': dest_path.full_path.rstrip('/')}
-        aiohttpretty.register_json_uri('POST', url1, data=data1,
-                                       body=intra_copy_fixtures['intra_copy_other_provider_file_metadata'])
+        aiohttpretty.register_json_uri(
+            'POST',
+            url1,
+            data=data1,
+            body=intra_copy_fixtures['intra_copy_other_provider_file_metadata']
+        )
 
         result = await provider.intra_copy(other_provider, src_path, dest_path)
         expected = (DropboxFileMetadata(
@@ -648,7 +662,9 @@ async def test_intra_move_replace_file(self, provider, root_provider_fixtures, e
                 {
                     'headers': {'Content-Type': 'application/json'},
                     'data': data,
-                    'body': json.dumps(error_fixtures['rename_conflict_file_metadata']).encode('utf-8'),
+                    'body': json.dumps(
+                        error_fixtures['rename_conflict_file_metadata']
+                    ).encode('utf-8'),
                     'status': 409
                 },
                 {
@@ -689,7 +705,9 @@ async def test_intra_move_replace_folder(self, provider, root_provider_fixtures,
                 {
                     'headers': {'Content-Type': 'application/json'},
                     'data': data,
-                    'body': json.dumps(error_fixtures['rename_conflict_folder_metadata']).encode('utf-8'),
+                    'body': json.dumps(
+                        error_fixtures['rename_conflict_folder_metadata']
+                    ).encode('utf-8'),
                     'status': 409
                 },
                 {
@@ -719,23 +737,20 @@ async def test_intra_move_casing_change(self, provider):
 
         assert e.value.code == 400
 
-    @pytest.mark.asyncio
-    @pytest.mark.aiohttpretty
-    async def test_intra_overwrite_error(self, provider):
-        src_path = WaterButlerPath('/pfile.txt', prepend=provider.folder)
 
-        with pytest.raises(exceptions.IntraCopyError) as e:
-            await provider.intra_move(provider, src_path, src_path)
-
-        assert e.value.code == 409
-
-        with pytest.raises(exceptions.IntraCopyError) as e:
-            await provider.intra_copy(provider, src_path, src_path)
+class TestOperations:
 
-        assert e.value.code == 409
+    def test_will_self_overwrite(self, provider, other_provider):
+        src_path = WaterButlerPath('/50 shades of nope.txt',
+                                   _ids=(provider.folder, '12231'))
+        dest_path = WaterButlerPath('/50 shades of nope2223.txt',
+                                    _ids=(provider.folder, '2342sdfsd'))
 
+        result = provider.will_self_overwrite(other_provider, src_path, dest_path)
+        assert result is False
 
-class TestOperations:
+        result = provider.will_self_overwrite(other_provider, src_path, src_path)
+        assert result is True
 
     def test_can_intra_copy(self, provider):
         assert provider.can_intra_copy(provider)
@@ -747,7 +762,7 @@ def test_can_intra_move(self, provider):
         assert provider.can_intra_move(provider)
 
     def test_cannot_intra_move_other(self, provider, other_provider):
-        assert provider.can_intra_move(other_provider) == False
+        assert provider.can_intra_move(other_provider) is False
 
     def test_conflict_error_handler_not_found(self, provider, error_fixtures):
         error_path = '/Photos/folder/file'