-
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 e83af8e
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) | ||
|
||
def send(self, email: Email): | ||
return self._conn.send_message( | ||
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 services.smtp import get_smtp_service | ||
|
||
from unittest.mock import call, MagicMock | ||
from ssl import create_default_context | ||
|
||
|
||
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) |