diff --git a/fittrackee/__init__.py b/fittrackee/__init__.py index 7ffd1967..b5f6492f 100644 --- a/fittrackee/__init__.py +++ b/fittrackee/__init__.py @@ -16,13 +16,13 @@ from flask_dramatiq import Dramatiq from flask_migrate import Migrate from flask_sqlalchemy import SQLAlchemy -from fittrackee.emails.email import Email +from fittrackee.emails.email import EmailService VERSION = __version__ = '0.5.7' db = SQLAlchemy() bcrypt = Bcrypt() migrate = Migrate() -email_service = Email() +email_service = EmailService() dramatiq = Dramatiq() log_file = os.getenv('APP_LOG') logging.basicConfig( diff --git a/fittrackee/emails/email.py b/fittrackee/emails/email.py index df84b069..187f3750 100644 --- a/fittrackee/emails/email.py +++ b/fittrackee/emails/email.py @@ -7,8 +7,9 @@ from typing import Dict, Optional, Type, Union from flask import Flask from jinja2 import Environment, FileSystemLoader, select_autoescape +from urllib3.util import parse_url -from .utils_email import parse_email_url +from .exceptions import InvalidEmailUrlScheme email_log = logging.getLogger('fittrackee_api_email') email_log.setLevel(logging.DEBUG) @@ -69,7 +70,7 @@ class EmailTemplate: return message.generate_message() -class Email: +class EmailService: def __init__(self, app: Optional[Flask] = None) -> None: self.host = 'localhost' self.port = 25 @@ -83,7 +84,7 @@ class Email: self.init_email(app) def init_email(self, app: Flask) -> None: - parsed_url = parse_email_url(app.config['EMAIL_URL']) + parsed_url = self.parse_email_url(app.config['EMAIL_URL']) self.host = parsed_url['host'] self.port = parsed_url['port'] self.use_tls = parsed_url['use_tls'] @@ -93,6 +94,23 @@ class Email: self.sender_email = app.config['SENDER_EMAIL'] self.email_template = EmailTemplate(app.config['TEMPLATES_FOLDER']) + @staticmethod + def parse_email_url(email_url: str) -> Dict: + parsed_url = parse_url(email_url) + if parsed_url.scheme != 'smtp': + raise InvalidEmailUrlScheme() + credentials = ( + parsed_url.auth.split(':') if parsed_url.auth else [None, None] + ) + return { + 'host': parsed_url.host, + 'port': 25 if parsed_url.port is None else parsed_url.port, + 'use_tls': True if parsed_url.query == 'tls=True' else False, + 'use_ssl': True if parsed_url.query == 'ssl=True' else False, + 'username': credentials[0], + 'password': credentials[1], + } + @property def smtp(self) -> Type[Union[smtplib.SMTP_SSL, smtplib.SMTP]]: return smtplib.SMTP_SSL if self.use_ssl else smtplib.SMTP diff --git a/fittrackee/tasks.py b/fittrackee/emails/tasks.py similarity index 100% rename from fittrackee/tasks.py rename to fittrackee/emails/tasks.py diff --git a/fittrackee/emails/utils_email.py b/fittrackee/emails/utils_email.py deleted file mode 100644 index f6a58f42..00000000 --- a/fittrackee/emails/utils_email.py +++ /dev/null @@ -1,22 +0,0 @@ -from typing import Dict - -from urllib3.util import parse_url - -from .exceptions import InvalidEmailUrlScheme - - -def parse_email_url(email_url: str) -> Dict: - parsed_url = parse_url(email_url) - if parsed_url.scheme != 'smtp': - raise InvalidEmailUrlScheme() - credentials = ( - parsed_url.auth.split(':') if parsed_url.auth else [None, None] - ) - return { - 'host': parsed_url.host, - 'port': 25 if parsed_url.port is None else parsed_url.port, - 'use_tls': True if parsed_url.query == 'tls=True' else False, - 'use_ssl': True if parsed_url.query == 'ssl=True' else False, - 'username': credentials[0], - 'password': credentials[1], - } diff --git a/fittrackee/tests/emails/test_email.py b/fittrackee/tests/emails/test_email_service.py similarity index 59% rename from fittrackee/tests/emails/test_email.py rename to fittrackee/tests/emails/test_email_service.py index 371caae4..9d999d71 100644 --- a/fittrackee/tests/emails/test_email.py +++ b/fittrackee/tests/emails/test_email_service.py @@ -1,9 +1,11 @@ from unittest.mock import Mock, patch +import pytest from flask import Flask from fittrackee import email_service from fittrackee.emails.email import EmailMessage +from fittrackee.emails.exceptions import InvalidEmailUrlScheme from ..api_test_case import CallArgsMixin from .template_results.password_reset_request import expected_en_text_body @@ -32,7 +34,62 @@ class TestEmailMessage: assert 'Hello !' in message_string -class TestEmailSending(CallArgsMixin): +class TestEmailServiceUrlParser(CallArgsMixin): + def test_it_raises_error_if_url_scheme_is_invalid(self) -> None: + url = 'stmp://username:password@localhost:587' + with pytest.raises(InvalidEmailUrlScheme): + email_service.parse_email_url(url) + + @staticmethod + def assert_parsed_email(url: str) -> None: + parsed_email = email_service.parse_email_url(url) + assert parsed_email['username'] is None + assert parsed_email['password'] is None + assert parsed_email['host'] == 'localhost' + assert parsed_email['port'] == 25 + assert parsed_email['use_tls'] is False + assert parsed_email['use_ssl'] is False + + def test_it_parses_email_url_without_port(self) -> None: + url = 'smtp://localhost' + self.assert_parsed_email(url) + + def test_it_parses_email_url_without_authentication(self) -> None: + url = 'smtp://localhost:25' + self.assert_parsed_email(url) + + def test_it_parses_email_url(self) -> None: + url = 'smtp://test@example.com:12345678@localhost:25' + parsed_email = email_service.parse_email_url(url) + assert parsed_email['username'] == 'test@example.com' + assert parsed_email['password'] == '12345678' + assert parsed_email['host'] == 'localhost' + assert parsed_email['port'] == 25 + assert parsed_email['use_tls'] is False + assert parsed_email['use_ssl'] is False + + def test_it_parses_email_url_with_tls(self) -> None: + url = 'smtp://test@example.com:12345678@localhost:587?tls=True' + parsed_email = email_service.parse_email_url(url) + assert parsed_email['username'] == 'test@example.com' + assert parsed_email['password'] == '12345678' + assert parsed_email['host'] == 'localhost' + assert parsed_email['port'] == 587 + assert parsed_email['use_tls'] is True + assert parsed_email['use_ssl'] is False + + def test_it_parses_email_url_with_ssl(self) -> None: + url = 'smtp://test@example.com:12345678@localhost:465?ssl=True' + parsed_email = email_service.parse_email_url(url) + assert parsed_email['username'] == 'test@example.com' + assert parsed_email['password'] == '12345678' + assert parsed_email['host'] == 'localhost' + assert parsed_email['port'] == 465 + assert parsed_email['use_tls'] is False + assert parsed_email['use_ssl'] is True + + +class TestEmailServiceSend(CallArgsMixin): email_data = { 'expiration_delay': '3 seconds', diff --git a/fittrackee/tests/emails/test_email_utils.py b/fittrackee/tests/emails/test_email_utils.py deleted file mode 100644 index c543246f..00000000 --- a/fittrackee/tests/emails/test_email_utils.py +++ /dev/null @@ -1,61 +0,0 @@ -import pytest - -from fittrackee.emails.utils_email import ( - InvalidEmailUrlScheme, - parse_email_url, -) - - -class TestEmailUrlParser: - def test_it_raises_error_if_url_scheme_is_invalid(self) -> None: - url = 'stmp://username:password@localhost:587' - with pytest.raises(InvalidEmailUrlScheme): - parse_email_url(url) - - @staticmethod - def assert_parsed_email(url: str) -> None: - parsed_email = parse_email_url(url) - assert parsed_email['username'] is None - assert parsed_email['password'] is None - assert parsed_email['host'] == 'localhost' - assert parsed_email['port'] == 25 - assert parsed_email['use_tls'] is False - assert parsed_email['use_ssl'] is False - - def test_it_parses_email_url_without_port(self) -> None: - url = 'smtp://localhost' - self.assert_parsed_email(url) - - def test_it_parses_email_url_without_authentication(self) -> None: - url = 'smtp://localhost:25' - self.assert_parsed_email(url) - - def test_it_parses_email_url(self) -> None: - url = 'smtp://test@example.com:12345678@localhost:25' - parsed_email = parse_email_url(url) - assert parsed_email['username'] == 'test@example.com' - assert parsed_email['password'] == '12345678' - assert parsed_email['host'] == 'localhost' - assert parsed_email['port'] == 25 - assert parsed_email['use_tls'] is False - assert parsed_email['use_ssl'] is False - - def test_it_parses_email_url_with_tls(self) -> None: - url = 'smtp://test@example.com:12345678@localhost:587?tls=True' - parsed_email = parse_email_url(url) - assert parsed_email['username'] == 'test@example.com' - assert parsed_email['password'] == '12345678' - assert parsed_email['host'] == 'localhost' - assert parsed_email['port'] == 587 - assert parsed_email['use_tls'] is True - assert parsed_email['use_ssl'] is False - - def test_it_parses_email_url_with_ssl(self) -> None: - url = 'smtp://test@example.com:12345678@localhost:465?ssl=True' - parsed_email = parse_email_url(url) - assert parsed_email['username'] == 'test@example.com' - assert parsed_email['password'] == '12345678' - assert parsed_email['host'] == 'localhost' - assert parsed_email['port'] == 465 - assert parsed_email['use_tls'] is False - assert parsed_email['use_ssl'] is True diff --git a/fittrackee/users/auth.py b/fittrackee/users/auth.py index de58731a..98564c3d 100644 --- a/fittrackee/users/auth.py +++ b/fittrackee/users/auth.py @@ -10,6 +10,7 @@ from werkzeug.exceptions import RequestEntityTooLarge from werkzeug.utils import secure_filename from fittrackee import appLog, bcrypt, db +from fittrackee.emails.tasks import reset_password_email from fittrackee.files import get_absolute_file_path from fittrackee.responses import ( ForbiddenErrorResponse, @@ -21,7 +22,6 @@ from fittrackee.responses import ( get_error_response_if_file_is_invalid, handle_error_and_return_response, ) -from fittrackee.tasks import reset_password_email from fittrackee.utils import get_readable_duration from fittrackee.workouts.models import Sport