NXDRIVE-2860: Code Coverage

tests/functional/test_concurrent_synchronization.py

@@ -0,0 +1,41 @@
+import time
+
+from nxdrive.constants import WINDOWS
+
+from .conftest import TwoUsersTest
+
+
+class TestConcurrentSynchronization(TwoUsersTest):
+    def create_docs(self, parent, number, name_pattern=None, delay=1.0):
+        return self.root_remote.execute(
+            command="NuxeoDrive.CreateTestDocuments",
+            input_obj=f"doc:{parent}",
+            namePattern=name_pattern,
+            number=number,
+            delay=int(delay * 1000),
+        )
+
+    def test_rename_local_folder(self):
+        # Get local and remote clients
+        local1 = self.local_1
+        local2 = self.local_2
+
+        # Launch first synchronization
+        self.engine_1.start()
+        self.engine_2.start()
+        self.wait_sync(wait_for_async=True, wait_for_engine_2=True)
+
+        # Test workspace should be created locally
+        assert local1.exists("/")
+        assert local2.exists("/")
+
+        # Create a local folder in the test workspace and a file inside
+        # this folder, then synchronize
+        local1.make_folder("/", "Test folder")
+        if WINDOWS:
+            # Too fast folder create-then-rename are not well handled
+            time.sleep(1)
+        local1.rename("/Test folder", "Renamed folder")
+        self.wait_sync(wait_for_async=True, wait_for_engine_2=True)
+        assert local1.exists("/Renamed folder")
+        assert local2.exists("/Renamed folder")

tests/old_functional/test_copy.py → tests/functional/test_copy.py

@@ -1,4 +1,4 @@
-from .common import OneUserTest
+from .conftest import OneUserTest
 
 
 class TestCopy(OneUserTest):

tests/functional/test_encoding.py

@@ -0,0 +1,42 @@
+import os
+from pathlib import Path
+
+from nxdrive.client.local import FileInfo
+
+from ..markers import not_mac
+from .conftest import OneUserTest
+
+
+class TestEncoding(OneUserTest):
+    def test_name_normalization(self):
+        local = self.local_1
+        remote = self.remote_document_client_1
+
+        self.engine_1.start()
+        self.wait_sync(wait_for_async=True)
+
+        filename = "space\xa0 et TM\u2122.doc"
+        local.make_file("/", filename)
+        self.wait_sync(wait_for_async=True)
+
+        assert remote.get_info("/" + filename).name == filename
+
+    @not_mac(reason="Normalization does not work on macOS")
+    def test_fileinfo_normalization(self):
+        local = self.local_1
+
+        self.engine_1.start()
+        self.wait_sync(wait_for_async=True)
+        self.engine_1.stop()
+
+        name = "Teste\u0301"
+        local.make_file("/", name, content=b"Test")
+
+        # FileInfo() will normalize the filename
+        assert FileInfo(local.base_folder, Path(name), False, 0).name != name
+
+        # The encoding should be different,
+        # cannot trust the get_children as they use FileInfo
+        children = os.listdir(local.abspath("/"))
+        assert len(children) == 1
+        assert children[0] != name

tests/functional/test_group_changes.py

@@ -0,0 +1,148 @@
+from logging import getLogger
+
+from nuxeo.exceptions import HTTPError
+from nuxeo.models import Document, Group
+
+from .. import env
+from .conftest import OneUserTest, root_remote, salt
+
+log = getLogger(__name__)
+
+
+class TestGroupChanges(OneUserTest):
+    """
+    Test that changes on groups are detected by Drive.
+    See https://jira.nuxeo.com/browse/NXP-14830.
+    """
+
+    def setUp(self):
+        self.group1 = salt("group1")
+        self.group2 = salt("group2")
+        self.parent_group = salt("parentGroup")
+        self.grand_parent_group = salt("grandParentGroup")
+        self.new_groups = (
+            Group(groupname=self.group1, memberUsers=[self.user_1]),
+            Group(groupname=self.group2, memberUsers=[self.user_1]),
+            Group(groupname=self.parent_group, memberGroups=[self.group1]),
+            Group(groupname=self.grand_parent_group, memberGroups=[self.parent_group]),
+        )
+        for group in self.new_groups:
+            self.root_remote.groups.create(group)
+
+        # Create test workspace
+        workspace_name = salt("groupChangesTestWorkspace")
+        self.workspace_group = self.root_remote.documents.create(
+            Document(
+                name=workspace_name,
+                type="Workspace",
+                properties={"dc:title": workspace_name},
+            ),
+            parent_path=env.WS_DIR,
+        )
+        self.workspace_path = self.workspace_group.path
+
+        self.admin_remote = root_remote(base_folder=self.workspace_path)
+
+        self.engine_1.start()
+        self.wait_sync(wait_for_async=True)
+
+    def tearDown(self):
+        self.workspace_group.delete()
+        for group in reversed(self.new_groups):
+            try:
+                self.root_remote.groups.delete(group.groupname)
+            except HTTPError as exc:
+                if exc.status == 404:
+                    continue
+                raise
+
+    def set_ace(self, user, doc):
+        log.info(f"Grant ReadWrite permission to  {user} on {doc}")
+        self.admin_remote.execute(
+            command="Document.SetACE",
+            input_obj=f"doc:{doc}",
+            user=user,
+            permission="ReadWrite",
+        )
+
+    def _test_group_changes(self, folder_path, group_name, need_parent=False):
+        """
+        Tests changes on the given group that has access to the given folder:
+            - Remove the test user from the group.
+            - Add the test user to the group.
+            - Delete the group.
+            - Create the group including the test user.
+        """
+        log.info(
+            "Test changes on %s for %s with need_parent=%r",
+            group_name,
+            folder_path,
+            need_parent,
+        )
+        remote = self.admin_remote
+        local = self.local_root_client_1
+
+        log.info("Remove driveuser_1 from %s", group_name)
+        group = remote.groups.get(group_name)
+        group.memberUsers = []
+        group.save()
+
+        log.info("Check that %s is deleted locally", folder_path)
+        self.wait_sync(wait_for_async=True)
+        assert not local.exists(folder_path)
+
+        log.info("Add driveuser_1 to %s", group_name)
+        group.memberUsers = [self.user_1]
+        group.save()
+
+        log.info("Check that %s is created locally", folder_path)
+        self.wait_sync(wait_for_async=True)
+        assert local.exists(folder_path)
+
+        log.info("Delete %s", group_name)
+        remote.groups.delete(group_name)
+
+        log.info("Check that %s is deleted locally", folder_path)
+        self.wait_sync(wait_for_async=True)
+        assert not local.exists(folder_path)
+
+        log.info("Create %s", group_name)
+        remote.groups.create(Group(groupname=group_name, memberUsers=[self.user_1]))
+
+        if need_parent:
+            log.info(
+                "%s should not be created locally since "
+                "the newly created group has not been added yet "
+                "as a subgroup of parentGroup",
+                folder_path,
+            )
+            self.wait_sync(wait_for_async=True)
+            assert not local.exists(folder_path)
+
+            log.debug("Add %s as a subgroup of parentGroup", group_name)
+            group = remote.groups.get(self.parent_group)
+            group.memberGroups = [group_name]
+            group.save()
+
+        log.info("Check that %s is created locally", folder_path)
+        self.wait_sync(wait_for_async=True)
+        assert local.exists(folder_path)
+
+    def _test_group_changes_with_ancestor_groups(self, ancestor_group):
+        """
+        Test changes on a descendant group of the given group
+        that has access to a synchronization root.
+        """
+        log.info("Create syncRoot folder")
+        sync_root_id = self.admin_remote.make_folder("/", "syncRoot")
+
+        self.set_ace(ancestor_group, sync_root_id)
+
+        log.info("Register syncRoot for driveuser_1")
+        self.remote_1.register_as_root(sync_root_id)
+
+        log.info("Check that syncRoot is created locally")
+        self.wait_sync(wait_for_async=True)
+        assert self.local_root_client_1.exists("/syncRoot")
+
+        self._test_group_changes("/syncRoot", self.group1, need_parent=True)

tests/functional/test_local_changes_when_offline.py

@@ -0,0 +1,59 @@
+""""
+Test if changes made to local file system when Drive is offline sync's back
+later when Drive becomes online.
+"""
+from .conftest import FILE_CONTENT, OneUserTest
+
+
+class TestOfflineChangesSync(OneUserTest):
+    def setUp(self):
+        self.engine_1.start()
+        self.wait_sync(wait_for_async=True)
+
+        self.local = self.local_1
+        self.remote = self.remote_document_client_1
+
+        # Create a folder and a file on the server
+        self.folder1_remote = self.remote.make_folder("/", "Folder1")
+        self.file1_remote = self.remote.make_file(
+            self.folder1_remote, "File1.txt", FILE_CONTENT
+        )
+        self.wait_sync(wait_for_async=True)
+
+    def copy_past_and_rename(self, stop_engine: bool = False):
+        if stop_engine:
+            # Make Drive offline (by suspend)
+            self.engine_1.suspend()
+
+        # Make a copy of the file (with xattr included)
+        self.local_1.copy("/Folder1/File1.txt", "/Folder1/File1 - Copy.txt")
+
+        # Rename the original file
+        self.local.rename("/Folder1/File1.txt", "File1_renamed.txt")
+
+        if stop_engine:
+            # Bring Drive online (by resume)
+            self.engine_1.resume()
+
+        self.wait_sync()
+
+        # Verify there is no local changes
+        assert self.local.exists("/Folder1/File1_renamed.txt")
+        assert self.local.exists("/Folder1/File1 - Copy.txt")
+        assert not self.local.exists("/Folder1/File1.txt")
+
+        # Verify that local changes are uploaded to server successfully
+        if self.remote.exists("/Folder1/File1 - Copy.txt"):
+            # '/Folder1/File1 - Copy.txt' is uploaded to server.
+            # So original file named should be changed as 'File_renamed.txt'
+            remote_info = self.remote.get_info(self.file1_remote)
+            assert remote_info.name == "File1 - Copy.txt"
+
+        else:
+            # Original file is renamed as 'File1 - Copy.txt'.
+            # This is a bug only if Drive is online during copy + rename
+            assert self.remote.exists("/Folder1/File1_renamed.txt")
+            remote_info = self.remote.get_info(self.file1_remote)
+            assert remote_info.name == "File1_renamed.txt"
+
+        assert not self.remote.exists("/Folder1/File1.txt")

tests/old_functional/test_local_copy_paste.py → tests/functional/test_local_copy_paste.py

@@ -1,6 +1,6 @@
 import shutil
 
-from .common import FILE_CONTENT, OneUserTest
+from .conftest import FILE_CONTENT, OneUserTest
 
 
 class TestLocalCopyPaste(OneUserTest):
@@ -71,12 +71,6 @@ def setUp(self):
             == self.NUMBER_OF_LOCAL_FILES_TOTAL
         )
 
-    def test_local_copy_paste_files(self):
-        self._local_copy_paste_files()
-
-    def test_local_copy_paste_files_stopped(self):
-        self._local_copy_paste_files(stopped=True)
-
     def _local_copy_paste_files(self, stopped=False):
         if not stopped:
             self.engine_1.start()