Merge pull request #58 from tracim/fix/better_user_lib_for_authentication

tracim/exceptions.py

     pass
 
 
-class LoginFailed(TracimException):
+class AuthenticationFailed(TracimException):
     pass
+
+
+class WrongUserPassword(TracimException):
+    pass
+
+class UserNotExist(TracimException):
+    pass

tracim/lib/core/user.py

 import transaction
 import typing as typing
 
+from sqlalchemy.orm import Session
+
+from tracim import CFG
 from tracim.models.auth import User
+from sqlalchemy.orm.exc import NoResultFound
+from tracim.exceptions import WrongUserPassword, UserNotExist
+from tracim.exceptions import AuthenticationFailed
+from tracim.models.context_models import UserInContext
 
 
 class UserApi(object):
 
-    def __init__(self, current_user: typing.Optional[User], session, config):
+    def __init__(
+            self,
+            current_user: typing.Optional[User],
+            session: Session,
+            config: CFG,
+    ) -> None:
         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_user_with_context(self, user: User) -> UserInContext:
+        """
+        Return UserInContext object from User
+        """
+        user = UserInContext(
+            user=user,
+            dbsession=self._session,
+            config=self._config,
+        )
+        return user
+
+    # Getters
+
+    def get_one(self, user_id: int) -> User:
+        """
+        Get one user by user id
+        """
+        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_email(self, email: str) -> User:
+        """
+        Get one user by email
+        :param email: Email of the user
+        :return: one user
+        """
+        return self._base_query().filter(User.email == email).one()
 
+    # FIXME - G.M - 24-04-2018 - Duplicate method with get_one.
     def get_one_by_id(self, id: int) -> User:
-        return self._base_query().filter(User.user_id==id).one()
+        return self.get_one(user_id=id)
+
+    def get_current_user(self) -> User:
+        """
+        Get current_user
+        """
+        if not self._user:
+            raise UserNotExist()
+        return self._user
+
+    def get_all(self) -> typing.Iterable[User]:
+        return self._session.query(User).order_by(User.display_name).all()
+
+    # Check methods
+
+    def user_with_email_exists(self, email: str) -> bool:
+        try:
+            self.get_one_by_email(email)
+            return True
+        # TODO - G.M - 09-04-2018 - Better exception
+        except:
+            return False
+
+    def authenticate_user(self, email: str, password: str) -> User:
+        """
+        Authenticate user with email and password, raise AuthenticationFailed
+        if uncorrect.
+        :param email: email of the user
+        :param password: cleartext password of the user
+        :return: User who was authenticated.
+        """
+        try:
+            user = self.get_one_by_email(email)
+            if user.validate_password(password):
+                return user
+            else:
+                raise WrongUserPassword()
+        except (WrongUserPassword, NoResultFound):
+            raise AuthenticationFailed()
+
+    # Actions
 
     def update(
             self,
             email: str=None,
             do_save=True,
             timezone: str='',
-    ):
+    ) -> None:
         if name is not None:
             user.display_name = name
 
         if do_save:
             self.save(user)
 
-    def user_with_email_exists(self, email: str):
-        try:
-            self.get_one_by_email(email)
-            return True
-        # TODO - G.M - 09-04-2018 - Better exception
-        except:
-            return False
-
     def create_user(self, email=None, groups=[], save_now=False) -> User:
         user = User()
 

tracim/models/context_models.py

 from tracim.models.auth import Profile
 
 
+class LoginCredentials(object):
+    """
+    Login credentials model for login
+    """
+
+    def __init__(self, email: str, password: str):
+        self.email = email
+        self.password = password
+
+
 class UserInContext(object):
     """
     Interface to get User data and User data related to context.

tracim/tests/library/test_user_api.py

 
 import transaction
 
+from tracim.exceptions import UserNotExist, AuthenticationFailed
 from tracim.lib.core.user import UserApi
+from tracim.models import User
+from tracim.models.context_models import UserInContext
 from tracim.tests import DefaultTest
 from tracim.tests import eq_
 
 
 class TestUserApi(DefaultTest):
 
-    def test_create_and_update_user(self):
+    def test_unit__create_and_update_user__ok__nominal_case(self):
         api = UserApi(
             current_user=None,
             session=self.session,
         eq_('bob@bob', nu.email)
         eq_('bob', nu.display_name)
 
-    def test_user_with_email_exists(self):
+    def test_unit__user_with_email_exists__ok__nominal_case(self):
         api = UserApi(
             current_user=None,
             session=self.session,
         eq_(True, api.user_with_email_exists('bibi@bibi'))
         eq_(False, api.user_with_email_exists('unknown'))
 
-    def test_get_one_by_email(self):
+    def test_unit__get_one_by_email__ok__nominal_case(self):
         api = UserApi(
             current_user=None,
             session=self.session,
 
         eq_(uid, api.get_one_by_email('bibi@bibi').user_id)
 
-    def test_get_one_by_email_exception(self):
+    def test_unit__get_one_by_email__err__user_does_not_exist(self):
         api = UserApi(
             current_user=None,
             session=self.session,
         with pytest.raises(NoResultFound):
             api.get_one_by_email('unknown')
 
-    def test_get_all(self):
-        # TODO - G.M - 29-03-2018 Check why this method is not enabled
+    # def test_unit__get_all__ok__nominal_case(self):
+    #     # TODO - G.M - 29-03-2018 Check why this method is not enabled
+    #     api = UserApi(
+    #         current_user=None,
+    #         session=self.session,
+    #         config=self.config,
+    #     )
+    #     u1 = api.create_user(True)
+    #     u2 = api.create_user(True)
+    #     users = api.get_all()
+    #     assert 2==len(users)
+
+    def test_unit__get_one__ok__nominal_case(self):
         api = UserApi(
             current_user=None,
             session=self.session,
             config=self.config,
         )
-        # u1 = api.create_user(True)
-        # u2 = api.create_user(True)
+        u = api.create_user()
+        api.update(u, 'titi', 'titi@titi', True)
+        one = api.get_one(u.user_id)
+        eq_(u.user_id, one.user_id)
 
-        # users = api.get_all()
-        # ok_(2==len(users))
+    def test_unit__get_user_with_context__nominal_case(self):
+        user = User(
+            email='admin@tracim.tracim',
+            display_name='Admin',
+            is_active=True,
+        )
+        api = UserApi(
+            current_user=None,
+            session=self.session,
+            config=self.config,
+        )
+        new_user = api.get_user_with_context(user)
+        assert isinstance(new_user, UserInContext)
+        assert new_user.user == user
+        assert new_user.profile.name == 'nobody'
+        assert new_user.user_id == user.user_id
+        assert new_user.email == 'admin@tracim.tracim'
+        assert new_user.display_name == 'Admin'
+        assert new_user.is_active is True
+        # TODO - G.M - 03-05-2018 - [avatar][calendar] Should test this
+        # with true value when those param will be available.
+        assert new_user.avatar_url is None
+        assert new_user.calendar_url is None
 
-    def test_get_one(self):
+    def test_unit__get_current_user_ok__nominal_case(self):
+        user = User(email='admin@tracim.tracim')
+        api = UserApi(
+            current_user=user,
+            session=self.session,
+            config=self.config,
+        )
+        new_user = api.get_current_user()
+        assert isinstance(new_user, User)
+        assert user == new_user
+
+    def test_unit__get_current_user__err__user_not_exist(self):
         api = UserApi(
             current_user=None,
             session=self.session,
             config=self.config,
         )
-        u = api.create_user()
-        api.update(u, 'titi', 'titi@titi', True)
-        one = api.get_one(u.user_id)
-        eq_(u.user_id, one.user_id)
+        with pytest.raises(UserNotExist):
+            api.get_current_user()
+
+    def test_unit__authenticate_user___ok__nominal_case(self):
+        api = UserApi(
+            current_user=None,
+            session=self.session,
+            config=self.config,
+        )
+        user = api.authenticate_user('admin@admin.admin', 'admin@admin.admin')
+        assert isinstance(user, User)
+        assert user.email == 'admin@admin.admin'
+
+    def test_unit__authenticate_user___err__wrong_password(self):
+        api = UserApi(
+            current_user=None,
+            session=self.session,
+            config=self.config,
+        )
+        with pytest.raises(AuthenticationFailed):
+            api.authenticate_user('admin@admin.admin', 'wrong_password')
+
+    def test_unit__authenticate_user___err__wrong_user(self):
+        api = UserApi(
+            current_user=None,
+            session=self.session,
+            config=self.config,
+        )
+        with pytest.raises(AuthenticationFailed):
+            api.authenticate_user('unknown_user', 'wrong_password')

tracim/views/core_api/schemas.py

 # coding=utf-8
 import marshmallow
+from marshmallow import post_load
+
+from tracim.models.context_models import LoginCredentials, UserInContext
 
 
 class ProfileSchema(marshmallow.Schema):
 
 
 class UserSchema(marshmallow.Schema):
+
     user_id = marshmallow.fields.Int(dump_only=True)
     email = marshmallow.fields.Email(required=True)
     display_name = marshmallow.fields.String()
 
 
 class BasicAuthSchema(marshmallow.Schema):
+
     email = marshmallow.fields.Email(required=True)
     password = marshmallow.fields.String(required=True, load_only=True)
 
+    @post_load
+    def make_login(self, data):
+        return LoginCredentials(**data)
+
 
 class LoginOutputHeaders(marshmallow.Schema):
     expire_after = marshmallow.fields.String()

tracim/views/core_api/session_controller.py

 # coding=utf-8
 from pyramid.config import Configurator
-from sqlalchemy.orm.exc import NoResultFound
 try:  # Python 3.5+
     from http import HTTPStatus
 except ImportError:
     from http import client as HTTPStatus
 
-
 from tracim import TracimRequest
 from tracim.extensions import hapic
 from tracim.lib.core.user import UserApi
-from tracim.models.context_models import UserInContext
 from tracim.views.controllers import Controller
 from tracim.views.core_api.schemas import UserSchema
 from tracim.views.core_api.schemas import NoContentSchema
 from tracim.views.core_api.schemas import LoginOutputHeaders
 from tracim.views.core_api.schemas import BasicAuthSchema
 from tracim.exceptions import NotAuthentificated
-from tracim.exceptions import LoginFailed
+from tracim.exceptions import AuthenticationFailed
 
 
 class SessionController(Controller):
     @hapic.with_api_doc()
     @hapic.input_headers(LoginOutputHeaders())
     @hapic.input_body(BasicAuthSchema())
-    @hapic.handle_exception(LoginFailed, http_code=HTTPStatus.BAD_REQUEST)
+    @hapic.handle_exception(AuthenticationFailed, HTTPStatus.BAD_REQUEST)
     # TODO - G.M - 17-04-2018 - fix output header ?
     # @hapic.output_headers()
-    @hapic.output_body(
-        NoContentSchema(),
-        default_http_code=HTTPStatus.NO_CONTENT
-    )
+    @hapic.output_body(NoContentSchema(), default_http_code=HTTPStatus.NO_CONTENT)  # nopep8
     def login(self, context, request: TracimRequest, hapic_data=None):
         """
         Logs user into the system
         """
-        email = request.json_body['email']
-        password = request.json_body['password']
+
+        login = hapic_data.body
         app_config = request.registry.settings['CFG']
-        try:
-            uapi = UserApi(
-                None,
-                session=request.dbsession,
-                config=app_config,
-            )
-            user = uapi.get_one_by_email(email)
-            valid_password = user.validate_password(password)
-            if not valid_password:
-                # Bad password
-                raise LoginFailed('Bad Credentials')
-        except NoResultFound:
-            # User does not exist
-            raise LoginFailed('Bad Credentials')
-        return
+        uapi = UserApi(
+            None,
+            session=request.dbsession,
+            config=app_config,
+        )
+        return uapi.authenticate_user(login.email, login.password)
 
     @hapic.with_api_doc()
-    @hapic.output_body(
-        NoContentSchema(),
-        default_http_code=HTTPStatus.NO_CONTENT
-    )
+    @hapic.output_body(NoContentSchema(), default_http_code=HTTPStatus.NO_CONTENT)  # nopep8
     def logout(self, context, request: TracimRequest, hapic_data=None):
         """
         Logs out current logged in user session
         return
 
     @hapic.with_api_doc()
-    @hapic.handle_exception(
-        NotAuthentificated,
-        http_code=HTTPStatus.UNAUTHORIZED
-    )
-    @hapic.output_body(
-        UserSchema(),
-    )
+    @hapic.handle_exception(NotAuthentificated, HTTPStatus.UNAUTHORIZED)
+    @hapic.output_body(UserSchema(),)
     def whoami(self, context, request: TracimRequest, hapic_data=None):
         """
         Return current logged in user or 401
         """
         app_config = request.registry.settings['CFG']
-        return UserInContext(
-            user=request.current_user,
-            dbsession=request.dbsession,
+        uapi = UserApi(
+            request.current_user,
+            session=request.dbsession,
             config=app_config,
         )
+        user = uapi.get_current_user()  # User
+        return uapi.get_user_with_context(user)
 
     def bind(self, configurator: Configurator):