add UserAPI and fix most of the model tests

tracim/fixtures/__init__.py

     """ Fixture classes (list) required for this fixtures"""
     require = NotImplemented
 
-    def __init__(self, session):
+    def __init__(self, session, config):
         self._session = session
+        self._config = config
 
     def insert(self):
         raise NotImplementedError()
     Fixtures loader. Load each fixture once.
     """
 
-    def __init__(self, session, loaded=None):
+    def __init__(self, session, config, loaded=None):
         loaded = [] if loaded is None else loaded
         self._loaded = loaded
         self._session = session
+        self._config = config
 
     def loads(self, fixtures_classes):
         for fixture_class in fixtures_classes:
 
     def _load(self, fixture_class: Fixture):
         if fixture_class not in self._loaded:
-            fixture = fixture_class(self._session)
+            fixture = fixture_class(
+                session=self._session,
+                config=self._config,
+            )
             fixture.insert()
             self._loaded.append(fixture_class)
             self._session.flush()

tracim/fixtures/users_and_groups.py

 # -*- coding: utf-8 -*-
 from tracim import models
 from tracim.fixtures import Fixture
-#from tracim.lib.user import UserApi
+from tracim.lib.user import UserApi
 
 
 class Base(Fixture):
         u.email = 'admin@admin.admin'
         u.password = 'admin@admin.admin'
         self._session.add(u)
-        #uapi = UserApi(u)
-        #uapi.execute_created_user_actions(u)
+        uapi = UserApi(
+            session=self._session,
+            config=self._config,
+            current_user=u)
+        uapi.execute_created_user_actions(u)
 
         g1 = models.Group()
         g1.group_id = 1

tracim/lib/user.py

+# -*- coding: utf-8 -*-
+import threading
+
+import transaction
+import typing as typing
+
+from tracim.models.auth import User
+
+# TODO - G.M -28-03-2018 - Check if "current user" stuff is always needed for tracimv2
+# CURRENT_USER_WEB = 'WEB'
+# CURRENT_USER_WSGIDAV = 'WSGIDAV'
+
+
+class UserApi(object):
+
+    def __init__(self, current_user: User, session, config):
+        self._session = session
+        self._user = current_user
+        self._config = config
+
+    def get_all(self):
+        return self._session.query(User).order_by(User.display_name).all()
+
+    def _base_query(self):
+        return self._session.query(User)
+
+    def get_one(self, user_id: int):
+        return self._base_query().filter(User.user_id==user_id).one()
+
+    def get_one_by_email(self, email: str):
+        return self._base_query().filter(User.email==email).one()
+
+    def get_one_by_id(self, id: int) -> User:
+        return self._base_query().filter(User.user_id==id).one()
+
+    def update(
+            self,
+            user: User,
+            name: str=None,
+            email: str=None,
+            do_save=True,
+            timezone: str='',
+    ):
+        if name is not None:
+            user.display_name = name
+
+        if email is not None:
+            user.email = email
+
+        user.timezone = timezone
+
+        if do_save:
+            self.save(user)
+
+        if email and self._user and user.user_id==self._user.user_id:
+            pass
+            # this is required for the _session to keep on being up-to-date
+            # TODO - G.M - 28-03-2018 - Check for pyramid equivalent
+            # tg.request.identity['repoze.who.userid'] = email
+            # tg.auth_force_login(email)
+
+    def user_with_email_exists(self, email: str):
+        try:
+            self.get_one_by_email(email)
+            return True
+        except:
+            return False
+
+    def create_user(self, email=None, groups=[], save_now=False) -> User:
+        user = User()
+
+        if email:
+            user.email = email
+
+        for group in groups:
+            user.groups.append(group)
+
+        self._session.add(user)
+
+        if save_now:
+            self._session.flush()
+
+        return user
+
+    def save(self, user: User):
+        self._session.flush()
+
+    def execute_created_user_actions(self, created_user: User) -> None:
+        """
+        Execute actions when user just been created
+        :return:
+        """
+        # NOTE: Cyclic import
+        # TODO - G.M - 28-03-2018 - Reenable Calendar stuff
+        #from tracim.lib.calendar import CalendarManager
+        #from tracim.model.organisational import UserCalendar
+
+        created_user.ensure_auth_token(dbsession=self._session, validity_seconds=self._config.USER_AUTH_TOKEN_VALIDITY)
+
+        # Ensure database is up-to-date
+        self._session.flush()
+        transaction.commit()
+
+        # calendar_manager = CalendarManager(created_user)
+        # calendar_manager.create_then_remove_fake_event(
+        #     calendar_class=UserCalendar,
+        #     related_object_id=created_user.user_id,
+        # )
+
+
+# class CurrentUserGetterInterface(object):
+#     def get_current_user(self) -> typing.Union[None, User]:
+#         raise NotImplementedError()
+#
+#
+# class BaseCurrentUserGetter(CurrentUserGetterInterface):
+#     def __init__(self) -> None:
+#         self.api = UserApi(None)
+
+# TODO - G.M - 28-03-2018 - Check for pyramid equivalent
+# class WebCurrentUserGetter(BaseCurrentUserGetter):
+#     def get_current_user(self) -> typing.Union[None, User]:
+#         # HACK - D.A. - 2015-09-02
+#         # In tests, the tg.request.identity may not be set
+#         # (this is a buggy case, but for now this is how the software is;)
+#         if tg.request is not None:
+#             if hasattr(tg.request, 'identity'):
+#                 if tg.request.identity is not None:
+#                     return self.api.get_one_by_email(
+#                         tg.request.identity['repoze.who.userid'],
+#                     )
+#
+#         return None
+
+# TODO - G.M - 28-03-2018 - Reenable Webdav stuff
+# class WsgidavCurrentUserGetter(BaseCurrentUserGetter):
+#     def get_current_user(self) -> typing.Union[None, User]:
+#         if hasattr(cherrypy.request, 'current_user_email'):
+#             return self.api.get_one_by_email(
+#                 cherrypy.request.current_user_email,
+#             )
+#
+#         return None
+
+
+# class CurrentUserGetterApi(object):
+#     thread_local = threading.local()
+#     matches = {
+#         CURRENT_USER_WEB: WebCurrentUserGetter,
+#         CURRENT_USER_WSGIDAV: WsgidavCurrentUserGetter,
+#     }
+#     default = CURRENT_USER_WEB
+#
+#     @classmethod
+#     def get_current_user(cls) -> User:
+#         try:
+#             return cls.thread_local.getter.get_current_user()
+#         except AttributeError:
+#             return cls.factory(cls.default).get_current_user()
+#
+#     @classmethod
+#     def set_thread_local_getter(cls, name) -> None:
+#         if not hasattr(cls.thread_local, 'getter'):
+#             cls.thread_local.getter = cls.factory(name)
+#
+#     @classmethod
+#     def factory(cls, name: str) -> CurrentUserGetterInterface:
+#         return cls.matches[name]()

tracim/models/__init__.py

 from .revision_protection import prevent_content_revision_delete
 # import or define all models here to ensure they are attached to the
 # Base.metadata prior to any initialization routines
-from .mymodel import MyModel  # flake8: noqa
-from .auth import User, Group, Permission
-from .data import Content, ContentRevisionRO
+from tracim.models.mymodel import MyModel  # flake8: noqa
+from tracim.models.auth import User, Group, Permission
+from tracim.models.data import Content, ContentRevisionRO
 
 # run configure_mappers after defining all of the models to ensure
 # all relationships can be setup
     """
     Get a ``sqlalchemy.orm.Session`` instance backed by a transaction.
 
-    This function will hook the session to the transaction manager which
+    This function will hook the _session to the transaction manager which
     will take care of committing any changes.
 
     - When using pyramid_tm it will automatically be committed or aborted
       depending on whether an exception is raised.
 
-    - When using scripts you should wrap the session in a manager yourself.
+    - When using scripts you should wrap the _session in a manager yourself.
       For example::
 
           import transaction

tracim/models/data.py

 import json
 import os
 from datetime import datetime
+from typing import TYPE_CHECKING
 
 from babel.dates import format_timedelta
 from bs4 import BeautifulSoup
         if key in ('_sa_instance_state', ):  # Prevent infinite loop from SQLAlchemy code and altered set
             return super().__setattr__(key, value)
 
+        # FIXME - G.M - 28-03-2018 - Cycling Import
+        from tracim.models.revision_protection import RevisionsIntegrity
         if inspect(self).has_identity \
                 and key in self._cloned_columns \
                 and not RevisionsIntegrity.is_updatable(self):
 
     def has_new_information_for(self, user: User) -> bool:
         """
-        :param user: the session current user
+        :param user: the _session current user
         :return: bool, True if there is new information for given user else False
                        False if the user is None
         """
 
     def has_new_information_for(self, user: User) -> bool:
         """
-        :param user: the session current user
+        :param user: the _session current user
         :return: bool, True if there is new information for given user else False
                        False if the user is None
         """

tracim/models/revision_protection.py

 from tracim.exceptions import ContentRevisionUpdateError
 from tracim.exceptions import SameValueError
 
-from .data import ContentRevisionRO
-from .data import Content
-from .meta import DeclarativeBase
+from tracim.models.data import ContentRevisionRO
+from tracim.models.data import Content
+from tracim.models.meta import DeclarativeBase
 
 
 def prevent_content_revision_delete(
     """
     Prepare context to update a Content. It will add a new updatable revision
     to the content.
-    :param dbsession: Database session
+    :param dbsession: Database _session
     :param tm: TransactionManager
     :param content: Content instance to update
     :param force_create_new_revision: Decide if new_rev should or should not
             yield content
         except SameValueError or ValueError as e:
             # INFO - 20-03-2018 - renew transaction when error happened
-            # This avoid bad session data like new "temporary" revision
+            # This avoid bad _session data like new "temporary" revision
             # to be add when problem happen.
             tm.abort()
             tm.begin()

tracim/tests/__init__.py

 from tracim.logger import logger
 from tracim.fixtures import FixturesLoader
 from tracim.fixtures.users_and_groups import Base as BaseFixture
-
+from tracim.config import CFG
 
 class BaseTest(unittest.TestCase):
     """
     def setUp(self):
         logger.debug(self, 'Setup Test...')
         self.config = testing.setUp(settings={
-            'sqlalchemy.url': 'sqlite:///:memory:'
+            'sqlalchemy.url': 'sqlite:///:memory:',
+            'user.auth_token.validity':'604800',
+            'depot_storage_dir': '/tmp/test/depot',
+            'depot_storage_name': 'test',
+            'preview_cache_dir': '/tmp/test/preview_cache',
+
         })
         self.config.include('tracim.models')
         settings = self.config.get_settings()
 #         """
 #         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()
+#         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:
 #         """
 #         content = Content(*args, **kwargs)
 #         content.label = name
 #         content.workspace = workspace
-#         self.session.add(content)
-#         self.session.flush()
+#         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
-#     """
-#     fixtures = [BaseFixture]
-#
-#     def init_database(self):
-#         BaseTest.init_database(self)
-#         fixtures_loader = FixturesLoader([BaseFixture])
-#         fixtures_loader.loads(self.fixtures)
+
+
+class StandardTest(BaseTest):
+    """
+    BaseTest with default fixtures
+    """
+    fixtures = [BaseFixture]
+
+    def init_database(self):
+        BaseTest.init_database(self)
+        fixtures_loader = FixturesLoader(
+            session=self.session,
+            config=CFG(self.config.get_settings()))
+        fixtures_loader.loads(self.fixtures)

tracim/tests/models/test_content.py

 from nose.tools import raises
 from sqlalchemy.sql.elements import and_
 from sqlalchemy.testing import eq_
+import transaction
 
 # from tracim.lib.content import ContentApi
 from tracim.exceptions import ContentRevisionUpdateError
 from tracim.models.data import ContentRevisionRO
 from tracim.models.data import ContentType
 from tracim.models.data import Workspace
-from tracim.tests import BaseTest
+from tracim.tests import StandardTest
 
 
-class TestContent(BaseTest):
+class TestContent(StandardTest):
 
     @raises(ContentRevisionUpdateError)
     def test_update_without_prepare(self):
         content = self.session.query(Content).filter(Content.id == created_content.id).one()
         eq_(1, self.session.query(ContentRevisionRO).filter(ContentRevisionRO.label == 'TEST_CONTENT_1').count())
 
-        with new_revision(content):
+        with new_revision(
+                dbsession=self.session,
+                tm=transaction.manager,
+                content=content
+        ):
             time.sleep(0.00001)
             content.description = 'TEST_CONTENT_DESCRIPTION_1_UPDATED'
         self.session.flush()
         eq_(2, self.session.query(ContentRevisionRO).filter(ContentRevisionRO.label == 'TEST_CONTENT_1').count())
         eq_(1, self.session.query(Content).filter(Content.id == created_content.id).count())
 
-        with new_revision(content):
+        with new_revision(
+                dbsession=self.session,
+                tm=transaction.manager,
+                content=content
+        ):
             time.sleep(0.00001)
             content.description = 'TEST_CONTENT_DESCRIPTION_1_UPDATED_2'
             content.label = 'TEST_CONTENT_1_UPDATED_2'

tracim/tests/models/test_content_revision.py

 
     # TODO - G.M - 28-03-2018 - Reenable this test
     # def test_new_revision(self):
-    #     admin = self.session.query(User).filter(User.email == 'admin@admin.admin').one()
+    #     admin = self._session.query(User).filter(User.email == 'admin@admin.admin').one()
     #     workspace = self._create_workspace_and_test(name='workspace_1', user=admin)
     #     folder = self._create_content_and_test(name='folder_1', workspace=workspace, type=ContentType.Folder)
     #     page = self._create_content_and_test(
     #         owner=admin
     #     )
     #
-    #     self.session.flush()
+    #     self._session.flush()
     #
     #     # Model create a new instance with list of column
     #     new_revision_by_model = ContentRevisionRO.new_from(page.revision)