'Refactored by Sourcery'

nxdrive/engine/workers.py

@@ -193,12 +193,11 @@ def run(self) -> None:
         self.thread_id = current_thread_id()
 
         try:
-            try:
-                self._execute()
-            except ThreadInterrupt:
-                log.debug("Thread INTERRUPT")
-            except Exception:
-                log.exception("Thread EXCEPTION")
+            self._execute()
+        except ThreadInterrupt:
+            log.debug("Thread INTERRUPT")
+        except Exception:
+            log.exception("Thread EXCEPTION")
         finally:
             self.quit()
             self._running = False

tests/functional/conftest.py

@@ -848,7 +848,7 @@ def generate_report(self, exceptions: List[Exception]) -> None:
         self.manager_1.generate_report(path=self._get_report_file())
 
     def _set_read_permission(self, user, doc_path, grant):
-        input_obj = "doc:" + doc_path
+        input_obj = f"doc:{doc_path}"
         remote = self.root_remote
         if grant:
             remote.execute(
@@ -881,7 +881,7 @@ def set_readonly(self, user: str, doc_path: str, grant: bool = True):
         :param grant: Set RO if True else RW.
         """
         remote = self.root_remote
-        input_obj = "doc:" + doc_path
+        input_obj = f"doc:{doc_path}"
         if grant:
             remote.execute(
                 command="Document.SetACE",

tests/functional/test_direct_transfer.py

@@ -731,7 +731,7 @@ def checks(self, created):
         assert not self.engine_1.dao.get_errors(limit=0)
 
     def direct_transfer(self, folder, duplicate_behavior: str = "create") -> None:
-        paths = {path: size for path, size in get_tree_list(folder)}  # noqa
+        paths = dict(get_tree_list(folder))
         # paths = dict([(path, size) for path, size in get_tree_list(folder)])
         self.engine_1.direct_transfer(
             paths,
@@ -768,12 +768,10 @@ def test_sub_folders(self):
         # There is no upload, right now
         assert not list(self.engine_1.dao.get_dt_uploads())
 
-        created = []
-
         root_folder = self.tmpdir / str(uuid4())[:6]
         root_folder.mkdir()
 
-        created.append(root_folder.as_posix())
+        created = [root_folder.as_posix()]
         for _ in range(3):
             sub_folder = root_folder / f"folder_{str(uuid4())[:4]}"
             sub_folder.mkdir()
@@ -795,13 +793,10 @@ def test_same_name_folders(self):
         # There is no upload, right now
         assert not list(self.engine_1.dao.get_dt_uploads())
 
-        created = []
-
         root_folder = self.tmpdir / str(uuid4())[:6]
         root_folder.mkdir()
 
-        created.append(root_folder.as_posix())
-
+        created = [root_folder.as_posix()]
         folder_a = root_folder / "folder_a"
         folder_a.mkdir()
         created.append(folder_a.as_posix())
@@ -846,11 +841,9 @@ def test_sessions(self):
         )
 
         for x in range(4):
-            created = []
             root_folder = self.tmpdir / str(uuid4())[:6]
             root_folder.mkdir()
-            created.append(root_folder)
-
+            created = [root_folder]
             sub_file = root_folder / f"file_{str(uuid4())[:4]}"
             sub_file.write_text("test", encoding="utf-8")
             created.append(sub_file)
@@ -1115,12 +1108,10 @@ def test_sub_files(self):
         # There is no upload, right now
         assert not list(self.engine_1.dao.get_dt_uploads())
 
-        created = []
-
         root_folder = self.tmpdir / str(uuid4())[:6]
         root_folder.mkdir()
 
-        created.append(root_folder.as_posix())
+        created = [root_folder.as_posix()]
         for _ in range(5):
             sub_file = root_folder / f"file_{str(uuid4())[:4]}"
             sub_file.write_text("test", encoding="utf-8")

tests/functional/test_encoding.py

@@ -100,7 +100,7 @@ def test_name_normalization(self):
         local.make_file("/", filename)
         self.wait_sync(wait_for_async=True)
 
-        assert remote.get_info("/" + filename).name == filename
+        assert remote.get_info(f"/{filename}").name == filename
 
     @not_mac(reason="Normalization does not work on macOS")
     def test_fileinfo_normalization(self):

tests/functional/test_local_deletion.py

@@ -19,21 +19,21 @@ def test_untrash_file(self):
 
         local.make_file("/", file1, content=b"This is a content")
         self.wait_sync()
-        assert remote.exists("/" + file1)
+        assert remote.exists(f"/{file1}")
 
         old_info = remote.get_info(f"/{file1}")
         abs_path = local.abspath(f"/{file1}")
 
         # Pretend we had trash the file
         shutil.move(abs_path, self.local_test_folder_1 / file1)
         self.wait_sync(wait_for_async=True)
-        assert not remote.exists("/" + file1)
-        assert not local.exists("/" + file1)
+        assert not remote.exists(f"/{file1}")
+        assert not local.exists(f"/{file1}")
         # See if it untrash or recreate
         shutil.move(self.local_test_folder_1 / file1, local.abspath("/"))
         self.wait_sync(wait_for_async=True)
         assert remote.exists(old_info.uid)
-        assert local.exists("/" + file1)
+        assert local.exists(f"/{file1}")
 
     def test_untrash_file_with_rename(self):
         local = self.local_1
@@ -50,8 +50,8 @@ def test_untrash_file_with_rename(self):
         # Pretend we had trash the file
         shutil.move(abs_path, self.local_test_folder_1 / file2)
         self.wait_sync(wait_for_async=True)
-        assert not remote.exists("/" + file1)
-        assert not local.exists("/" + file1)
+        assert not remote.exists(f"/{file1}")
+        assert not local.exists(f"/{file1}")
         (self.local_test_folder_1 / file2).write_bytes(b"New content")
         if WINDOWS:
             # Python API overwrite the tag by default
@@ -62,9 +62,9 @@ def test_untrash_file_with_rename(self):
         shutil.move(self.local_test_folder_1 / file2, local.abspath("/"))
         self.wait_sync(wait_for_async=True)
         assert remote.exists(old_info.uid)
-        assert local.exists("/" + file2)
-        assert not local.exists("/" + file1)
-        assert local.get_content("/" + file2) == b"New content"
+        assert local.exists(f"/{file2}")
+        assert not local.exists(f"/{file1}")
+        assert local.get_content(f"/{file2}") == b"New content"
 
     def test_move_untrash_file_on_parent(self):
         local = self.local_1
@@ -165,7 +165,7 @@ def test_move_untrash_file_on_parent_with_no_rights_on_destination(self):
 
         # Remove rights
         folder_path = f"{self.ws.path}/ToCopy"
-        input_obj = "doc:" + folder_path
+        input_obj = f"doc:{folder_path}"
         self.root_remote.execute(
             command="Document.SetACE",
             input_obj=input_obj,

tests/functional/test_readonly.py

@@ -245,7 +245,6 @@ def test_folder_add(self):
         On Windows, those folders are ignored.
         """
 
-        remote = self.remote_document_client_1
         folder = self.local_nxdrive_folder_1 / "foo" / "test.txt"
 
         if not WINDOWS:
@@ -262,6 +261,7 @@ def test_folder_add(self):
             ]
             assert sorted(ignored) == ["foo", "foo/test.txt"]
 
+            remote = self.remote_document_client_1
             # Check there is nothing uploaded to the server
             assert not remote.get_children_info("/")
 

tests/functional/test_remote_client.py

@@ -507,7 +507,7 @@ def mocked_move(*args, **kwargs):
 
     dummy_file_path = env.WS_DIR
 
-    dummy_file_path = str(env.WS_DIR) + "#"
+    dummy_file_path = f"{str(env.WS_DIR)}#"
 
     with patch.object(
         remote,

tests/functional/test_remote_client_old.py

@@ -67,7 +67,7 @@ def test_get_fs_info(self):
         assert info.download_url is None
 
         # Check non existing file info
-        fs_item_id = FS_ITEM_ID_PREFIX + "fakeId"
+        fs_item_id = f"{FS_ITEM_ID_PREFIX}fakeId"
         with pytest.raises(NotFound):
             remote.get_fs_info(fs_item_id)
 
@@ -185,10 +185,10 @@ def test_scroll_descendants(self):
             )
             assert isinstance(scroll_res, dict)
             scroll_id = scroll_res["scroll_id"]
-            partial_descendants = scroll_res["descendants"]
-            if not partial_descendants:
+            if partial_descendants := scroll_res["descendants"]:
+                descendants.extend(partial_descendants)
+            else:
                 break
-            descendants.extend(partial_descendants)
         descendants = sorted(descendants, key=operator.attrgetter("name"))
         assert len(descendants) == 4
 
@@ -312,7 +312,7 @@ def test_exists(self):
         assert remote.fs_exists(fs_item_id)
 
         # Check non existing file system item (non existing document)
-        fs_item_id = FS_ITEM_ID_PREFIX + "fakeId"
+        fs_item_id = f"{FS_ITEM_ID_PREFIX}fakeId"
         assert not remote.fs_exists(fs_item_id)
 
         # Check non existing file system item (document without content)
@@ -353,7 +353,7 @@ def test_get_fs_item(self):
         assert fs_item["folder"]
 
         # Check non existing file system item
-        fs_item_id = FS_ITEM_ID_PREFIX + "fakeId"
+        fs_item_id = f"{FS_ITEM_ID_PREFIX}fakeId"
         assert remote.get_fs_item(fs_item_id) is None
 
     def test_streaming_upload(self):

tests/functional/test_remote_move_and_rename.py

@@ -757,7 +757,7 @@ def test_remote_create_and_move(self):
         # Create a folder with some stuff inside, and sync
         a1 = remote.make_folder("/", "a1")
         for idx in range(5):
-            fname = "file-{}.txt".format(idx)
+            fname = f"file-{idx}.txt"
             remote.make_file(a1, fname, content=b"Content of " + fname.encode("utf-8"))
         engine.start()
         self.wait_sync(wait_for_async=True)

tests/functional/test_shared_folders.py

@@ -29,7 +29,7 @@ def test_move_sync_root_child_to_user_workspace(self):
             parent_uid = remote_1.make_folder(uid, "Parent")
 
             # As user1 grant Everything permission to user2 on parent folder
-            input_obj = "doc:" + parent_uid
+            input_obj = f"doc:{parent_uid}"
             self.root_remote.execute(
                 command="Document.SetACE",
                 input_obj=input_obj,

tests/functional/test_synchronization.py

@@ -768,8 +768,8 @@ def _raise_for_second_file_only(*args, **kwargs):
         with patch.object(self.engine_1, "remote", new=bad_remote):
             # Create 2 files locally
             base = "A" * 40
-            file1 = base + "1.txt"
-            file2 = base + "2.txt"
+            file1 = f"{base}1.txt"
+            file2 = f"{base}2.txt"
             local.make_file("/", file1, content=b"foo")
             local.make_file("/", file2, content=b"bar")
 

tests/functional/test_watchers.py

@@ -102,9 +102,7 @@ def test_reconcile_scan(self):
             assert item.pair_state == "synchronized"
 
     def test_remote_scan(self):
-        total = len(self.make_server_tree())
-        # Add the workspace folder + the root
-        total += 2
+        total = len(self.make_server_tree()) + 2
         # Wait for ES indexing
         self.wait()
         self.queue_manager_1.suspend()