-
Notifications
You must be signed in to change notification settings - Fork 10
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Email helper class that wraps EmailMessage * SMTPService class that creates connection to SMTP server using config options in services.smtp * send method on SMTPService that takes Email as an arg to send an email using the SMTP connection Signed-off-by: joseph-sentry <[email protected]>
- Loading branch information
1 parent
7c81440
commit 2ef32d9
Showing
4 changed files
with
89 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,13 @@ | ||
from email.message import EmailMessage | ||
|
||
|
||
class Email: | ||
def __init__( | ||
self, to_addr=None, from_addr=None, subject=None, text=None, html=None | ||
): | ||
self.message = EmailMessage() | ||
self.message["To"] = to_addr | ||
self.message["From"] = from_addr | ||
self.message["Subject"] = subject | ||
self.message.set_content(text) | ||
self.message.add_alternative(html, "text/html") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,44 @@ | ||
import smtplib | ||
import ssl | ||
from functools import cached_property | ||
|
||
from shared.config import get_config | ||
|
||
from helpers.email import Email | ||
|
||
_smtp_service = None | ||
|
||
|
||
def get_smtp_service(): | ||
if len(get_config("services", "smtp", default={})) == 0: | ||
return None | ||
return _get_cached_smtp_service() | ||
|
||
|
||
def _get_cached_smtp_service(): | ||
global _smtp_service | ||
if _smtp_service is None: | ||
_smtp_service = SMTPService() | ||
return _smtp_service | ||
|
||
|
||
class SMTPService: | ||
def __init__(self): | ||
self.host = get_config("services", "smtp", "host", default="mailhog") | ||
self.port = get_config("services", "smtp", "port", default=1025) | ||
self.username = get_config("services", "smtp", "username", default=None) | ||
self.password = get_config("services", "smtp", "password", default=None) | ||
self.ssl_context = ssl.create_default_context() | ||
|
||
self._conn = smtplib.SMTP( | ||
host=self.host, | ||
port=self.port, | ||
) | ||
self._conn.starttls(context=self.ssl_context) | ||
if self.username and self.password: | ||
self._conn.login(self.username, self.password) | ||
Check warning on line 39 in services/smtp.py Codecov Public QA / codecov/patchservices/smtp.py#L39
Check warning on line 39 in services/smtp.py Codecov - Staging / codecov/patchservices/smtp.py#L39
|
||
|
||
def send(self, email: Email): | ||
return self._conn.send_message( | ||
Check warning on line 42 in services/smtp.py Codecov Public QA / codecov/patchservices/smtp.py#L42
Check warning on line 42 in services/smtp.py Codecov - Staging / codecov/patchservices/smtp.py#L42
|
||
email.message, | ||
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,26 @@ | ||
from ssl import create_default_context | ||
from unittest.mock import MagicMock, call | ||
|
||
from services.smtp import get_smtp_service | ||
|
||
|
||
class TestStorage(object): | ||
def test_correct_init(self, mocker, mock_configuration): | ||
mocker.patch("smtplib.SMTP") | ||
m = mocker.patch("ssl.create_default_context", return_value=MagicMock()) | ||
service = get_smtp_service() | ||
service._conn.starttls.assert_called_with(context=m.return_value) | ||
|
||
def test_idempotent_service(self, mocker, mock_configuration): | ||
mocker.patch("smtplib.SMTP") | ||
first = get_smtp_service() | ||
second = get_smtp_service() | ||
assert id(first) == id(second) | ||
|
||
def test_idempotent_connection(self, mocker, mock_configuration): | ||
mocker.patch("smtplib.SMTP") | ||
first = get_smtp_service() | ||
first_conn = first._conn | ||
second = get_smtp_service() | ||
second_conn = second._conn | ||
assert id(first_conn) == id(second_conn) |