NXDRIVE-2824: Code Coverage for Direct Edit Flow (#3841)

* NXDRIVE-2824: Code Coverage for Direct Edit Flow

docs/changes/5.3.2.md

@@ -8,6 +8,7 @@ Release date: `2023-xx-xx`
 
 ### Direct Edit
 
+- [NXDRIVE-2824](https://jira.nuxeo.com/browse/NXDRIVE-2824): Code Coverage for Direct Edit Flow
 - [NXDRIVE-2](https://jira.nuxeo.com/browse/NXDRIVE-2):
 
 ### Direct Transfer

tests/functional/test_direct_edit.py

@@ -534,3 +534,48 @@ def test_url_resolver(manager_factory, nuxeo_url):
         manager.engines["0"] = MockUrlTestEngine("https://localhost/nuxeo", user)
         assert get_engine("https://localhost:443/nuxeo/", user=user)
         assert get_engine("https://localhost/nuxeo", user=user)
+
+
+def test_lock_unlock(
+    direct_edit,
+    manager_factory,
+):
+    manager, engine = manager_factory()
+
+    def lock_(id):
+        return {id: "Locked"}
+
+    def lock_error(*args, **kwargs):
+        raise HTTPError(status=500, message="internal_server_error")
+
+    def unlock_(*args, **kwargs):
+        return False
+
+    def send_notification_(ref):
+        return
+
+    def _guess_user_from_http_error_(*args, **kwargs):
+        return engine.remote.user_id
+
+    direct_edit.is_already_locked = False
+    with patch.object(engine.remote, "lock", new=lock_):
+        with patch.object(direct_edit, "send_notification", new=send_notification_):
+            data = direct_edit._lock(engine.remote, "id", "ref")
+    assert data is not None
+
+    with patch.object(engine.remote, "lock", new=lock_error):
+        with patch.object(
+            direct_edit, "_guess_user_from_http_error", new=_guess_user_from_http_error_
+        ):
+            data = direct_edit._lock(engine.remote, "id", "ref")
+    assert not data
+
+    direct_edit.is_already_locked = True
+    data = direct_edit._lock(engine.remote, "id", "ref")
+    assert data is None
+
+    direct_edit._file_metrics = {"ref": "MOCK"}
+    with patch.object(engine.remote, "unlock", new=unlock_):
+        data = direct_edit._unlock(engine.remote, "id", "ref")
+
+    assert not data

tests/functional/test_remote_client.py

@@ -1,8 +1,11 @@
+from unittest.mock import Mock, patch
+
 import pytest
 from nuxeo.models import Document
 
+from nxdrive.engine.activity import Action, DownloadAction
 from nxdrive.metrics.constants import GLOBAL_METRICS
-from nxdrive.objects import RemoteFileInfo
+from nxdrive.objects import RemoteFileInfo, SubTypeEnricher
 from nxdrive.options import Options
 from nxdrive.utils import shortify
 
@@ -162,3 +165,100 @@ def test_expand_sync_root_name_length(option, manager_factory, obj_factory):
         assert sync_root.name == final_name
         assert sync_root.name.count("…") == Options.sync_root_max_level or 1
         assert len(sync_root.name) <= 250
+
+
+def test_upload_folder_type(manager_factory):
+    manager, engine = manager_factory()
+    remote = engine.remote
+
+    def request_(*args, **kwargs):
+        return "mocked-value"
+
+    with manager:
+        with patch.object(remote.client, "request", new=request_):
+            folder_type = remote.upload_folder_type("string", {"key": "value"})
+        assert folder_type == "mocked-value"
+
+
+def test_cancel_batch(manager_factory):
+    manager, engine = manager_factory()
+    remote = engine.remote
+
+    return_val = None
+    with manager:
+        return_val = remote.cancel_batch({"key": "value"})
+        assert not return_val
+
+
+def test_filter_schema(manager_factory):
+    manager, engine = manager_factory()
+    remote = engine.remote
+
+    def get_config_types_(*args, **kwargs):
+        configTypes = {"doctypes": {1: {"schemas": "file"}}}
+        return configTypes
+
+    returned_val = None
+    obj_ = SubTypeEnricher("uid", "path", "title", ["str"], {"key": "val"})
+    with manager:
+        with patch.object(remote, "get_config_types", new=get_config_types_):
+            returned_val = remote.filter_schema(obj_)
+        assert returned_val == ["1"]
+
+
+def test_get_note(manager_factory):
+    manager, engine = manager_factory()
+    remote = engine.remote
+
+    def fetch_(*args, **kwargs):
+        return {"properties": {"key": "val"}}
+
+    returned_val = None
+    with manager:
+        with patch.object(remote, "fetch", new=fetch_):
+            returned_val = remote.get_note("")
+    assert returned_val == b""
+
+
+def test_is_filtered(manager_factory):
+    manager, engine = manager_factory()
+    remote = engine.remote
+
+    def is_filter_(*args, **kwargs):
+        return True
+
+    returned_val = None
+    with manager:
+        with patch.object(remote.dao, "is_filter", new=is_filter_):
+            returned_val = remote.is_filtered("")
+    assert returned_val
+
+
+def test_transfer_start_callback(manager_factory):
+    manager, engine = manager_factory()
+    remote = engine.remote
+
+    obj1_ = Mock()
+    with manager:
+        returned_val = remote.transfer_start_callback(obj1_)
+    assert not returned_val
+
+
+def test_transfer_end_callback(manager_factory):
+    manager, engine = manager_factory()
+    remote = engine.remote
+
+    def get_current_action_(*args, **kwargs):
+        obj = DownloadAction("path", 1)
+        return obj
+
+    def get_download_(*args, **kwargs):
+        return False
+
+    obj1_ = Mock()
+    returned_val = None
+    with manager:
+        with patch.object(Action, "get_current_action", new=get_current_action_):
+            with patch.object(remote.dao, "get_download", new=get_download_):
+                returned_val = remote.transfer_end_callback(obj1_)
+    assert not returned_val