NXDRIVE-2844: Nuxeo Drive transfer failed after removing hardware (Sourcery refactored)

docs/changes/5.3.2.md

@@ -6,6 +6,7 @@ Release date: `2023-xx-xx`
 
 - [NXDRIVE-2812](https://jira.nuxeo.com/browse/NXDRIVE-2812): Update Github Action for set-output and save-state
 - [NXDRIVE-2838](https://jira.nuxeo.com/browse/NXDRIVE-2838): Fix Release build
+- [NXDRIVE-2844](https://jira.nuxeo.com/browse/NXDRIVE-2844): Nuxeo Drive transfer failed after removing hardware
 - [NXDRIVE-2363](https://jira.nuxeo.com/browse/NXDRIVE-2363): MacOS Application auto-restart does not work since 4.3.0
 
 ### Direct Edit

nxdrive/engine/processor.py

@@ -291,8 +291,7 @@
         try:
             return self.dao.acquire_state(self.thread_id, item.id)
         except sqlite3.OperationalError:
-            state = self.dao.get_state_from_id(item.id)
-            if state:
+            if state := self.dao.get_state_from_id(item.id):
                 if (
                     WINDOWS
                     and state.pair_state == "locally_moved"
@@ -460,9 +459,7 @@
                     log.info("The document does not exist anymore locally")
                     self.dao.remove_state(doc_pair)
                 elif error in LONG_FILE_ERRORS:
-                    self.dao.remove_filter(
-                        doc_pair.remote_parent_path + "/" + doc_pair.remote_ref
-                    )
+                    self.dao.remove_filter(f"{doc_pair.remote_parent_path}/{doc_pair.remote_ref}")
                     self.engine.longPathError.emit(doc_pair)
                 elif hasattr(exc, "trash_issue"):
                     """
@@ -475,18 +472,19 @@
                 else:
                     self._handle_pair_handler_exception(doc_pair, handler_name, exc)
             except RuntimeError as exc:
-                if "but the refreshed credentials are still expired" in str(exc):
-                    log.warning(
-                        "AWS credentials were refreshed, but the refreshed credentials are still expired"
-                    )
-                    log.info("Reinitializing the upload")
-                    self.dao.remove_transfer(
-                        "upload",
-                        doc_pair=doc_pair.id,
-                        is_direct_transfer=doc_pair.local_state == "direct",
-                    )
-                else:
+                if "but the refreshed credentials are still expired" not in str(
+                    exc
+                ):
                     raise
+                log.warning(
+                    "AWS credentials were refreshed, but the refreshed credentials are still expired"
+                )
+                log.info("Reinitializing the upload")
+                self.dao.remove_transfer(
+                    "upload",
+                    doc_pair=doc_pair.id,
+                    is_direct_transfer=doc_pair.local_state == "direct",
+                )
             except Exception as exc:
                 # Workaround to forward unhandled exceptions to sys.excepthook between all Qthreads
                 sys.excepthook(*sys.exc_info())
@@ -562,18 +560,20 @@
             log.debug(f"The session is paused, skipping <DocPair[{doc_pair.id}]>")
             return
 
-        if WINDOWS:
-            path = doc_pair.local_path
-        else:
-            # The path retrieved from the database will have its starting slash trimmed, restore it
-            path = Path(f"/{doc_pair.local_path}")
-
+        path = doc_pair.local_path if WINDOWS else Path(f"/{doc_pair.local_path}")
         if not path.exists():
-            log.warning(
-                f"Cancelling Direct Transfer of {path!r} because it does not exist anymore"
-            )
-            self._direct_transfer_cancel(doc_pair)
             self.engine.directTranferError.emit(path)
+            if session:
+                log.warning(
+                    f"Pausing Direct Transfer of {path!r} because it does not exist. \
+                        Please validate the path and resume."
+                )
+                self.dao.pause_session(session.uid)
+            else:
+                log.warning(
+                    f"Cancelling Direct Transfer of {path!r} because it does not exist."
+                )
+                self._direct_transfer_cancel(doc_pair)
             return
 
         # Do the upload
@@ -608,9 +608,7 @@
         # Clean-up
         self.dao.remove_state(doc_pair, recursive=recursive)
 
-        # Update session then handle the status
-        session = self.dao.get_session(doc_pair.session)
-        if session:
+        if session := self.dao.get_session(doc_pair.session):
             if (
                 not cancelled_transfer
                 and session.status is not TransferStatus.CANCELLED
@@ -649,8 +647,7 @@
             remote_info.name != doc_pair.local_name
             or remote_info.digest != doc_pair.local_digest
         ):
-            modified = self.dao.get_state_from_local(doc_pair.local_path)
-            if modified:
+            if modified := self.dao.get_state_from_local(doc_pair.local_path):
                 log.info(
                     f"Forcing remotely_modified for pair={modified!r} "
                     f"with info={remote_info!r}"

tests/functional/test_processor.py

@@ -0,0 +1,52 @@
+from collections import namedtuple
+from pathlib import Path
+from unittest.mock import patch
+
+from nxdrive.engine.processor import Processor
+
+
+def test_synchronize_direct_transfer(manager_factory):
+    manager, engine = manager_factory()
+    dao = engine.dao
+    remote = engine.remote
+    Mocked_Session = namedtuple(
+        "session",
+        "status, uid",
+        defaults=("ok", "1"),
+    )
+    session = Mocked_Session()
+    str_path = "unknownPath"
+    path = Path(str_path)
+    DocPair = namedtuple(
+        "DocPair",
+        "local_path, session",
+        defaults=(path, session),
+    )
+    doc_pair = DocPair()
+
+    def mocked_get_session(*args, **kwargs):
+        return session
+
+    def mocked_get_none_session(*args, **kwargs):
+        return None
+
+    def mocked_upload(*args, **kwargs):
+        return
+
+    def mocked_direct_transfer_end(*args, **kwargs):
+        return
+
+    def mocked_pause_session(*args, **kwargs):
+        return
+
+    processor = Processor(engine, True)
+
+    with patch.object(dao, "pause_session", new=mocked_pause_session):
+        with patch.object(remote, "upload", new=mocked_upload):
+            with patch.object(
+                processor, "_direct_transfer_end", new=mocked_direct_transfer_end
+            ):
+                with patch.object(dao, "get_session", new=mocked_get_session):
+                    assert processor._synchronize_direct_transfer(doc_pair) is None
+                with patch.object(dao, "get_session", new=mocked_get_none_session):
+                    assert processor._synchronize_direct_transfer(doc_pair) is None