Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Habitability, coverage.

  • Loading branch information...
commit fb037e472e00a2bc03695e79133211c6e9a7a18d 1 parent 337ef20
@tseaver tseaver authored
Showing with 321 additions and 508 deletions.
  1. +95 −101 pyramid_mailer/mailer.py
  2. +226 −407 pyramid_mailer/tests/test_mailer.py
View
196 pyramid_mailer/mailer.py
@@ -1,115 +1,125 @@
+from datetime import datetime
+from os import makedirs
+from os.path import exists
+from os.path import join
+from random import sample
import smtplib
+from pyramid.settings import asbool
from repoze.sendmail.mailer import SMTPMailer
from repoze.sendmail.mailer import SendmailMailer
from repoze.sendmail.delivery import DirectMailDelivery
from repoze.sendmail.delivery import QueuedMailDelivery
-from pyramid.settings import asbool
-
from pyramid_mailer._compat import SMTP_SSL
-from os.path import exists, join
-from os import makedirs
-from datetime import datetime
-from random import sample
+
class DebugMailer(object):
+ """ Debug mailer for testing
+
+ Stores messages as files in the specified directory.
"""
- Debug mailer for testing
- """
- def __init__(self, top_level_directory='/tmp/app-messages'):
+ def __init__(self, top_level_directory):
if not exists(top_level_directory):
makedirs(top_level_directory)
self.tld = top_level_directory
@classmethod
def from_settings(cls, settings, prefix='mail.'):
- return cls()
+ """Create a new instance of 'DebugMailer' from settings dict.
- def send_impl(self, message, fail_silently=False):
+ :param settings: a settings dict-like
+ :param prefix: prefix separating 'pyramid_mailer' settings
"""
- Sends message to a file for debugging
+ settings = settings or {}
+ top_level_directory = settings.get(prefix+'top_level_directory')
+ if top_level_directory is None:
+ raise ValueError("DebugMailer: must specify "
+ "'%stop_level_directory'" % prefix)
+ return cls(top_level_directory)
+
+ def _send(self, message, fail_silently=False):
+ """Save message to a file for debugging
"""
seeds = '1234567890qwertyuiopasdfghjklzxcvbnm'
file_part1 = datetime.now().strftime('%Y%m%d%H%M%S')
file_part2 = ''.join(sample(seeds, 4))
- filename = join(self.tld,'%s_%s.msg' % (file_part1, file_part2))
+ filename = join(self.tld, '%s_%s.msg' % (file_part1, file_part2))
with open(filename, 'w') as fd:
fd.write(str(message.to_message()))
-
- send = send_impl
- send_immediately = send_impl
- send_to_queue = send_impl
- send_sendmail = send_impl
- send_immediately_sendmail = send_impl
+
+ send = _send
+ send_immediately = _send
+ send_to_queue = _send
+ send_sendmail = _send
+ send_immediately_sendmail = _send
+
class DummyMailer(object):
- """
- Dummy mailing instance, used for example in unit tests.
+ """Dummy mailer instance, used for example in unit tests.
+
+ Sent messages are appended to 'outbox' list.
- Keeps all sent messages internally in list as **outbox** property.
- Queued messages are instead added to **queue** property.
+ Queued messages are appended to 'queue' list.
"""
def __init__(self):
self.outbox = []
self.queue = []
- def send(self, message):
- """
- Mocks sending a transactional message. The message is added to the
- **outbox** list.
+ def send(self, message):
+ """Mock sending a transactional message via SMTP.
- :param message: a **Message** instance.
+ The message is appended to the 'outbox' list.
+
+ :param message: a 'Message' instance.
"""
self.outbox.append(message)
def send_immediately(self, message, fail_silently=False):
- """
- Mocks sending an immediate (non-transactional) message. The message
- is added to the **outbox** list.
+ """Mock sending an immediate (non-transactional) message.
+
+ The message is appended to the 'outbox' list.
:versionadded: 0.3
- :param message: a **Message** instance.
+ :param message: a 'Message' instance.
:param fail_silently: swallow connection errors (ignored here)
"""
self.outbox.append(message)
def send_to_queue(self, message):
- """
- Mocks sending to a maildir queue. The message is added to the **queue**
- list.
+ """Mock sending to a maildir queue.
- :param message: a **Message** instance.
+ The message is appended to the 'queue' list.
+
+ :param message: a 'Message' instance.
"""
self.queue.append(message)
def send_sendmail(self, message ):
- """
- Mocks sending a transactional message. The message is added to the
- **outbox** list.
+ """Mock sending a transactional message via sendmail.
- :param message: a **Message** instance.
+ The message is added to the 'outbox' list.
+
+ :param message: a 'Message' instance.
"""
self.outbox.append(message)
def send_immediately_sendmail(self, message, fail_silently=False):
- """
- Mocks sending an immediate (non-transactional) message. The message
- is added to the **outbox** list.
+ """Mock sending an immediate (non-transactional) message.
- :param message: a **Message** instance.
+ The message is added to the 'outbox' list.
+
+ :param message: a 'Message' instance.
:param fail_silently: swallow connection errors (ignored here)
"""
self.outbox.append(message)
class SMTP_SSLMailer(SMTPMailer):
+ """Subclass of SMTPMailer enabling SSL.
"""
- Subclass of SMTPMailer enabling SSL.
- """
-
smtp = SMTP_SSL
def __init__(self, *args, **kwargs):
@@ -119,22 +129,21 @@ def __init__(self, *args, **kwargs):
def smtp_factory(self):
if self.smtp is None:
- raise RuntimeError('No SMTP_SSL support in Python usable by mailer')
-
+ raise RuntimeError(
+ 'No SMTP_SSL support in Python usable by mailer')
+
connection = self.smtp(
self.hostname,
str(self.port),
keyfile=self.keyfile,
certfile=self.certfile
)
-
connection.set_debuglevel(self.debug_smtp)
return connection
class Mailer(object):
- """
- Manages sending of email messages.
+ """Manages sending of email messages.
:param host: SMTP hostname
:param port: SMTP port
@@ -142,7 +151,7 @@ class Mailer(object):
:param password: SMPT password
:param tls: use TLS
:param ssl: use SSL
- :param keyfile: SSL key file
+ :param keyfile: SSL key file
:param certfile: SSL certificate file
:param queue_path: path to maildir for queued messages
:param default_sender: default "from" address
@@ -153,11 +162,11 @@ class Mailer(object):
:param debug: SMTP debug level
"""
- def __init__(self,
- host='localhost',
- port=25,
+ def __init__(self,
+ host='localhost',
+ port=25,
username=None,
- password=None,
+ password=None,
tls=False,
ssl=False,
keyfile=None,
@@ -167,10 +176,7 @@ def __init__(self,
sendmail_app=None,
sendmail_template=None,
debug=0):
-
-
if ssl:
-
self.smtp_mailer = SMTP_SSLMailer(
hostname=host,
port=port,
@@ -181,16 +187,14 @@ def __init__(self,
debug_smtp=debug,
keyfile=keyfile,
certfile=certfile)
-
else:
-
self.smtp_mailer = SMTPMailer(
- hostname=host,
- port=port,
- username=username,
- password=password,
- no_tls=not(tls),
- force_tls=tls,
+ hostname=host,
+ port=port,
+ username=username,
+ password=password,
+ no_tls=not(tls),
+ force_tls=tls,
debug_smtp=debug)
self.direct_delivery = DirectMailDelivery(self.smtp_mailer)
@@ -199,28 +203,25 @@ def __init__(self,
self.queue_delivery = QueuedMailDelivery(queue_path)
else:
self.queue_delivery = None
-
+
self.sendmail_mailer = SendmailMailer(sendmail_app, sendmail_template)
self.sendmail_delivery = DirectMailDelivery(self.sendmail_mailer)
-
self.default_sender = default_sender
@classmethod
def from_settings(cls, settings, prefix='mail.'):
- """
- Creates a new instance of **Mailer** from settings dict.
+ """Create a new instance of 'Mailer' from settings dict.
:param settings: a settings dict-like
- :param prefix: prefix separating **pyramid_mailer** settings
+ :param prefix: prefix separating 'pyramid_mailer' settings
"""
-
settings = settings or {}
kwarg_names = [prefix + k for k in (
'host', 'port', 'username',
- 'password', 'tls', 'ssl', 'keyfile',
+ 'password', 'tls', 'ssl', 'keyfile',
'certfile', 'queue_path', 'debug', 'default_sender')]
-
+
size = len(prefix)
kwargs = dict(((k[size:], settings[k]) for k in settings.keys() if
@@ -234,30 +235,28 @@ def from_settings(cls, settings, prefix='mail.'):
return cls(**kwargs)
def send(self, message):
- """
- Sends a message. The message is handled inside a transaction, so
- in case of failure (or the message fails) the message will not be sent.
+ """Send a message.
- :param message: a **Message** instance.
- """
+ The message is handled inside a transaction, so in case of failure
+ (or the message fails) the message will not be sent.
+ :param message: a 'Message' instance.
+ """
return self.direct_delivery.send(*self._message_args(message))
def send_immediately(self, message, fail_silently=False):
- """
- Sends a message immediately, outside the transaction manager.
+ """Send a message immediately, outside the transaction manager.
- If there is a connection error to the mail server this will have to
+ If there is a connection error to the mail server this will have to
be handled manually. However if you pass ``fail_silently`` the error
will be swallowed.
:versionadded: 0.3
- :param message: a **Message** instance.
+ :param message: a 'Message' instance.
:param fail_silently: silently handle connection errors.
"""
-
try:
return self.smtp_mailer.send(*self._message_args(message))
except smtplib.socket.error:
@@ -265,40 +264,36 @@ def send_immediately(self, message, fail_silently=False):
raise
def send_to_queue(self, message):
- """
- Adds a message to a maildir queue.
-
- In order to handle this, the setting **mail.queue_path** must be
+ """Add a message to a maildir queue.
+
+ In order to handle this, the setting 'mail.queue_path' must be
provided and must point to a valid maildir.
- :param message: a **Message** instance.
+ :param message: a 'Message' instance.
"""
-
if not self.queue_delivery:
raise RuntimeError("No queue_path provided")
-
+
return self.queue_delivery.send(*self._message_args(message))
def _message_args(self, message):
message.sender = message.sender or self.default_sender
# convert Lamson message to Python email package msessage
- msg = message.to_message()
+ msg = message.to_message()
return (message.sender, message.send_to, msg)
def send_sendmail(self, message ):
- """
- Sends a message within the transaction manager.
+ """Send a message within the transaction manager.
Uses the local sendmail option
- :param message: a **Message** instance.
+ :param message: a 'Message' instance.
"""
return self.sendmail_delivery.send(*self._message_args(message))
def send_immediately_sendmail(self, message, fail_silently=False):
- """
- Sends a message immediately, outside the transaction manager.
+ """Send a message immediately, outside the transaction manager.
Uses the local sendmail option
@@ -306,11 +301,10 @@ def send_immediately_sendmail(self, message, fail_silently=False):
be handled manually. However if you pass ``fail_silently`` the error
will be swallowed.
- :param message: a **Message** instance.
+ :param message: a 'Message' instance.
:param fail_silently: silently handle connection errors.
"""
-
try:
return self.sendmail_mailer.send(*self._message_args(message))
except:
View
633 pyramid_mailer/tests/test_mailer.py
@@ -1,425 +1,297 @@
-import errno
import unittest
-class TestMailerSendmail(unittest.TestCase):
- def test_dummy_send_immediately_sendmail(self):
+class _Base(unittest.TestCase):
- from pyramid_mailer.mailer import DummyMailer
- from pyramid_mailer.message import Message
-
- mailer = DummyMailer()
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
+ _tempdir = None
- mailer.send_immediately_sendmail(msg)
-
- self.assertEqual(len(mailer.outbox), 1)
+ def tearDown(self):
+ if self._tempdir is not None:
+ from shutil import rmtree
+ rmtree(self._tempdir)
- def test_dummy_send_immediately_sendmail_and_fail_silently(self):
+ def _makeTempdir(self):
+ from tempfile import mkdtemp
+ if self._tempdir is None:
+ self._tempdir = mkdtemp()
+ return self._tempdir
- from pyramid_mailer.mailer import DummyMailer
- from pyramid_mailer.message import Message
- mailer = DummyMailer()
+class DebugMailerTests(_Base):
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
-
- mailer.send_immediately_sendmail(msg, True)
-
- self.assertEqual(len(mailer.outbox), 1)
+ def _getTargetClass(self):
+ from pyramid_mailer.mailer import DebugMailer
+ return DebugMailer
- def test_dummy_send_sendmail(self):
+ def _makeOne(self, tld=None):
+ if tld is None:
+ tld = self._makeTempdir()
+ return self._getTargetClass()(tld)
- from pyramid_mailer.mailer import DummyMailer
- from pyramid_mailer.message import Message
+ def _listFiles(self):
+ from os import listdir
+ return listdir(self._tempdir)
- mailer = DummyMailer()
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
+ def test___init___wo_existing_directory(self):
+ from os.path import isdir
+ from shutil import rmtree
+ tempdir = self._makeTempdir()
+ rmtree(tempdir)
+ mailer = self._makeOne(tempdir)
+ self.assertTrue(isdir(tempdir))
+ def test_from_settings(self):
+ tempdir = self._makeTempdir()
+ settings = {'mail.top_level_directory': tempdir}
+ mailer = self._getTargetClass().from_settings(settings, 'mail.')
+ self.assertEqual(mailer.tld, tempdir)
+
+ def test_from_settings_wo_tld(self):
+ tempdir = self._makeTempdir()
+ self.assertRaises(ValueError,
+ self._getTargetClass().from_settings, None)
+
+ def test__send(self):
+ mailer = self._makeOne()
+ msg = _makeMessage()
mailer.send_sendmail(msg)
+ files = self._listFiles()
+ self.assertEqual(len(files), 1)
- self.assertEqual(len(mailer.outbox), 1)
-
- def test_send_sendmail(self):
- from pyramid_mailer.mailer import Mailer
- from pyramid_mailer.message import Message
-
- mailer = Mailer()
-
- msg = Message(subject="test_send_sendmail",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="body-test_send_sendmail")
- mailer.send_sendmail(msg)
- def test_send_immediately_sendmail(self):
- email_sender = "sender@example.com"
- email_recipient = "tester@example.com"
- from pyramid_mailer.mailer import Mailer
- from pyramid_mailer.message import Message
-
- mailer = Mailer()
- sendmail_mailer = DummyMailer()
- mailer.sendmail_mailer = sendmail_mailer
-
- msg = Message(subject="test_send_immediately_sendmail",
- sender = email_sender,
- recipients = [ email_recipient ],
- body="body-test_send_immediately_sendmail")
- mailer.send_immediately_sendmail(msg)
- out = sendmail_mailer.out
- self.assertEqual(len(out), 1)
- first = out[0]
- self.assertEqual(first[0], 'sender@example.com')
- self.assertEqual(first[1], set(['tester@example.com']))
-
- def test_send_immediately_sendmail_with_exc_fail_silently(self):
- email_sender = "sender@example.com"
- email_recipient = "tester@example.com"
- from pyramid_mailer.mailer import Mailer
- from pyramid_mailer.message import Message
-
- mailer = Mailer()
- sendmail_mailer = DummyMailer(ValueError())
- mailer.sendmail_mailer = sendmail_mailer
-
- msg = Message(subject="test_send_immediately_sendmail",
- sender = email_sender,
- recipients = [ email_recipient ],
- body="body-test_send_immediately_sendmail")
- mailer.send_immediately_sendmail(msg, fail_silently=True)
- out = sendmail_mailer.out
- self.assertEqual(len(out), 0)
-
- def test_send_immediately_sendmail_with_exc_fail_loudly(self):
- email_sender = "sender@example.com"
- email_recipient = "tester@example.com"
- from pyramid_mailer.mailer import Mailer
- from pyramid_mailer.message import Message
-
- mailer = Mailer()
- sendmail_mailer = DummyMailer(ValueError())
- mailer.sendmail_mailer = sendmail_mailer
-
- msg = Message(subject="test_send_immediately_sendmail",
- sender = email_sender,
- recipients = [ email_recipient ],
- body="body-test_send_immediately_sendmail")
- self.assertRaises(ValueError, mailer.send_immediately_sendmail, msg)
-
-class TestMailer(unittest.TestCase):
-
- def test_dummy_send_immediately(self):
-
- from pyramid_mailer.mailer import DummyMailer
- from pyramid_mailer.message import Message
-
- mailer = DummyMailer()
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
-
- mailer.send_immediately(msg)
-
- self.assertEqual(len(mailer.outbox), 1)
-
- def test_dummy_send_immediately_and_fail_silently(self):
-
- from pyramid_mailer.mailer import DummyMailer
- from pyramid_mailer.message import Message
-
- mailer = DummyMailer()
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
-
- mailer.send_immediately(msg, True)
-
- self.assertEqual(len(mailer.outbox), 1)
-
- def test_dummy_send(self):
+class DummyMailerTests(unittest.TestCase):
+ def _getTargetClass(self):
from pyramid_mailer.mailer import DummyMailer
- from pyramid_mailer.message import Message
-
- mailer = DummyMailer()
+ return DummyMailer
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
+ def _makeOne(self):
+ return self._getTargetClass()()
+ def test_send(self):
+ mailer = self._makeOne()
+ msg = _makeMessage()
mailer.send(msg)
-
- self.assertEqual(len(mailer.outbox), 1)
-
- def test_dummy_send_to_queue(self):
-
- from pyramid_mailer.mailer import DummyMailer
- from pyramid_mailer.message import Message
-
- mailer = DummyMailer()
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
-
- mailer.send_to_queue(msg)
-
- self.assertEqual(len(mailer.queue), 1)
+ self.assertEqual(mailer.outbox, [msg])
def test_send_immediately(self):
+ mailer = self._makeOne()
+ msg = _makeMessage()
+ mailer.send_immediately(msg)
+ self.assertEqual(mailer.outbox, [msg])
- import socket
-
- from pyramid_mailer.mailer import Mailer
- from pyramid_mailer.message import Message
-
- mailer = Mailer(host='localhost', port='28322')
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
-
- self.assertRaises(socket.error,
- mailer.send_immediately,
- msg)
-
- def test_send_immediately_and_fail_silently(self):
-
- from pyramid_mailer.mailer import Mailer
- from pyramid_mailer.message import Message
-
- mailer = Mailer(host='localhost', port='28322')
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
-
- result = mailer.send_immediately(msg, True)
- self.assertEqual(result, None)
-
- def test_send_immediately_multipart(self):
-
- from pyramid_mailer.mailer import Mailer
- from pyramid_mailer.message import Message
-
- mailer = Mailer()
-
- utf_8_encoded = b'mo \xe2\x82\xac'
- utf_8 = utf_8_encoded.decode('utf_8')
-
- text_string = utf_8
- html_string = '<p>' + utf_8 + '</p>'
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body=text_string,
- html=html_string)
-
- smtp_mailer = DummyMailer()
-
- mailer.smtp_mailer = smtp_mailer
-
+ def test_send_immediately_w_fail_silently(self):
+ mailer = self._makeOne()
+ msg = _makeMessage()
mailer.send_immediately(msg, True)
+ self.assertEqual(mailer.outbox, [msg])
- self.assertEqual(len(smtp_mailer.out), 1)
+ def test_send_to_queue(self):
+ mailer = self._makeOne()
+ msg = _makeMessage()
+ mailer.send_to_queue(msg)
+ self.assertEqual(mailer.queue, [msg])
- def test_send(self):
+ def test_send_sendmail(self):
+ mailer = self._makeOne()
+ msg = _makeMessage()
+ mailer.send_sendmail(msg)
+ self.assertEqual(mailer.outbox, [msg])
- from pyramid_mailer.mailer import Mailer
- from pyramid_mailer.message import Message
+ def test_send_immediately_sendmail(self):
+ mailer = self._makeOne()
+ msg = _makeMessage()
+ mailer.send_immediately_sendmail(msg)
+ self.assertEqual(mailer.outbox, [msg])
- mailer = Mailer()
+ def test_send_immediately_sendmail_w_fail_silently(self):
+ mailer = self._makeOne()
+ msg = _makeMessage()
+ mailer.send_immediately_sendmail(msg, True)
+ self.assertEqual(mailer.outbox, [msg])
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
- smtp_mailer = DummyMailer()
- mailer.smtp_mailer = smtp_mailer
- mailer.send(msg)
- self.assertEqual(len(smtp_mailer.out), 0)
+class TestSMTP_SSLMailer(unittest.TestCase):
- def test_send_to_queue_unconfigured(self):
+ def _getTargetClass(self):
+ from pyramid_mailer.mailer import SMTP_SSLMailer
+ return SMTP_SSLMailer
- from pyramid_mailer.mailer import Mailer
- from pyramid_mailer.message import Message
+ def _makeOne(self, *arg, **kw):
+ return self._getTargetClass()(*arg, **kw)
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
- mailer = Mailer()
+ def test_ctor(self):
+ inst = self._makeOne(keyfile='keyfile', certfile='certfile')
+ self.assertEqual(inst.keyfile, 'keyfile')
+ self.assertEqual(inst.certfile, 'certfile')
- self.assertRaises(RuntimeError, mailer.send_to_queue, msg)
+ def test_smtp_factory_smtp_is_None(self):
+ inst = self._makeOne()
+ inst.smtp = None
+ self.assertRaises(RuntimeError, inst.smtp_factory)
- def test_send_to_queue(self):
+ def test_smtp_factory_smtp_is_not_None(self):
+ inst = self._makeOne(
+ debug_smtp=9,
+ hostname='hostname',
+ port=25,
+ certfile='certfile',
+ keyfile='keyfile'
+ )
+ inst.smtp = DummyConnectionFactory
+ conn = inst.smtp_factory()
+ self.assertEqual(conn.hostname, 'hostname')
+ self.assertEqual(conn.port, '25')
+ self.assertEqual(conn.certfile, 'certfile')
+ self.assertEqual(conn.keyfile, 'keyfile')
+ self.assertEqual(conn.debuglevel, 9)
- import os
- import tempfile
- import shutil
+class MailerTests(_Base):
+ def _getTargetClass(self):
from pyramid_mailer.mailer import Mailer
- from pyramid_mailer.message import Message
-
- tmpdir = tempfile.mkdtemp()
- try:
- test_queue = os.path.join(tmpdir, 'test_queue')
- for dir in ('cur', 'new', 'tmp'):
- os.makedirs(os.path.join(test_queue, dir))
-
- mailer = Mailer(queue_path=test_queue)
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
+ return Mailer
- queuedelivery = DummyMailer()
- mailer.queue_delivery = queuedelivery
+ def _makeOne(self, *args, **kw):
+ return self._getTargetClass()(*args, **kw)
- mailer.send_to_queue(msg)
- self.assertEqual(len(queuedelivery.out), 1)
- finally:
- shutil.rmtree(tmpdir, ignore_errors=True)
-
- def test_use_ssl_mailer(self):
-
- from pyramid_mailer.mailer import Mailer
- from pyramid_mailer._compat import SMTP_SSL
+ def test___init___w_ssl(self):
from smtplib import SMTP
-
- mailer = Mailer(ssl=True)
+ from pyramid_mailer._compat import SMTP_SSL
+ mailer = self._makeOne(ssl=True)
mailer
if SMTP_SSL is not None:
self.assertEqual(mailer.direct_delivery.mailer.smtp, SMTP_SSL)
-
else: # pragma: no cover
self.assertEqual(mailer.direct_delivery.mailer.smtp, SMTP)
- def test_from_settings_factory(self):
-
- from pyramid_mailer._compat import SMTP_SSL
+ def test_from_settings(self):
from smtplib import SMTP
- from pyramid_mailer import mailer_factory_from_settings
-
+ from pyramid_mailer._compat import SMTP_SSL
settings = {'mymail.host': 'my.server.com',
'mymail.port': 123,
'mymail.username': 'tester',
'mymail.password': 'test',
- 'mymail.tls': True,
+ 'mymail.tls': 'false',
'mymail.ssl': True,
'mymail.keyfile': 'ssl.key',
'mymail.certfile': 'ssl.crt',
'mymail.queue_path': '/tmp',
'mymail.debug': 1}
-
- mailer = mailer_factory_from_settings(settings, prefix='mymail.')
-
+ mailer = self._getTargetClass().from_settings(settings,
+ prefix='mymail.')
self.assertEqual(mailer.direct_delivery.mailer.hostname,
'my.server.com')
self.assertEqual(mailer.direct_delivery.mailer.port, 123)
self.assertEqual(mailer.direct_delivery.mailer.username, 'tester')
self.assertEqual(mailer.direct_delivery.mailer.password, 'test')
- self.assertEqual(mailer.direct_delivery.mailer.force_tls, True)
+ self.assertEqual(mailer.direct_delivery.mailer.force_tls, False)
if SMTP_SSL is not None:
self.assertEqual(mailer.direct_delivery.mailer.smtp, SMTP_SSL)
else: # pragma: no cover
self.assertEqual(mailer.direct_delivery.mailer.smtp, SMTP)
-
self.assertEqual(mailer.direct_delivery.mailer.keyfile, 'ssl.key')
self.assertEqual(mailer.direct_delivery.mailer.certfile, 'ssl.crt')
self.assertEqual(mailer.queue_delivery.queuePath, '/tmp')
self.assertEqual(mailer.direct_delivery.mailer.debug_smtp, 1)
- def test_from_settings(self):
+ def test_send_immediately(self):
+ import socket
+ mailer = self._makeOne(host='localhost', port='28322')
+ msg = _makeMessage()
+ self.assertRaises(socket.error,
+ mailer.send_immediately,
+ msg)
- from pyramid_mailer._compat import SMTP_SSL
- from smtplib import SMTP
- from pyramid_mailer.mailer import Mailer
+ def test_send_immediately_and_fail_silently(self):
+ mailer = self._makeOne(host='localhost', port='28322')
+ msg = _makeMessage()
- settings = {'mymail.host': 'my.server.com',
- 'mymail.port': 123,
- 'mymail.username': 'tester',
- 'mymail.password': 'test',
- 'mymail.tls': 'false',
- 'mymail.ssl': True,
- 'mymail.keyfile': 'ssl.key',
- 'mymail.certfile': 'ssl.crt',
- 'mymail.queue_path': '/tmp',
- 'mymail.debug': 1}
+ result = mailer.send_immediately(msg, True)
+ self.assertEqual(result, None)
- mailer = Mailer.from_settings(settings, prefix='mymail.')
+ def test_send_immediately_multipart(self):
+ mailer = self._makeOne()
+ utf_8_encoded = b'mo \xe2\x82\xac'
+ utf_8 = utf_8_encoded.decode('utf_8')
+ text_string = utf_8
+ html_string = '<p>' + utf_8 + '</p>'
+ msg = _makeMessage(body=text_string, html=html_string)
+ smtp_mailer = DummyMailer()
+ mailer.smtp_mailer = smtp_mailer
+ mailer.send_immediately(msg, True)
+ self.assertEqual(len(smtp_mailer.out), 1)
- self.assertEqual(mailer.direct_delivery.mailer.hostname,
- 'my.server.com')
- self.assertEqual(mailer.direct_delivery.mailer.port, 123)
- self.assertEqual(mailer.direct_delivery.mailer.username, 'tester')
- self.assertEqual(mailer.direct_delivery.mailer.password, 'test')
- self.assertEqual(mailer.direct_delivery.mailer.force_tls, False)
- if SMTP_SSL is not None:
- self.assertEqual(mailer.direct_delivery.mailer.smtp, SMTP_SSL)
- else: # pragma: no cover
- self.assertEqual(mailer.direct_delivery.mailer.smtp, SMTP)
+ def test_send(self):
+ mailer = self._makeOne()
+ msg = _makeMessage()
+ smtp_mailer = DummyMailer()
+ mailer.smtp_mailer = smtp_mailer
+ mailer.send(msg)
+ self.assertEqual(len(smtp_mailer.out), 0)
- self.assertEqual(mailer.direct_delivery.mailer.keyfile, 'ssl.key')
- self.assertEqual(mailer.direct_delivery.mailer.certfile, 'ssl.crt')
- self.assertEqual(mailer.queue_delivery.queuePath, '/tmp')
- self.assertEqual(mailer.direct_delivery.mailer.debug_smtp, 1)
+ def test_send_to_queue_unconfigured(self):
+ msg = _makeMessage()
+ mailer = self._makeOne()
+ self.assertRaises(RuntimeError, mailer.send_to_queue, msg)
-class TestSMTP_SSLMailer(unittest.TestCase):
- def _makeOne(self, *arg, **kw):
- from pyramid_mailer.mailer import SMTP_SSLMailer
- return SMTP_SSLMailer(*arg, **kw)
-
- def test_ctor(self):
- inst = self._makeOne(keyfile='keyfile', certfile='certfile')
- self.assertEqual(inst.keyfile, 'keyfile')
- self.assertEqual(inst.certfile, 'certfile')
-
- def test_smtp_factory_smtp_is_None(self):
- inst = self._makeOne()
- inst.smtp = None
- self.assertRaises(RuntimeError, inst.smtp_factory)
+ def test_send_to_queue(self):
+ import os
+ test_queue = os.path.join(self._makeTempdir(), 'test_queue')
+ for dir in ('cur', 'new', 'tmp'):
+ os.makedirs(os.path.join(test_queue, dir))
+ mailer = self._makeOne(queue_path=test_queue)
+ msg = _makeMessage()
+ queuedelivery = DummyMailer()
+ mailer.queue_delivery = queuedelivery
+ mailer.send_to_queue(msg)
+ self.assertEqual(len(queuedelivery.out), 1)
+
+ def test_send_sendmail(self):
+ mailer = self._makeOne()
+ msg = _makeMessage()
+ mailer.send_sendmail(msg)
+
+ def test_send_immediately_sendmail(self):
+ email_sender = "sender@example.com"
+ email_recipient = "tester@example.com"
+ mailer = self._makeOne()
+ sendmail_mailer = DummyMailer()
+ mailer.sendmail_mailer = sendmail_mailer
+ msg = _makeMessage(subject="test_send_immediately_sendmail",
+ body="body-test_send_immediately_sendmail")
+ mailer.send_immediately_sendmail(msg)
+ out = sendmail_mailer.out
+ self.assertEqual(len(out), 1)
+ first = out[0]
+ self.assertEqual(first[0], 'sender@example.com')
+ self.assertEqual(first[1], set(['tester@example.com']))
+
+ def test_send_immediately_sendmail_with_exc_fail_silently(self):
+ email_sender = "sender@example.com"
+ email_recipient = "tester@example.com"
+ mailer = self._makeOne()
+ sendmail_mailer = DummyMailer(ValueError())
+ mailer.sendmail_mailer = sendmail_mailer
+ msg = _makeMessage(subject="test_send_immediately_sendmail",
+ body="body-test_send_immediately_sendmail")
+ mailer.send_immediately_sendmail(msg, fail_silently=True)
+ out = sendmail_mailer.out
+ self.assertEqual(len(out), 0)
+
+ def test_send_immediately_sendmail_with_exc_fail_loudly(self):
+ email_sender = "sender@example.com"
+ email_recipient = "tester@example.com"
+ mailer = self._makeOne()
+ sendmail_mailer = DummyMailer(ValueError())
+ mailer.sendmail_mailer = sendmail_mailer
+ msg = _makeMessage(subject="test_send_immediately_sendmail",
+ body="body-test_send_immediately_sendmail")
+ self.assertRaises(ValueError, mailer.send_immediately_sendmail, msg)
- def test_smtp_factory_smtp_is_not_None(self):
- inst = self._makeOne(
- debug_smtp=9,
- hostname='hostname',
- port=25,
- certfile='certfile',
- keyfile='keyfile'
- )
- inst.smtp = DummyConnectionFactory
- conn = inst.smtp_factory()
- self.assertEqual(conn.hostname, 'hostname')
- self.assertEqual(conn.port, '25')
- self.assertEqual(conn.certfile, 'certfile')
- self.assertEqual(conn.keyfile, 'keyfile')
- self.assertEqual(conn.debuglevel, 9)
class DummyConnectionFactory(object):
+
def __init__(self, hostname, port, keyfile=None, certfile=None):
self.hostname = hostname
self.port = port
@@ -429,7 +301,9 @@ def __init__(self, hostname, port, keyfile=None, certfile=None):
def set_debuglevel(self, level):
self.debuglevel = level
+
class DummyMailer(object):
+
def __init__(self, raises=None):
self.out = []
self.raises = raises
@@ -440,69 +314,14 @@ def send(self, frm, to, msg):
self.out.append((frm, to, msg))
-class TestDebugMailer(unittest.TestCase):
-
- def setUp(self):
- from os.path import exists, join
- from os import makedirs, listdir, remove
- if exists('/tmp/app-messages'):
- for fl in listdir('/tmp/app-messages'):
- remove(join('/tmp/app-messages',fl))
-
- def check_number_files(self):
- from os.path import exists, join
- from os import makedirs, listdir, remove
- if exists('/tmp/app-messages'):
- return len(listdir('/tmp/app-messages'))
- else:
- return 0
-
-
- def test_debug_send_immediately_sendmail(self):
- from pyramid_mailer.mailer import DebugMailer
- from pyramid_mailer.message import Message
-
- mailer = DebugMailer()
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
-
- mailer.send_immediately_sendmail(msg)
-
- self.assertEqual(self.check_number_files(), 1)
-
- def test_debug_send_immediately_sendmail_and_fail_silently(self):
-
- from pyramid_mailer.mailer import DebugMailer
- from pyramid_mailer.message import Message
-
- mailer = DebugMailer()
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
-
- mailer.send_immediately_sendmail(msg, True)
-
- self.assertEqual(self.check_number_files(), 1)
-
- def test_debug_send_sendmail(self):
-
- from pyramid_mailer.mailer import DebugMailer
- from pyramid_mailer.message import Message
-
- mailer = DebugMailer()
-
- msg = Message(subject="testing",
- sender="sender@example.com",
- recipients=["tester@example.com"],
- body="test")
-
- mailer.send_sendmail(msg)
-
- self.assertEqual(self.check_number_files(), 1)
-
-
+def _makeMessage(subject="testing",
+ sender="sender@example.com",
+ recipients=["tester@example.com"],
+ body="test",
+ **kw):
+ from pyramid_mailer.message import Message
+ return Message(subject=subject,
+ sender=sender,
+ recipients=recipients,
+ body=body,
+ **kw)
Please sign in to comment.
Something went wrong with that request. Please try again.