add ContentAPI, UserApi, GroupApi and others core libApi + converted test to pyramid

tracim/lib/group.py

@@ -0,0 +1,23 @@
+# -*- coding: utf-8 -*-
+
+__author__ = 'damien'
+
+from tracim.models.auth import Group, User
+from sqlalchemy.orm import Query
+from sqlalchemy.orm import Session
+
+
+class GroupApi(object):
+
+    def __init__(self, session: Session, current_user: User):
+        self._user = current_user
+        self._session = session
+
+    def _base_query(self) -> Query:
+        return self._session.query(Group)
+
+    def get_one(self, group_id) -> Group:
+        return self._base_query().filter(Group.group_id == group_id).one()
+
+    def get_one_with_name(self, group_name) -> Group:
+        return self._base_query().filter(Group.group_name == group_name).one()

tracim/lib/userworkspace.py

@@ -0,0 +1,131 @@
+# -*- coding: utf-8 -*-
+
+__author__ = 'damien'
+
+from sqlalchemy.orm import Session
+from tracim.models.auth import User
+from tracim.models.data import Workspace
+from tracim.models.data import UserRoleInWorkspace
+from tracim.models.data import RoleType
+
+
+class RoleApi(object):
+
+    ALL_ROLE_VALUES = UserRoleInWorkspace.get_all_role_values()
+    # Dict containing readable members roles for given role
+    members_read_rights = {
+        UserRoleInWorkspace.NOT_APPLICABLE: [],
+        UserRoleInWorkspace.READER: [
+            UserRoleInWorkspace.WORKSPACE_MANAGER,
+        ],
+        UserRoleInWorkspace.CONTRIBUTOR: [
+            UserRoleInWorkspace.WORKSPACE_MANAGER,
+            UserRoleInWorkspace.CONTENT_MANAGER,
+            UserRoleInWorkspace.CONTRIBUTOR,
+        ],
+        UserRoleInWorkspace.CONTENT_MANAGER: [
+            UserRoleInWorkspace.WORKSPACE_MANAGER,
+            UserRoleInWorkspace.CONTENT_MANAGER,
+            UserRoleInWorkspace.CONTRIBUTOR,
+            UserRoleInWorkspace.READER,
+        ],
+        UserRoleInWorkspace.WORKSPACE_MANAGER: [
+            UserRoleInWorkspace.WORKSPACE_MANAGER,
+            UserRoleInWorkspace.CONTENT_MANAGER,
+            UserRoleInWorkspace.CONTRIBUTOR,
+            UserRoleInWorkspace.READER,
+        ],
+    }
+
+    @classmethod
+    def role_can_read_member_role(cls, reader_role: int, tested_role: int) \
+            -> bool:
+        """
+        :param reader_role: role as viewer
+        :param tested_role: role as viwed
+        :return: True if given role can view member role in workspace.
+        """
+        if reader_role in cls.members_read_rights:
+            return tested_role in cls.members_read_rights[reader_role]
+        return False
+
+    @classmethod
+    def create_role(cls) -> UserRoleInWorkspace:
+        role = UserRoleInWorkspace()
+
+        return role
+
+    def __init__(self, session: Session, current_user: User):
+        self._session = session
+        self._user = current_user
+
+    def _get_one_rsc(self, user_id, workspace_id):
+        """
+        :param user_id:
+        :param workspace_id:
+        :return: a Query object, filtered query but without fetching the object.
+        """
+        return self._session.query(UserRoleInWorkspace).\
+            filter(UserRoleInWorkspace.workspace_id == workspace_id).\
+            filter(UserRoleInWorkspace.user_id == user_id)
+
+    def get_one(self, user_id, workspace_id):
+        return self._get_one_rsc(user_id, workspace_id).one()
+
+    def create_one(
+            self,
+            user: User,
+            workspace: Workspace,
+            role_level: int,
+            with_notif: bool,
+            flush: bool=True
+    ) -> UserRoleInWorkspace:
+        role = self.create_role()
+        role.user_id = user.user_id
+        role.workspace = workspace
+        role.role = role_level
+        role.do_notify = with_notif
+        if flush:
+            self._session.flush()
+        return role
+
+    def delete_one(self, user_id, workspace_id, flush=True):
+        self._get_one_rsc(user_id, workspace_id).delete()
+        if flush:
+            self._session.flush()
+
+    def _get_all_for_user(self, user_id):
+        return self._session.query(UserRoleInWorkspace)\
+            .filter(UserRoleInWorkspace.user_id == user_id)
+
+    def get_all_for_user(self, user_id):
+        return self._get_all_for_user(user_id).all()
+
+    def get_all_for_user_order_by_workspace(
+            self,
+            user_id: int
+    ) -> UserRoleInWorkspace:
+        return self._get_all_for_user(user_id)\
+            .join(UserRoleInWorkspace.workspace).order_by(Workspace.label).all()
+
+    def get_all_for_workspace(self, workspace_id):
+        return self._session.query(UserRoleInWorkspace)\
+            .filter(UserRoleInWorkspace.workspace_id == workspace_id).all()
+
+    def save(self, role: UserRoleInWorkspace):
+        self._session.flush()
+
+    @classmethod
+    def get_roles_for_select_field(cls):
+        """
+
+        :return: list of DictLikeClass instances representing available Roles
+        (to be used in select fields)
+        """
+        result = list()
+
+        for role_id in UserRoleInWorkspace.get_all_role_values():
+            role = RoleType(role_id)
+            result.append(role)
+
+        return result

tracim/lib/utils.py

@@ -0,0 +1,35 @@
+import datetime
+
+def cmp_to_key(mycmp):
+    """
+    List sort related function
+
+    Convert a cmp= function into a key= function
+    """
+    class K(object):
+        def __init__(self, obj, *args):
+            self.obj = obj
+        def __lt__(self, other):
+            return mycmp(self.obj, other.obj) < 0
+        def __gt__(self, other):
+            return mycmp(self.obj, other.obj) > 0
+        def __eq__(self, other):
+            return mycmp(self.obj, other.obj) == 0
+        def __le__(self, other):
+            return mycmp(self.obj, other.obj) <= 0
+        def __ge__(self, other):
+            return mycmp(self.obj, other.obj) >= 0
+        def __ne__(self, other):
+            return mycmp(self.obj, other.obj) != 0
+    return K
+
+def current_date_for_filename() -> str:
+    """
+    ISO8601 current date, adapted to be used in filename (for
+    webdav feature for example), with trouble-free characters.
+    :return: current date as string like "2018-03-19T15.49.27.246592"
+    """
+    # INFO - G.M - 19-03-2018 - As ':' is in transform_to_bdd method in
+    # webdav utils, it may cause trouble. So, it should be replaced to
+    # a character which will not change in bdd.
+    return datetime.datetime.now().isoformat().replace(':', '.')

tracim/lib/workspace.py

@@ -0,0 +1,237 @@
+# -*- coding: utf-8 -*-
+import typing
+
+from sqlalchemy.orm import Query
+from sqlalchemy.orm import Session
+from tracim.translation import fake_translator as _
+
+from tracim.lib.userworkspace import RoleApi
+from tracim.models.auth import Group
+from tracim.models.auth import User
+from tracim.models.data import UserRoleInWorkspace
+from tracim.models.data import Workspace
+
+__author__ = 'damien'
+
+
+class WorkspaceApi(object):
+
+    def __init__(
+            self,
+            session: Session,
+            current_user: User,
+            force_role: bool=False
+    ):
+        """
+        :param current_user: Current user of context
+        :param force_role: If True, app role in queries even if admin
+        """
+        self._session = session
+        self._user = current_user
+        self._force_role = force_role
+
+    def _base_query_without_roles(self):
+        return self._session.query(Workspace).filter(Workspace.is_deleted == False)
+
+    def _base_query(self):
+        if not self._force_role and self._user.profile.id>=Group.TIM_ADMIN:
+            return self._base_query_without_roles()
+
+        return self._session.query(Workspace).\
+            join(Workspace.roles).\
+            filter(UserRoleInWorkspace.user_id == self._user.user_id).\
+            filter(Workspace.is_deleted == False)
+
+    def create_workspace(
+            self,
+            label: str='',
+            description: str='',
+            calendar_enabled: bool=False,
+            save_now: bool=False,
+    ) -> Workspace:
+        if not label:
+            label = self.generate_label()
+
+        workspace = Workspace()
+        workspace.label = label
+        workspace.description = description
+        workspace.calendar_enabled = calendar_enabled
+
+        # By default, we force the current user to be the workspace manager
+        # And to receive email notifications
+        role_api = RoleApi(
+            session=self._session,
+            current_user=self._user,
+        )
+
+        role = role_api.create_one(
+            self._user,
+            workspace,
+            UserRoleInWorkspace.WORKSPACE_MANAGER,
+            with_notif=True,
+        )
+
+        self._session.add(workspace)
+        self._session.add(role)
+
+        if save_now:
+            self._session.flush()
+
+        # TODO - G.M - 28-03-2018 - [Calendar] Reenable calendar stuff
+        # if calendar_enabled:
+        #     self._ensure_calendar_exist(workspace)
+        # else:
+        #     self._disable_calendar(workspace)
+
+        return workspace
+
+    def get_one(self, id):
+        return self._base_query().filter(Workspace.workspace_id == id).one()
+
+    def get_one_by_label(self, label: str) -> Workspace:
+        return self._base_query().filter(Workspace.label == label).one()
+
+    """
+    def get_one_for_current_user(self, id):
+        return self._base_query().filter(Workspace.workspace_id==id).\
+            session.query(ZKContact).filter(ZKContact.groups.any(ZKGroup.id.in_([1,2,3])))
+            filter(sqla.).one()
+    """
+
+    def get_all(self):
+        return self._base_query().all()
+
+    def get_all_for_user(self, user: User, ignored_ids=None):
+        workspaces = []
+
+        for role in user.roles:
+            if not role.workspace.is_deleted:
+                if not ignored_ids:
+                    workspaces.append(role.workspace)
+                elif role.workspace.workspace_id not in ignored_ids:
+                        workspaces.append(role.workspace)
+                else:
+                    pass  # do not return workspace
+
+        workspaces.sort(key=lambda workspace: workspace.label.lower())
+        return workspaces
+
+    def get_all_manageable(self) -> typing.List[Workspace]:
+        """Get all workspaces the current user has manager rights on."""
+        workspaces = []  # type: typing.List[Workspace]
+        if self._user.profile.id == Group.TIM_ADMIN:
+            workspaces = self._base_query().order_by(Workspace.label).all()
+        elif self._user.profile.id == Group.TIM_MANAGER:
+            workspaces = self._base_query() \
+                .filter(
+                    UserRoleInWorkspace.role ==
+                    UserRoleInWorkspace.WORKSPACE_MANAGER
+                ) \
+                .order_by(Workspace.label) \
+                .all()
+        return workspaces
+
+    def disable_notifications(self, user: User, workspace: Workspace):
+        for role in user.roles:
+            if role.workspace==workspace:
+                role.do_notify = False
+
+    def enable_notifications(self, user: User, workspace: Workspace):
+        for role in user.roles:
+            if role.workspace==workspace:
+                role.do_notify = True
+
+    def get_notifiable_roles(self, workspace: Workspace) -> [UserRoleInWorkspace]:
+        roles = []
+        for role in workspace.roles:
+            if role.do_notify==True \
+                    and role.user!=self._user \
+                    and role.user.is_active:
+                roles.append(role)
+        return roles
+
+    def save(self, workspace: Workspace):
+        self._session.flush()
+
+    def delete_one(self, workspace_id, flush=True):
+        workspace = self.get_one(workspace_id)
+        workspace.is_deleted = True
+
+        if flush:
+            self._session.flush()
+
+    def restore_one(self, workspace_id, flush=True):
+        workspace = self._session.query(Workspace)\
+            .filter(Workspace.is_deleted==True)\
+            .filter(Workspace.workspace_id==workspace_id).one()
+        workspace.is_deleted = False
+
+        if flush:
+            self._session.flush()
+
+        return workspace
+
+    def execute_created_workspace_actions(self, workspace: Workspace) -> None:
+        pass
+        # TODO - G.M - 28-03-2018 - [Calendar] Re-enable this calendar stuff
+        # self.ensure_calendar_exist(workspace)
+
+    # TODO - G.M - 28-03-2018 - [Calendar] Re-enable this calendar stuff
+    # def ensure_calendar_exist(self, workspace: Workspace) -> None:
+    #     # Note: Cyclic imports
+    #     from tracim.lib.calendar import CalendarManager
+    #     from tracim.model.organisational import WorkspaceCalendar
+    #
+    #     calendar_manager = CalendarManager(self._user)
+    #
+    #     try:
+    #         calendar_manager.enable_calendar_file(
+    #             calendar_class=WorkspaceCalendar,
+    #             related_object_id=workspace.workspace_id,
+    #             raise_=True,
+    #         )
+    #     # If previous calendar file no exist, calendar must be created
+    #     except FileNotFoundError:
+    #         self._user.ensure_auth_token()
+    #
+    #         # Ensure database is up-to-date
+    #         self.session.flush()
+    #         transaction.commit()
+    #
+    #         calendar_manager.create_then_remove_fake_event(
+    #             calendar_class=WorkspaceCalendar,
+    #             related_object_id=workspace.workspace_id,
+    #         )
+    #
+    # def disable_calendar(self, workspace: Workspace) -> None:
+    #     # Note: Cyclic imports
+    #     from tracim.lib.calendar import CalendarManager
+    #     from tracim.model.organisational import WorkspaceCalendar
+    #
+    #     calendar_manager = CalendarManager(self._user)
+    #     calendar_manager.disable_calendar_file(
+    #         calendar_class=WorkspaceCalendar,
+    #         related_object_id=workspace.workspace_id,
+    #         raise_=False,
+    #     )
+
+    def get_base_query(self) -> Query:
+        return self._base_query()
+
+    def generate_label(self) -> str:
+        """
+        :return: Generated workspace label
+        """
+        query = self._base_query_without_roles() \
+            .filter(Workspace.label.ilike('{0}%'.format(
+                _('Workspace'),
+            )))
+
+        return _('Workspace {}').format(
+            query.count() + 1,
+        )
+
+
+class UnsafeWorkspaceApi(WorkspaceApi):
+    def _base_query(self):
+        return self.session.query(Workspace).filter(Workspace.is_deleted==False)

tracim/models/__init__.py

@@ -21,7 +21,7 @@ def get_engine(settings, prefix='sqlalchemy.'):
 
 
 def get_session_factory(engine):
-    factory = sessionmaker()
+    factory = sessionmaker(expire_on_commit=False)
     factory.configure(bind=engine)
     return factory
 

tracim/models/revision_protection.py

@@ -63,7 +63,7 @@ class RevisionsIntegrity(object):
 
 @contextmanager
 def new_revision(
-        dbsession: Session,
+        session: Session,
         tm: TransactionManager,
         content: Content,
         force_create_new_revision: bool=False,
@@ -71,14 +71,14 @@ def new_revision(
     """
     Prepare context to update a Content. It will add a new updatable revision
     to the content.
-    :param dbsession: Database _session
+    :param session: Database _session
     :param tm: TransactionManager
     :param content: Content instance to update
     :param force_create_new_revision: Decide if new_rev should or should not
     be forced.
     :return:
     """
-    with dbsession.no_autoflush:
+    with session.no_autoflush:
         try:
             if force_create_new_revision \
                     or inspect(content.revision).has_identity:

tracim/tests/__init__.py

@@ -3,6 +3,9 @@ import transaction
 from depot.manager import DepotManager
 from pyramid import testing
 
+from nose.tools import eq_
+from tracim.lib.content import ContentApi
+from tracim.lib.workspace import WorkspaceApi
 from tracim.models.data import Workspace
 from tracim.models.data import Content
 from tracim.logger import logger
@@ -57,33 +60,6 @@ class BaseTest(unittest.TestCase):
         transaction.abort()
         DeclarativeBase.metadata.drop_all(self.engine)
 
-# class DefaultTest(object):
-#
-#     def _create_workspace_and_test(self, name, user) -> Workspace:
-#         """
-#         All extra parameters (*args, **kwargs) are for Workspace init
-#         :return: Created workspace instance
-#         """
-#         WorkspaceApi(user).create_workspace(name, save_now=True)
-#
-#         eq_(1, self._session.query(Workspace).filter(Workspace.label == name).count())
-#         return self._session.query(Workspace).filter(Workspace.label == name).one()
-#
-#     def _create_content_and_test(self, name, workspace, *args, **kwargs) -> Content:
-#         """
-#         All extra parameters (*args, **kwargs) are for Content init
-#         :return: Created Content instance
-#         """
-#         content = Content(*args, **kwargs)
-#         content.label = name
-#         content.workspace = workspace
-#         self._session.add(content)
-#         self._session.flush()
-#
-#         eq_(1, ContentApi.get_canonical_query().filter(Content.label == name).count())
-#         return ContentApi.get_canonical_query().filter(Content.label == name).one()
-
-
 class StandardTest(BaseTest):
     """
     BaseTest with default fixtures
@@ -96,3 +72,37 @@ class StandardTest(BaseTest):
             session=self.session,
             config=CFG(self.config.get_settings()))
         fixtures_loader.loads(self.fixtures)
+
+
+class DefaultTest(StandardTest):
+
+    def _create_workspace_and_test(self, name, user) -> Workspace:
+        """
+        All extra parameters (*args, **kwargs) are for Workspace init
+        :return: Created workspace instance
+        """
+        WorkspaceApi(
+            current_user=user,
+            session=self.session,
+        ).create_workspace(name, save_now=True)
+
+        eq_(1, self.session.query(Workspace).filter(Workspace.label == name).count())
+        return self.session.query(Workspace).filter(Workspace.label == name).one()
+
+    def _create_content_and_test(self, name, workspace, *args, **kwargs) -> Content:
+        """
+        All extra parameters (*args, **kwargs) are for Content init
+        :return: Created Content instance
+        """
+        content = Content(*args, **kwargs)
+        content.label = name
+        content.workspace = workspace
+        self.session.add(content)
+        self.session.flush()
+
+        content_api = ContentApi(
+            current_user=None,
+            session=self.session,
+        )
+        eq_(1, content_api.get_canonical_query().filter(Content.label == name).count())
+        return content_api.get_canonical_query().filter(Content.label == name).one()

tracim/tests/library/__init__.py

@@ -0,0 +1,2 @@
+# -*- coding: utf-8 -*-
+"""Unit test suite for the library of the application."""

tracim/tests/models/test_content.py

@@ -76,7 +76,7 @@ class TestContent(StandardTest):
         eq_(1, self.session.query(ContentRevisionRO).filter(ContentRevisionRO.label == 'TEST_CONTENT_1').count())
 
         with new_revision(
-                dbsession=self.session,
+                session=self.session,
                 tm=transaction.manager,
                 content=content
         ):
@@ -88,7 +88,7 @@ class TestContent(StandardTest):
         eq_(1, self.session.query(Content).filter(Content.id == created_content.id).count())
 
         with new_revision(
-                dbsession=self.session,
+                session=self.session,
                 tm=transaction.manager,
                 content=content
         ):