API - disable emails sending when EMAIL_URL is not initialized

fittrackee/__init__.py

@@ -41,7 +41,7 @@ class CustomFlask(Flask):
     request_class = CustomRequest
 
 
-def create_app() -> Flask:
+def create_app(init_email: bool = True) -> Flask:
     # instantiate the app
     app = CustomFlask(
         __name__, static_folder='dist/static', template_folder='dist'
@@ -64,8 +64,15 @@ def create_app() -> Flask:
     migrate.init_app(app, db)
     dramatiq.init_app(app)
 
-    # set up email
+    # set up email if 'EMAIL_URL' is initialized
-    email_service.init_email(app)
+    if init_email:
+        if app.config['EMAIL_URL']:
+            email_service.init_email(app)
+            app.config['CAN_SEND_EMAILS'] = True
+        else:
+            appLog.warning(
+                'EMAIL_URL is not provided, email sending is deactivated.'
+            )
 
     # get configuration from database
     from .application.utils import (

fittrackee/application/app_config.py

@@ -42,6 +42,7 @@ def get_application_config() -> Union[Dict, HttpResponse]:
         "data": {
           "admin_contact": "admin@example.com",
           "gpx_limit_import": 10,
+          "is_email_sending_enabled": true,
           "is_registration_enabled": false,
           "max_single_file_size": 1048576,
           "max_users": 0,
@@ -91,6 +92,7 @@ def update_application_config(auth_user: User) -> Union[Dict, HttpResponse]:
         "data": {
           "admin_contact": "admin@example.com",
           "gpx_limit_import": 10,
+          "is_email_sending_enabled": true,
           "is_registration_enabled": false,
           "max_single_file_size": 1048576,
           "max_users": 10,

fittrackee/application/models.py

@@ -46,6 +46,7 @@ class AppConfig(BaseModel):
         return {
             'admin_contact': self.admin_contact,
             'gpx_limit_import': self.gpx_limit_import,
+            'is_email_sending_enabled': current_app.config['CAN_SEND_EMAILS'],
             'is_registration_enabled': self.is_registration_enabled,
             'max_single_file_size': self.max_single_file_size,
             'max_zip_file_size': self.max_zip_file_size,

fittrackee/cli/app.py

@@ -1,3 +1,3 @@
 from fittrackee import create_app
 
-app = create_app()
+app = create_app(init_email=False)

fittrackee/config.py

@@ -29,6 +29,7 @@ class BaseConfig:
     UI_URL = os.environ.get('UI_URL')
     EMAIL_URL = os.environ.get('EMAIL_URL')
     SENDER_EMAIL = os.environ.get('SENDER_EMAIL')
+    CAN_SEND_EMAILS = False
     DRAMATIQ_BROKER = broker
     TILE_SERVER = {
         'URL': os.environ.get(

fittrackee/tests/application/test_app_config_model.py

@@ -24,6 +24,7 @@ class TestConfigModel:
             serialized_app_config['gpx_limit_import']
             == app_config.gpx_limit_import
         )
+        assert serialized_app_config['is_email_sending_enabled'] is True
         assert serialized_app_config['is_registration_enabled'] is True
         assert (
             serialized_app_config['max_single_file_size']
@@ -49,3 +50,11 @@ class TestConfigModel:
 
         assert app_config.is_registration_enabled is False
         assert serialized_app_config['is_registration_enabled'] is False
+
+    def test_it_returns_email_sending_disabled_when_no_email_url_provided(
+        self, app_wo_email_activation: Flask, user_1: User, user_2: User
+    ) -> None:
+        app_config = AppConfig.query.first()
+        serialized_app_config = app_config.serialize()
+
+        assert serialized_app_config['is_email_sending_enabled'] is False

fittrackee/tests/fixtures/fixtures_app.py

@@ -146,6 +146,12 @@ def app_wo_email_auth(monkeypatch: pytest.MonkeyPatch) -> Generator:
     yield from get_app(with_config=True)
 
 
+@pytest.fixture
+def app_wo_email_activation(monkeypatch: pytest.MonkeyPatch) -> Generator:
+    monkeypatch.setenv('EMAIL_URL', '')
+    yield from get_app(with_config=True)
+
+
 @pytest.fixture
 def app_wo_domain() -> Generator:
     yield from get_app(with_config=True)

fittrackee/tests/users/test_auth_api.py

@@ -294,6 +294,31 @@ class TestUserRegistration(ApiTestCaseMixin):
             },
         )
 
+    def test_it_does_not_call_account_confirmation_email_when_email_sending_is_disabled(  # noqa
+        self,
+        app_wo_email_activation: Flask,
+        account_confirmation_email_mock: Mock,
+    ) -> None:
+        client = app_wo_email_activation.test_client()
+        email = self.random_email()
+        username = self.random_string()
+
+        response = client.post(
+            '/api/auth/register',
+            data=json.dumps(
+                dict(
+                    username=username,
+                    email=email,
+                    password='12345678',
+                )
+            ),
+            content_type='application/json',
+            environ_base={'HTTP_USER_AGENT': USER_AGENT},
+        )
+
+        assert response.status_code == 200
+        account_confirmation_email_mock.send.assert_not_called()
+
     @pytest.mark.parametrize(
         'text_transformation',
         ['upper', 'lower'],
@@ -773,6 +798,36 @@ class TestUserAccountUpdate(ApiTestCaseMixin):
         assert new_email == user_1.email_to_confirm
         assert user_1.confirmation_token is not None
 
+    def test_it_updates_email_when_email_sending_is_disabled(
+        self,
+        app_wo_email_activation: Flask,
+        user_1: User,
+        email_updated_to_current_address_mock: MagicMock,
+        email_updated_to_new_address_mock: MagicMock,
+        password_change_email_mock: MagicMock,
+    ) -> None:
+        client, auth_token = self.get_test_client_and_auth_token(
+            app_wo_email_activation, user_1.email
+        )
+        new_email = 'new.email@example.com'
+
+        response = client.patch(
+            '/api/auth/profile/edit/account',
+            content_type='application/json',
+            data=json.dumps(
+                dict(
+                    email=new_email,
+                    password='12345678',
+                )
+            ),
+            headers=dict(Authorization=f'Bearer {auth_token}'),
+        )
+
+        assert response.status_code == 200
+        assert user_1.email == new_email
+        assert user_1.email_to_confirm is None
+        assert user_1.confirmation_token is None
+
     def test_it_calls_email_updated_to_current_email_send_when_new_email_provided(  # noqa
         self,
         app: Flask,
@@ -1107,6 +1162,37 @@ class TestUserAccountUpdate(ApiTestCaseMixin):
         email_updated_to_new_address_mock.send.assert_called_once()
         password_change_email_mock.send.assert_called_once()
 
+    def test_it_does_not_calls_all_email_send_when_email_sending_is_disabled(
+        self,
+        app_wo_email_activation: Flask,
+        user_1: User,
+        email_updated_to_current_address_mock: MagicMock,
+        email_updated_to_new_address_mock: MagicMock,
+        password_change_email_mock: MagicMock,
+    ) -> None:
+        client, auth_token = self.get_test_client_and_auth_token(
+            app_wo_email_activation, user_1.email
+        )
+
+        client.patch(
+            '/api/auth/profile/edit/account',
+            content_type='application/json',
+            data=json.dumps(
+                dict(
+                    email='new.email@example.com',
+                    password='12345678',
+                    new_password=self.random_string(),
+                )
+            ),
+            headers=dict(Authorization=f'Bearer {auth_token}'),
+        )
+
+        self.assert_no_emails_sent(
+            email_updated_to_current_address_mock,
+            email_updated_to_new_address_mock,
+            password_change_email_mock,
+        )
+
 
 class TestUserPreferencesUpdate(ApiTestCaseMixin):
     def test_it_returns_error_if_payload_is_empty(
@@ -1648,6 +1734,21 @@ class TestPasswordResetRequest(ApiTestCaseMixin):
 
         self.assert_400(response)
 
+    def test_it_returns_error_when_email_sending_is_disabled(
+        self, app_wo_email_activation: Flask
+    ) -> None:
+        client = app_wo_email_activation.test_client()
+
+        response = client.post(
+            '/api/auth/password/reset-request',
+            data=json.dumps(dict(email='test@test.com')),
+            content_type='application/json',
+        )
+
+        self.assert_404_with_message(
+            response, 'the requested URL was not found on the server'
+        )
+
     def test_it_requests_password_reset_when_user_exists(
         self, app: Flask, user_1: User, user_reset_password_email: Mock
     ) -> None:
@@ -1873,7 +1974,7 @@ class TestPasswordUpdate(ApiTestCaseMixin):
         assert data['status'] == 'success'
         assert data['message'] == 'password updated'
 
-    def test_it_send_email_after_successful_update(
+    def test_it_sends_email_after_successful_update(
         self,
         app: Flask,
         user_1: User,
@@ -1908,6 +2009,29 @@ class TestPasswordUpdate(ApiTestCaseMixin):
             },
         )
 
+    def test_it_does_not_send_email_when_email_sending_is_disabled(
+        self,
+        app_wo_email_activation: Flask,
+        user_1: User,
+        password_change_email_mock: MagicMock,
+    ) -> None:
+        token = get_user_token(user_1.id, password_reset=True)
+        client = app_wo_email_activation.test_client()
+
+        client.post(
+            '/api/auth/password/update',
+            data=json.dumps(
+                dict(
+                    token=token,
+                    password=self.random_string(),
+                )
+            ),
+            content_type='application/json',
+            environ_base={'HTTP_USER_AGENT': USER_AGENT},
+        )
+
+        password_change_email_mock.send.assert_not_called()
+
 
 class TestEmailUpdateWitUnauthenticatedUser(ApiTestCaseMixin):
     def test_it_returns_error_if_token_is_missing(self, app: Flask) -> None:
@@ -2138,3 +2262,18 @@ class TestResendAccountConfirmationEmail(ApiTestCaseMixin):
                 ),
             },
         )
+
+    def test_it_returns_error_if_email_sending_is_disabled(
+        self, app_wo_email_activation: Flask, inactive_user: User
+    ) -> None:
+        client = app_wo_email_activation.test_client()
+
+        response = client.post(
+            '/api/auth/account/resend-confirmation',
+            data=json.dumps(dict(email=inactive_user.email)),
+            content_type='application/json',
+        )
+
+        self.assert_404_with_message(
+            response, 'the requested URL was not found on the server'
+        )

fittrackee/tests/users/test_users_api.py

@@ -1077,6 +1077,27 @@ class TestUpdateUser(ApiTestCaseMixin):
             },
         )
 
+    def test_it_does_not_call_password_change_email_when_email_sending_is_disabled(  # noqa
+        self,
+        app_wo_email_activation: Flask,
+        user_1_admin: User,
+        user_2: User,
+        user_password_change_email_mock: MagicMock,
+    ) -> None:
+        client, auth_token = self.get_test_client_and_auth_token(
+            app_wo_email_activation, user_1_admin.email
+        )
+
+        response = client.patch(
+            f'/api/users/{user_2.username}',
+            content_type='application/json',
+            data=json.dumps(dict(reset_password=True)),
+            headers=dict(Authorization=f'Bearer {auth_token}'),
+        )
+
+        assert response.status_code == 200
+        user_password_change_email_mock.send.assert_not_called()
+
     def test_it_calls_reset_password_email_when_password_reset_is_successful(
         self,
         app: Flask,
@@ -1118,6 +1139,27 @@ class TestUpdateUser(ApiTestCaseMixin):
             },
         )
 
+    def test_it_does_not_call_reset_password_email_when_email_sending_is_disabled(  # noqa
+        self,
+        app_wo_email_activation: Flask,
+        user_1_admin: User,
+        user_2: User,
+        user_reset_password_email: MagicMock,
+    ) -> None:
+        client, auth_token = self.get_test_client_and_auth_token(
+            app_wo_email_activation, user_1_admin.email
+        )
+
+        response = client.patch(
+            f'/api/users/{user_2.username}',
+            content_type='application/json',
+            data=json.dumps(dict(reset_password=True)),
+            headers=dict(Authorization=f'Bearer {auth_token}'),
+        )
+
+        assert response.status_code == 200
+        user_reset_password_email.send.assert_not_called()
+
     def test_it_returns_error_when_updating_email_with_invalid_address(
         self, app: Flask, user_1_admin: User, user_2: User
     ) -> None:
@@ -1229,6 +1271,28 @@ class TestUpdateUser(ApiTestCaseMixin):
             },
         )
 
+    def test_it_does_not_call_email_updated_to_new_address_when_email_sending_is_disabled(  # noqa
+        self,
+        app_wo_email_activation: Flask,
+        user_1_admin: User,
+        user_2: User,
+        user_email_updated_to_new_address_mock: MagicMock,
+    ) -> None:
+        client, auth_token = self.get_test_client_and_auth_token(
+            app_wo_email_activation, user_1_admin.email
+        )
+        new_email = 'new.' + user_2.email
+
+        response = client.patch(
+            f'/api/users/{user_2.username}',
+            content_type='application/json',
+            data=json.dumps(dict(new_email=new_email)),
+            headers=dict(Authorization=f'Bearer {auth_token}'),
+        )
+
+        assert response.status_code == 200
+        user_email_updated_to_new_address_mock.send.assert_not_called()
+
     def test_it_activates_user_account(
         self, app: Flask, user_1_admin: User, inactive_user: User
     ) -> None:

fittrackee/users/auth.py

@@ -40,25 +40,27 @@ from .utils.token import decode_user_token
 auth_blueprint = Blueprint('auth', __name__)
 
 HEX_COLOR_REGEX = regex = "^#([A-Fa-f0-9]{6}|[A-Fa-f0-9]{3})$"
+NOT_FOUND_MESSAGE = 'the requested URL was not found on the server'
 
 
 def send_account_confirmation_email(user: User) -> None:
-    ui_url = current_app.config['UI_URL']
+    if current_app.config['CAN_SEND_EMAILS']:
-    email_data = {
+        ui_url = current_app.config['UI_URL']
-        'username': user.username,
+        email_data = {
-        'fittrackee_url': ui_url,
+            'username': user.username,
-        'operating_system': request.user_agent.platform,  # type: ignore  # noqa
+            'fittrackee_url': ui_url,
-        'browser_name': request.user_agent.browser,  # type: ignore
+            'operating_system': request.user_agent.platform,  # type: ignore  # noqa
-        'account_confirmation_url': (
+            'browser_name': request.user_agent.browser,  # type: ignore
-            f'{ui_url}/account-confirmation'
+            'account_confirmation_url': (
-            f'?token={user.confirmation_token}'
+                f'{ui_url}/account-confirmation'
-        ),
+                f'?token={user.confirmation_token}'
-    }
+            ),
-    user_data = {
+        }
-        'language': 'en',
+        user_data = {
-        'email': user.email,
+            'language': 'en',
-    }
+            'email': user.email,
-    account_confirmation_email.send(user_data, email_data)
+        }
+        account_confirmation_email.send(user_data, email_data)
 
 
 @auth_blueprint.route('/auth/register', methods=['POST'])
@@ -505,7 +507,7 @@ def update_user_account(auth_user: User) -> Union[Dict, HttpResponse]:
     """
     update authenticated user email and password
 
-    It sends emails:
+    It sends emails if sending is enabled:
 
     - Password change
     - Email change:
@@ -634,8 +636,12 @@ def update_user_account(auth_user: User) -> Union[Dict, HttpResponse]:
     try:
         if email_to_confirm != auth_user.email:
             if is_valid_email(email_to_confirm):
-                auth_user.email_to_confirm = email_to_confirm
+                if current_app.config['CAN_SEND_EMAILS']:
-                auth_user.confirmation_token = secrets.token_urlsafe(30)
+                    auth_user.email_to_confirm = email_to_confirm
+                    auth_user.confirmation_token = secrets.token_urlsafe(30)
+                else:
+                    auth_user.email = email_to_confirm
+                    auth_user.confirmation_token = None
             else:
                 error_messages = 'email: valid email must be provided\n'
 
@@ -652,44 +658,48 @@ def update_user_account(auth_user: User) -> Union[Dict, HttpResponse]:
 
         db.session.commit()
 
-        ui_url = current_app.config['UI_URL']
+        if current_app.config['CAN_SEND_EMAILS']:
-        user_data = {
+            ui_url = current_app.config['UI_URL']
-            'language': (
+            user_data = {
-                'en' if auth_user.language is None else auth_user.language
+                'language': (
-            ),
+                    'en' if auth_user.language is None else auth_user.language
-            'email': auth_user.email,
+                ),
-        }
+                'email': auth_user.email,
-        data = {
-            'username': auth_user.username,
-            'fittrackee_url': ui_url,
-            'operating_system': request.user_agent.platform,
-            'browser_name': request.user_agent.browser,
-        }
-
-        if new_password is not None:
-            password_change_email.send(user_data, data)
-
-        if (
-            auth_user.email_to_confirm is not None
-            and auth_user.email_to_confirm != auth_user.email
-        ):
-            email_data = {
-                **data,
-                **{'new_email_address': email_to_confirm},
             }
-            email_updated_to_current_address.send(user_data, email_data)
+            data = {
-
+                'username': auth_user.username,
-            email_data = {
+                'fittrackee_url': ui_url,
-                **data,
+                'operating_system': request.user_agent.platform,
-                **{
+                'browser_name': request.user_agent.browser,
-                    'email_confirmation_url': (
-                        f'{ui_url}/email-update'
-                        f'?token={auth_user.confirmation_token}'
-                    )
-                },
             }
-            user_data = {**user_data, **{'email': auth_user.email_to_confirm}}
+
-            email_updated_to_new_address.send(user_data, email_data)
+            if new_password is not None:
+                password_change_email.send(user_data, data)
+
+            if (
+                auth_user.email_to_confirm is not None
+                and auth_user.email_to_confirm != auth_user.email
+            ):
+                email_data = {
+                    **data,
+                    **{'new_email_address': email_to_confirm},
+                }
+                email_updated_to_current_address.send(user_data, email_data)
+
+                email_data = {
+                    **data,
+                    **{
+                        'email_confirmation_url': (
+                            f'{ui_url}/email-update'
+                            f'?token={auth_user.confirmation_token}'
+                        )
+                    },
+                }
+                user_data = {
+                    **user_data,
+                    **{'email': auth_user.email_to_confirm},
+                }
+                email_updated_to_new_address.send(user_data, email_data)
 
         return {
             'status': 'success',
@@ -1139,6 +1149,8 @@ def request_password_reset() -> Union[Dict, HttpResponse]:
     """
     handle password reset request
 
+    If email sending is disabled, this endpoint is not available
+
     **Example request**:
 
     .. sourcecode:: http
@@ -1162,8 +1174,12 @@ def request_password_reset() -> Union[Dict, HttpResponse]:
 
     :statuscode 200: password reset request processed
     :statuscode 400: invalid payload
+    :statuscode 404: the requested URL was not found on the server
 
     """
+    if not current_app.config['CAN_SEND_EMAILS']:
+        return NotFoundErrorResponse(NOT_FOUND_MESSAGE)
+
     post_data = request.get_json()
     if not post_data or post_data.get('email') is None:
         return InvalidPayloadErrorResponse()
@@ -1203,6 +1219,8 @@ def update_password() -> Union[Dict, HttpResponse]:
     """
     update user password after password reset request
 
+    It sends emails if sending is enabled
+
     **Example request**:
 
     .. sourcecode:: http
@@ -1259,18 +1277,21 @@ def update_password() -> Union[Dict, HttpResponse]:
         ).decode()
         db.session.commit()
 
-        password_change_email.send(
+        if current_app.config['CAN_SEND_EMAILS']:
-            {
+            password_change_email.send(
-                'language': ('en' if user.language is None else user.language),
+                {
-                'email': user.email,
+                    'language': (
-            },
+                        'en' if user.language is None else user.language
-            {
+                    ),
-                'username': user.username,
+                    'email': user.email,
-                'fittrackee_url': current_app.config['UI_URL'],
+                },
-                'operating_system': request.user_agent.platform,
+                {
-                'browser_name': request.user_agent.browser,
+                    'username': user.username,
-            },
+                    'fittrackee_url': current_app.config['UI_URL'],
-        )
+                    'operating_system': request.user_agent.platform,
+                    'browser_name': request.user_agent.browser,
+                },
+            )
 
         return {
             'status': 'success',
@@ -1406,6 +1427,8 @@ def resend_account_confirmation_email() -> Union[Dict, HttpResponse]:
     """
     resend email with instructions to confirm account
 
+    If email sending is disabled, this endpoint is not available
+
     **Example request**:
 
     .. sourcecode:: http
@@ -1429,9 +1452,13 @@ def resend_account_confirmation_email() -> Union[Dict, HttpResponse]:
 
     :statuscode 200: confirmation email resent
     :statuscode 400: invalid payload
+    :statuscode 404: the requested URL was not found on the server
     :statuscode 500: error, please try again or contact the administrator
 
     """
+    if not current_app.config['CAN_SEND_EMAILS']:
+        return NotFoundErrorResponse(NOT_FOUND_MESSAGE)
+
     post_data = request.get_json()
     if not post_data or post_data.get('email') is None:
         return InvalidPayloadErrorResponse()

fittrackee/users/users.py

@@ -400,8 +400,9 @@ def update_user(auth_user: User, user_name: str) -> Union[Dict, HttpResponse]:
     Update user account
 
     - add/remove admin rights (regardless user account status)
-    - reset password (and send email to update user password)
+    - reset password (and send email to update user password,
-    - update user email (and send email to update user password)
+      if sending enabled)
+    - update user email (and send email to new user email, if sending enabled)
     - activate account for an inactive user
 
     Only user with admin rights can modify another user
@@ -527,52 +528,56 @@ def update_user(auth_user: User, user_name: str) -> Union[Dict, HttpResponse]:
             new_email=new_email,
         )
 
-        user_language = 'en' if user.language is None else user.language
+        if current_app.config['CAN_SEND_EMAILS']:
-        ui_url = current_app.config['UI_URL']
+            user_language = 'en' if user.language is None else user.language
-        if reset_password:
+            ui_url = current_app.config['UI_URL']
-            user_data = {
+            if reset_password:
-                'language': user_language,
+                user_data = {
-                'email': user.email,
+                    'language': user_language,
-            }
+                    'email': user.email,
-            password_change_email.send(
+                }
-                user_data,
+                password_change_email.send(
-                {
+                    user_data,
-                    'username': user.username,
+                    {
-                    'fittrackee_url': ui_url,
+                        'username': user.username,
-                },
+                        'fittrackee_url': ui_url,
-            )
+                    },
-            password_reset_token = user.encode_password_reset_token(user.id)
+                )
-            reset_password_email.send(
+                password_reset_token = user.encode_password_reset_token(
-                user_data,
+                    user.id
-                {
+                )
-                    'expiration_delay': get_readable_duration(
+                reset_password_email.send(
-                        current_app.config[
+                    user_data,
-                            'PASSWORD_TOKEN_EXPIRATION_SECONDS'
+                    {
-                        ],
+                        'expiration_delay': get_readable_duration(
-                        user_language,
+                            current_app.config[
-                    ),
+                                'PASSWORD_TOKEN_EXPIRATION_SECONDS'
-                    'username': user.username,
+                            ],
-                    'password_reset_url': (
+                            user_language,
-                        f'{ui_url}/password-reset?token={password_reset_token}'
+                        ),
-                    ),
+                        'username': user.username,
-                    'fittrackee_url': ui_url,
+                        'password_reset_url': (
-                },
+                            f'{ui_url}/password-reset?'
-            )
+                            f'token={password_reset_token}'
+                        ),
+                        'fittrackee_url': ui_url,
+                    },
+                )
 
-        if new_email:
+            if new_email:
-            user_data = {
+                user_data = {
-                'language': user_language,
+                    'language': user_language,
-                'email': user.email_to_confirm,
+                    'email': user.email_to_confirm,
-            }
+                }
-            email_data = {
+                email_data = {
-                'username': user.username,
+                    'username': user.username,
-                'fittrackee_url': ui_url,
+                    'fittrackee_url': ui_url,
-                'email_confirmation_url': (
+                    'email_confirmation_url': (
-                    f'{ui_url}/email-update'
+                        f'{ui_url}/email-update'
-                    f'?token={user.confirmation_token}'
+                        f'?token={user.confirmation_token}'
-                ),
+                    ),
-            }
+                }
-            email_updated_to_new_address.send(user_data, email_data)
+                email_updated_to_new_address.send(user_data, email_data)
 
         return {
             'status': 'success',