dev

tracim/tracim/fixtures/content.py

@@ -0,0 +1,114 @@
+# -*- coding: utf-8 -*-
+from tracim import model
+from tracim.fixtures import Fixture
+from tracim.fixtures.users_and_groups import Test
+from tracim.lib.content import ContentApi
+from tracim.lib.userworkspace import RoleApi
+from tracim.lib.workspace import WorkspaceApi
+from tracim.model.data import ContentType
+from tracim.model.data import UserRoleInWorkspace
+
+
+class Content(Fixture):
+    require = [Test]
+
+    def insert(self):
+        admin = self._session.query(model.User) \
+            .filter(model.User.email == 'admin@admin.admin') \
+            .one()
+        bob = self._session.query(model.User) \
+            .filter(model.User.email == 'bob@fsf.local') \
+            .one()
+        workspace_api = WorkspaceApi(admin)
+        content_api = ContentApi(admin)
+        role_api = RoleApi(admin)
+
+        # Workspaces
+        w1 = workspace_api.create_workspace('w1', save_now=True)
+        w2 = workspace_api.create_workspace('w2', save_now=True)
+        w3 = workspace_api.create_workspace('w3', save_now=True)
+
+        # Workspaces roles
+        role_api.create_one(
+            user=bob,
+            workspace=w1,
+            role_level=UserRoleInWorkspace.CONTENT_MANAGER,
+            with_notif=False,
+        )
+        role_api.create_one(
+            user=bob,
+            workspace=w2,
+            role_level=UserRoleInWorkspace.CONTENT_MANAGER,
+            with_notif=False,
+        )
+
+        w1f1 = content_api.create(
+            content_type=ContentType.Folder,
+            workspace=w1,
+            label='w1f1',
+            do_save=True,
+        )
+        w1f2 = content_api.create(
+            content_type=ContentType.Folder,
+            workspace=w1,
+            label='w1f2',
+            do_save=True,
+        )
+
+        # Folders
+        w2f1 = content_api.create(
+            content_type=ContentType.Folder,
+            workspace=w2,
+            label='w1f1',
+            do_save=True,
+        )
+        w2f2 = content_api.create(
+            content_type=ContentType.Folder,
+            workspace=w2,
+            label='w2f2',
+            do_save=True,
+        )
+
+        w3f1 = content_api.create(
+            content_type=ContentType.Folder,
+            workspace=w3,
+            label='w3f3',
+            do_save=True,
+        )
+
+        # Pages, threads, ..
+        w1f1p1 = content_api.create(
+            content_type=ContentType.Page,
+            workspace=w1,
+            parent=w1f1,
+            label='w1f1p1',
+            do_save=True,
+        )
+        w1f1t1 = content_api.create(
+            content_type=ContentType.Thread,
+            workspace=w1,
+            parent=w1f1,
+            label='w1f1t1',
+            do_save=False,
+        )
+        w1f1t1.description = 'w1f1t1 description'
+        self._session.add(w1f1t1)
+        w1f1d1 = content_api.create(
+            content_type=ContentType.File,
+            workspace=w1,
+            parent=w1f1,
+            label='w1f1d1',
+            do_save=False,
+        )
+        w1f1d1.file_extension = '.txt'
+        w1f1d1.file_content = b'w1f1d1 content'
+        self._session.add(w1f1d1)
+
+        w2f1p1 = content_api.create(
+            content_type=ContentType.Page,
+            workspace=w2,
+            parent=w2f1,
+            label='w2f1p1',
+            do_save=True,
+        )
+        self._session.flush()

tracim/tracim/lib/webdav/sql_dav_provider.py

@@ -241,11 +241,13 @@ class Provider(DAVProvider):
         path = self.reduce_path(path)
         parent_path = dirname(path)
 
-        blbl = parent_path.replace('/'+workspace.label, '')
+        relative_parents_path = parent_path[len(workspace.label)+1:]
+        parents = relative_parents_path.split('/')
 
-        parents = blbl.split('/')
-
-        parents.remove('')
+        try:
+            parents.remove('')
+        except ValueError:
+            pass
         parents = [transform_to_bdd(x) for x in parents]
 
         try:

tracim/tracim/lib/webdav/sql_resources.py

@@ -964,6 +964,8 @@ class File(DAVNonCollection):
                 self.content.file_extension = new_file_extension
                 self.content_api.save(self.content)
             else:
+                # TODO: ???le new parent ne peut être le parent actuel ?
+
                 self.content_api.move(
                     item=self.content,
                     new_parent=parent,

tracim/tracim/tests/library/test_webdav.py

@@ -0,0 +1,440 @@
+# -*- coding: utf-8 -*-
+import io
+from nose.tools import eq_
+from nose.tools import ok_
+from tracim.lib.webdav.sql_dav_provider import Provider
+from tracim.lib.webdav.sql_resources import Root
+from tracim.model import Content
+from tracim.model import ContentRevisionRO
+from tracim.model import DBSession
+from tracim.tests import TestStandard
+from tracim.fixtures.content import Content as ContentFixtures
+from wsgidav import util
+
+
+class TestWebDav(TestStandard):
+    fixtures = [ContentFixtures]
+
+    def _get_provider(self):
+        return Provider(
+            show_archived=False,
+            show_deleted=False,
+            show_history=False,
+        )
+
+    def _get_environ(
+            self,
+            provider: Provider,
+            username: str,
+    ) -> dict:
+        return {
+            'http_authenticator.username': username,
+            'http_authenticator.realm': '/',
+            'wsgidav.provider': provider,
+        }
+
+    def _put_new_text_file(
+            self,
+            provider,
+            environ,
+            file_path,
+            file_content,
+    ):
+        # This part id a reproduction of
+        # wsgidav.request_server.RequestServer#doPUT
+
+        # Grab parent folder where create file
+        parentRes = provider.getResourceInst(
+            util.getUriParent(file_path),
+            environ,
+        )
+        ok_(parentRes, msg='we should found folder for {0}'.format(file_path))
+
+        new_resource = parentRes.createEmptyResource(
+            util.getUriName(file_path),
+        )
+        write_object = new_resource.beginWrite(
+            contentType='application/octet-stream',
+        )
+        write_object.write(b'hello\n')
+        write_object.close()
+        new_resource.endWrite(withErrors=False)
+
+        # Now file should exist
+        return provider.getResourceInst(
+            file_path,
+            environ,
+        )
+
+    def test_unit__get_root__ok(self):
+        provider = self._get_provider()
+        root = provider.getResourceInst(
+            '/',
+            self._get_environ(
+                provider,
+                'bob@fsf.local',
+            )
+        )
+        ok_(root, msg='Path / should return a Root instance')
+        ok_(isinstance(root, Root))
+
+    def test_unit__list_workspaces_with_admin__ok(self):
+        provider = self._get_provider()
+        root = provider.getResourceInst(
+            '/',
+            self._get_environ(
+                provider,
+                'bob@fsf.local',
+            )
+        )
+        ok_(root, msg='Path / should return a Root instance')
+        ok_(isinstance(root, Root), msg='Path / should return a Root instance')
+
+        children = root.getMemberList()
+        eq_(
+            2,
+            len(children),
+            msg='Root should return 2 workspaces instead {0}'.format(
+                len(children),
+            )
+        )
+
+        workspaces_names = [w.name for w in children]
+        ok_('w1' in workspaces_names, msg='w1 should be in names ({0})'.format(
+            workspaces_names,
+        ))
+        ok_('w2' in workspaces_names, msg='w2 should be in names ({0})'.format(
+            workspaces_names,
+        ))
+
+    def test_unit__list_workspace_folders__ok(self):
+        provider = self._get_provider()
+        w1 = provider.getResourceInst(
+            '/w1/',
+            self._get_environ(
+                provider,
+                'bob@fsf.local',
+            )
+        )
+        ok_(w1, msg='Path /w1 should return a Wrkspace instance')
+
+        children = w1.getMemberList()
+        eq_(
+            2,
+            len(children),
+            msg='w1 should list 2 folders instead {0}'.format(
+                len(children),
+            ),
+        )
+
+        folders_names = [f.name for f in children]
+        ok_(
+            'w1f1' in folders_names,
+            msg='w1f1 should be in names ({0})'.format(
+                folders_names,
+            )
+        )
+        ok_(
+            'w1f2' in folders_names,
+            msg='w1f2 should be in names ({0})'.format(
+                folders_names,
+            )
+        )
+
+    def test_unit__list_content__ok(self):
+        provider = self._get_provider()
+        w1f1 = provider.getResourceInst(
+            '/w1/w1f1',
+            self._get_environ(
+                provider,
+                'bob@fsf.local',
+            )
+        )
+        ok_(w1f1, msg='Path /w1f1 should return a Wrkspace instance')
+
+        children = w1f1.getMemberList()
+        eq_(
+            3,
+            len(children),
+            msg='w1f1 should list 3 folders instead {0}'.format(
+                len(children),
+            ),
+        )
+
+        content_names = [c.name for c in children]
+        ok_(
+            'w1f1p1.html' in content_names,
+            msg='w1f1.html should be in names ({0})'.format(
+                content_names,
+            )
+        )
+        ok_(
+            'w1f1t1.html' in content_names,
+            msg='w1f1t1.html should be in names ({0})'.format(
+                content_names,
+            )
+        )
+        ok_(
+            'w1f1d1.txt' in content_names,
+            msg='w1f1d1.txt should be in names ({0})'.format(
+                content_names,
+            )
+        )
+
+    def test_unit__get_content__ok(self):
+        provider = self._get_provider()
+        w1f1d1 = provider.getResourceInst(
+            '/w1/w1f1/w1f1d1.txt',
+            self._get_environ(
+                provider,
+                'bob@fsf.local',
+            )
+        )
+
+        ok_(w1f1d1, msg='w1f1d1 should be found')
+        eq_('w1f1d1.txt', w1f1d1.name)
+
+    def test_unit__delete_content__ok(self):
+        provider = self._get_provider()
+        w1f1d1 = provider.getResourceInst(
+            '/w1/w1f1/w1f1d1.txt',
+            self._get_environ(
+                provider,
+                'bob@fsf.local',
+            )
+        )
+
+        content_w1f1d1 = DBSession.query(ContentRevisionRO) \
+            .filter(Content.label == 'w1f1d1') \
+            .one()  # It must exist only one revision, cf fixtures
+        eq_(
+            False,
+            content_w1f1d1.is_deleted,
+            msg='Content should not be deleted !'
+        )
+        content_w1f1d1_id = content_w1f1d1.content_id
+
+        w1f1d1.delete()
+
+        DBSession.flush()
+        content_w1f1d1 = DBSession.query(ContentRevisionRO) \
+            .filter(Content.content_id == content_w1f1d1_id) \
+            .order_by(Content.revision_id.desc()) \
+            .first()
+        eq_(
+            True,
+            content_w1f1d1.is_deleted,
+            msg='Content should be deleted !'
+        )
+
+        result = provider.getResourceInst(
+            '/w1/w1f1/w1f1d1.txt',
+            self._get_environ(
+                provider,
+                'bob@fsf.local',
+            )
+        )
+        eq_(None, result, msg='Result should be None instead {0}'.format(
+            result
+        ))
+
+    def test_unit__create_content__ok(self):
+        provider = self._get_provider()
+        environ = self._get_environ(
+            provider,
+            'bob@fsf.local',
+        )
+        result = provider.getResourceInst(
+            '/w1/w1f1/new_file.txt',
+            environ,
+        )
+
+        eq_(None, result, msg='Result should be None instead {0}'.format(
+            result
+        ))
+
+        result = self._put_new_text_file(
+            provider,
+            environ,
+            '/w1/w1f1/new_file.txt',
+            b'hello\n',
+        )
+
+        ok_(result, msg='Result should not be None instead {0}'.format(
+            result
+        ))
+        eq_(
+            b'hello\n',
+            result.content.file_content,
+            msg='fiel content should be "hello\n" but it is {0}'.format(
+                result.content.file_content
+            )
+        )
+
+    def test_unit__create_delete_and_create_file__ok(self):
+        provider = self._get_provider()
+        environ = self._get_environ(
+            provider,
+            'bob@fsf.local',
+        )
+        new_file = provider.getResourceInst(
+            '/w1/w1f1/new_file.txt',
+            environ,
+        )
+
+        eq_(None, new_file, msg='Result should be None instead {0}'.format(
+            new_file
+        ))
+
+        # create it
+        new_file = self._put_new_text_file(
+            provider,
+            environ,
+            '/w1/w1f1/new_file.txt',
+            b'hello\n',
+        )
+        ok_(new_file, msg='Result should not be None instead {0}'.format(
+            new_file
+        ))
+
+        content_new_file = DBSession.query(ContentRevisionRO) \
+            .filter(Content.label == 'new_file') \
+            .one()  # It must exist only one revision
+        eq_(
+            False,
+            content_new_file.is_deleted,
+            msg='Content should not be deleted !'
+        )
+        content_new_file_id = content_new_file.content_id
+
+        # Delete if
+        new_file.delete()
+
+        DBSession.flush()
+        content_w1f1d1 = DBSession.query(ContentRevisionRO) \
+            .filter(Content.content_id == content_new_file_id) \
+            .order_by(Content.revision_id.desc()) \
+            .first()
+        eq_(
+            True,
+            content_w1f1d1.is_deleted,
+            msg='Content should be deleted !'
+        )
+
+        result = provider.getResourceInst(
+            '/w1/w1f1/new_file.txt',
+            self._get_environ(
+                provider,
+                'bob@fsf.local',
+            )
+        )
+        eq_(None, result, msg='Result should be None instead {0}'.format(
+            result
+        ))
+
+        # Then create it again
+        new_file = self._put_new_text_file(
+            provider,
+            environ,
+            '/w1/w1f1/new_file.txt',
+            b'hello\n',
+        )
+        ok_(new_file, msg='Result should not be None instead {0}'.format(
+            new_file
+        ))
+
+        # Previous file is still dleeted
+        DBSession.flush()
+        content_w1f1d1 = DBSession.query(ContentRevisionRO) \
+            .filter(Content.content_id == content_new_file_id) \
+            .order_by(Content.revision_id.desc()) \
+            .first()
+        eq_(
+            True,
+            content_w1f1d1.is_deleted,
+            msg='Content should be deleted !'
+        )
+
+        # And an other file exist for this name
+        content_new_new_file = DBSession.query(ContentRevisionRO) \
+            .filter(Content.label == 'new_file') \
+            .order_by(Content.revision_id.desc()) \
+            .first()
+        ok_(
+            content_new_new_file.content_id != content_new_file_id,
+            msg='Contents ids should not be same !'
+        )
+        eq_(
+            False,
+            content_new_new_file.is_deleted,
+            msg='Content should not be deleted !'
+        )
+
+    def test_unit__rename_content__ok(self):
+        provider = self._get_provider()
+        environ = self._get_environ(
+            provider,
+            'bob@fsf.local',
+        )
+        w1f1d1 = provider.getResourceInst(
+            '/w1/w1f1/w1f1d1.txt',
+            environ,
+        )
+
+        content_w1f1d1 = DBSession.query(ContentRevisionRO) \
+            .filter(Content.label == 'w1f1d1') \
+            .one()  # It must exist only one revision, cf fixtures
+        ok_(content_w1f1d1, msg='w1f1d1 should be exist')
+        content_w1f1d1_id = content_w1f1d1.content_id
+
+        w1f1d1.moveRecursive('/w1/w1f1/w1f1d1_RENAMED.txt')
+
+        # Database content is renamed
+        content_w1f1d1 = DBSession.query(ContentRevisionRO) \
+            .filter(ContentRevisionRO.content_id == content_w1f1d1_id) \
+            .order_by(ContentRevisionRO.revision_id.desc()) \
+            .first()
+        eq_(
+            'w1f1d1_RENAMED',
+            content_w1f1d1.label,
+            msg='File should be labeled w1f1d1_RENAMED, not {0}'.format(
+                content_w1f1d1.label
+            )
+        )
+
+    def test_unit__move_content__ok(self):
+        provider = self._get_provider()
+        environ = self._get_environ(
+            provider,
+            'bob@fsf.local',
+        )
+        w1f1d1 = provider.getResourceInst(
+            '/w1/w1f1/w1f1d1.txt',
+            environ,
+        )
+
+        content_w1f1d1 = DBSession.query(ContentRevisionRO) \
+            .filter(Content.label == 'w1f1d1') \
+            .one()  # It must exist only one revision, cf fixtures
+        ok_(content_w1f1d1, msg='w1f1d1 should be exist')
+        content_w1f1d1_id = content_w1f1d1.content_id
+        content_w1f1d1_parent = content_w1f1d1.parent
+        eq_(
+            content_w1f1d1_parent.label,
+            'w1f1',
+            msg='field parent should be w1f1',
+        )
+
+        w1f1d1.moveRecursive('/w1/w1f2/w1f1d1.txt')  # move in f2
+
+        # Database content is moved
+        content_w1f1d1 = DBSession.query(ContentRevisionRO) \
+            .filter(ContentRevisionRO.content_id == content_w1f1d1_id) \
+            .order_by(ContentRevisionRO.revision_id.desc()) \
+            .first()
+        ok_(
+            content_w1f1d1.parent.label != content_w1f1d1_parent.label,
+            msg='file should be moved in w1f2 but is in {0}'.format(
+                content_w1f1d1.parent.label
+            )
+        )