diff --git a/@startuml b/@startuml
new file mode 100644
index 000000000..238e04290
--- /dev/null
+++ b/@startuml
@@ -0,0 +1,94 @@
+@startuml
+header Nuxeo Drive - Direct Transfer
+
+' uncomment the line below if you're using computer with a retina display
+' skinparam dpi 300
+!define Table(name,desc) class name as "desc" << (T,#FFAAAA) >>
+' we use bold for primary key
+' green color for unique
+' and underscore for not_null
+!define primary_key(x) <b>x</b>
+!define unique(x) <color:green>x</color>
+!define not_null(x) <u>x</u>
+hide methods
+hide stereotypes
+
+class Manager {
+    (The transfers manager)
+    ==
+    + sessions
+    + current_session
+    + decrease_priority()
+    + increase_priority()
+    + shutdown()
+    + start()
+    + stop()
+}
+
+class Session {
+    (A session manager)
+    ==
+    + transfers
+    + not_null(add())
+    + add_all()
+    + decrease_priority()
+    + increase_priority()
+}
+
+class UploadSession {
+    (An upload session manager)
+    ==
+    + func
+    + add()
+    + add_all()
+}
+
+class DownloadSession {
+    (A download session manager)
+    ==
+    + func
+    + add()
+    + add_all()
+}
+
+Manager::current_session --> Session
+Session --* DownloadSession
+Session --* UploadSession
+
+'
+'Table(Session, "Session\n(A transfers session)") {
+'    + primary_key(id) INTEGER
+'    + not_null(priority) INTEGER
+'    + not_null(state) INTEGER
+'    '.. termination status ..
+'    + not_null(status) INTEGER
+'    + finished DATETIME
+'    + remote_path VARCHAR
+'    + started DATETIME
+'}
+
+'Table(Transfer, "Transfer\n(An item to transfer)") {
+'    + primary_key(id) INTEGER
+'    + not_null(unique(session)) INTEGER
+'    + not_null(unique(local_path)) VARCHAR
+'    + not_null(unique(remote_path)) VARCHAR
+'    + error_count INTEGER
+'    + error_total_count INTEGER
+'    + last_error VARCHAR
+'    + file_size INTEGER
+'    + is_file BOOLEAN
+'    + is_upload BOOLEAN
+'    + remote_ref VARCHAR
+'    + status INTEGER
+'    + transferred_size INTEGER
+'    .. uploads only ..
+'    + batch VARCHAR
+'    + chunk_size INTEGER
+'    + doctype VARCHAR
+'    + replace_blob BOOLEAN
+'    + uploaded BOOLEAN
+'}
+
+'Manager::current_sesion --> Session::id
+'Session::id <-- Transfer::session
+@enduml
diff --git a/docs/changes/4.4.4.md b/docs/changes/4.4.4.md
index c9125fb45..d3a7b28a8 100644
--- a/docs/changes/4.4.4.md
+++ b/docs/changes/4.4.4.md
@@ -33,6 +33,12 @@ Release date: `2020-xx-xx`
 - [NXDRIVE-2155](https://jira.nuxeo.com/browse/NXDRIVE-2155): Improvements on the main popup
 - [NXDRIVE-2158](https://jira.nuxeo.com/browse/NXDRIVE-2158): Review the warning message on conflict popup
 
+### Direct Transfer
+
+- [NXDRIVE-1910](https://jira.nuxeo.com/browse/NXDRIVE-1910): Allow the feature when the application is paused
+- [NXDRIVE-2006](https://jira.nuxeo.com/browse/NXDRIVE-2006): Isolate the component
+- [NXDRIVE-2029](https://jira.nuxeo.com/browse/NXDRIVE-2029): Allow the feature when the synchronization is disabled
+
 ## GUI
 
 - [NXDRIVE-2138](https://jira.nuxeo.com/browse/NXDRIVE-2138): Add settings for features
@@ -59,6 +65,7 @@ Release date: `2020-xx-xx`
 ## Minor Changes
 
 - Added `importlib-metadata` 1.6.1
+- Added `peewee` 3.13.3
 - Added `wheel` 0.34.2
 - Added `pip-tools` 5.2.1
 - Added `zipp` 3.1.0
@@ -87,12 +94,20 @@ Release date: `2020-xx-xx`
 
 - Added `Application.exit_app()`
 - Added `USER_AGENT` in `constants.py`
+- Added `Engine.dt_manager`
+- Added `Engine.direct_transfer_duplicate_callback`
+- Renamed `Engine.directTranferDuplicateError` to `directTransferDuplicateError`
 - Added `EngineDAO.update_upload()`
 - Added `NuxeoDocumentInfo.is_proxy`
 - Removed `Remote.direct_transfer()`. Use `DirectTransferUploader.upload()` instead.
 - Removed `Remote.get_document_or_none()`. Use `DirectTransferUploader.get_document_or_none()` instead.
 - Removed `Remote.link_blob_to_doc()`. Use `BaseUploader.link_blob_to_doc()` instead.
 - Removed `Remote.upload_chunks()`. Use `BaseUploader.upload_chunks()` instead.
+- Added `Report.copy_direct_transfer_dbs()`
+- Added `TransferStatus.ABORTED`
+- Removed exceptions.py::`DirectTransferDuplicateFoundError`
+- Removed notification.py::`DirectTransferStatus()`
+- Added `state.py`
 - Added utils.py::`disk_space()`
 - Removed options.py::`handle_feat_direct_transfer()`
 - Removed options.py::`handle_feat_direct_edit()`
@@ -112,6 +127,7 @@ Release date: `2020-xx-xx`
 - Added features.py::`Beta`
 - Added `state.py`
 - Added client/uploader
+- Added the `direct_transfer` folder
 - Removed `conf_name` keyword argument to `CLIHandler.load_config()`
 - Removed engine/dao/sqlite.py::`str_to_path()`
 - Changed order of elements returned by utils.py::`get_tree_list()`
diff --git a/nxdrive/client/uploader/direct_transfer.py b/nxdrive/client/uploader/direct_transfer.py
index 27bbe84d3..f62bb47e8 100644
--- a/nxdrive/client/uploader/direct_transfer.py
+++ b/nxdrive/client/uploader/direct_transfer.py
@@ -9,10 +9,10 @@
 from nuxeo.models import Document
 
 from ...engine.activity import LinkingAction, UploadAction
-from ...exceptions import DirectTransferDuplicateFoundError
 from ...objects import Upload
 from ..local import LocalClient
 from . import BaseUploader
+from .exceptions import DirectTransferDuplicateFoundError
 
 log = getLogger(__name__)
 
diff --git a/nxdrive/data/i18n/i18n.json b/nxdrive/data/i18n/i18n.json
index bc608a47c..2c60fff85 100644
--- a/nxdrive/data/i18n/i18n.json
+++ b/nxdrive/data/i18n/i18n.json
@@ -86,13 +86,11 @@
     "DIRECT_EDIT_UNLOCK_ERROR_DESCRIPTION": "The document „%1“ was not unlocked",
     "DIRECT_EDIT_UPDATED_FILE": "The file „%1“ has been updated",
     "DIRECT_EDIT_VERSION": "You are trying to perform a Direct Edit on version %1 of „%2“. It is not the current version of the document, so it is readonly and cannot be edited.",
-    "DIRECT_TRANSFER_END": "Transfer done: „%1“",
     "DIRECT_TRANSFER_ERROR": "Transfer error: „%1“",
     "DIRECT_TRANSFER_NO_ACCOUNT": "Cannot use the Direct Transfer feature with no account, aborting.",
     "DIRECT_TRANSFER_NOT_ALLOWED": "Direct Transfer of „%1“ is not allowed for synced files.",
     "DIRECT_TRANSFER_NOT_ENABLED": "The Direct Transfer feature is not enabled.",
     "DIRECT_TRANSFER_NOT_POSSIBLE": "The Direct Transfer feature is not usable right now.<br>Please try again later.",
-    "DIRECT_TRANSFER_START": "Transfer start: „%1“",
     "DIRECT_TRANSFER_WINDOW_TITLE": "Direct Transfer - %1",
     "DISCONNECT": "Remove account",
     "DISCONNECTING": "Disconnecting …",
diff --git a/nxdrive/direct_transfer/__init__.py b/nxdrive/direct_transfer/__init__.py
new file mode 100644
index 000000000..82d3b04c6
--- /dev/null
+++ b/nxdrive/direct_transfer/__init__.py
@@ -0,0 +1 @@
+"""The Direct Transfer feature."""
diff --git a/nxdrive/direct_transfer/adapters/__init__.py b/nxdrive/direct_transfer/adapters/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/nxdrive/direct_transfer/adapters/notifications.py b/nxdrive/direct_transfer/adapters/notifications.py
new file mode 100644
index 000000000..a92ca2bb3
--- /dev/null
+++ b/nxdrive/direct_transfer/adapters/notifications.py
@@ -0,0 +1,7 @@
+import abc
+
+
+class AbstractNotifications(abc.ABC):
+    @abc.abstractmethod
+    def send(self, destination, message):
+        raise NotImplementedError
diff --git a/nxdrive/direct_transfer/adapters/orm.py b/nxdrive/direct_transfer/adapters/orm.py
new file mode 100644
index 000000000..98261211f
--- /dev/null
+++ b/nxdrive/direct_transfer/adapters/orm.py
@@ -0,0 +1,80 @@
+from logging import getLogger
+
+from sqlalchemy import (
+    Boolean,
+    Column,
+    Date,
+    Enum,
+    ForeignKey,
+    Integer,
+    JSON,
+    MetaData,
+    String,
+    Table,
+    Text,
+)
+from sqlalchemy.orm import mapper, relationship
+
+from ..domain import model
+from .. import constants
+
+log = getLogger(__name__)
+
+metadata = MetaData()
+
+transfer = Table(
+    "transfer",
+    metadata,
+    Column("uid", Integer, primary_key=True, autoincrement=True),
+    Column("state", Enum(constants.SessionState), default=constants.SessionState.DRAFT),
+    Column("priority", Integer, default=0),
+)
+
+# transfer_item = Table(
+#     "transfer_item",
+#     metadata,
+#     Column("uid", Integer, primary_key=True, autoincrement=True),
+#     Column("size", Integer),
+#     Column("is_file", Boolean),
+#     Column("transferred", Boolean),
+#     Column("state", Enum(constants.TransferStatus)),
+#     Column("remote_ref", Text),
+#     Column("remote_path", Text),
+#     Column("local_path", Text),
+#     Column("batch", JSON),
+#     Column("doctype", String(32)),
+# )
+
+# transfer_item_action = Table(
+#     "transfer_item_action",
+#     metadata,
+#     Column("uid", Integer, primary_key=True, autoincrement=True),
+#     Column("transfer_id", ForeignKey("transfer.uid")),
+#     Column("transfer_item_id", ForeignKey("transfer_item.uid")),
+#     Column("action", Enum(constants.TransferAction)),
+# )
+
+# transfer_item_error = Table(
+#     "transfer_item_error",
+#     metadata,
+#     Column("uid", Integer, primary_key=True, autoincrement=True),
+#     Column("transfer_item_id", ForeignKey("transfer_item.uid")),
+#     Column("error_count", Integer),
+#     Column("error_count_total", Integer),
+#     Column("last_error", Text),
+# )
+
+
+def start_mappers():
+    log.info("Starting mappers")
+    transfer_mapper = mapper(model.Transfer, transfer)
+    # transfer_item_mapper = mapper(
+    #     model.TransferItem,
+    #     transfer_item,
+    #     properties={"transfer": relationship(transfer_mapper)},
+    # )
+    # mapper(
+    #     model.TransferItemAction,
+    #     transfer_item_action,
+    #     properties={"transfer_item": relationship(transfer_item_mapper)},
+    # )
diff --git a/nxdrive/direct_transfer/adapters/qt_eventpublisher.py b/nxdrive/direct_transfer/adapters/qt_eventpublisher.py
new file mode 100644
index 000000000..8589cf77c
--- /dev/null
+++ b/nxdrive/direct_transfer/adapters/qt_eventpublisher.py
@@ -0,0 +1,12 @@
+import json
+from logging import getLogger
+from dataclasses import asdict
+
+from ..domain import events
+
+logger = getLogger(__name__)
+
+
+def publish(channel, event: events.Event):
+    logger.info("publishing: channel=%s, event=%s", channel, event)
+    # r.publish(channel, json.dumps(asdict(event)))
diff --git a/nxdrive/direct_transfer/adapters/repository.py b/nxdrive/direct_transfer/adapters/repository.py
new file mode 100644
index 000000000..07847af1e
--- /dev/null
+++ b/nxdrive/direct_transfer/adapters/repository.py
@@ -0,0 +1,58 @@
+import abc
+from typing import Set
+
+from ..adapters import orm
+from ..domain import model
+
+
+class AbstractRepository(abc.ABC):
+    def __init__(self):
+        self.seen: Set[model.Transfer] = set()
+
+    def add(self, transfer: model.Transfer):
+        self._add(transfer)
+        self.seen.add(transfer)
+
+    def get(self, uid) -> model.Transfer:
+        transfer = self._get(uid)
+        if transfer:
+            self.seen.add(transfer)
+        return transfer
+
+    def get_by_batchref(self, batchref) -> model.Transfer:
+        transfer = self._get_by_batchref(batchref)
+        if transfer:
+            self.seen.add(transfer)
+        return transfer
+
+    @abc.abstractmethod
+    def _add(self, transfer: model.Transfer):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def _get(self, uid) -> model.Transfer:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def _get_by_batchref(self, batchref) -> model.Transfer:
+        raise NotImplementedError
+
+
+class SqlAlchemyRepository(AbstractRepository):
+    def __init__(self, session):
+        super().__init__()
+        self.session = session
+
+    def _add(self, transfer):
+        self.session.add(transfer)
+
+    def _get(self, uid: int):
+        return self.session.query(model.Transfer).filter_by(uid=uid).first()
+
+    def _get_by_batchref(self, batchref):
+        return (
+            self.session.query(model.Transfer)
+            .join(model.Batch)
+            .filter(orm.batches.c.reference == batchref,)
+            .first()
+        )
diff --git a/nxdrive/direct_transfer/bootstrap.py b/nxdrive/direct_transfer/bootstrap.py
new file mode 100644
index 000000000..4b50c568a
--- /dev/null
+++ b/nxdrive/direct_transfer/bootstrap.py
@@ -0,0 +1,43 @@
+import inspect
+from typing import Callable
+
+from .adapters import orm, qt_eventpublisher
+from .adapters.notifications import AbstractNotifications
+from .service_layer import handlers, messagebus, unit_of_work
+
+
+def bootstrap(
+    start_orm: bool = True,
+    uow: unit_of_work.AbstractUnitOfWork = unit_of_work.SqlAlchemyUnitOfWork(),
+    notifications: AbstractNotifications = None,
+    publish: Callable = qt_eventpublisher.publish,
+) -> messagebus.MessageBus:
+
+    if start_orm:
+        orm.start_mappers()
+
+    dependencies = {"uow": uow, "notifications": notifications, "publish": publish}
+    injected_event_handlers = {
+        event_type: [
+            inject_dependencies(handler, dependencies) for handler in event_handlers
+        ]
+        for event_type, event_handlers in handlers.EVENT_HANDLERS.items()
+    }
+    injected_command_handlers = {
+        command_type: inject_dependencies(handler, dependencies)
+        for command_type, handler in handlers.COMMAND_HANDLERS.items()
+    }
+
+    return messagebus.MessageBus(
+        uow=uow,
+        event_handlers=injected_event_handlers,
+        command_handlers=injected_command_handlers,
+    )
+
+
+def inject_dependencies(handler, dependencies):
+    params = inspect.signature(handler).parameters
+    deps = {
+        name: dependency for name, dependency in dependencies.items() if name in params
+    }
+    return lambda message: handler(message, **deps)
diff --git a/nxdrive/direct_transfer/constants.py b/nxdrive/direct_transfer/constants.py
new file mode 100644
index 000000000..573eb66be
--- /dev/null
+++ b/nxdrive/direct_transfer/constants.py
@@ -0,0 +1,46 @@
+"""
+The Direct Transfer feature.
+
+What: constants.
+"""
+
+from enum import Enum
+
+
+class TransferAction(Enum):
+    """Represent a session state."""
+
+    DOWNLOAD = "DOWN"
+    UPLOAD = "UP"
+    REPLACE_BLOB = "REPLACE"
+
+
+class SessionState(Enum):
+    """Represent a session state."""
+
+    DRAFT = 0
+    PENDING = 1
+    RUNNING = 2
+    SUSPENDED = 3
+    ON_HOLD = 4
+    ENDED = 5
+
+
+class SessionTerminationStatus(Enum):
+    """Represent a session final status."""
+
+    NONE = 0
+    SUCCESS = 1
+    FAILURE = 2
+    CANCELLED = 3
+
+
+class TransferStatus(Enum):
+    """Represent a single item transfer status."""
+
+    NONE = 0
+    ONGOING = 1
+    PAUSED = 2
+    SUSPENDED = 3
+    DONE = 4
+    ABORTED = 5
diff --git a/nxdrive/direct_transfer/domain/__init__.py b/nxdrive/direct_transfer/domain/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/nxdrive/direct_transfer/domain/commands.py b/nxdrive/direct_transfer/domain/commands.py
new file mode 100644
index 000000000..42fb16e89
--- /dev/null
+++ b/nxdrive/direct_transfer/domain/commands.py
@@ -0,0 +1,30 @@
+from dataclasses import dataclass
+
+
+class Command:
+    pass
+
+
+@dataclass
+class Cancel(Command):
+    uid: int
+
+
+@dataclass
+class Force(Command):
+    uid: int
+
+
+@dataclass
+class Pause(Command):
+    uid: int
+
+
+@dataclass
+class Resume(Command):
+    uid: int
+
+
+@dataclass
+class Transfer(Command):
+    uid: int
diff --git a/nxdrive/direct_transfer/domain/events.py b/nxdrive/direct_transfer/domain/events.py
new file mode 100644
index 000000000..0dda01b96
--- /dev/null
+++ b/nxdrive/direct_transfer/domain/events.py
@@ -0,0 +1,46 @@
+from dataclasses import dataclass
+
+
+class Event:
+    pass
+
+
+@dataclass
+class Added(Event):
+    uid: int
+
+
+@dataclass
+class Cancelled(Event):
+    uid: int
+
+
+@dataclass
+class ChunkSent(Event):
+    uid: int
+
+
+@dataclass
+class Done(Event):
+    uid: int
+
+
+@dataclass
+class Paused(Event):
+    uid: int
+
+
+@dataclass
+class Removed(Event):
+    uid: int
+
+
+@dataclass
+class Resumed(Event):
+    uid: int
+
+
+@dataclass
+class TransferError(Event):
+    uid: int
+    error: str
diff --git a/nxdrive/direct_transfer/domain/model.py b/nxdrive/direct_transfer/domain/model.py
new file mode 100644
index 000000000..283f45dc5
--- /dev/null
+++ b/nxdrive/direct_transfer/domain/model.py
@@ -0,0 +1,84 @@
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Set
+
+from ..domain import events
+from ..constants import SessionState, SessionTerminationStatus
+
+
+@dataclass(unsafe_hash=True)
+class TransferItem:
+    uid: int
+    local_path: str
+    is_file: Optional[bool] = field(default=True, hash=False)
+    size: Optional[bool] = field(default=0, hash=False)
+    transferred: Optional[bool] = field(default=False, hash=False)
+    status: Optional[int] = field(default=SessionState.DRAFT, hash=False)
+    remote_ref: Optional[str] = field(default="", hash=False)
+    remote_path: Optional[str] = field(default="", hash=False)
+    batch: Optional[Dict[str, Any]] = field(default_factory=dict, hash=False)
+
+    def transferred_size(self) -> int:
+        if self.transferred:
+            return self.size
+        return 0
+
+
+class Transfer:
+    def __init__(
+        self,
+        uid: str,
+        state: Optional[SessionState] = SessionState.DRAFT,
+        priority: Optional[int] = 0,
+    ):
+        self.uid = uid
+        self.state = state
+        self.priority = priority
+        self._items: Set[TransferItem] = set()
+        self.events: List[events.Event] = []
+
+    def __repr__(self) -> str:
+        return f"Transfer(uid={self.uid}, priority={self.priority})"
+
+    def __eq__(self, other) -> bool:
+        if not isinstance(other, Transfer):
+            return False
+        return other.uid == self.uid
+
+    def __hash__(self) -> int:
+        return hash(self.uid)
+
+    def __gt__(self, other) -> bool:
+        return self.priority > other.priority or self.uid > other.uid
+
+    @property
+    def status(self) -> SessionTerminationStatus:
+        if self.state is not SessionState.ENDED:
+            return SessionTerminationStatus.NONE
+        if all(item.transferred for item in self._items):
+            return SessionTerminationStatus.SUCCESS
+        return SessionTerminationStatus.FAILURE
+
+    def add(self, item: TransferItem) -> None:
+        self._items.add(item)
+        self.events.append(events.Added(item.uid))
+
+    def remove(self, item: TransferItem) -> None:
+        self._items.remove(item)
+        self.events.append(events.Removed(item.uid))
+
+
+@dataclass(frozen=True)
+class TransferItemAction:
+    uid: int
+    transfer_id: int
+    transfer_item_id: int
+    action: str
+
+
+@dataclass(frozen=True)
+class TransferItemError:
+    uid: int
+    transfer_item_id: int
+    error_count: int
+    error_count_total: int
+    last_error: str
diff --git a/nxdrive/direct_transfer/manager.py b/nxdrive/direct_transfer/manager.py
new file mode 100644
index 000000000..e9f4a7c05
--- /dev/null
+++ b/nxdrive/direct_transfer/manager.py
@@ -0,0 +1,63 @@
+"""
+The Direct Transfer feature.
+
+What: the transfers manager.
+
+For now, transfers are done in sequence as it eases the implementation.
+For better performances, one should really use Nuxeo Drive.
+"""
+from logging import getLogger
+from pathlib import Path
+from typing import TYPE_CHECKING, Callable
+
+from . import bootstrap
+
+if TYPE_CHECKING:
+    from nuxeo.handlers.default import Uploader  # noqa
+    from nuxeo.models import Document  # noqa
+
+    from .client.remote_client import Remote  # noqa
+
+
+log = getLogger(__name__)
+
+
+class DirectTransferManager:
+    """Direct Transfer manager.
+    This is the core of the feature, separated from Nuxeo Drive.
+    It is used to create a new Direct Transfer, keep track of the progression,
+    pause/resume transfers and it is connected to a (beautiful) custom QML component.
+    """
+
+    def __init__(
+        self,
+        db: str,
+        engine_uid: str,
+        remote: "Remote",
+        chunk_callback: Callable[["Uploader"], None] = lambda *_: None,
+        done_callback: Callable[[bool, int], None] = lambda *_: None,
+        dupe_callback: Callable[[Path, "Document"], None] = lambda *_: None,
+    ) -> None:
+        """
+        *chunk_callback* is triggered when a chunk is successfully transferred.
+        *done_callback* is triggered when a transfer was successfully done.
+        *dupe_callback* is triggered when a given upload would generate a duplicate on the server.
+        """
+        self.db = db
+        self.engine_uid = engine_uid
+        self.remote = remote
+        self.chunk_callback = chunk_callback
+        self.done_callback = done_callback
+        self.dupe_callback = dupe_callback
+
+        self.bus = bootstrap.bootstrap()
+
+    def __repr__(self) -> str:
+        return (
+            f"{type(self).__name__}<uid={self.engine_uid!r}"
+            f", started={self.is_started!r}"
+            f", sessions={len(self.sessions)}"
+            f", session={self.current_session}"
+            f", db={self.db!r}"
+            ">"
+        )
diff --git a/nxdrive/direct_transfer/runner.py b/nxdrive/direct_transfer/runner.py
new file mode 100644
index 000000000..5b93b8def
--- /dev/null
+++ b/nxdrive/direct_transfer/runner.py
@@ -0,0 +1,27 @@
+"""
+The Direct Transfer feature.
+
+What: threaded runner.
+"""
+from threading import Thread
+from typing import Any, Optional
+
+
+class Runner(Thread):
+    """Simple runner used for every single upload.
+    Any exception will be forwarded to the caller via the .error attribute.
+    """
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """Simple init to declare .error that is used to communicate eventual exceptions to the caller."""
+        super().__init__(*args, **kwargs)
+        self.error: Optional[Exception] = None
+
+    def run(self) -> None:
+        """Catch any error and store it inside the .error attribute instead of throwing it."""
+        try:
+            super().run()
+        except Exception as exc:
+            self.error = exc
+            # Break reference cyle
+            exc = None  # type: ignore
diff --git a/nxdrive/direct_transfer/service_layer/__init__.py b/nxdrive/direct_transfer/service_layer/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/nxdrive/direct_transfer/service_layer/handlers.py b/nxdrive/direct_transfer/service_layer/handlers.py
new file mode 100644
index 000000000..9a8aaf214
--- /dev/null
+++ b/nxdrive/direct_transfer/service_layer/handlers.py
@@ -0,0 +1,107 @@
+from typing import Callable, Dict, List, Type
+from pathlib import Path
+
+from nuxeo.models import Document
+
+from . import unit_of_work
+from ..adapters import notifications
+from ..domain import commands, events
+
+
+class DirectTransferPaused(Exception):
+    """A transfer has been paused, the file's processing should stop."""
+
+
+class DirectTransferDuplicateFoundError(ValueError):
+    """
+    Exception raised when a duplicate file already exists on the server
+    and trying to Direct Transfer a local file with the same name.
+    """
+
+    def __init__(self, file: Path, doc: Document) -> None:
+        self.file = file
+        self.doc = doc
+
+    def __repr__(self) -> str:
+        return f"{type(self).__name__}<file={self.file!r}, doc={self.doc!r}>"
+
+    def __str__(self) -> str:
+        return (
+            f"Document with the name {self.file.name!r} already found on the server: {self.doc}."
+            f"Direct Transfer of {self.file!r} postponed after the user decided what to do."
+        )
+
+
+#
+# Events
+#
+
+
+def chunk_sent(
+    event: events.ChunkSent, notifications: notifications.AbstractNotifications
+):
+    pass
+
+
+def send_transfer_error_notification(
+    event: events.TransferError, notifications: notifications.AbstractNotifications
+):
+    pass
+
+
+def transfer_cancelled(event: events.Cancelled, publish: Callable):
+    pass
+
+
+def transfer_done(event: events.Done, publish: Callable):
+    pass
+
+
+def transfer_paused(event: events.Paused, publish: Callable):
+    pass
+
+
+def transfer_resumed(event: events.Resumed, publish: Callable):
+    pass
+
+
+#
+# Commands
+#
+
+
+def cancel(cmd: commands.Cancel, uow: unit_of_work.AbstractUnitOfWork):
+    pass
+
+
+def force_transfer(cmd: commands.Force, uow: unit_of_work.AbstractUnitOfWork):
+    pass
+
+
+def pause_transfer(cmd: commands.Pause, uow: unit_of_work.AbstractUnitOfWork):
+    pass
+
+
+def resume_transfer(cmd: commands.Resume, uow: unit_of_work.AbstractUnitOfWork):
+    pass
+
+
+def transfer(cmd: commands.Transfer, uow: unit_of_work.AbstractUnitOfWork):
+    pass
+
+
+EVENT_HANDLERS: Dict[Type[events.Event], List[Callable]] = {
+    events.Cancelled: [transfer_cancelled],
+    events.Done: [transfer_done],
+    events.Paused: [transfer_paused],
+    events.Resumed: [transfer_resumed],
+    events.TransferError: [send_transfer_error_notification],
+}
+
+COMMAND_HANDLERS: Dict[Type[commands.Command], Callable] = {
+    commands.Cancel: cancel,
+    commands.Force: force_transfer,
+    commands.Pause: pause_transfer,
+    commands.Resume: resume_transfer,
+    commands.Transfer: transfer,
+}
diff --git a/nxdrive/direct_transfer/service_layer/messagebus.py b/nxdrive/direct_transfer/service_layer/messagebus.py
new file mode 100644
index 000000000..d201111c9
--- /dev/null
+++ b/nxdrive/direct_transfer/service_layer/messagebus.py
@@ -0,0 +1,52 @@
+from logging import getLogger
+from typing import Callable, Dict, List, Union, Type
+
+from ..domain import commands, events
+from ..service_layer import unit_of_work
+
+logger = getLogger(__name__)
+
+Message = Union[commands.Command, events.Event]
+
+
+class MessageBus:
+    def __init__(
+        self,
+        uow: unit_of_work.AbstractUnitOfWork,
+        event_handlers: Dict[Type[events.Event], List[Callable]],
+        command_handlers: Dict[Type[commands.Command], Callable],
+    ):
+        self.uow = uow
+        self.event_handlers = event_handlers
+        self.command_handlers = command_handlers
+
+    def handle(self, message: Message):
+        self.queue = [message]
+        while self.queue:
+            message = self.queue.pop(0)
+            if isinstance(message, events.Event):
+                self.handle_event(message)
+            elif isinstance(message, commands.Command):
+                self.handle_command(message)
+            else:
+                raise Exception(f"{message} was not an Event or Command")
+
+    def handle_event(self, event: events.Event):
+        for handler in self.event_handlers[type(event)]:
+            try:
+                logger.debug("handling event %s with handler %s", event, handler)
+                handler(event)
+                self.queue.extend(self.uow.collect_new_events())
+            except Exception:
+                logger.exception("Exception handling event %s", event)
+                continue
+
+    def handle_command(self, command: commands.Command):
+        logger.debug("handling command %s", command)
+        try:
+            handler = self.command_handlers[type(command)]
+            handler(command)
+            self.queue.extend(self.uow.collect_new_events())
+        except Exception:
+            logger.exception("Exception handling command %s", command)
+            raise
diff --git a/nxdrive/direct_transfer/service_layer/unit_of_work.py b/nxdrive/direct_transfer/service_layer/unit_of_work.py
new file mode 100644
index 000000000..76bd7b485
--- /dev/null
+++ b/nxdrive/direct_transfer/service_layer/unit_of_work.py
@@ -0,0 +1,57 @@
+import abc
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm.session import Session
+
+from ..adapters import repository
+
+
+class AbstractUnitOfWork(abc.ABC):
+    products: repository.AbstractRepository
+
+    def __enter__(self) -> "AbstractUnitOfWork":
+        return self
+
+    def __exit__(self, *args):
+        self.rollback()
+
+    def commit(self):
+        self._commit()
+
+    def collect_new_events(self):
+        for product in self.products.seen:
+            while product.events:
+                yield product.events.pop(0)
+
+    @abc.abstractmethod
+    def _commit(self):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def rollback(self):
+        raise NotImplementedError
+
+
+class SqlAlchemyUnitOfWork(AbstractUnitOfWork):
+    def __init__(self, session_factory=None):
+        if session_factory is None:
+            self.session_factory = sessionmaker(
+                bind=create_engine("sqlite:///:memory:"),
+            )
+        else:
+            self.session_factory: Session = session_factory
+
+    def __enter__(self):
+        self.session = self.session_factory()
+        self.products = repository.SqlAlchemyRepository(self.session)
+        return super().__enter__()
+
+    def __exit__(self, *args):
+        super().__exit__(*args)
+        self.session.close()
+
+    def _commit(self):
+        self.session.commit()
+
+    def rollback(self):
+        self.session.rollback()
diff --git a/nxdrive/direct_transfer/tests/e2e/test_direct_transfer.py b/nxdrive/direct_transfer/tests/e2e/test_direct_transfer.py
new file mode 100644
index 000000000..69a39cf26
--- /dev/null
+++ b/nxdrive/direct_transfer/tests/e2e/test_direct_transfer.py
@@ -0,0 +1,569 @@
+"""
+Test the Direct Transfer feature in different scenarii.
+"""
+from pathlib import Path
+from shutil import copyfile, copytree
+from time import sleep
+from unittest.mock import patch
+from uuid import uuid4
+
+import pytest
+from nuxeo.exceptions import HTTPError
+from nuxeo.models import Document
+from nxdrive.direct_transfer import DirectTransferDuplicateFoundError
+from nxdrive.direct_transfer.constants import TransferStatus
+from nxdrive.options import Options
+from nxdrive.utils import get_tree_list
+from requests.exceptions import ConnectionError
+
+from .. import ensure_no_exception
+from ..markers import not_windows
+
+LOCATION = Path(__file__).parent.parent
+
+
+def has_blob(engine, path: str) -> bool:
+    """Check that *path* exists on the server and has a blob attached.
+    As when doing a Direct Transfer, the document is first created on the server,
+    this is the only way to check if the blob upload has been finished successfully.
+    """
+    try:
+        doc = engine.remote.documents.get(path=path)
+    except Exception:
+        return False
+    return bool(doc.properties.get("file:content"))
+
+
+def sync_and_check(engine, should_have_blob: bool = True) -> None:
+    # Sync
+    wait_sync(engine)
+
+    # Check the file exists on the server and has a blob attached
+    path = f"{engine.ws_root.path}/{engine.file.name}"
+    blob = has_blob(engine, path)
+    assert blob is should_have_blob
+
+
+def wait_sync(engine):
+    """Wait for the Direct transfer session to finish."""
+    for _ in range(30):  # 30 sec maxi
+        if not engine.dt_manager.is_started:
+            break
+        sleep(1)
+
+
+@pytest.fixture
+def engine(tmp_path, app, server, manager_factory, obj_factory):
+    Options.synchronization_enabled = False
+
+    manager, engine = manager_factory()
+
+    # Attach the fake workspace folder
+    engine.ws_root = obj_factory()
+
+    # Grant RW ACLs for the current user
+    server.operations.execute(
+        command="Document.SetACE",
+        input_obj=engine.ws_root.uid,
+        user=engine.remote_user,
+        permission="ReadWrite",
+        grant=True,
+    )
+
+    # Start the engine!
+    engine.start()
+
+    # Start an empty DT session to create the .dt_manager attribute
+    engine.direct_transfer([], "")
+
+    # Lower chunk_* options to have chunked uploads without having to create big files
+    default_chunk_limit = Options.chunk_limit
+    default_chunk_size = Options.chunk_size
+    Options.chunk_limit = 1
+    Options.chunk_size = 1
+
+    # The file used for the Direct Transfer (10 MiB)
+    engine.file = tmp_path / f"{uuid4()}.bin"
+    engine.file.write_bytes(b"0" * 1024 * 1024 * 10)
+
+    try:
+        with manager:
+            yield engine
+    finally:
+        # Restore options
+        Options.chunk_limit = default_chunk_limit
+        Options.chunk_size = default_chunk_size
+
+
+def test_simple(engine):
+    """Simple Direct Transfer test."""
+    with ensure_no_exception():
+        engine.direct_transfer([engine.file], engine.ws_root.path)
+        sync_and_check(engine)
+
+    # Ensure the remote path is saved for next times
+    assert engine.dao.get_config("dt_last_remote_location") == engine.ws_root.path
+
+
+def test_with_engine_not_started(engine):
+    """A Direct Transfer should work even if engines are stopped."""
+    engine.stop()
+
+    with ensure_no_exception():
+        engine.direct_transfer([engine.file], engine.ws_root.path)
+        sync_and_check(engine)
+
+
+def test_duplicate_file_but_no_blob_attached(engine):
+    """The file already exists on the server but has no blob attached yet."""
+
+    def create(*_, **__):
+        """Patch Nuxeo.server.documents.create() to be able to check that nothing will be done."""
+        assert 0, "No twice creation should be done!"
+
+    # Create the document on the server, with no blob attached
+    new_doc = Document(
+        name=engine.file.name, type="File", properties={"dc:title": engine.file.name}
+    )
+    doc = engine.remote.documents.create(new_doc, parent_path=engine.ws_root.path)
+    assert doc.properties.get("file:content") is None
+
+    # The upload should work: the doc will be retrieved and the blob uploaded and attached
+    with patch.object(engine.dt_manager.remote.documents, "create", new=create):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+    # Ensure there is only 1 document on the server
+    children = engine.remote.documents.get_children(path=engine.ws_root.path)
+    assert len(children) == 1
+    assert children[0].title == engine.file.name
+
+
+def test_duplicate_file_cancellation(engine):
+    """The file already exists on the server and has a blob attached.
+    Then, the user wants to cancel the transfer.
+    """
+
+    def cancel(file: Path, doc: Document) -> None:
+        """Mimic the user choosing to cancel the upload."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        engine.direct_transfer_cancel(file)
+
+    def upload(*_, **__):
+        """Patch Remote.upload() to be able to check that nothing will be uploaded."""
+        assert 0, "No twice upload should be done!"
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager, "dupe_callback", new=cancel):
+        with ensure_no_exception():
+            # 1st upload: OK
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # 2nd upload: it should be cancelled by the user
+            with patch.object(engine.dt_manager.remote, "upload", new=upload):
+                engine.direct_transfer([engine.file], engine.ws_root.path)
+                wait_sync(engine)
+
+    assert checkpoint
+
+    # Ensure there is only 1 document on the server
+    children = engine.remote.documents.get_children(path=engine.ws_root.path)
+    assert len(children) == 1
+    assert children[0].title == engine.file.name
+
+
+def test_duplicate_file_replace_blob(engine):
+    """The file already exists on the server and has a blob attached.
+    Then, the user wants to replace the blob.
+    """
+
+    def replace(file: Path, doc: Document) -> None:
+        """Mimic the user choosing to replace the blob on the server."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        engine.direct_transfer_replace_blob(file, doc)
+
+    checkpoint = False
+    content = b"blob changed!"
+
+    with patch.object(engine.dt_manager, "dupe_callback", new=replace):
+        with ensure_no_exception():
+            # 1st upload: OK
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # To ease testing, we change local file content
+            engine.file.write_bytes(content)
+
+            # 2nd upload: the blob should be replaced on the server
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # Manual upload to test the exception raised
+            dt_man = engine.dt_manager
+            engine.direct_transfer([engine.file], engine.ws_root.path, start=False)
+            with pytest.raises(DirectTransferDuplicateFoundError) as exc:
+                dt_man.remote.dt_upload(dt_man.remote, dt_man.transfers[0])
+            assert repr(exc)
+            assert str(exc)
+
+    assert checkpoint
+
+    # Ensure there is only 1 document on the server
+    children = engine.remote.documents.get_children(path=engine.ws_root.path)
+    assert len(children) == 1
+    assert children[0].title == engine.file.name
+    assert engine.remote.get_blob(children[0].uid) == content
+
+
+def test_pause_upload_manually(engine):
+    """
+    Pause the transfer by simulating a click on the pause/resume icon
+    on the current upload in the systray menu.
+    """
+
+    def callback(*_):
+        """
+        This will mimic what is done in SystrayTranfer.qml:
+            - call API.pause_transfer() that will call:
+                - engine.dao.pause_transfer(nature, transfer_uid)
+        Then the upload will be paused in Remote.upload().
+        """
+        nonlocal checkpoint
+        checkpoint = True
+
+        # Ensure we have 1 ongoing upload
+        transfers = engine.dt_manager.uploading
+        assert len(transfers) == 1
+
+        # Pause the upload
+        transfers[0].status = TransferStatus.PAUSED
+        transfers[0].save()
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager, "chunk_callback", new=callback):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+        assert len(engine.dt_manager.uploading) == 0
+
+    assert checkpoint
+
+    # Resume the upload
+    engine.dt_manager.start()
+    sync_and_check(engine)
+
+
+def test_pause_upload_automatically(engine):
+    """
+    Pause the transfer by simulating an application exit
+    or clicking on the Suspend menu entry from the systray.
+    """
+
+    def callback(*_):
+        """This will mimic what is done in SystrayMenu.qml: suspend the app."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        # Ensure we have 1 ongoing upload
+        assert len(engine.dt_manager.uploading) == 1
+
+        # Suspend!
+        engine.suspend()
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager, "chunk_callback", new=callback):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+        assert len(engine.dt_manager.uploading) == 0
+
+    assert checkpoint
+
+    # Resume the upload
+    engine.resume()
+    sync_and_check(engine)
+
+
+def test_modifying_paused_upload(engine):
+    """Modifying a paused upload should discard the current upload."""
+
+    def callback(*_):
+        """Pause the upload and apply changes to the document."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        # Ensure we have 1 ongoing upload
+        transfers = engine.dt_manager.uploading
+        assert len(transfers) == 1
+
+        # Pause the upload
+        transfers[0].status = TransferStatus.PAUSED
+        transfers[0].save()
+
+        # Apply changes to the file
+        engine.file.write_bytes(content)
+
+    checkpoint = False
+    content = b"locally changed"
+
+    with patch.object(engine.dt_manager, "chunk_callback", new=callback):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+    assert checkpoint
+
+    # Resume the upload
+    engine.dt_manager.start()
+
+    # Check the local content is correct
+    sync_and_check(engine)
+    assert engine.file.read_bytes() == content
+
+    # And check the remote content is correct too
+    blob = engine.remote.get_blob(engine.dt_manager.transfers[0].remote_ref)
+    assert blob == content
+
+
+@not_windows(
+    reason="Cannot test the behavior as the local deletion is blocked by the OS."
+)
+def test_deleting_paused_upload(engine):
+    """Deleting a paused upload should discard the current upload."""
+
+    def callback(*_):
+        """Pause the upload and delete the document."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        # Ensure we have 1 ongoing upload
+        transfers = engine.dt_manager.uploading
+        assert len(transfers) == 1
+
+        # Pause the upload
+        transfers[0].status = TransferStatus.PAUSED
+        transfers[0].save()
+
+        # Remove the document
+        # (this is the problematic part on Windows, because for the
+        #  file descriptor to be released we need to escape from
+        #  Remote.upload(), which is not possible from here)
+        engine.file.unlink()
+        assert not engine.file.exists()
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager, "chunk_callback", new=callback):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+    assert checkpoint
+
+    # Resume the upload
+    engine.dt_manager.start()
+    sync_and_check(engine, should_have_blob=False)
+
+
+def test_not_server_error_upload(engine):
+    """Test an error happening after chunks were uploaded."""
+
+    def bad(*args, **kwargs):
+        """Simulate an exception that is not handled by the Processor."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        raise ValueError("Mocked exception")
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager.remote, "dt_link_blob_to_doc", new=bad):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # There should be 1 upload with ONGOING transfer status
+            assert len(engine.dt_manager.uploading) == 1
+
+            # The file exists on the server but has no blob yet
+            path = f"{engine.ws_root.path}/{engine.file.name}"
+            assert not has_blob(engine, path)
+
+    assert checkpoint
+
+    # Check thransfer error details
+    transfer = engine.dt_manager.uploading[0]
+    assert transfer.error_count == 3
+    assert transfer.error_count_total == 3
+
+    # Reset the error
+    engine.dt_manager.reset_error(transfer)
+    assert transfer.error_count == 0
+    assert transfer.error_count_total == 3
+
+    # Restart the upload
+    engine.dt_manager.start()
+    sync_and_check(engine)
+
+
+def test_server_error_but_upload_ok(engine):
+    """
+    Test an error happening after chunks were uploaded and the Blob.AttachOnDocument operation call.
+    This could happen if a proxy does not understand well the final requests as seen in NXDRIVE-1753.
+    """
+
+    def bad(*args, **kwargs):
+        # Call the original method to effectively end the upload process
+        execute_orig(*args, **kwargs)
+
+        if kwargs.get("command", "") == "Blob.AttachOnDocument":
+            # The file should be present on the server
+            children = engine.remote.documents.get_children(path=engine.ws_root.path)
+            assert len(children) == 1
+            assert children[0].title == engine.file.name
+
+            # There should be 1 upload with ONGOING transfer status
+            assert len(engine.dt_manager.uploading) == 1
+
+            # And throw an error
+            stack = (
+                "The proxy server received an invalid response from an upstream server."
+            )
+            raise HTTPError(status=502, message="Mocked Proxy Error", stacktrace=stack)
+
+    execute_orig = engine.remote.execute
+
+    with patch.object(engine.dt_manager.remote, "execute", new=bad):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+        assert len(engine.dt_manager.uploading) == 0
+
+    engine.dt_manager.start()
+    sync_and_check(engine)
+
+
+def test_server_error_upload(engine):
+    """Test a server error happening after chunks were uploaded, at the Blob.AttachOnDocument operation call."""
+
+    def bad(*args, **kwargs):
+        """Simulate a server error."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        raise ConnectionError("Mocked exception")
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager.remote, "dt_link_blob_to_doc", new=bad):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # There should be 1 upload with ONGOING transfer status
+            assert len(engine.dt_manager.uploading) == 1
+
+            # The file exists on the server but has no blob yet
+            path = f"{engine.ws_root.path}/{engine.file.name}"
+            assert not has_blob(engine, path)
+
+    assert checkpoint
+
+    # Reset the error
+    transfer = engine.dt_manager.uploading[0]
+    engine.dt_manager.reset_error(transfer)
+
+    # Restart the upload
+    engine.dt_manager.start()
+    sync_and_check(engine)
+
+
+def test_chunk_upload_error(engine):
+    """Test a server error happening while uploading chunks."""
+
+    def bad(*args, **kwargs):
+        """Simulate an error after the upload of chunk n°1."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        nonlocal number
+
+        if number == 0:
+            number += 1
+            return send_data_orig(*args, **kwargs)
+        else:
+            raise ConnectionError("Mocked error")
+
+    checkpoint = False
+    number = 0
+    send_data_orig = engine.dt_manager.remote.uploads.send_data
+
+    with patch.object(engine.dt_manager.remote.uploads, "send_data", new=bad):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # There should be 1 upload with ONGOING transfer status
+            assert len(engine.dt_manager.uploading) == 1
+
+            # The file exists on the server but has no blob yet
+            path = f"{engine.ws_root.path}/{engine.file.name}"
+            assert not has_blob(engine, path)
+
+    assert checkpoint
+
+    # Reset the error
+    transfer = engine.dt_manager.uploading[0]
+    engine.dt_manager.reset_error(transfer)
+
+    # Restart the upload
+    engine.dt_manager.start()
+    sync_and_check(engine)
+
+
+def test_folder(tmp_path, engine):
+    """Test the Direct Transfer on a folder containing files and a sufolder."""
+
+    # We need more data for this test
+    folder = tmp_path / str(uuid4())
+    copytree(LOCATION / "resources", folder, copy_function=copyfile)
+
+    # Get folders and files tests will handle
+    tree = {path: rpath for rpath, path in get_tree_list(folder, engine.ws_root.path)}
+    files = [path for path in tree.keys() if path.is_file()]
+    folders = [path for path in tree.keys() if path.is_dir()]
+    engine.tree = tree
+
+    def check() -> None:
+        # Sync
+        wait_sync(engine)
+
+        # Check files exist on the server with their attached blob
+        for file in files:
+            path = f"{tree[file]}/{file.name}"
+            assert has_blob(engine, path)
+
+        # Check subfolders
+        for folder in folders:
+            assert engine.remote.documents.get(path=tree[folder])
+
+    with ensure_no_exception():
+        engine.direct_transfer([folder], engine.ws_root.path)
+        check()
+
+    # Ensure there is only 1 folder created at the workspace root
+    children = engine.remote.documents.get_children(path=engine.ws_root.path)
+    assert len(children) == 1
+    assert children[0].title == folder.name
+
+    # All has been uploaded
+    assert engine.dt_manager.is_completed
diff --git a/nxdrive/direct_transfer/tests/integration/test_direct_transfer_manager.py b/nxdrive/direct_transfer/tests/integration/test_direct_transfer_manager.py
new file mode 100644
index 000000000..7a9fb0647
--- /dev/null
+++ b/nxdrive/direct_transfer/tests/integration/test_direct_transfer_manager.py
@@ -0,0 +1,291 @@
+# coding: utf-8
+"""
+Test the Direct Transfer manager.
+"""
+from datetime import datetime
+from pathlib import Path
+from typing import List
+from unittest.mock import patch
+from uuid import uuid4
+
+import pytest
+from nuxeo.models import Batch
+from nxdrive.direct_transfer import DirectTransferManager
+from nxdrive.direct_transfer.constants import TransferStatus
+from nxdrive.direct_transfer.models import Session, Transfer
+from nxdrive.options import Options
+
+
+class FakeRemote:
+    """Fake Remote client."""
+
+
+@pytest.fixture
+def manager(tmp_path):
+    db = tmp_path / f"{uuid4()}.db"
+    remote = FakeRemote()
+    yield DirectTransferManager(str(db), "UID", remote)
+
+
+def create_tree(root: Path) -> List[Path]:
+    """Create the tree for tests."""
+    paths = []
+    size = 0
+
+    # Folders
+    folder = root / "one folder"
+    subfolder = folder / "subfolder"
+    subfolder.mkdir(parents=True)
+    paths.append(folder)
+    paths.append(subfolder)
+
+    # Files
+    file = root / "root-file.bin"
+    size += file.write_bytes(b"0" * Options.chunk_size * 1024 * 1024 + b"0" * 1024)
+    paths.append(file)
+
+    file = folder / "one file.txt"
+    size += file.write_text("azerty")
+    paths.append(file)
+    file = folder / "two files.txt"
+    size += file.write_text("azerty" * 2)
+    paths.append(file)
+
+    file = subfolder / "file1.txt"
+    size += file.write_text("azerty")
+    paths.append(file)
+    file = subfolder / "file2.txt"
+    size += file.write_text("azerty" * 2)
+    paths.append(file)
+
+    return paths, size
+
+
+def test_manager_empty(manager):
+    assert repr(manager)
+    assert manager.engine_uid == "UID"
+    assert len(manager.transfers) == 0
+    assert manager.is_completed
+    assert not manager.is_started
+    assert manager.progress == 100.0
+    assert manager.size == 0.0
+    assert manager.uploaded == 0.0
+
+    # Check the session
+    session = manager.session
+    assert session == Session.get(Session.id == 1)
+    assert session.started is None
+    assert session.finished is None
+    assert session.priority == 0
+    assert session.status is None
+
+    # Check the transfers
+    assert isinstance(manager.transfers, list)
+    assert len(manager.transfers) == 0
+
+    # Check methods were injected into the remote client
+    for meth in (
+        "dt_upload",
+        "dt_do_upload",
+        "dt_link_blob_to_doc",
+        "dt_upload_chunks",
+    ):
+        assert hasattr(manager.remote, meth)
+
+    # Check priority changes
+    manager.increase_priority(session)
+    manager.increase_priority(session)
+    assert session.priority == 2
+    manager.decrease_priority(session)
+    assert session.priority == 1
+    manager.decrease_priority(session)
+    assert session.priority == 0
+    manager.decrease_priority(session)
+    assert session.priority == 0
+
+
+def test_manager_session_resume(tmp_path, manager):
+    """Test resuming a session with 1 transfer."""
+    paths, _ = create_tree(tmp_path)
+    remote_path = "/default-domain/something"
+
+    manager.add(paths[-1], remote_path)
+    assert isinstance(manager.transfers, list)
+    assert len(manager.transfers) == 1
+
+    manager.reinit()
+
+    assert isinstance(manager.transfers, list)
+    assert len(manager.transfers) == 1
+    assert manager.transfers[0].local_path == paths[-1]
+    assert manager.session.id == 1
+
+    manager.add_all(paths, remote_path)
+    assert isinstance(manager.transfers, list)
+    assert len(manager.transfers) == len(paths)
+
+
+def test_manager_with_transfers(tmp_path, manager):
+    paths, size = create_tree(tmp_path)
+    remote_path = "/default-domain/something"
+
+    # Starting an empty session is useless
+    assert not manager.transfers
+    manager.start()
+
+    # Add several paths
+    manager.add_all(paths, remote_path)
+    assert len(manager.transfers) == len(paths)
+
+    # Ensure the total size is well-computed
+    assert manager.size == size
+
+    # Add again all paths, this should not affect the actual paths list
+    for path in paths:
+        assert manager.add(path, remote_path) is None
+
+    # Add a problematic path, this should not affect the actual paths list
+    assert manager.add(Path("inexistent"), remote_path) is None
+
+    # Check transfers
+    transfer = manager.transfers[0]
+    assert isinstance(transfer, Transfer)
+    assert transfer.session == Session.get(Session.id == 1)
+    assert isinstance(transfer.local_path, Path)
+    assert transfer.remote_path == remote_path
+    assert not transfer.is_file
+    assert not transfer.file_size
+    assert transfer.status is TransferStatus.ONGOING
+    assert isinstance(transfer.batch, Batch)
+    assert not transfer.remote_ref
+    assert not transfer.uploaded
+    assert not transfer.uploaded_size
+    assert not transfer.chunk_size
+    assert not transfer.replace_blob
+
+
+def test_manager_reset_transfers(tmp_path, manager):
+    paths, _ = create_tree(tmp_path)
+    remote_path = "/default-domain/something"
+
+    # Add several paths
+    manager.add_all(paths, remote_path)
+    assert len(manager.transfers) == len(paths)
+
+    # Now, reset the manager
+    assert manager.session == Session.get(Session.id == 1)
+    manager.reset()
+    # It should have no effect as the session is not finished
+    assert manager.session == Session.get(Session.id == 1)
+
+    # Mimic a finish session
+    manager.session.started = datetime.now()
+    manager.session.finished = datetime.now()
+    manager.session.status is TransferStatus.DONE
+
+    # And reset for good!
+    manager.reset()
+    # A new session is created
+    assert manager.session == Session.get(Session.id == 2)
+    assert not manager.transfers
+
+
+def test_manager_start_stop_transfers(
+    tmp_path, app, server, manager_factory, obj_factory
+):
+    manager, engine = manager_factory()
+    dt_manager = engine.dt_manager
+
+    # Create a remote folder for our tests
+    remote_folder = obj_factory()
+    # Grant RW ACLs for the current user
+    server.operations.execute(
+        command="Document.SetACE",
+        input_obj=remote_folder.uid,
+        user=engine.remote_user,
+        permission="ReadWrite",
+        grant=True,
+    )
+
+    # Add several paths
+    paths, size = create_tree(tmp_path)
+    dt_manager.add_all(paths, remote_folder.path)
+
+    # Stopping now has no effect
+    assert not dt_manager.is_started
+    dt_manager.stop()
+    assert not dt_manager.is_started
+
+    # Ensure session times are not set
+    assert not dt_manager.session.started
+    assert not dt_manager.session.finished
+    assert dt_manager.session.status is TransferStatus.ONGOING
+
+    # Simulate a stop() while working
+
+    def should_stop_at_0() -> bool:
+        """Stop at the start."""
+        # Ensure the session is started
+        assert dt_manager.is_started
+        assert isinstance(dt_manager.session.started, datetime)
+
+        # Check we cannot start again
+        dt_manager.start()
+
+        # Check we cannot reset
+        dt_manager.reset()
+
+        # Mimic a call to stop() from the outside
+        dt_manager.stop()
+        assert not dt_manager.is_started
+        return True
+
+    def should_stop_at_4() -> bool:
+        """Stop at the 4th upload."""
+        # Ensure the session is started
+        assert dt_manager.is_started
+        assert isinstance(dt_manager.session.started, datetime)
+
+        nonlocal count
+        count += 1
+        return count >= 4
+
+    with patch.object(dt_manager, "should_stop", new=should_stop_at_0):
+        dt_manager.start()
+    assert dt_manager.uploaded == 0
+    assert dt_manager.progress == 0.0
+
+    count = 0
+    with patch.object(dt_manager, "should_stop", new=should_stop_at_4):
+        dt_manager.start()
+    assert dt_manager.uploaded > 0
+    assert dt_manager.progress > 0.0
+
+    # The finished time is still not yet set
+    assert not dt_manager.session.finished
+
+    # Now, really process transfers
+    date_original = dt_manager.session.started
+    dt_manager.start()
+
+    # Check sizes
+    assert dt_manager.uploaded == size
+    assert dt_manager.progress == 100.0
+
+    # Ensure the started time is not altered
+    assert dt_manager.session.started == date_original
+
+    # Ensure the session is completed
+    assert not dt_manager.is_started
+    assert dt_manager.is_completed
+    assert isinstance(dt_manager.session.finished, datetime)
+    assert dt_manager.session.status is TransferStatus.DONE
+
+    # Trying to start again is useless
+    dt_manager.start()
+
+    # Finally, check the latest transfer (to test SQL models completely)
+    transfer = dt_manager.transfers[-1]
+    assert transfer.is_file
+    assert transfer.batch.uid
+    assert transfer == Transfer.get(Transfer.id == transfer.id)
diff --git a/nxdrive/direct_transfer/tests/integration/test_direct_transfer_session.py b/nxdrive/direct_transfer/tests/integration/test_direct_transfer_session.py
new file mode 100644
index 000000000..96ad7b908
--- /dev/null
+++ b/nxdrive/direct_transfer/tests/integration/test_direct_transfer_session.py
@@ -0,0 +1,89 @@
+"""
+Test the Direct Transfer manager: session lifecycle.
+"""
+from uuid import uuid4
+
+import pytest
+from nxdrive.direct_transfer.constants import SessionState, SessionTerminationStatus
+from nxdrive.direct_transfer.manager import DirectTransferManager
+from nxdrive.direct_transfer.session import UploadSession, create_session, get_sessions
+
+
+class FakeRemote:
+    """Fake Remote client."""
+
+
+@pytest.fixture
+def manager(tmp_path):
+    uid = str(uuid4())
+    db = tmp_path / f"{uid}.db"
+    remote = FakeRemote()
+    yield DirectTransferManager(str(db), uid, remote)
+
+
+def test_empty_sessions(manager):
+    assert not get_sessions()
+
+
+def test_one_session(manager):
+    # Create one upload session
+    create_session(is_upload=True)
+    sessions = get_sessions()
+    assert len(sessions) == 1
+    assert isinstance(sessions[0], UploadSession)
+
+    # The manager was init before the new session, so it is empty
+    assert not manager.sessions
+    assert not manager.current_session
+
+    # Reload session, it should see the new one
+    manager.reload()
+    assert len(manager.sessions) == 1
+    assert isinstance(manager.current_session, UploadSession)
+
+
+def test_session_lifecycle(manager, tmp_path):
+    """Test the session lifecycle."""
+    session = create_session(is_upload=True)
+
+    # By default, this is a draft and it has no termination status
+    assert session.state is SessionState.DRAFT
+    assert session.status is SessionTerminationStatus.NONE
+
+    # Test forcing a state
+    session.change_state(SessionState.PENDING)
+    assert session.state is SessionState.PENDING
+    assert session.status is SessionTerminationStatus.NONE
+    # (reset the state for next tests)
+    session.change_state(SessionState.DRAFT)
+    assert session.state is SessionState.DRAFT
+
+    # The session cannot move forward if there is no transfers
+    session.change_state()
+    assert session.state is SessionState.DRAFT
+    assert session.status is SessionTerminationStatus.NONE
+
+    # Let's add a transfer
+    file = tmp_path / "foo.ods"
+    file.write_text("bla" * 42)
+    session.add(file, "/default-domain/workspaces")
+    # And let's say the user decided it is good to go
+    session.change_state()
+    assert session.state is SessionState.PENDING
+    assert session.status is SessionTerminationStatus.NONE
+    # (from here, the sessions manager will prioritize the session)
+
+    # Mimic other actions/states
+
+    # The user clicked on the pause button
+    session.pause()
+    session.change_state()
+    assert session.state is SessionState.SUSPENDED
+    assert session.status is SessionTerminationStatus.NONE
+    session.change_state(SessionState.PENDING)  # Reset
+
+    # The user clicked on the pause button
+    session.is_started = True
+    session.change_state()
+    assert session.state is SessionState.SUSPENDED
+    assert session.status is SessionTerminationStatus.NONE
diff --git a/nxdrive/direct_transfer/tests/integration/test_repository.py b/nxdrive/direct_transfer/tests/integration/test_repository.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/nxdrive/direct_transfer/tests/integration/test_uow.py b/nxdrive/direct_transfer/tests/integration/test_uow.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/nxdrive/direct_transfer/tests/unit/test_handlers.py b/nxdrive/direct_transfer/tests/unit/test_handlers.py
new file mode 100644
index 000000000..23fd3ea35
--- /dev/null
+++ b/nxdrive/direct_transfer/tests/unit/test_handlers.py
@@ -0,0 +1,114 @@
+from datetime import date
+
+import pytest
+from nxdrive.direct_transfer import bootstrap
+from nxdrive.direct_transfer.adapters import repository
+from nxdrive.direct_transfer.domain import commands
+from nxdrive.direct_transfer.service_layer import handlers, unit_of_work
+
+
+class FakeRepository(repository.AbstractRepository):
+    def __init__(self, products):
+        super().__init__()
+        self._products = set(products)
+
+    def _add(self, product):
+        self._products.add(product)
+
+    def _get(self, sku):
+        return next((p for p in self._products if p.sku == sku), None)
+
+    def _get_by_batchref(self, batchref):
+        return next(
+            (p for p in self._products for b in p.batches if b.reference == batchref),
+            None,
+        )
+
+
+class FakeUnitOfWork(unit_of_work.AbstractUnitOfWork):
+    def __init__(self):
+        self.products = FakeRepository([])
+        self.committed = False
+
+    def _commit(self):
+        self.committed = True
+
+    def rollback(self):
+        pass
+
+
+def bootstrap_test_app():
+    return bootstrap.bootstrap(
+        start_orm=False,
+        uow=FakeUnitOfWork(),
+        notifications=lambda *args: None,
+        publish=lambda *args: None,
+    )
+
+
+class TestAddBatch:
+    def test_for_new_product(self):
+        bus = bootstrap_test_app()
+        bus.handle(commands.CreateBatch("b1", "CRUNCHY-ARMCHAIR", 100, None))
+        assert bus.uow.products.get("CRUNCHY-ARMCHAIR") is not None
+        assert bus.uow.committed
+
+    def test_for_existing_product(self):
+        bus = bootstrap_test_app()
+        bus.handle(commands.CreateBatch("b1", "GARISH-RUG", 100, None))
+        bus.handle(commands.CreateBatch("b2", "GARISH-RUG", 99, None))
+        assert "b2" in [b.reference for b in bus.uow.products.get("GARISH-RUG").batches]
+
+
+class TestAllocate:
+    def test_allocates(self):
+        bus = bootstrap_test_app()
+        bus.handle(commands.CreateBatch("batch1", "COMPLICATED-LAMP", 100, None))
+        bus.handle(commands.Allocate("o1", "COMPLICATED-LAMP", 10))
+        [batch] = bus.uow.products.get("COMPLICATED-LAMP").batches
+        assert batch.available_quantity == 90
+
+    def test_errors_for_invalid_sku(self):
+        bus = bootstrap_test_app()
+        bus.handle(commands.CreateBatch("b1", "AREALSKU", 100, None))
+
+        with pytest.raises(handlers.InvalidSku, match="Invalid sku NONEXISTENTSKU"):
+            bus.handle(commands.Allocate("o1", "NONEXISTENTSKU", 10))
+
+    def test_commits(self):
+        bus = bootstrap_test_app()
+        bus.handle(commands.CreateBatch("b1", "OMINOUS-MIRROR", 100, None))
+        bus.handle(commands.Allocate("o1", "OMINOUS-MIRROR", 10))
+        assert bus.uow.committed
+
+
+class TestChangeBatchQuantity:
+    def test_changes_available_quantity(self):
+        bus = bootstrap_test_app()
+        bus.handle(commands.CreateBatch("batch1", "ADORABLE-SETTEE", 100, None))
+        [batch] = bus.uow.products.get(sku="ADORABLE-SETTEE").batches
+        assert batch.available_quantity == 100
+
+        bus.handle(commands.ChangeBatchQuantity("batch1", 50))
+        assert batch.available_quantity == 50
+
+    def test_reallocates_if_necessary(self):
+        bus = bootstrap_test_app()
+        history = [
+            commands.CreateBatch("batch1", "INDIFFERENT-TABLE", 50, None),
+            commands.CreateBatch("batch2", "INDIFFERENT-TABLE", 50, date.today()),
+            commands.Allocate("order1", "INDIFFERENT-TABLE", 20),
+            commands.Allocate("order2", "INDIFFERENT-TABLE", 20),
+        ]
+        for msg in history:
+            bus.handle(msg)
+        [batch1, batch2] = bus.uow.products.get(sku="INDIFFERENT-TABLE").batches
+        assert batch1.available_quantity == 10
+        assert batch2.available_quantity == 50
+
+        bus.handle(commands.ChangeBatchQuantity("batch1", 25))
+
+        # order1 or order2 will be deallocated, so we'll have 25 - 20
+        assert batch1.available_quantity == 5
+        # and 20 will be reallocated to the next batch
+        assert batch2.available_quantity == 30
diff --git a/nxdrive/direct_transfer/tests/unit/test_transfer.py b/nxdrive/direct_transfer/tests/unit/test_transfer.py
new file mode 100644
index 000000000..76bb47a81
--- /dev/null
+++ b/nxdrive/direct_transfer/tests/unit/test_transfer.py
@@ -0,0 +1,37 @@
+from nxdrive.direct_transfer.domain.model import Transfer, TransferItem
+
+
+def test_transfer_add_items():
+    transfer = Transfer(42)
+    item1 = TransferItem(0, "/foo")
+    item2 = TransferItem(1, "/foo")
+    item1_dupe = TransferItem(0, "/foo")
+    transfer.add(item1)
+    transfer.add(item1_dupe)
+    transfer.add(item2)
+    assert transfer._items == {item1, item2}
+
+
+def test_transfer_remove_items():
+    transfer = Transfer(42)
+    item1 = TransferItem(0, "/foo")
+    item2 = TransferItem(1, "/foo")
+    transfer.add(item1)
+    transfer.add(item2)
+    transfer.remove(item1)
+    assert transfer._items == {item2}
+
+
+def test_transfer_with_same_priorities():
+    transfer1 = Transfer(0)
+    transfer2 = Transfer(1)
+    transfer3 = Transfer(2)
+    assert transfer1.priority == transfer2.priority == transfer3.priority
+    assert transfer1 < transfer2 < transfer3
+
+
+def test_transfer_with_different_priorities():
+    transfer1 = Transfer(0, priority=1)
+    transfer2 = Transfer(1, priority=2)
+    transfer3 = Transfer(2, priority=0)
+    assert transfer3 < transfer1 < transfer2
diff --git a/nxdrive/direct_transfer/tests/unit/test_transfer_item_action.py b/nxdrive/direct_transfer/tests/unit/test_transfer_item_action.py
new file mode 100644
index 000000000..666fa86cd
--- /dev/null
+++ b/nxdrive/direct_transfer/tests/unit/test_transfer_item_action.py
@@ -0,0 +1,29 @@
+from nxdrive.direct_transfer.domain.model import TransferItemAction
+from nxdrive.direct_transfer.transfer import add_uploads
+
+
+def test_add_uploads_one_file(tmp_path):
+    file = tmp_path / "file.txt"
+    file.touch()
+
+    expected = [TransferItemAction("UP", file, "/remote-path")]
+    got = list(add_uploads(file, "/remote-path"))
+    assert got == expected
+
+
+def test_add_uploads_recursive(tmp_path):
+    file = tmp_path / "file.txt"
+    file.touch()
+    subfolder = tmp_path / "subfolder"
+    subfolder.mkdir()
+    subfile = tmp_path / "subfolder" / "file.txt"
+    subfile.touch()
+
+    expected = [
+        TransferItemAction("UP", tmp_path, "/remote-path"),
+        TransferItemAction("UP", file, f"/remote-path/{tmp_path.name}"),
+        TransferItemAction("UP", subfolder, f"/remote-path/{tmp_path.name}"),
+        TransferItemAction("UP", subfile, f"/remote-path/{tmp_path.name}/subfolder"),
+    ]
+    got = list(add_uploads(tmp_path, "/remote-path"))
+    assert got == expected
diff --git a/nxdrive/direct_transfer/transfer.py b/nxdrive/direct_transfer/transfer.py
new file mode 100644
index 000000000..0b90c0d56
--- /dev/null
+++ b/nxdrive/direct_transfer/transfer.py
@@ -0,0 +1,20 @@
+from logging import getLogger
+from pathlib import Path
+from typing import Generator
+
+from ..utils import get_tree_list
+from .domain import model
+
+log = getLogger(__name__)
+
+
+def add_uploads(
+    local_path: Path, remote_path: str
+) -> Generator[model.TransferItemAction, None, None]:
+    """Recursively plan *local_paths* upload to the given *remote_path*."""
+    if local_path.is_file():
+        yield model.TransferItemAction("UP", local_path, remote_path)
+    elif local_path.is_dir():
+        tree = get_tree_list(local_path, remote_path)
+        for path, computed_remote_path in sorted(tree):
+            yield model.TransferItemAction("UP", path, computed_remote_path)
diff --git a/nxdrive/engine/dao/sqlite.py b/nxdrive/engine/dao/sqlite.py
index 5cd9da4f6..b8e171231 100644
--- a/nxdrive/engine/dao/sqlite.py
+++ b/nxdrive/engine/dao/sqlite.py
@@ -115,9 +115,6 @@
     ("unsynchronized", "moved"): "unsynchronized",
     ("unsynchronized", "synchronized"): "unsynchronized",
     ("unsynchronized", "deleted"): "remotely_deleted",
-    # Direct Transfer
-    ("direct", "unknown"): "direct_transfer",
-    ("direct", "deleted"): "direct_transfer_replace_blob",
 }
 
 
@@ -630,7 +627,7 @@ def __init__(self, db: Path) -> None:
         self.reinit_processors()
 
     def get_schema_version(self) -> int:
-        return 10
+        return 11
 
     def _migrate_state(self, cursor: Cursor) -> None:
         try:
@@ -656,6 +653,7 @@ def _migrate_db(self, cursor: Cursor, version: int) -> None:
                 "   AND folderish = 0"
             )
             self.store_int(SCHEMA_VERSION, 1)
+
         if version < 2:
             cursor.execute(
                 "CREATE TABLE if not exists ToRemoteScan ("
@@ -664,16 +662,20 @@ def _migrate_db(self, cursor: Cursor, version: int) -> None:
                 ")"
             )
             self.store_int(SCHEMA_VERSION, 2)
+
         if version < 3:
             self._migrate_state(cursor)
             self.store_int(SCHEMA_VERSION, 3)
+
         if version < 4:
             self._migrate_state(cursor)
             cursor.execute("UPDATE States SET creation_date = last_remote_updated")
             self.store_int(SCHEMA_VERSION, 4)
+
         if version < 5:
             self._create_transfer_tables(cursor)
             self.store_int(SCHEMA_VERSION, 5)
+
         if version < 6:
             # Add the *filesize* field to the Downloads table,
             # used to display download metrics in the systray menu.
@@ -686,6 +688,7 @@ def _migrate_db(self, cursor: Cursor, version: int) -> None:
                 # so we can bypass the error
                 pass
             self.store_int(SCHEMA_VERSION, 6)
+
         if version < 7:
             # Remove the no-more-used *idx* field of Uploads.
             # SQLite does not support column deletion, we need to recreate
@@ -710,8 +713,8 @@ def _migrate_db(self, cursor: Cursor, version: int) -> None:
 
             # Delete the table
             cursor.execute("DROP TABLE Uploads_backup;")
-
             self.store_int(SCHEMA_VERSION, 7)
+
         if version < 8:
             if WINDOWS:
                 # Update the tmpname column to add the long path prefix on Windows
@@ -770,6 +773,27 @@ def _migrate_db(self, cursor: Cursor, version: int) -> None:
 
             self.store_int(SCHEMA_VERSION, 10)
 
+        if version < 11:
+            # With NXDRIVE-2006, Direct Transfer entries were removed from the sync database.
+            # Here we just do the cleanup without managing the conversion using the new database as
+            # the new database is only known from the Direct Transfer manager (outside of the current
+            # code scope) and it would require too many actions for a beta feature (at the time).
+            log.warning(
+                "Removing all Direct Transfers for the database migration, sorry :/"
+            )
+
+            for state in cursor.execute(
+                "SELECT * FROM States WHERE local_state = 'direct'"
+            ):
+                # Remove the eventual upload associated to the Direct Transfer ...
+                cursor.execute(
+                    "DELETE FROM Uploads WHERE path = ?", (state.local_path,)
+                )
+                # ... And remove the Direct Transfer
+                cursor.execute("DELETE FROM States WHERE id = ?", (state.id,))
+
+            self.store_int(SCHEMA_VERSION, 11)
+
     def _create_table(self, cursor: Cursor, name: str, force: bool = False) -> None:
         if name == "States":
             self._create_state_table(cursor, force)
@@ -1226,21 +1250,6 @@ def update_local_modification_time(self, row: DocPair, info: FileInfo) -> None:
                 (info.last_modification_time, row.id),
             )
 
-    def update_pair_state(self, row: DocPair) -> None:
-        """Update local, remote and pair states of a given *doc_pair*.
-        States should already be defined in the *doc_pair* attributes,
-        the goal is only to save them in the database.
-        """
-        row.pair_state = self._get_pair_state(row)
-
-        with self.lock:
-            con = self._get_write_connection()
-            c = con.cursor()
-            c.execute(
-                "UPDATE States SET local_state = ?, remote_state = ?, pair_state = ? WHERE id = ?",
-                (row.local_state, row.remote_state, row.pair_state, row.id),
-            )
-
     def get_valid_duplicate_file(self, digest: str) -> Optional[DocPair]:
         """Find a file already synced with the same digest as the given *digest*."""
         c = self._get_read_connection().cursor()
diff --git a/nxdrive/engine/engine.py b/nxdrive/engine/engine.py
index 13320d04e..e1f4e8581 100644
--- a/nxdrive/engine/engine.py
+++ b/nxdrive/engine/engine.py
@@ -28,6 +28,7 @@
     DelAction,
     TransferStatus,
 )
+from ..direct_transfer import DirectTransferManager
 from ..exceptions import (
     EngineInitError,
     InvalidDriveException,
@@ -41,7 +42,6 @@
     current_thread_id,
     find_icon,
     find_suitable_tmp_dir,
-    get_tree_list,
     if_frozen,
     safe_filename,
     safe_long_path,
@@ -102,9 +102,8 @@ class Engine(QObject):
     online = pyqtSignal()
 
     # Direct Transfer
-    directTranferDuplicateError = pyqtSignal(Path, Document)
+    directTransferDuplicateError = pyqtSignal(Path, Document)
     directTranferError = pyqtSignal(Path)
-    directTranferStatus = pyqtSignal(Path, bool)
 
     type = "NXDRIVE"
     # Folder locker - LocalFolder processor can prevent
@@ -187,6 +186,19 @@ def __init__(
         # Pause in case of no more space on the device
         self.noSpaceLeftOnDevice.connect(self.suspend)
 
+        # Set the Direct Transfer database file
+        db = self.manager.home / "direct_transfer" / f"{self.uid}.db"
+        db.parent.mkdir(exist_ok=True)
+
+        # And create the Direct Transfer manager
+        self.dt_manager = DirectTransferManager(
+            str(db),
+            self.uid,
+            self.remote,
+            done_callback=self.manager.directTransferStats.emit,
+            dupe_callback=self.directTransferDuplicateError.emit,
+        )
+
     def __repr__(self) -> str:
         return (
             f"<{type(self).__name__} "
@@ -399,70 +411,32 @@ def delete_doc(self, path: Path, mode: DelAction = None) -> None:
                     f"{doc_pair.remote_parent_path}/{doc_pair.remote_ref}"
                 )
 
-    def direct_transfer(self, local_paths: Set[Path], remote_ref: str) -> None:
+    def direct_transfer(
+        self, local_paths: Set[Path], remote_path: str, start: bool = True
+    ) -> None:
         """Plan the Direct Transfer."""
-        # self.directTranferStatus.emit(local_path[0], True)
-
-        def plan(path: Path, remote_uid: str) -> None:
-            """Actions to do (refactored in a function to prevent duplicate code between files and folders)."""
-            # Save the remote folder's reference into the file/folder xattrs
-            try:
-                self.local.set_remote_id(path, remote_uid)
-            except PermissionError:
-                log.warning(
-                    f"Cannot set the remote ID on {path!r}, skipping the upload"
-                )
-                return
-
-            # Add the path into the database to plan the upload
-            info = self.local.get_info(path, check=False)
-            self.dao.insert_local_state(info, parent_path=None, local_state="direct")
-
         # Save the remote location for next times
-        self.dao.update_config("dt_last_remote_location", remote_ref)
+        self.dao.update_config("dt_last_remote_location", remote_path)
 
-        for local_path in sorted(local_paths):
-            if local_path.is_file():
-                plan(local_path, remote_ref)
-            else:
-                tree = sorted(get_tree_list(local_path, remote_ref))
-                for path, remote_path in tree:
-                    plan(path, remote_path)
+        if self.dt_manager.is_completed:
+            # Create a new transfers session
+            self.dt_manager.reset()
+
+        # Add paths recursively and start uploading!
+        if local_paths:
+            self.dt_manager.add_all(local_paths, remote_path)
+            if start:
+                self.dt_manager.start()
 
     def direct_transfer_cancel(self, file: Path) -> None:
         """Cancel the Direct Transfer of the given local *file*."""
-        log.info(f"Direct Transfer of {file!r}, user choice: cancel the upload")
-
-        doc_pair = self.dao.get_state_from_local(file)
-        if not doc_pair:
-            # Magic teleportation?
-            log.warning("The doc pair disappeared?! Direct Transfer cancelled.")
-            return
-
-        # Cancel the upload, clean-up the database and local file
-        self.dao.remove_state(doc_pair)
-        self.local.remove_remote_id(file)
-        self.local.remove_remote_id(file, name="remote")
+        if self.dt_manager:
+            self.dt_manager.cancel(file)
 
     def direct_transfer_replace_blob(self, file: Path, doc: Document) -> None:
         """Replace the document's blob on the server."""
-        log.info(
-            f"Direct Transfer of {file!r}, user choice: "
-            f"replace the document's attached file (UID is {doc.uid!r})"
-        )
-
-        doc_pair = self.dao.get_state_from_local(file)
-        if not doc_pair:
-            # Magic teleportation?
-            log.warning("The doc pair disappeared?! Direct Transfer cancelled.")
-            return
-
-        # Plan the replacement of the document's blob on the server
-        doc_pair.remote_state = "deleted"
-        self.dao.update_pair_state(doc_pair)
-
-        # Repush the pair to be traited later
-        self.queue_manager.push(doc_pair)
+        if self.dt_manager:
+            self.dt_manager.replace_blob(file, doc.uid)
 
     def rollback_delete(self, path: Path) -> None:
         """ Re-synchronize a document when a deletion is cancelled. """
@@ -554,6 +528,8 @@ def resume(self) -> None:
             else:
                 thread.start()
         self.resume_suspended_transfers()
+        if self.dt_manager:
+            self.dt_manager.start()
         self.syncResumed.emit()
 
     def resume_transfer(self, nature: str, uid: int) -> None:
@@ -601,6 +577,8 @@ def suspend(self) -> None:
         log.info(f"Engine {self.uid} is suspending")
         self._pause = True
         self.dao.suspend_transfers()
+        if self.dt_manager:
+            self.dt_manager.stop()
         self.queue_manager.suspend()
         for thread in self._threads:
             thread.worker.suspend()
diff --git a/nxdrive/engine/processor.py b/nxdrive/engine/processor.py
index 151cd7b3e..ae921ab9a 100644
--- a/nxdrive/engine/processor.py
+++ b/nxdrive/engine/processor.py
@@ -22,7 +22,6 @@
 
 from ..behavior import Behavior
 from ..client.local import FileInfo
-from ..client.uploader.direct_transfer import DirectTransferUploader
 from ..constants import (
     CONNECTION_ERROR,
     LONG_FILE_ERRORS,
@@ -33,7 +32,6 @@
     TransferStatus,
 )
 from ..exceptions import (
-    DirectTransferDuplicateFoundError,
     DownloadPaused,
     DuplicationDisabledError,
     NotFound,
@@ -44,7 +42,6 @@
     UploadPaused,
 )
 from ..objects import DocPair, RemoteFileInfo
-from ..options import Options
 from ..utils import is_generated_tmp_file, lock_path, safe_filename, unlock_path
 from .workers import EngineWorker
 
@@ -314,16 +311,11 @@ def _execute(self) -> None:
                 self._postpone_pair(doc_pair, "MAX_RETRY_ERROR")
             except HTTPError as exc:
                 if exc.status == 404:
-                    if doc_pair.local_state == "direct":
-                        # Happening while Direct Transfer'ring a document,
-                        # that means the parent folder is not yet created. Postpone.
-                        self._postpone_pair(doc_pair, "Parent not yet synced")
-                    else:
-                        # We saw it happened once a migration is done.
-                        # Nuxeo kept the document reference but it does
-                        # not exist physically anywhere.
-                        log.info(f"The document does not exist anymore: {doc_pair!r}")
-                        self.dao.remove_state(doc_pair)
+                    # We saw it happened once a migration is done.
+                    # Nuxeo kept the document reference but it does
+                    # not exist physically anywhere.
+                    log.info(f"The document does not exist anymore: {doc_pair!r}")
+                    self.dao.remove_state(doc_pair)
                 elif exc.status == 409:  # Conflict
                     # It could happen on multiple files drag'n drop
                     # starting with identical characters.
@@ -401,23 +393,9 @@ def _execute(self) -> None:
                     self._postpone_pair(doc_pair, "Trashing not possible")
                 else:
                     self._handle_pair_handler_exception(doc_pair, handler_name, exc)
-            except DirectTransferDuplicateFoundError as exc:
-                # Ask the user what to do when a possible duplicate can be created by a Direct Transfer call
-                log.info(str(exc))
-                self.engine.directTranferDuplicateError.emit(exc.file, exc.doc)
             except Exception as exc:
                 # Workaround to forward unhandled exceptions to sys.excepthook between all Qthreads
                 sys.excepthook(*sys.exc_info())
-
-                # Show a notification for Direct Transfer errors
-                if doc_pair.pair_state.startswith("direct_transfer"):
-                    file = (
-                        doc_pair.local_path
-                        if WINDOWS
-                        else Path(f"/{doc_pair.local_path}")
-                    )
-                    self.engine.directTranferError.emit(file)
-
                 self._handle_pair_handler_exception(doc_pair, handler_name, exc)
             finally:
                 if soft_lock:
@@ -484,60 +462,6 @@ def _handle_pair_handler_exception(
             log.exception("Unknown error")
             self.increase_error(doc_pair, f"SYNC_HANDLER_{handler_name}", exception=e)
 
-    def _synchronize_direct_transfer(
-        self, doc_pair: DocPair, replace_blob: bool = False
-    ) -> None:
-        """Direct Transfer of a local file."""
-        if WINDOWS:
-            file = doc_pair.local_path
-        else:
-            # The path retrieved from the database will have its starting slash trimmed, restore it
-            file = Path(f"/{doc_pair.local_path}")
-
-        if not file.exists():
-            self.engine.directTranferError.emit(file)
-            log.warning(
-                f"Cancelling Direct Transfer of {file!r} because it does not exist anymore"
-            )
-            self.dao.remove_state(doc_pair)
-            self.dao.remove_transfer("upload", file)
-            return
-
-        # The remote path is stored as the remote ref in xattrs of the file
-        parent_path = self.local.get_remote_id(file)
-        if not parent_path:
-            self.engine.directTranferError.emit(file)
-            log.warning(
-                f"Cancelling Direct Transfer of {file!r} because it has no remote path set"
-            )
-            self.dao.remove_state(doc_pair)
-            self.dao.remove_transfer("upload", file)
-            return
-
-        # Do the upload
-        self.remote.upload(
-            file,
-            parent_path=parent_path,
-            engine_uid=self.engine.uid,
-            uploader=DirectTransferUploader,
-            replace_blob=replace_blob,
-        )
-
-        # Clean-up
-        self.dao.remove_state(doc_pair)
-        self.local.remove_remote_id(file)
-        self.local.remove_remote_id(file, name="remote")
-
-        # Display a notification only for big files (to prevent notifications flood)
-        if not doc_pair.folderish and doc_pair.size >= Options.big_file * 1024 * 1024:
-            self.engine.directTranferStatus.emit(file, False)
-
-        self.engine.manager.directTransferStats.emit(doc_pair.folderish, doc_pair.size)
-
-    def _synchronize_direct_transfer_replace_blob(self, doc_pair: DocPair) -> None:
-        """Force the blob replacement of the remote document (choice done by the user)."""
-        self._synchronize_direct_transfer(doc_pair, replace_blob=True)
-
     def _synchronize_conflicted(self, doc_pair: DocPair) -> None:
         if doc_pair.local_state == "moved" and doc_pair.remote_state in (
             "moved",
diff --git a/nxdrive/engine/queue_manager.py b/nxdrive/engine/queue_manager.py
index 3ea1b8b51..dc29e30b9 100644
--- a/nxdrive/engine/queue_manager.py
+++ b/nxdrive/engine/queue_manager.py
@@ -208,13 +208,6 @@ def push(self, state: Union[DocPair, QueueItem]) -> None:
                     f"{self._remote_file_queue.qsize()}"
                 )
             self.newItem.emit(row_id)
-        elif state.pair_state.startswith("direct_transfer"):
-            self._local_file_queue.put(state)
-            log.debug(
-                "Pushed to _local_file_queue, now of size: "
-                f"{self._local_file_queue.qsize()}"
-            )
-            self.newItem.emit(row_id)
         else:
             # deleted and conflicted
             log.info(f"Not processable state: {state!r}")
diff --git a/nxdrive/exceptions.py b/nxdrive/exceptions.py
index bf18737ef..9058bcbf5 100644
--- a/nxdrive/exceptions.py
+++ b/nxdrive/exceptions.py
@@ -1,9 +1,6 @@
 # coding: utf-8
-from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
-from nuxeo.models import Document
-
 if TYPE_CHECKING:
     from .engine.engine import Engine  # noqa
 
@@ -14,26 +11,6 @@ class DriveError(Exception):
     pass
 
 
-class DirectTransferDuplicateFoundError(ValueError):
-    """
-    Exception raised when a duplicate file already exists on the server
-    and trying to Direct Transfer a local file with the same name.
-    """
-
-    def __init__(self, file: Path, doc: Document) -> None:
-        self.file = file
-        self.doc = doc
-
-    def __repr__(self) -> str:
-        return (
-            f"Document with the name {self.file.name!r} already found on the server: {self.doc}."
-            f"Direct Transfer of {self.file!r} postponed after the user decided what to do."
-        )
-
-    def __str__(self) -> str:
-        return repr(self)
-
-
 class DocumentAlreadyLocked(DriveError):
     """ In Direct Edit, a document is locked by someone else. """
 
diff --git a/nxdrive/gui/application.py b/nxdrive/gui/application.py
index a875653ac..37e0c7a95 100644
--- a/nxdrive/gui/application.py
+++ b/nxdrive/gui/application.py
@@ -845,7 +845,7 @@ def auth() -> None:
 
     @pyqtSlot(object)
     def _connect_engine(self, engine: Engine) -> None:
-        engine.directTranferDuplicateError.connect(
+        engine.directTransferDuplicateError.connect(
             self._direct_transfer_duplicate_error
         )
         engine.syncStarted.connect(self.change_systray_icon)
diff --git a/nxdrive/logging_config.py b/nxdrive/logging_config.py
index d8c7c5885..549e61085 100644
--- a/nxdrive/logging_config.py
+++ b/nxdrive/logging_config.py
@@ -193,6 +193,9 @@ def configure(
     # NXDRIVE-1918: filter out botocore logging (too verbose)
     logging.getLogger("botocore").setLevel(logging.ERROR)
 
+    # NXDRIVE-2006: filter out peewee logging (too verbose)
+    logging.getLogger("peewee").setLevel(logging.ERROR)
+
 
 def get_handler(name: str) -> Optional[logging.Handler]:
     for handler in logging.getLogger().handlers:
diff --git a/nxdrive/manager.py b/nxdrive/manager.py
index a003c29ac..9142931e2 100644
--- a/nxdrive/manager.py
+++ b/nxdrive/manager.py
@@ -407,6 +407,7 @@ def stop(self, euid: str = None) -> None:
                 continue
             if engine.is_started():
                 engine.stop()
+            engine.dt_manager.shutdown()
         self.osi.cleanup()
         self.dispose_db()
         self.stopped.emit()
diff --git a/nxdrive/notification.py b/nxdrive/notification.py
index ba7fb9ebb..f67a1ff2f 100644
--- a/nxdrive/notification.py
+++ b/nxdrive/notification.py
@@ -435,21 +435,6 @@ def __init__(self, file: Path) -> None:
         )
 
 
-class DirectTransferStatus(Notification):
-    """Display a notification when a Direct Transfer is starting or finished."""
-
-    def __init__(self, file: Path, status: bool) -> None:
-        label = "DIRECT_TRANSFER_START" if status else "DIRECT_TRANSFER_END"
-        values = [file.name]
-        super().__init__(
-            label,
-            title="Direct Transfer",
-            description=Translator.get(label, values=values),
-            level=Notification.LEVEL_INFO,
-            flags=Notification.FLAG_VOLATILE | Notification.FLAG_BUBBLE,
-        )
-
-
 class ErrorOpenedFile(Notification):
     def __init__(self, path: str, is_folder: bool) -> None:
         values = [short_name(path), APP_NAME]
@@ -518,16 +503,11 @@ def _connect_engine(self, engine: "Engine") -> None:
         engine.errorOpenedFile.connect(self._errorOpenedFile)
         engine.longPathError.connect(self._longPathError)
         engine.directTranferError.connect(self._direct_transfer_error)
-        engine.directTranferStatus.connect(self._direct_transfer_status)
 
     def _direct_transfer_error(self, file: Path) -> None:
         """Display a notification when a Direct Transfer is in error."""
         self.send_notification(DirectTransferError(file))
 
-    def _direct_transfer_status(self, file: Path, status: bool) -> None:
-        """Display a notification when a Direct Transfer is starting or finished."""
-        self.send_notification(DirectTransferStatus(file, status))
-
     def _errorOpenedFile(self, doc: DocPair) -> None:
         self.send_notification(ErrorOpenedFile(str(doc.local_path), doc.folderish))
 
diff --git a/nxdrive/report.py b/nxdrive/report.py
index cd3a16fef..a023941f6 100644
--- a/nxdrive/report.py
+++ b/nxdrive/report.py
@@ -71,6 +71,28 @@ def copy_logs(self, myzip: ZipFile) -> None:
             except Exception:
                 log.exception(f"Impossible to copy the log {rel_path!r}")
 
+    def copy_direct_transfer_dbs(self, myzip: ZipFile) -> None:
+        """Copy all log Direct Transfer databases to the ZIP report."""
+
+        folder = self._manager.home / "direct_transfer"
+        if not folder.is_dir():
+            return
+
+        for path in folder.iterdir():
+            if not path.is_file():
+                continue
+            if path.suffix != ".db":
+                continue
+
+            comp = ZIP_DEFLATED
+            rel_path = path.relative_to(self._manager.home)
+            try:
+                myzip.write(str(path), str(rel_path), compress_type=comp)
+            except Exception:
+                log.exception(
+                    f"Impossible to copy the Direct Transfer database {rel_path!r}"
+                )
+
     @staticmethod
     def copy_db(myzip: ZipFile, dao: "EngineDAO") -> None:
         """
@@ -125,6 +147,9 @@ def generate(self) -> None:
             for engine in self._manager.engines.values():
                 self.copy_db(zip_, engine.dao)
 
+            # Direct Transfer databases
+            self.copy_direct_transfer_dbs(zip_)
+
             # Logs
             self.copy_logs(zip_)
 
diff --git a/tests/conftest.py b/tests/conftest.py
index 367bab260..db03c2453 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -81,6 +81,9 @@ def no_warnings(recwarn):
         elif "(rm_rf) error removing" in message:
             # First appeared with pytest 5.4.1
             continue
+        elif "resolve package from __spec__ or __package__" in message:
+            # Not related to our stuff (Peewee)
+            continue
 
         warn = f"{warning.filename}:{warning.lineno} {message}"
         print(warn, file=sys.stderr)
diff --git a/tests/functional/conftest.py b/tests/functional/conftest.py
index ef18d1e20..6671c7cfa 100644
--- a/tests/functional/conftest.py
+++ b/tests/functional/conftest.py
@@ -1,7 +1,6 @@
 # coding: utf-8
 from logging import getLogger
 from pathlib import Path
-from random import randint
 from typing import Callable, Optional
 from uuid import uuid4
 
@@ -12,6 +11,7 @@
 from nxdrive.manager import Manager
 
 from .. import env
+from ..utils import salt
 
 log = getLogger(__name__)
 
@@ -83,7 +83,8 @@ def user_factory(request, server, faker):
 
     def _make_user(username: str = "", password: str = "Administrator"):
         first_name, last_name = fake.name().split(" ", 1)
-        username = username or f"{first_name.lower()}-{randint(1, 99_999)}"
+        username = username or first_name.lower()
+        username = salt(username)
         properties = {
             "lastName": last_name,
             "firstName": first_name,
@@ -117,6 +118,7 @@ def _make(
         enable_sync: bool = False,
     ):
         title = title or str(uuid4())
+        title = salt(title)
         new = Document(name=title, type=nature, properties={"dc:title": title})
         obj = server.documents.create(new, parent_path=env.WS_DIR)
         request.addfinalizer(obj.delete)
diff --git a/tests/functional/test_direct_edit.py b/tests/functional/test_direct_edit.py
index f5b083f07..c316e23cd 100644
--- a/tests/functional/test_direct_edit.py
+++ b/tests/functional/test_direct_edit.py
@@ -20,6 +20,7 @@ def __init__(self, url: str, user: str):
         self._user = user
         self._stopped = True
         self._invalid_credentials = False
+        self.dt_manager = None
 
     def get_binder(self):
         """
diff --git a/tests/functional/test_remote_client.py b/tests/functional/test_remote_client.py
index f342b07e0..c7d5b6e35 100644
--- a/tests/functional/test_remote_client.py
+++ b/tests/functional/test_remote_client.py
@@ -6,10 +6,10 @@
 @pytest.mark.parametrize(
     "username",
     [
-        "ndt-Alice",
-        "ndt-bob@bar.com",
-        # "ndt-éléonor",
-        # "ndt-東京スカイツリー",
+        "Alice",
+        "bob@bar.com",
+        # "éléonor",
+        # "東京スカイツリー",
     ],
 )
 def test_personal_space(manager_factory, tmp, nuxeo_url, user_factory, username):
diff --git a/tests/old_functional/direct_transfer/test_signals.py b/tests/old_functional/direct_transfer/test_signals.py
new file mode 100644
index 000000000..149e150e0
--- /dev/null
+++ b/tests/old_functional/direct_transfer/test_signals.py
@@ -0,0 +1,149 @@
+"""
+Test the Direct Transfer feature.
+This is not a simple copy of the same test file from functional test.
+Here, we test signals transition from within the application.
+"""
+from contextlib import suppress
+from shutil import copyfile
+from time import sleep
+from unittest.mock import patch
+from uuid import uuid4
+
+from nxdrive.options import Options
+
+from .. import ensure_no_exception
+from .common import OneUserTest
+
+
+class TestDirectTransfer(OneUserTest):
+    def setUp(self):
+        # No sync root, to ease testing
+        self.remote_1.unregister_as_root(self.workspace)
+        self.engine_1.start()
+        self.engine_1.direct_transfer([], "")
+        self.dt_manager = self.engine_1.dt_manager
+
+        # Lower chunk_* options to have chunked uploads without having to create big files
+        self.default_chunk_limit = Options.chunk_limit
+        self.default_chunk_size = Options.chunk_size
+        Options.chunk_limit = 1
+        Options.chunk_size = 1
+
+        # The file used for the Direct Transfer (must be > 1 MiB)
+        source = (
+            self.location / "resources" / "databases" / "engine_migration_duplicate.db"
+        )
+        assert source.stat().st_size > 1024 * 1024
+
+        # Work with a copy of the file to allow parallel testing
+        self.file = self.tmpdir / f"{uuid4()}.bin"
+        copyfile(source, self.file)
+
+    def tearDown(self):
+        # Restore options
+        Options.chunk_limit = self.default_chunk_limit
+        Options.chunk_size = self.default_chunk_size
+
+        # Disconnect eventual signals to prevent failures when tests are run in parallel
+        with suppress(TypeError):
+            self.engine_1.directTransferDuplicateError.disconnect(
+                self.app.user_choice_cancel
+            )
+        with suppress(TypeError):
+            self.engine_1.directTransferDuplicateError.disconnect(
+                self.app.user_choice_replace
+            )
+
+    def wait_sync(self) -> None:
+        """Wait for the Direct transfer session to finish."""
+        for _ in range(30):  # 30 sec maxi
+            if not self.dt_manager.is_started:
+                break
+            sleep(1)
+
+    def sync_and_check(self, should_have_blob: bool = True) -> None:
+        # Let some time for the signal to transit (as this is async, the test would fail)
+        sleep(3)
+
+        # Sync
+        self.wait_sync()
+
+        # Check the error count
+        assert not self.engine_1.dao.get_errors(limit=0)
+
+        # Check the file exists on the server and has a blob attached
+        blob = self.has_blob()
+        assert blob is should_have_blob
+
+    def has_blob(self) -> bool:
+        """Check that *self.file* exists on the server and has a blob attached.
+        As when doing a Direct Transfer, the document is first created on the server,
+        this is the only way to check if the blob upload has been finished successfully.
+        """
+        try:
+            doc = self.root_remote.documents.get(
+                path=f"{self.ws.path}/{self.file.name}"
+            )
+        except Exception:
+            return False
+        return bool(doc.properties.get("file:content"))
+
+    def test_duplicate_file_cancellation(self):
+        """The file already exists on the server and has a blob attached.
+        Then, the user wants to cancel the transfer.
+        """
+
+        # Mimic the user clicking on "Cancel"
+        self.engine_1.directTransferDuplicateError.connect(self.app.user_choice_cancel)
+
+        def upload(*_, **__):
+            """Patch Remote.upload() to be able to check that nothing will be uploaded."""
+            assert 0, "No twice upload should be done!"
+
+        with ensure_no_exception():
+            # 1st upload: OK
+            self.engine_1.direct_transfer([self.file], self.ws.path)
+            self.sync_and_check()
+
+            # 2nd upload: it should be cancelled by the user
+            with patch.object(self.dt_manager.remote, "upload", new=upload):
+                self.engine_1.direct_transfer([self.file], self.ws.path)
+                self.sync_and_check()
+
+        # Ensure the signal was emitted
+        assert self.app.emitted
+
+        # Ensure there is only 1 document on the server
+        children = self.remote_document_client_1.get_children_info(self.workspace)
+        assert len(children) == 1
+        assert children[0].name == self.file.name
+
+    def test_duplicate_file_replace_blob(self):
+        """The file already exists on the server and has a blob attached.
+        Then, the user wants to replace the blob.
+        """
+
+        # Mimic the user clicking on "Replace"
+        self.engine_1.directTransferDuplicateError.connect(self.app.user_choice_replace)
+
+        with ensure_no_exception():
+            # 1st upload: OK
+            self.engine_1.direct_transfer([self.file], self.ws.path)
+            self.sync_and_check()
+
+            # To ease testing, we change local file content
+            self.file.write_bytes(b"blob changed!")
+
+            # 2nd upload: the blob should be replaced on the server
+            self.engine_1.direct_transfer([self.file], self.ws.path)
+            self.sync_and_check()
+
+        # Ensure the signal was emitted
+        assert self.app.emitted
+
+        # Ensure there is only 1 document on the server
+        children = self.engine_1.remote.documents.get_children(path=self.ws.path)
+        assert len(children) == 1
+        assert children[0].title == self.file.name
+        # Ensure the blob content was updated
+        assert self.engine_1.remote.get_blob(children[0].uid) == b"blob changed!"
diff --git a/tests/old_functional/test_direct_transfer.py b/tests/old_functional/test_direct_transfer.py
deleted file mode 100644
index c7f7b88d9..000000000
--- a/tests/old_functional/test_direct_transfer.py
+++ /dev/null
@@ -1,636 +0,0 @@
-"""
-Test the Direct Transfer feature in different scenarii.
-"""
-from contextlib import suppress
-from shutil import copyfile, copytree
-from time import sleep
-from unittest.mock import patch
-from uuid import uuid4
-
-import pytest
-from nuxeo.exceptions import HTTPError
-from nuxeo.models import Document
-from nxdrive.client.uploader.direct_transfer import DirectTransferUploader
-from nxdrive.constants import TransferStatus
-from nxdrive.options import Options
-from nxdrive.utils import get_tree_list
-from requests.exceptions import ConnectionError
-
-from .. import ensure_no_exception
-from ..markers import not_windows
-from .common import OneUserNoSync, OneUserTest
-
-
-class DirectTransfer:
-    def setUp(self):
-        # No sync root, to ease testing
-        self.remote_1.unregister_as_root(self.workspace)
-        self.engine_1.start()
-
-        # Lower chunk_* options to have chunked uploads without having to create big files
-        self.default_chunk_limit = Options.chunk_limit
-        self.default_chunk_size = Options.chunk_size
-        Options.chunk_limit = 1
-        Options.chunk_size = 1
-
-        # The file used for the Direct Transfer (must be > 1 MiB)
-        source = (
-            self.location / "resources" / "databases" / "engine_migration_duplicate.db"
-        )
-        assert source.stat().st_size > 1024 * 1024
-
-        # Work with a copy of the file to allow parallel testing
-        self.file = self.tmpdir / f"{uuid4()}.bin"
-        copyfile(source, self.file)
-
-    def tearDown(self):
-        # Restore options
-        Options.chunk_limit = self.default_chunk_limit
-        Options.chunk_size = self.default_chunk_size
-
-        # Disconnect eventual signals to prevent failures when tests are run in parallel
-        with suppress(TypeError):
-            self.engine_1.directTranferDuplicateError.disconnect(
-                self.app.user_choice_cancel
-            )
-        with suppress(TypeError):
-            self.engine_1.directTranferDuplicateError.disconnect(
-                self.app.user_choice_replace
-            )
-
-    def has_blob(self) -> bool:
-        """Check that *self.file* exists on the server and has a blob attached.
-        As when doing a Direct Transfer, the document is first created on the server,
-        this is the only way to check if the blob upload has been finished successfully.
-        """
-        try:
-            doc = self.root_remote.documents.get(
-                path=f"{self.ws.path}/{self.file.name}"
-            )
-        except Exception:
-            return False
-        return bool(doc.properties.get("file:content"))
-
-    def no_uploads(self) -> bool:
-        """Check there is no ongoing uploads."""
-        assert not any(
-            u for u in self.engine_1.dao.get_uploads() if u.path == self.file
-        )
-
-    def sync_and_check(self, should_have_blob: bool = True) -> None:
-        # Let time for uploads to be planned
-        sleep(3)
-
-        # Sync
-        self.wait_sync()
-
-        # Check the error count
-        assert not self.engine_1.dao.get_errors(limit=0)
-
-        # Check the uploads count
-        assert not list(self.engine_1.dao.get_uploads())
-
-        # Check the file exists on the server and has a blob attached
-        if should_have_blob:
-            assert self.has_blob()
-        else:
-            assert not self.has_blob()
-
-    def test_with_engine_not_started(self):
-        """A Direct Transfer should work even if engines are stopped."""
-        pytest.xfail("Waiting for NXDRIVE-1910")
-
-        engine = self.engine_1
-        engine.stop()
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with ensure_no_exception():
-            engine.direct_transfer([self.file], self.ws.path)
-            self.sync_and_check()
-
-    def test_duplicate_file_but_no_blob_attached(self):
-        """The file already exists on the server but has no blob attached yet."""
-
-        # Create the document on the server, with no blob attached
-        new_doc = Document(
-            name=self.file.name, type="File", properties={"dc:title": self.file.name}
-        )
-        doc = self.root_remote.documents.create(new_doc, parent_path=self.ws.path)
-        assert doc.properties.get("file:content") is None
-
-        with ensure_no_exception():
-            # The upload should work: the doc will be retrieved and the blob uploaded and attached
-            self.engine_1.direct_transfer([self.file], self.ws.path)
-            self.sync_and_check()
-
-        # Ensure there is only 1 document on the server
-        children = self.remote_document_client_1.get_children_info(self.workspace)
-        assert len(children) == 1
-        assert children[0].name == self.file.name
-
-    def test_duplicate_file_cancellation(self):
-        """The file already exists on the server and has a blob attached.
-        Then, the user wants to cancel the transfer.
-        """
-
-        # Mimic the user clicking on "Cancel"
-        self.engine_1.directTranferDuplicateError.connect(self.app.user_choice_cancel)
-
-        class NoChunkUpload(DirectTransferUploader):
-            def upload_chunks(self, *_, **__):
-                """Patch Remote.upload() to be able to check that nothing will be uploaded."""
-                assert 0, "No twice upload should be done!"
-
-        def upload(*args, **kwargs):
-            """Set our specific uploader to check for twice upload."""
-            kwargs.pop("uploader")
-            return upload_orig(*args, uploader=NoChunkUpload, **kwargs)
-
-        engine = self.engine_1
-        upload_orig = engine.remote.upload
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with ensure_no_exception():
-            # 1st upload: OK
-            engine.direct_transfer([self.file], self.ws.path)
-            self.sync_and_check()
-
-            # 2nd upload: it should be cancelled by the user
-            with patch.object(engine.remote, "upload", new=upload):
-                engine.direct_transfer([self.file], self.ws.path)
-                self.sync_and_check()
-
-        # Ensure the signal was emitted
-        assert self.app.emitted
-
-        # Ensure there is only 1 document on the server
-        children = self.remote_document_client_1.get_children_info(self.workspace)
-        assert len(children) == 1
-        assert children[0].name == self.file.name
-
-    def test_duplicate_file_replace_blob(self):
-        """The file already exists on the server and has a blob attached.
-        Then, the user wants to replace the blob.
-        """
-
-        # Mimic the user clicking on "Replace"
-        self.engine_1.directTranferDuplicateError.connect(self.app.user_choice_replace)
-
-        with ensure_no_exception():
-            # 1st upload: OK
-            self.engine_1.direct_transfer([self.file], self.ws.path)
-            self.sync_and_check()
-
-            # To ease testing, we change local file content
-            self.file.write_bytes(b"blob changed!")
-
-            # 2nd upload: the blob should be replaced on the server
-            self.engine_1.direct_transfer([self.file], self.ws.path)
-            self.sync_and_check()
-
-        # Ensure the signal was emitted
-        assert self.app.emitted
-
-        # Ensure there is only 1 document on the server
-        children = self.remote_document_client_1.get_children_info(self.workspace)
-        assert len(children) == 1
-        assert children[0].name == self.file.name
-
-        # Ensure the blob content was updated
-        doc = self.app.doc
-        assert isinstance(doc, Document)
-        assert doc.fetch_blob(xpath="file:content") == b"blob changed!"
-
-    def test_pause_upload_manually(self):
-        """
-        Pause the transfer by simulating a click on the pause/resume icon
-        on the current upload in the systray menu.
-        """
-
-        def callback(*_):
-            """
-            This will mimic what is done in SystrayTranfer.qml:
-                - call API.pause_transfer() that will call:
-                    - engine.dao.pause_transfer(nature, transfer_uid)
-            Then the upload will be paused in Remote.upload().
-            """
-            # Ensure we have 1 ongoing upload
-            uploads = list(dao.get_uploads())
-            assert uploads
-            upload = uploads[0]
-            assert upload.status == TransferStatus.ONGOING
-
-            # Pause the upload
-            dao.pause_transfer("upload", upload.uid, 50.0)
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "upload_callback", new=callback):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-            assert dao.get_uploads_with_status(TransferStatus.PAUSED)
-
-        # Resume the upload
-        engine.resume_transfer("upload", list(dao.get_uploads())[0].uid)
-        self.sync_and_check()
-
-    def test_pause_upload_automatically(self):
-        """
-        Pause the transfer by simulating an application exit
-        or clicking on the Suspend menu entry from the systray.
-        """
-
-        def callback(*_):
-            """This will mimic what is done in SystrayMenu.qml: suspend the app."""
-            # Ensure we have 1 ongoing upload
-            uploads = list(dao.get_uploads())
-            assert uploads
-            upload = uploads[0]
-            assert upload.status == TransferStatus.ONGOING
-
-            # Suspend!
-            self.manager_1.suspend()
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "upload_callback", new=callback):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-            assert dao.get_uploads_with_status(TransferStatus.SUSPENDED)
-
-        # Resume the upload
-        self.manager_1.resume()
-        self.sync_and_check()
-
-    def test_modifying_paused_upload(self):
-        """Modifying a paused upload should discard the current upload."""
-
-        def callback(*_):
-            """Pause the upload and apply changes to the document."""
-            # Ensure we have 1 ongoing upload
-            uploads = list(dao.get_uploads())
-            assert uploads
-            upload = uploads[0]
-            assert upload.status == TransferStatus.ONGOING
-
-            # Pause the upload
-            dao.pause_transfer("upload", upload.uid, 50.0)
-
-            # Apply changes to the file
-            self.file.write_bytes(b"locally changed")
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "upload_callback", new=callback):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-        # Resume the upload
-        engine.resume_transfer("upload", list(dao.get_uploads())[0].uid)
-        self.sync_and_check()
-        # Check the local content is correct
-        assert self.file.read_bytes() == b"locally changed"
-
-    @not_windows(
-        reason="Cannot test the behavior as the local deletion is blocked by the OS."
-    )
-    def test_deleting_paused_upload(self):
-        """Deleting a paused upload should discard the current upload."""
-
-        def callback(*_):
-            """Pause the upload and delete the document."""
-            # Ensure we have 1 ongoing upload
-            uploads = list(dao.get_uploads())
-            assert uploads
-            upload = uploads[0]
-            assert upload.status == TransferStatus.ONGOING
-
-            # Pause the upload
-            dao.pause_transfer("upload", upload.uid, 50.0)
-
-            # Remove the document
-            # (this is the problematic part on Windows, because for the
-            #  file descriptor to be released we need to escape from
-            #  Remote.upload(), which is not possible from here)
-            self.file.unlink()
-            assert not self.file.exists()
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "upload_callback", new=callback):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-        # Resume the upload
-        engine.resume_transfer("upload", list(dao.get_uploads())[0].uid)
-        self.sync_and_check(should_have_blob=False)
-
-    def test_not_server_error_upload(self):
-        """Test an error happening after chunks were uploaded, at the NuxeoDrive.CreateFile operation call."""
-
-        class BadUploader(DirectTransferUploader):
-            """Used to simulate bad server responses."""
-
-            def link_blob_to_doc(self, *args, **kwargs):
-                """Simulate a server error."""
-                raise ValueError("Mocked exception")
-
-        def upload(*args, **kwargs):
-            """Set our specific uploader to simulate server error."""
-            kwargs.pop("uploader")
-            return upload_orig(*args, uploader=BadUploader, **kwargs)
-
-        engine = self.engine_1
-        dao = engine.dao
-        upload_orig = engine.remote.upload
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "upload", new=upload):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-                # There should be 1 upload with ONGOING transfer status
-                uploads = list(dao.get_uploads())
-                assert len(uploads) == 1
-                upload = uploads[0]
-                assert upload.status == TransferStatus.DONE
-
-                # The file exists on the server but has no blob yet
-                assert not self.has_blob()
-
-                # The doc should be in error
-                assert len(dao.get_errors(limit=0)) == 1
-
-        # Reset the error
-        for state in dao.get_errors():
-            dao.reset_error(state)
-
-        self.sync_and_check()
-
-    def test_server_error_but_upload_ok(self):
-        """
-        Test an error happening after chunks were uploaded and the Blob.AttachOnDocument operation call.
-        This could happen if a proxy does not understand well the final requests as seen in NXDRIVE-1753.
-        """
-        pytest.skip("Not yet implemented.")
-
-        class BadUploader(DirectTransferUploader):
-            """Used to simulate bad server responses."""
-
-            def link_blob_to_doc(self, *args, **kwargs):
-                """Simulate a server error."""
-                # Call the original method to effectively end the upload process
-                super().link_blob_to_doc(*args, **kwargs)
-
-                # The file should be present on the server
-                assert self.remote.exists(file_path)
-
-                # There should be 1 upload with DONE transfer status
-                uploads = list(dao.get_uploads())
-                assert len(uploads) == 1
-                upload = uploads[0]
-                assert upload.status == TransferStatus.DONE
-
-                # And throw an error
-                stack = "The proxy server received an invalid response from an upstream server."
-                raise HTTPError(
-                    status=502, message="Mocked Proxy Error", stacktrace=stack
-                )
-
-        def upload(*args, **kwargs):
-            """Set our specific uploader to simulate server error."""
-            kwargs.pop("uploader")
-            return upload_orig(*args, uploader=BadUploader, **kwargs)
-
-        file_path = f"{self.ws.path}/{self.file.name}"
-        engine = self.engine_1
-        dao = engine.dao
-        upload_orig = engine.remote.upload
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "upload", new=upload):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-                # There should be no upload as the Processor has checked the file existence
-                # on the server and so deleted the upload from the database
-                self.no_uploads()
-
-        self.sync_and_check()
-
-    def test_server_error_upload(self):
-        """Test a server error happening after chunks were uploaded, at the Blob.AttachOnDocument operation call."""
-
-        class BadUploader(DirectTransferUploader):
-            """Used to simulate bad server responses."""
-
-            def link_blob_to_doc(self, *args, **kwargs):
-                """Simulate a server error."""
-                raise ConnectionError("Mocked exception")
-
-        def upload(*args, **kwargs):
-            """Set our specific uploader to simulate server error."""
-            kwargs.pop("uploader")
-            return upload_orig(*args, uploader=BadUploader, **kwargs)
-
-        engine = self.engine_1
-        dao = engine.dao
-        upload_orig = engine.remote.upload
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "upload", new=upload):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-                # There should be 1 upload with ONGOING transfer status
-                uploads = list(dao.get_uploads())
-                assert len(uploads) == 1
-                upload = uploads[0]
-                assert upload.status == TransferStatus.DONE
-
-                # The file exists on the server but has no blob yet
-                assert not self.has_blob()
-
-        self.sync_and_check()
-
-    def test_chunk_upload_error(self):
-        """Test a server error happening while uploading chunks."""
-
-        def send_data(*args, **kwargs):
-            """Simulate an error."""
-            raise ConnectionError("Mocked error")
-
-        def callback(upload):
-            """Patch send_data() after chunk 1 is sent."""
-            if len(upload.blob.uploadedChunkIds) == 1:
-                upload.service.send_data = send_data
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-        bad_remote = self.get_bad_remote()
-        bad_remote.upload_callback = callback
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine, "remote", new=bad_remote):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-                # There should be 1 upload with ONGOING transfer status
-                uploads = list(dao.get_uploads())
-                assert len(uploads) == 1
-                upload = uploads[0]
-                assert upload.status == TransferStatus.ONGOING
-
-                # The file exists on the server but has no blob yet
-                assert not self.has_blob()
-
-        self.sync_and_check()
-
-
-class TestDirectTransfer(OneUserTest, DirectTransfer):
-    """Direct Transfer in "normal" mode, i.e.: when synchronization features are enabled."""
-
-    def setUp(self):
-        DirectTransfer.setUp(self)
-
-
-class TestDirectTransferNoSync(OneUserNoSync, DirectTransfer):
-    """Direct Transfer should work when synchronization features are not enabled."""
-
-    def setUp(self):
-        DirectTransfer.setUp(self)
-
-
-class DirectTransferFolder:
-    def setUp(self):
-        # No sync root, to ease testing
-        self.remote_1.unregister_as_root(self.workspace)
-        self.engine_1.start()
-
-        # The folder used for the Direct Transfer (must be > 1 MiB)
-        folder = self.location / "resources"
-        # Work with a copy of the folder to allow parallel testing
-        self.folder = self.tmpdir / str(uuid4())
-        copytree(folder, self.folder, copy_function=copyfile)
-
-        # Get folders and files tests will handle
-        self.tree = {
-            path: rpath for path, rpath in get_tree_list(self.folder, self.ws.path)
-        }
-        paths = self.tree.keys()
-        self.files = [path for path in paths if path.is_file()]
-        self.folders = [path for path in paths if path not in self.files]
-
-        # Lower chunk_* options to have chunked uploads without having to create big files
-        self.default_chunk_limit = Options.chunk_limit
-        self.default_chunk_size = Options.chunk_size
-        Options.chunk_limit = 1
-        Options.chunk_size = 1
-
-    def tearDown(self):
-        # Restore options
-        Options.chunk_limit = self.default_chunk_limit
-        Options.chunk_size = self.default_chunk_size
-
-    def has_blob(self, file: str) -> bool:
-        """Check that *file* exists on the server and has a blob attached.
-        As when doing a Direct Transfer, the document is first created on the server,
-        this is the only way to check if the blob upload has been finished successfully.
-        """
-        try:
-            doc = self.root_remote.documents.get(path=f"{self.tree[file]}/{file.name}")
-        except Exception:
-            return False
-        return bool(doc.properties.get("file:content"))
-
-    def sync_and_check(self) -> None:
-        # Let time for uploads to be planned
-        sleep(3)
-
-        # Sync
-        self.wait_sync()
-
-        # Check the error count
-        assert not self.engine_1.dao.get_errors(limit=0)
-
-        # Check the uploads count
-        assert not list(self.engine_1.dao.get_uploads())
-
-        # Check files exist on the server with their attached blob
-        for file in self.files:
-            assert self.has_blob(file)
-
-        # Check subfolders
-        for folder in self.folders:
-            assert self.root_remote.documents.get(path=self.tree[folder])
-
-    def test_folder(self):
-        """Test the Direct Transfer on a folder containing files and a sufolder."""
-
-        # There is no upload, right now
-        assert not list(self.engine_1.dao.get_uploads())
-
-        with ensure_no_exception():
-            self.engine_1.direct_transfer([self.folder], self.ws.path)
-            self.sync_and_check()
-
-        # Ensure there is only 1 folder created at the workspace root
-        children = self.remote_1.get_children(self.ws.path)["entries"]
-        assert len(children) == 1
-        assert children[0]["title"] == self.folder.name
-
-        # All has been uploaded
-        assert not list(self.engine_1.dao.get_uploads())
-
-
-# NXDRIVE-2019
-
-
-class _TestDirectTransferFolder(OneUserTest, DirectTransferFolder):
-    """Direct Transfer in "normal" mode, i.e.: when synchronization features are enabled."""
-
-    def setUp(self):
-        DirectTransferFolder.setUp(self)
-
-
-class _TestDirectTransferFolderNoSync(OneUserNoSync, DirectTransferFolder):
-    """Direct Transfer should work when synchronization features are not enabled."""
-
-    def setUp(self):
-        DirectTransferFolder.setUp(self)
diff --git a/tests/resources/databases/engine_migration_11.db b/tests/resources/databases/engine_migration_11.db
new file mode 100644
index 000000000..e190c95b6
Binary files /dev/null and b/tests/resources/databases/engine_migration_11.db differ
diff --git a/tests/resources/subfolder/README.md b/tests/resources/subfolder/README.md
new file mode 100644
index 000000000..084c7005a
--- /dev/null
+++ b/tests/resources/subfolder/README.md
@@ -0,0 +1 @@
+Used by Direct Transfer manager tests.
diff --git a/tests/resources/subfolder/subfile.txt b/tests/resources/subfolder/subfile.txt
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/resources/subfolder/subfile2.txt b/tests/resources/subfolder/subfile2.txt
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/resources/subfolder/subfile3.txt b/tests/resources/subfolder/subfile3.txt
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/unit/test_engine_dao.py b/tests/unit/test_engine_dao.py
index fc32f3cf5..6e7981cf9 100644
--- a/tests/unit/test_engine_dao.py
+++ b/tests/unit/test_engine_dao.py
@@ -358,3 +358,18 @@ def test_migration_db_v10(engine_dao):
             Path("/Tests Drive/Live Connect/Test document Live Connect")
         )
         assert not bad_digest_file
+
+
+def test_migration_db_v11(engine_dao):
+    """
+    With NXDRIVE-2006, Direct Transfers were removed from the sync database.
+    Ensure that assertion is true.
+    """
+    # The test file contains 1 record of an ongoing Direct Transfer.
+    with engine_dao("engine_migration_11.db") as dao:
+        # With the v11 migration, it should have been deleted ...
+        states = list(dao.get_states_from_partial_local(Path()))
+        assert len(states) == 1  # The root
+
+        # ... And so for its related upload
+        assert not list(dao.get_uploads())
diff --git a/tools/deps/requirements.txt b/tools/deps/requirements.txt
index 10f604801..dea0a173f 100644
--- a/tools/deps/requirements.txt
+++ b/tools/deps/requirements.txt
@@ -85,6 +85,12 @@ tools/deps/nuxeo-python-client-bb6ee6b.zip[s3] \
 pathtools==0.1.2 \
     --hash=sha256:7c35c5421a39bb82e58018febd90e3b6e5db34c5443aaaf742b3f33d4655f1c0
     # via watchdog
+sqlalchemy==1.3.17 \
+    --hash=sha256:2a12f8be25b9ea3d1d5b165202181f2b7da4b3395289000284e5bb86154ce87c \
+    --hash=sha256:e8aa395482728de8bdcca9cc0faf3765ab483e81e01923aaa736b42f0294f570 \
+    --hash=sha256:f35248f7e0d63b234a109dd72fbfb4b5cb6cb6840b221d0df0ecbf54ab087654 \
+    --hash=sha256:ce1ddaadee913543ff0154021d31b134551f63428065168e756d90bdc4c686f5 \
+    --hash=sha256:9cb1819008f0225a7c066cac8bb0cf90847b2c4a6eb9ebb7431dbd00c56c06c5
 psutil==5.6.7 \
     --hash=sha256:1b1575240ca9a90b437e5a40db662acd87bbf181f6aa02f0204978737b913c6b \
     --hash=sha256:28f771129bfee9fc6b63d83a15d857663bbdcae3828e1cb926e91320a9b5b5cd \
diff --git a/tools/whitelist.py b/tools/whitelist.py
index dc899d83c..944e7090f 100644
--- a/tools/whitelist.py
+++ b/tools/whitelist.py
@@ -15,6 +15,17 @@
 CliHandler.unbind_root  # Used by the arguments parser
 CliHandler.unbind_server  # Used by the arguments parser
 _.close_settings_too  # Used by Appiclation.show_filters()
+direct_transfer.manager.decrease_priority  # Used by QML
+direct_transfer.manager.increase_priority  # Used by QML
+direct_transfer.models.BaseModel.Meta  # Used by Peewee
+direct_transfer.models.BaseModel.Meta.only_save_dirty  # Used by Peewee
+direct_transfer.models.BatchField.field_type  # Used by Peewee
+direct_transfer.models.BatchField.db_value  # Used by Peewee
+direct_transfer.models.BatchField.python_value  # Used by Peewee
+direct_transfer.models.Transfer.doctype  # Used by Peewee
+direct_transfer.models.Transfer.error_count_total  # Used by Peewee
+direct_transfer.uploader.transfer.batch.service  # Used by the Nuxeo Python client
+direct_transfer.uploader.transfer.doctype  # Used by Peewee
 DocPair.last_sync_error_date  # Check NXDRIVE-1804
 Download.transfer_type  # Used in QML
 Engine.folder  # Used in QML
@@ -36,8 +47,6 @@
 on_any_event  # Used by Watchdog
 _.row_factory  # Internal use of SQLite
 Processor._synchronize_conflicted  # Used by Processor._execute()
-Processor._synchronize_direct_transfer  # Used by Processor._execute()
-Processor._synchronize_direct_transfer_replace_blob  # Used by Processor._execute()
 Processor._synchronize_deleted_unknown  # Used by Processor._execute()
 Processor._synchronize_locally_deleted  # Used by Processor._execute()
 Processor._synchronize_locally_resolved  # Used by Processor._execute()
