Merge pull request #347 from SamR1/update-users-cli

Add CLI command to create user account

docsrc/source/cli.rst

@@ -100,6 +100,29 @@ Remove blacklisted tokens expired for more than provided number of days.
      - Number of days.
 
 
+``ftcli users create``
+""""""""""""""""""""""
+.. versionadded:: 0.7.15
+
+Create a user account.
+**Note**: the newly created account is active.
+
+.. cssclass:: table-bordered
+.. list-table::
+   :widths: 25 50
+   :header-rows: 1
+
+   * - Arguments/options
+     - Description
+   * - ``USERNAME``
+     - Username.
+   * - ``--email EMAIL``
+     - User email (mandatory).
+   * - ``--password PASSWORD``
+     - User password (if not provided, a random password is generated).
+
+
+
 ``ftcli users export_archives``
 """""""""""""""""""""""""""""""
 .. versionadded:: 0.7.13
@@ -129,8 +152,10 @@ Modify a user account (admin rights, active status, email and password).
    :widths: 25 50
    :header-rows: 1
 
-   * - Options
+   * - Arguments/options
      - Description
+   * - ``USERNAME``
+     - Username.
    * - ``--set-admin BOOLEAN``
      - Add/remove admin rights (when adding admin rights, it also activates user account if not active).
    * - ``--activate``

fittrackee/tests/users/test_users_utils.py

@@ -13,6 +13,7 @@ from flask import Flask
 from fittrackee import bcrypt, db
 from fittrackee.users.exceptions import (
     InvalidEmailException,
+    UserCreationException,
     UserNotFoundException,
 )
 from fittrackee.users.models import BlacklistedToken, User
@@ -32,7 +33,7 @@ from fittrackee.users.utils.token import (
 from ..utils import random_email, random_int, random_string
 
 
-class TestUserManagerService:
+class TestUserManagerServiceUserUpdate:
     def test_it_raises_exception_if_user_does_not_exist(
         self, app: Flask
     ) -> None:
@@ -190,6 +191,155 @@ class TestUserManagerService:
         assert user_1.confirmation_token is None
 
 
+class TestUserManagerServiceUserCreation:
+    def test_it_raises_exception_if_provided_username_is_invalid(
+        self, app: Flask
+    ) -> None:
+        user_manager_service = UserManagerService(username='.admin')
+        with pytest.raises(
+            UserCreationException,
+            match=(
+                'username: only alphanumeric characters and '
+                'the underscore character "_" allowed\n'
+            ),
+        ):
+            user_manager_service.create(email=random_email())
+
+    def test_it_raises_exception_if_a_user_exists_with_same_username(
+        self, app: Flask, user_1: User
+    ) -> None:
+        user_manager_service = UserManagerService(username=user_1.username)
+        with pytest.raises(
+            UserCreationException,
+            match='sorry, that username is already taken',
+        ):
+            user_manager_service.create(email=random_email())
+
+    def test_it_raises_exception_if_provided_email_is_invalid(
+        self, app: Flask
+    ) -> None:
+        user_manager_service = UserManagerService(username=random_string())
+        with pytest.raises(
+            UserCreationException, match='valid email must be provided'
+        ):
+            user_manager_service.create(email=random_string())
+
+    def test_it_raises_exception_if_a_user_exists_with_same_email(
+        self, app: Flask, user_1: User
+    ) -> None:
+        user_manager_service = UserManagerService(username=random_string())
+        with pytest.raises(
+            UserCreationException,
+            match='This user already exists. No action done.',
+        ):
+            user_manager_service.create(email=user_1.email)
+
+    def test_it_creates_user_with_provided_password(self, app: Flask) -> None:
+        username = random_string()
+        email = random_email()
+        password = random_string()
+        user_manager_service = UserManagerService(username=username)
+
+        new_user, user_password = user_manager_service.create(email, password)
+
+        assert new_user
+        assert new_user.username == username
+        assert new_user.email == email
+        assert bcrypt.check_password_hash(new_user.password, password)
+        assert user_password == password
+
+    def test_it_creates_user_when_password_is_not_provided(
+        self, app: Flask
+    ) -> None:
+        username = random_string()
+        email = random_email()
+        user_manager_service = UserManagerService(username=username)
+
+        new_user, user_password = user_manager_service.create(email)
+
+        assert new_user
+        assert new_user.username == username
+        assert new_user.email == email
+        assert bcrypt.check_password_hash(new_user.password, user_password)
+
+    def test_it_creates_when_registration_is_not_enabled(
+        self,
+        app_with_3_users_max: Flask,
+        user_1: User,
+        user_2: User,
+        user_3: User,
+    ) -> None:
+        username = random_string()
+        email = random_email()
+        user_manager_service = UserManagerService(username=username)
+
+        new_user, user_password = user_manager_service.create(email)
+
+        assert new_user
+        assert new_user.username == username
+        assert new_user.email == email
+        assert bcrypt.check_password_hash(new_user.password, user_password)
+
+    def test_created_user_is_inactive(self, app: Flask) -> None:
+        username = random_string()
+        user_manager_service = UserManagerService(username=username)
+
+        new_user, _ = user_manager_service.create(email=random_email())
+
+        assert new_user
+        assert new_user.is_active is False
+        assert new_user.confirmation_token is not None
+
+    def test_created_user_has_no_admin_rights(self, app: Flask) -> None:
+        username = random_string()
+        user_manager_service = UserManagerService(username=username)
+
+        new_user, _ = user_manager_service.create(email=random_email())
+
+        assert new_user
+        assert new_user.admin is False
+
+    def test_created_user_does_not_accept_privacy_policy(
+        self, app: Flask
+    ) -> None:
+        username = random_string()
+        user_manager_service = UserManagerService(username=username)
+
+        new_user, _ = user_manager_service.create(email=random_email())
+
+        assert new_user
+        assert new_user.accepted_policy_date is None
+
+    def test_created_user_timezone_is_europe_paris(self, app: Flask) -> None:
+        username = random_string()
+        user_manager_service = UserManagerService(username=username)
+
+        new_user, _ = user_manager_service.create(email=random_email())
+
+        assert new_user
+        assert new_user.timezone == 'Europe/Paris'
+
+    def test_created_user_date_format_is_MM_dd_yyyy(  # noqa
+        self, app: Flask
+    ) -> None:
+        username = random_string()
+        user_manager_service = UserManagerService(username=username)
+
+        new_user, _ = user_manager_service.create(email=random_email())
+
+        assert new_user
+        assert new_user.date_format == 'MM/dd/yyyy'
+
+    def test_created_user_language_is_en(self, app: Flask) -> None:
+        username = random_string()
+        user_manager_service = UserManagerService(username=username)
+
+        new_user, _ = user_manager_service.create(email=random_email())
+
+        assert new_user
+        assert new_user.language == 'en'
+
+
 class TestIsValidEmail:
     @pytest.mark.parametrize(
         ('input_email',),

fittrackee/users/auth.py

@@ -40,9 +40,11 @@ from fittrackee.responses import (
 from fittrackee.utils import get_readable_duration
 from fittrackee.workouts.models import Sport
 
+from .exceptions import UserControlsException, UserCreationException
 from .models import BlacklistedToken, User, UserDataExport, UserSportPreference
 from .tasks import export_data
-from .utils.controls import check_password, is_valid_email, register_controls
+from .utils.admin import UserManagerService
+from .utils.controls import check_password, is_valid_email
 from .utils.language import get_language
 from .utils.token import decode_user_token
 
@@ -160,32 +162,12 @@ def register_user() -> Union[Tuple[Dict, int], HttpResponse]:
     language = get_language(post_data.get('language'))
 
     try:
-        ret = register_controls(username, email, password)
+        user_manager_service = UserManagerService(username=username)
-    except TypeError as e:
+        new_user, _ = user_manager_service.create_user(email, password)
-        return handle_error_and_return_response(e, db=db)
+        # if a user exists with same email address (returned new_user is None),
-
+        # no error is returned since a user has to confirm his email to
-    if ret != '':
+        # activate his account
-        return InvalidPayloadErrorResponse(ret)
+        if new_user:
-
-    try:
-        user = User.query.filter(
-            func.lower(User.username) == func.lower(username)
-        ).first()
-        if user:
-            return InvalidPayloadErrorResponse(
-                'sorry, that username is already taken'
-            )
-
-        # if a user exists with same email address, no error is returned
-        # since a user has to confirm his email to activate his account
-        user = User.query.filter(
-            func.lower(User.email) == func.lower(email)
-        ).first()
-        if not user:
-            new_user = User(username=username, email=email, password=password)
-            new_user.timezone = 'Europe/Paris'
-            new_user.date_format = 'MM/dd/yyyy'
-            new_user.confirmation_token = secrets.token_urlsafe(30)
             new_user.language = language
             new_user.accepted_policy_date = datetime.datetime.utcnow()
             db.session.add(new_user)
@@ -195,7 +177,14 @@ def register_user() -> Union[Tuple[Dict, int], HttpResponse]:
 
         return {'status': 'success'}, 200
     # handler errors
-    except (exc.IntegrityError, exc.OperationalError, ValueError) as e:
+    except (UserControlsException, UserCreationException) as e:
+        return InvalidPayloadErrorResponse(str(e))
+    except (
+        exc.IntegrityError,
+        exc.OperationalError,
+        TypeError,
+        ValueError,
+    ) as e:
         return handle_error_and_return_response(e, db=db)
 
 

fittrackee/users/commands.py

@@ -4,6 +4,7 @@ from typing import Optional
 import click
 from humanize import naturalsize
 
+from fittrackee import db
 from fittrackee.cli.app import app
 from fittrackee.users.exceptions import UserNotFoundException
 from fittrackee.users.export_data import (
@@ -25,6 +26,32 @@ def users_cli() -> None:
     pass
 
 
+@users_cli.command('create')
+@click.argument('username')
+@click.option('--email', type=str, required=True, help='User email.')
+@click.option(
+    '--password',
+    type=str,
+    help='User password. If not provided, a random password is generated.',
+)
+def create_user(username: str, email: str, password: Optional[str]) -> None:
+    """Create an active user account."""
+    with app.app_context():
+        try:
+            user_manager_service = UserManagerService(username)
+            user, user_password = user_manager_service.create_user(
+                email=email, password=password
+            )
+            db.session.add(user)
+            db.session.commit()
+            user_manager_service.update(activate=True)
+            click.echo(f"User '{username}' created.")
+            if not password:
+                click.echo(f"The user password is: {user_password}")
+        except Exception as e:
+            click.echo(f'Error(s) occurred:\n{e}', err=True)
+
+
 @users_cli.command('update')
 @click.argument('username')
 @click.option(

fittrackee/users/exceptions.py

@@ -2,5 +2,13 @@ class InvalidEmailException(Exception):
     ...
 
 
+class UserControlsException(Exception):
+    ...
+
+
+class UserCreationException(Exception):
+    ...
+
+
 class UserNotFoundException(Exception):
     ...

fittrackee/users/utils/admin.py

@@ -1,11 +1,18 @@
 import secrets
 from typing import Optional, Tuple
 
+from sqlalchemy import func
+
 from fittrackee import db
 
-from ..exceptions import InvalidEmailException, UserNotFoundException
+from ..exceptions import (
+    InvalidEmailException,
+    UserControlsException,
+    UserCreationException,
+    UserNotFoundException,
+)
 from ..models import User
-from ..utils.controls import is_valid_email
+from ..utils.controls import is_valid_email, register_controls
 
 
 class UserManagerService:
@@ -80,3 +87,62 @@ class UserManagerService:
 
         db.session.commit()
         return user, user_updated, new_password
+
+    def create_user(
+        self,
+        email: str,
+        password: Optional[str] = None,
+        check_email: bool = False,
+    ) -> Tuple[Optional[User], Optional[str]]:
+        if not password:
+            password = secrets.token_urlsafe(30)
+
+        ret = register_controls(self.username, email, password)
+
+        if ret != '':
+            raise UserControlsException(ret)
+
+        user = User.query.filter(
+            func.lower(User.username) == func.lower(self.username)
+        ).first()
+        if user:
+            raise UserCreationException(
+                'sorry, that username is already taken'
+            )
+
+        # if a user exists with same email address, no error is returned
+        # since a user has to confirm his email to activate his account
+        user = User.query.filter(
+            func.lower(User.email) == func.lower(email)
+        ).first()
+        if user:
+            if check_email:
+                raise UserCreationException(
+                    'This user already exists. No action done.'
+                )
+            return None, None
+
+        new_user = User(username=self.username, email=email, password=password)
+        new_user.timezone = 'Europe/Paris'
+        new_user.date_format = 'MM/dd/yyyy'
+        new_user.confirmation_token = secrets.token_urlsafe(30)
+        db.session.add(new_user)
+        db.session.flush()
+
+        return new_user, password
+
+    def create(
+        self,
+        email: str,
+        password: Optional[str] = None,
+    ) -> Tuple[Optional[User], Optional[str]]:
+        try:
+            new_user, password = self.create_user(
+                email, password, check_email=True
+            )
+            if new_user:
+                new_user.language = 'en'
+                db.session.commit()
+        except UserControlsException as e:
+            raise UserCreationException(str(e))
+        return new_user, password