Finish work on orphan overwrites

Author: NyanHelsing

tests/core/test_provider.py

@@ -40,32 +40,44 @@ def test_can_intra_move(self, provider2):
 
     @pytest.mark.asyncio
     async def test_will_orphan_dest_is_file(self, provider1):
-        src_path =  WaterButlerPath('/folder1/folder1/',
-                                    _ids=['root', 'folder', 'Folder'],
-                                    folder=True)
-        dest_path = WaterButlerPath('/folder1/',
-                                    _ids=['root','folder'],
-                                    folder=False)
+        src_path = WaterButlerPath(
+            '/folder1/folder1/',
+            _ids=['root', 'folder', 'Folder'],
+            folder=True
+        )
+        dest_path = WaterButlerPath(
+            '/folder1/',
+            _ids=['root','folder'],
+            folder=False
+        )
         assert provider1.replace_will_orphan(src_path, dest_path) == False
 
     @pytest.mark.asyncio
     async def test_will_orphan_dest_different_names(self, provider1):
-        src_path =  WaterButlerPath('/folder1/folder1/',
-                                    _ids=['root', 'folder', 'Folder'],
-                                    folder=True)
-        dest_path = WaterButlerPath('/folder2/',
-                                    _ids=['root','folder'],
-                                    folder=True)
+        src_path =  WaterButlerPath(
+            '/folder1/folder1/',
+            _ids=['root', 'folder', 'Folder'],
+            folder=True
+        )
+        dest_path = WaterButlerPath(
+            '/folder2/',
+            _ids=['root','folder'],
+            folder=True
+        )
         assert provider1.replace_will_orphan(src_path, dest_path) == False
 
     @pytest.mark.asyncio
     async def test_will_orphan_dest_different_branch(self, provider1):
-        src_path =  WaterButlerPath('/other_folder/folder1/',
-                                    _ids=['root', 'other_folder', 'Folder'],
-                                    folder=True)
-        dest_path = WaterButlerPath('/folder1/',
-                                    _ids=['root','folder'],
-                                    folder=True)
+        src_path =  WaterButlerPath(
+            '/other_folder/folder1/',
+            _ids=['root', 'other_folder', 'Folder'],
+            folder=True
+        )
+        dest_path = WaterButlerPath(
+            '/folder1/',
+            _ids=['root','folder'],
+            folder=True
+        )
         assert provider1.replace_will_orphan(src_path, dest_path) == False
 
     @pytest.mark.asyncio
@@ -482,6 +494,7 @@ async def test_intra_move_folder_orphan(self, provider1):
 
         with pytest.raises(exceptions.OrphanSelfError) as exc:
             await provider1.move(provider1, src_path, dest_path)
+
         assert exc.value.code == 400
         assert exc.typename == 'OrphanSelfError'
 

tests/providers/box/test_provider.py

@@ -164,13 +164,17 @@ async def test_validate_path_bad_path(self, provider):
     async def test_validate_path(self, provider, root_provider_fixtures):
         provider.folder = '0'
         folder_id = '0'
+        response_body = root_provider_fixtures['revalidate_metadata']
+
+        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)
+        aiohttpretty.register_json_uri('GET', good_url, body=response_body, status=200)
 
         result = await provider.validate_path('/bulbasaur')
         assert result == WaterButlerPath('/bulbasaur', folder=False)
@@ -377,22 +381,23 @@ async def test_delete_root_no_confirm(self, provider, root_provider_fixtures):
     @pytest.mark.aiohttpretty
     async def test_delete_root(self, provider, root_provider_fixtures):
         item = root_provider_fixtures['file_metadata']['entries'][0]
+        list_response_body = root_provider_fixtures['one_entry_folder_list_metadata']
         path = WaterButlerPath('/newfile', _ids=(provider.folder, item['id']))
         root_path = WaterButlerPath('/', _ids=('0'))
 
-        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']
+        list_url = provider.build_url(
+            'folders',
+            root_path.identifier,
+            'items',
+            fields='id,name,size,modified_at,etag,total_count',
+            offset=(0),
+            limit=1000
         )
-
         url = provider.build_url('files', item['id'], fields='id,name,path_collection')
         delete_url = provider.build_url('files', path.identifier)
-        aiohttpretty.register_json_uri('get', url,
-                                       body=root_provider_fixtures['file_metadata']['entries'][0])
+
+        aiohttpretty.register_json_uri('GET', list_url, body=list_response_body)
+        aiohttpretty.register_json_uri('GET', url, body=item)
         aiohttpretty.register_json_uri('DELETE', delete_url, status=204)
 
         await provider.delete(root_path, 1)
@@ -647,9 +652,10 @@ 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)
@@ -660,10 +666,12 @@ 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']
 
@@ -672,9 +680,14 @@ async def test_intra_copy_folder_replace(self,
 
         file_url = provider.build_url('folders', src_path.identifier, 'copy')
         delete_url = provider.build_url('folders', dest_path.identifier, recursive=True)
-        list_url = provider.build_url('folders', item['id'], 'items',
-                                      fields='id,name,size,modified_at,etag,total_count',
-                                      offset=0, limit=1000)
+        list_url = provider.build_url(
+            'folders',
+            item['id'],
+            'items',
+            fields='id,name,size,modified_at,etag,total_count',
+            offset=0,
+            limit=1000
+        )
 
         aiohttpretty.register_json_uri('GET', list_url, body=list_metadata)
         aiohttpretty.register_uri('DELETE', delete_url, status=204)
@@ -683,9 +696,10 @@ async def test_intra_copy_folder_replace(self,
         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)
@@ -718,8 +732,10 @@ async def test_intra_move_file(self, provider, root_provider_fixtures):
     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)
@@ -765,21 +781,31 @@ 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']
 
         src_path = WaterButlerPath('/name/', _ids=(provider, item['id']))
-        dest_path = WaterButlerPath('/charmander/name/', _ids=(provider, item['id'], '7759994812'))
+        dest_path = WaterButlerPath(
+            '/charmander/name/',
+            _ids=(provider, item['id'], '7759994812')
+        )
 
         file_url = provider.build_url('folders', src_path.identifier)
         delete_url = provider.build_url('folders', dest_path.identifier, recursive=True)
-        list_url = provider.build_url('folders', item['id'], 'items',
-                                      fields='id,name,size,modified_at,etag,total_count',
-                                      offset=0, limit=1000)
+        list_url = provider.build_url(
+            'folders',
+            item['id'],
+            'items',
+            fields='id,name,size,modified_at,etag,total_count',
+            offset=0,
+            limit=1000
+        )
 
         aiohttpretty.register_json_uri('PUT', file_url, body=item)
         aiohttpretty.register_uri('DELETE', delete_url, status=204)
@@ -788,9 +814,10 @@ async def test_intra_move_folder_replace(self,
         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)
@@ -875,10 +902,14 @@ async def test_returns_metadata(self, provider, root_provider_fixtures):
 class TestOperations:
 
     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'))
+        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

tests/providers/dropbox/test_provider.py

@@ -659,45 +659,45 @@ async def test_intra_copy_replace_file(
             provider_fixtures,
             error_fixtures
     ):
-        url = provider.build_url('files', 'delete_v2')
-        path = await provider.validate_path('/The past')
-        data = {'path': path.full_path}
-        aiohttpretty.register_json_uri('POST', url, data=data, status=HTTPStatus.OK)
+        delete_path = await provider.validate_path('/The past')
+        delete_data = {'path': path.full_path}
+        delete_url = provider.build_url('files', 'delete_v2')
+        aiohttpretty.register_json_uri(
+            'POST',
+            delete_url,
+            data=delete_data,
+            status=HTTPStatus.OK
+        )
 
         src_path = WaterButlerPath('/pfile', prepend=provider.folder)
         dest_path = WaterButlerPath('/pfile_renamed', prepend=provider.folder)
-
-        url = provider.build_url('files', 'copy_v2')
-        data = {
+        rename_conflict_folder_metadata = error_fixtures['rename_conflict_folder_metadata']
+        imc_metadata = provider_fixtures['intra_move_copy_file_metadata_v2']
+        copy_data = {
             'from_path': src_path.full_path.rstrip('/'),
             'to_path': dest_path.full_path.rstrip('/')
         }
-        aiohttpretty.register_json_uri('POST', url, **{
-            "responses": [
-                {
-                    'headers': {'Content-Type': 'application/json'},
-                    'data': data,
-                    'body': json.dumps(error_fixtures['rename_conflict_folder_metadata']).encode('utf-8'),
-                    'status': HTTPStatus.CONFLICT
-                },
-                {
-                    'headers': {'Content-Type': 'application/json'},
-                    'data': data,
-                    'body': json.dumps(provider_fixtures['intra_move_copy_file_metadata_v2']).encode('utf-8')
-                },
-            ]
-        })
+        copy_responses = [
+            {
+                'headers': {'Content-Type': 'application/json'},
+                'data': copy_data,
+                'body': json.dumps(rename_conflict_folder_metadata).encode('utf-8'),
+                'status': HTTPStatus.CONFLICT
+            },
+            {
+                'headers': {'Content-Type': 'application/json'},
+                'data': copy_data,
+                'body': json.dumps(imc_metadata).encode('utf-8')
+            },
+        ]
+        copy_url = provider.build_url('files', 'copy_v2')
+        aiohttpretty.register_json_uri('POST', copy_url, responses=copy_responses)
+
+        expected = (DropboxFileMetadata(imc_metadata['metadata'], provider.folder), False)
 
         result = await provider.intra_copy(provider, src_path, dest_path)
-        expected = (
-            DropboxFileMetadata(
-                provider_fixtures['intra_move_copy_file_metadata_v2']['metadata'],
-                provider.folder
-            ),
-            False
-        )
 
-        assert expected == result
+        assert result == expected
 
     @pytest.mark.asyncio
     @pytest.mark.aiohttpretty
@@ -919,17 +919,27 @@ async def test_intra_move_casing_change(self, provider):
 class TestOperations:
 
     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
+        src_path = WaterButlerPath(
+            '/50 shades of nope.txt',
+            _ids=(provider.folder, '12231')
+        )
 
         result = provider.will_self_overwrite(other_provider, src_path, src_path)
         assert result is True
 
+    def test_wont_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
+
     def test_can_intra_copy(self, provider):
         assert provider.can_intra_copy(provider)