Merge pull request #542 from inkhey/webdav_copy_fix

tracim/tracim/lib/content.py

 
         item.revision_type = ActionDescription.MOVE
 
+    def copy(
+        self,
+        item: Content,
+        new_parent: Content=None,
+        new_label: str=None,
+        do_save: bool=True,
+        do_notify: bool=True,
+    ) -> Content:
+        """
+        Copy nearly all content, revision included. Children not included, see
+        "copy_children" for this.
+        :param item: Item to copy
+        :param new_parent: new parent of the new copied item
+        :param new_label: new label of the new copied item
+        :param do_notify: notify copy or not
+        :return: Newly copied item
+        """
+        if (not new_parent and not new_label) or (new_parent == item.parent and new_label == item.label):  # nopep8
+            # TODO - G.M - 08-03-2018 - Use something else than value error
+            raise ValueError("You can't copy file into itself")
+        if new_parent:
+            workspace = new_parent.workspace
+            parent = new_parent
+        else:
+            workspace = item.workspace
+            parent = item.parent
+        label = new_label or item.label
+
+        content = item.copy(parent)
+        # INFO - GM - 15-03-2018 - add "copy" revision
+        with new_revision(content, force_create_new_revision=True) as rev:
+            rev.parent = parent
+            rev.workspace = workspace
+            rev.label = label
+            rev.revision_type = ActionDescription.COPY
+            rev.properties['origin'] = {
+                'content': item.id,
+                'revision': item.last_revision.revision_id,
+            }
+        if do_save:
+            self.save(content, ActionDescription.COPY, do_notify=do_notify)
+        return content
+
+    def copy_children(self, origin_content: Content, new_content: Content):
+        for child in origin_content.children:
+            self.copy(child, new_content)
+
     def move_recursively(self, item: Content,
                          new_parent: Content, new_workspace: Workspace):
         self.move(item, new_parent, False, new_workspace)

tracim/tracim/lib/webdav/design.py

     'unarchiving': 'Item unarchived',
     'undeletion': 'Item undeleted',
     'move': 'Item moved',
-    'comment': 'Comment'
+    'comment': 'Comment',
+    'copy' : 'Item copied',
 }
 
 

tracim/tracim/lib/webdav/sql_resources.py

 
         transaction.commit()
 
+    def copyMoveSingle(self, destpath, isMove):
+        if isMove:
+            # INFO - G.M - 12-03-2018 - This case should not happen
+            # As far as moveRecursive method exist, all move should not go
+            # through this method. If such case appear, try replace this to :
+            ####
+            # self.move_file(destpath)
+            # return
+            ####
+
+            raise NotImplemented
+
+        new_file_name = None
+        new_file_extension = None
+
+        # Inspect destpath
+        if basename(destpath) != self.getDisplayName():
+            new_given_file_name = transform_to_bdd(basename(destpath))
+            new_file_name, new_file_extension = \
+                os.path.splitext(new_given_file_name)
+
+        workspace_api = WorkspaceApi(self.user)
+        content_api = ContentApi(self.user)
+        destination_workspace = self.provider.get_workspace_from_path(
+            destpath,
+            workspace_api,
+        )
+        destination_parent = self.provider.get_parent_from_path(
+            destpath,
+            content_api,
+            destination_workspace,
+        )
+        workspace = self.content.workspace
+        parent = self.content.parent
+        new_content = self.content_api.copy(
+            item=self.content,
+            new_label=new_file_name,
+            new_parent=destination_parent,
+        )
+        self.content_api.copy_children(self.content, new_content)
+        transaction.commit()
+
     def supportRecursiveMove(self, destPath):
         return True
 

tracim/tracim/model/__init__.py

 
 
 @contextmanager
-def new_revision(content: Content) -> Content:
+def new_revision(
+        content: Content,
+        force_create_new_revision: bool=False,
+) -> Content:
     """
     Prepare context to update a Content. It will add a new updatable revision to the content.
     :param content: Content instance to update
     """
     with DBSession.no_autoflush:
         try:
-            if inspect(content.revision).has_identity:
+            if force_create_new_revision \
+                    or inspect(content.revision).has_identity:
                 content.new_revision()
             RevisionsIntegrity.add_to_updatable(content.revision)
             yield content

tracim/tracim/model/data.py

     - closed-deprecated
     """
 
+    COPY = 'copy'
     ARCHIVING = 'archiving'
     COMMENT = 'content-comment'
     CREATION = 'creation'
         'status-update': 'fa-random',
         'unarchiving': 'fa-file-archive-o',
         'undeletion': 'fa-trash-o',
-        'move': 'fa-arrows'
+        'move': 'fa-arrows',
+        'copy': 'fa-files-o',
     }
 
     _LABELS = {
         'status-update': l_('New status'),
         'unarchiving': l_('Item unarchived'),
         'undeletion': l_('Item undeleted'),
-        'move': l_('Item moved')
+        'move': l_('Item moved'),
+        'copy': l_('Item copied'),
     }
 
     def __init__(self, id):
                 cls.STATUS_UPDATE,
                 cls.UNARCHIVING,
                 cls.UNDELETION,
-                cls.MOVE]
+                cls.MOVE,
+                cls.COPY,
+                ]
 
 
 class ContentStatus(object):
                 return False
             return True
 
+        # TODO - G.M - 15-03-2018 - Choose only correct Content-type for origin
+        # Only content who can be copied need this
+        if item.type == ContentType.Any:
+            if 'origin' in properties.keys():
+                return True
         raise NotImplementedError
 
     @classmethod
         'is_archived',
         'is_deleted',
         'label',
-        'node',
         'owner',
         'owner_id',
         'parent',
 
         return new_rev
 
+    @classmethod
+    def copy(
+            cls,
+            revision: 'ContentRevisionRO',
+            parent: 'Content'
+    ) -> 'ContentRevisionRO':
+
+        copy_rev = cls()
+        import copy
+        copy_columns = cls._cloned_columns
+        for column_name in copy_columns:
+            # INFO - G-M - 15-03-2018 - set correct parent
+            if column_name == 'parent_id':
+                column_value = copy.copy(parent.id)
+            elif column_name == 'parent':
+                column_value = copy.copy(parent)
+            else:
+                column_value = copy.copy(getattr(revision, column_name))
+            setattr(copy_rev, column_name, column_value)
+
+        # copy attached_file
+        if revision.depot_file:
+            copy_rev.depot_file = FileIntent(
+                revision.depot_file.file.read(),
+                revision.file_name,
+                revision.file_mimetype,
+            )
+        return copy_rev
+
+
     def __setattr__(self, key: str, value: 'mixed'):
         """
         ContentRevisionUpdateError is raised if tried to update column and revision own identity
         revisions = sorted(self.revisions, key=lambda revision: revision.revision_id)
         return revisions[-1]
 
-    def new_revision(self) -> None:
+    def new_revision(self) -> ContentRevisionRO:
         """
         Return and assign to this content a new revision.
         If it's a new content, revision is totally new.
         cid = content.content_id
         return url_template.format(wid=wid, fid=fid, ctype=ctype, cid=cid)
 
+    def copy(self, parent):
+        cpy_content = Content()
+        for rev in self.revisions:
+            cpy_rev = ContentRevisionRO.copy(rev, parent)
+            cpy_content.revisions.append(cpy_rev)
+        return cpy_content
+
 
 class RevisionReadStatus(DeclarativeBase):
 

tracim/tracim/tests/library/test_content_api.py

 from nose.tools import eq_, ok_
 from nose.tools import raises
 
+from depot.io.utils import FileIntent
 import transaction
 
 from tracim.lib.content import compare_content_for_sorting_by_type_and_name
         eq_('', c.label)
         eq_(ActionDescription.COMMENT, c.revision_type)
 
+    def test_unit_copy_file_different_label_different_parent_ok(self):
+        uapi = UserApi(None)
+        groups = [
+            GroupApi(None).get_one(Group.TIM_USER),
+            GroupApi(None).get_one(Group.TIM_MANAGER),
+            GroupApi(None).get_one(Group.TIM_ADMIN)
+        ]
+
+        user = uapi.create_user(
+            email='user1@user',
+            groups=groups,
+            save_now=True
+        )
+        user2 = uapi.create_user(
+            email='user2@user',
+            groups=groups,
+            save_now=True
+        )
+        workspace = WorkspaceApi(user).create_workspace(
+            'test workspace',
+            save_now=True
+        )
+        RoleApi(user).create_one(
+            user2,
+            workspace,
+            UserRoleInWorkspace.WORKSPACE_MANAGER,
+            with_notif=False
+        )
+        api = ContentApi(user)
+        foldera = api.create(
+            ContentType.Folder,
+            workspace,
+            None,
+            'folder a',
+            True
+        )
+        with DBSession.no_autoflush:
+            text_file = api.create(
+                content_type=ContentType.File,
+                workspace=workspace,
+                parent=foldera,
+                label='test_file',
+                do_save=False,
+            )
+            api.update_file_data(
+                text_file,
+                'test_file',
+                'text/plain',
+                b'test_content'
+            )
+
+        api.save(text_file, ActionDescription.CREATION)
+        api2 = ContentApi(user2)
+        workspace2 = WorkspaceApi(user2).create_workspace(
+            'test workspace2',
+            save_now=True
+        )
+        folderb = api2.create(
+            ContentType.Folder,
+            workspace2,
+            None,
+            'folder b',
+            True
+        )
+
+        api2.copy(
+            item=text_file,
+            new_parent=folderb,
+            new_label='test_file_copy'
+        )
+
+        transaction.commit()
+        text_file_copy = api2.get_one_by_label_and_parent(
+            'test_file_copy',
+            folderb,
+        )
+
+        assert text_file != text_file_copy
+        assert text_file_copy.content_id != text_file.content_id
+        assert text_file_copy.workspace_id == workspace2.workspace_id
+        assert text_file_copy.depot_file.file.read() == text_file.depot_file.file.read()   # nopep8
+        assert text_file_copy.depot_file.path != text_file.depot_file.path
+        assert text_file_copy.label == 'test_file_copy'
+        assert text_file_copy.type == text_file.type
+        assert text_file_copy.parent.content_id == folderb.content_id
+        assert text_file_copy.owner.user_id == user.user_id
+        assert text_file_copy.description == text_file.description
+        assert text_file_copy.file_extension == text_file.file_extension
+        assert text_file_copy.file_mimetype == text_file.file_mimetype
+        assert text_file_copy.revision_type == ActionDescription.COPY
+        assert len(text_file_copy.revisions) == len(text_file.revisions) + 1
+
+    def test_unit_copy_file__same_label_different_parent_ok(self):
+        uapi = UserApi(None)
+        groups = [GroupApi(None).get_one(Group.TIM_USER),
+                  GroupApi(None).get_one(Group.TIM_MANAGER),
+                  GroupApi(None).get_one(Group.TIM_ADMIN)]
+
+        user = uapi.create_user(
+            email='user1@user',
+            groups=groups,
+            save_now=True
+        )
+        user2 = uapi.create_user(
+            email='user2@user',
+            groups=groups,
+            save_now=True
+        )
+        workspace = WorkspaceApi(user).create_workspace(
+            'test workspace',
+            save_now=True
+        )
+        RoleApi(user).create_one(
+            user2,
+            workspace,
+            UserRoleInWorkspace.WORKSPACE_MANAGER,
+            with_notif=False
+        )
+        api = ContentApi(user)
+        foldera = api.create(
+            ContentType.Folder,
+            workspace,
+            None,
+            'folder a',
+            True
+        )
+        with DBSession.no_autoflush:
+            text_file = api.create(
+                content_type=ContentType.File,
+                workspace=workspace,
+                parent=foldera,
+                label='test_file',
+                do_save=False,
+            )
+            api.update_file_data(
+                text_file,
+                'test_file',
+                'text/plain',
+                b'test_content'
+            )
+
+        api.save(text_file, ActionDescription.CREATION)
+        api2 = ContentApi(user2)
+        workspace2 = WorkspaceApi(user2).create_workspace(
+            'test workspace2',
+            save_now=True
+        )
+        folderb = api2.create(
+            ContentType.Folder,
+            workspace2,
+            None,
+            'folder b',
+            True
+        )
+        api2.copy(
+            item=text_file,
+            new_parent=folderb,
+        )
+
+        transaction.commit()
+        text_file_copy = api2.get_one_by_label_and_parent(
+            'test_file',
+            folderb,
+        )
+
+        assert text_file != text_file_copy
+        assert text_file_copy.content_id != text_file.content_id
+        assert text_file_copy.workspace_id == workspace2.workspace_id
+        assert text_file_copy.depot_file.file.read() == text_file.depot_file.file.read()  # nopep8
+        assert text_file_copy.depot_file.path != text_file.depot_file.path
+        assert text_file_copy.label == text_file.label
+        assert text_file_copy.type == text_file.type
+        assert text_file_copy.parent.content_id == folderb.content_id
+        assert text_file_copy.owner.user_id == user.user_id
+        assert text_file_copy.description == text_file.description
+        assert text_file_copy.file_extension == text_file.file_extension
+        assert text_file_copy.file_mimetype == text_file.file_mimetype
+        assert text_file_copy.revision_type == ActionDescription.COPY
+        assert len(text_file_copy.revisions) == len(text_file.revisions) + 1
+
+    def test_unit_copy_file_different_label_same_parent_ok(self):
+        uapi = UserApi(None)
+        groups = [
+            GroupApi(None).get_one(Group.TIM_USER),
+            GroupApi(None).get_one(Group.TIM_MANAGER),
+            GroupApi(None).get_one(Group.TIM_ADMIN),
+        ]
+
+        user = uapi.create_user(
+            email='user1@user',
+            groups=groups,
+            save_now=True,
+        )
+        user2 = uapi.create_user(
+            email='user2@user',
+            groups=groups,
+            save_now=True
+        )
+        workspace = WorkspaceApi(user).create_workspace(
+            'test workspace',
+            save_now=True
+        )
+        RoleApi(user).create_one(
+            user2, workspace,
+            UserRoleInWorkspace.WORKSPACE_MANAGER,
+            with_notif=False
+        )
+        api = ContentApi(user)
+        foldera = api.create(
+            ContentType.Folder,
+            workspace,
+            None,
+            'folder a',
+            True
+        )
+        with DBSession.no_autoflush:
+            text_file = api.create(
+                content_type=ContentType.File,
+                workspace=workspace,
+                parent=foldera,
+                label='test_file',
+                do_save=False,
+            )
+            api.update_file_data(
+                text_file,
+                'test_file',
+                'text/plain',
+                b'test_content'
+            )
+
+        api.save(
+            text_file,
+            ActionDescription.CREATION
+        )
+        api2 = ContentApi(user2)
+
+        api2.copy(
+            item=text_file,
+            new_label='test_file_copy'
+        )
+
+        transaction.commit()
+        text_file_copy = api2.get_one_by_label_and_parent(
+            'test_file_copy',
+            foldera,
+        )
+
+        assert text_file != text_file_copy
+        assert text_file_copy.content_id != text_file.content_id
+        assert text_file_copy.workspace_id == workspace.workspace_id
+        assert text_file_copy.depot_file.file.read() == text_file.depot_file.file.read()  # nopep8
+        assert text_file_copy.depot_file.path != text_file.depot_file.path
+        assert text_file_copy.label == 'test_file_copy'
+        assert text_file_copy.type == text_file.type
+        assert text_file_copy.parent.content_id == foldera.content_id
+        assert text_file_copy.owner.user_id == user.user_id
+        assert text_file_copy.description == text_file.description
+        assert text_file_copy.file_extension == text_file.file_extension
+        assert text_file_copy.file_mimetype == text_file.file_mimetype
+        assert text_file_copy.revision_type == ActionDescription.COPY
+        assert len(text_file_copy.revisions) == len(text_file.revisions) + 1
 
     def test_mark_read__workspace(self):
         uapi = UserApi(None)

tracim/tracim/tests/models/test_content_revision.py

 class TestContentRevision(BaseTest, TestStandard):
 
     def _new_from(self, revision):
-        excluded_columns = ('revision_id', '_sa_instance_state', 'depot_file')
+        excluded_columns = (
+            'revision_id',
+            '_sa_instance_state',
+            'depot_file',
+            'node',
+            'revision_read_statuses',
+        )
         revision_columns = [attr.key for attr in inspect(revision).attrs if not attr.key in excluded_columns]
         new_revision = ContentRevisionRO()