Idea for using composition rather than inheritance

This commit is contained in:
richard-dds 2018-10-16 13:04:30 -04:00
parent f13cc03d24
commit b930c85c3f
2 changed files with 65 additions and 59 deletions

View File

@ -1,8 +1,19 @@
from contextlib import contextmanager
import smtplib import smtplib
from email.message import EmailMessage from email.message import EmailMessage
class _HostConnection: class MailConnection(object):
def send(self, message):
raise NotImplementedError()
@property
def messages(self):
raise NotImplementedError()
class SMTPConnection(MailConnection):
def __init__(self, server, port, username, password, use_tls=False): def __init__(self, server, port, username, password, use_tls=False):
self.server = server self.server = server
self.port = port self.port = port
@ -11,7 +22,10 @@ class _HostConnection:
self.use_tls = use_tls self.use_tls = use_tls
self.host = None self.host = None
def __enter__(self): @contextmanager
def _host(self):
host = None
if self.use_tls: if self.use_tls:
self.host = smtplib.SMTP(self.server, self.port) self.host = smtplib.SMTP(self.server, self.port)
self.host.starttls() self.host.starttls()
@ -19,52 +33,20 @@ class _HostConnection:
self.host = smtplib.SMTP_SSL(self.server, self.port) self.host = smtplib.SMTP_SSL(self.server, self.port)
self.host.login(self.username, self.password) self.host.login(self.username, self.password)
return self.host yield host
def __exit__(self, *args): host.quit()
if self.host:
self.host.quit()
class BaseMailer:
def __init__(self, server, port, sender, password, use_tls=False):
self.server = server
self.port = port
self.sender = sender
self.password = password
self.use_tls = use_tls
def _message(self, recipients, subject, body):
msg = EmailMessage()
msg.set_content(body)
msg["From"] = self.sender
msg["To"] = ", ".join(recipients)
msg["Subject"] = subject
return msg
def send(self, recipients, subject, body):
pass
# do not collect messages by default
@property @property
def messages(self): def messages(self):
return [] return []
def send(self, message):
class Mailer(BaseMailer): with self._host() as host:
def connection(self): host.send_message(message)
return _HostConnection(
self.server, self.port, self.sender, self.password, use_tls=self.use_tls
)
def send(self, recipients, subject, body):
message = self._message(recipients, subject, body)
with self.connection() as conn:
conn.send_message(message)
class RedisMailer(BaseMailer): class RedisConnection(MailConnection):
def __init__(self, redis, **kwargs): def __init__(self, redis, **kwargs):
super().__init__(**kwargs) super().__init__(**kwargs)
self.redis = redis self.redis = redis
@ -77,6 +59,34 @@ class RedisMailer(BaseMailer):
def messages(self): def messages(self):
return [msg.decode() for msg in self.redis.lrange("atat_inbox", 0, -1)] return [msg.decode() for msg in self.redis.lrange("atat_inbox", 0, -1)]
def send(self, message):
self.redis.lpush("atat_inbox", str(message))
class Mailer(object):
def __init__(self, connection, sender):
self.connection = connection
self.sender = sender
def _message(self, recipients, subject, body):
msg = EmailMessage()
msg.set_content(body)
msg["From"] = self.sender
msg["To"] = ", ".join(recipients)
msg["Subject"] = subject
return msg
def send(self, recipients, subject, body): def send(self, recipients, subject, body):
message = self._message(recipients, subject, body) message = self._message(recipients, subject, body)
self.redis.lpush("atat_inbox", str(message)) self.connection.send(message)
@property
def messages(self):
return self.connection.messages
class RedisMailer(object):
def __init__(self, *args, **kwargs):
pass

View File

@ -1,44 +1,40 @@
import pytest import pytest
from atst.utils.mailer import Mailer, RedisMailer from atst.utils.mailer import Mailer, Mailer, MailConnection, RedisConnection
class MockHost: class MockConnection(MailConnection):
def __init__(self): def __init__(self):
self.messages = [] self._messages = []
def __enter__(self): def send(self, message):
return self self._messages.append(message)
def __exit__(self, *args): @property
pass def messages(self):
return self._messages
def send_message(self, message):
self.messages.append(message)
@pytest.fixture @pytest.fixture
def mail_host(): def mailer():
return MockHost() return Mailer(MockConnection(), "test@atat.com")
def test_mailer_can_send_mail(monkeypatch, mail_host): def test_mailer_can_send_mail(mailer):
monkeypatch.setattr("atst.utils.mailer.Mailer.connection", lambda *args: mail_host)
mailer = Mailer("localhost", 456, "leia@rebellion.net", "droidsyourelookingfor")
message_data = { message_data = {
"recipients": ["ben@tattoine.org"], "recipients": ["ben@tattoine.org"],
"subject": "help", "subject": "help",
"body": "you're my only hope", "body": "you're my only hope",
} }
mailer.send(**message_data) mailer.send(**message_data)
assert len(mail_host.messages) == 1 assert len(mailer.messages) == 1
message = mail_host.messages[0] message = mailer.messages[0]
assert message["To"] == message_data["recipients"][0] assert message["To"] == message_data["recipients"][0]
assert message["Subject"] == message_data["subject"] assert message["Subject"] == message_data["subject"]
assert message.get_content().strip() == message_data["body"] assert message.get_content().strip() == message_data["body"]
def test_redis_mailer_can_save_messages(app): def test_redis_mailer_can_save_messages(app):
mailer = RedisMailer(app.redis, server=None, port=None, sender=None, password=None) mailer = Mailer(RedisConnection(app.redis), "test@atat.com")
message_data = { message_data = {
"recipients": ["ben@tattoine.org"], "recipients": ["ben@tattoine.org"],
"subject": "help", "subject": "help",