API - email service refactoring

This commit is contained in:
Sam 2022-02-16 13:35:38 +01:00
parent 04a89fafd8
commit 0b2e2ed5dd
7 changed files with 82 additions and 90 deletions

View File

@ -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(

View File

@ -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

View File

@ -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],
}

View File

@ -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',

View File

@ -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

View File

@ -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