Commits

Matthew Frazier committed 8e767fc

added cartero.senders.console

Comments (0)

Files changed (2)

cartero/senders/console.py

+# -*- coding: utf-8 -*-
+"""
+cartero.senders.console
+=======================
+This contains the `ConsoleSender` class, which dumps messages received to
+stdout or stderr. It's mostly only useful for debugging.
+
+:copyright: 2010 Matthew "LeafStorm" Frazier
+:license:   MIT/X11, see LICENSE for details
+"""
+import sys
+from cartero.encoding import MIMEEncodingSender
+from cartero.message import dump_text
+from cartero.sender import Sender, send_wrapper, senderlog
+from cartero.utils.config import Option, Choice, Boolean
+from cartero.utils.general import contextmanager
+from threading import RLock
+
+
+class ConsoleSender(MIMEEncodingSender):
+    """
+    This dumps mail messages to `sys.stdout` or `sys.stderr`. It takes the
+    options:
+    
+    `stream`
+        Either ``"stdout"`` or ``"stderr"``, indicating the stream to dump the
+        messages to.
+    
+    `format`
+        Either ``"friendly"`` or ``"mime"``. Friendly is human-readable (the
+        output of `cartero.message.dump_text`), and mime is the full
+        MIME-encoded version as would go through an SMTP server.
+    
+    `attachment_data`
+        Whether or not to include an attachment's full data in "friendly"
+        mode, or just its name, content type, and length.
+    """
+    options = {
+        'stream': Choice(('stdout', 'stderr'), default='stdout'),
+        'format': Choice(('friendly', 'mime'), default='friendly'),
+        'attachment_data': Boolean(default=False)
+    }
+    
+    def _setup(self):
+        self._lock = RLock()
+        if self.config.stream == 'stdout':
+            self.stream = sys.stdout
+        elif self.config.stream == 'stderr':
+            self.stream = sys.stderr
+    
+    @send_wrapper
+    def send(self, messages):
+        self._lock.acquire()
+        try:
+            for msg in messages:
+                self.stream.write(self.format(msg))
+                self.stream.write('\n\n')
+        finally:
+            self._lock.release()
+        return len(messages)
+    
+    def format(self, msg):
+        if self.config.format == 'friendly':
+            return dump_text(msg, self.config.attachment_data)
+        else:
+            return self.encode_message(msg)

tests/sender-tests/test-console-sender.py

+# -*- coding: utf-8 -*-
+"""
+sender-tests/test-console-sender.py
+===================================
+This file tests the `cartero.senders.console.ConsoleSender`.
+
+:copyright: 2010 Matthew "LeafStorm" Frazier
+:license:   MIT/X11, see LICENSE for details
+"""
+import sys
+from cartero.encoding import build_mime
+from cartero.message import Message, dump_text
+from cartero.senders.console import ConsoleSender
+try:
+    from cStringIO import StringIO
+except ImportError:
+    from StringIO import StringIO
+
+
+class StreamTrapper(object):
+    "Hack of epic proportions."
+    def trap_stdout(self):
+        self.old_stdout = sys.stdout
+        sys.stdout = StringIO()
+    
+    def untrap_stdout(self):
+        buf = sys.stdout
+        sys.stdout = self.old_stdout
+        return buf.getvalue()
+    
+    def trap_stderr(self):
+        self.old_stderr = sys.stderr
+        sys.stderr = StringIO()
+    
+    def untrap_stderr(self):
+        buf = sys.stderr
+        sys.stderr = self.old_stderr
+        return buf.getvalue()
+
+
+def make_test_message():
+    text = (u'Get to Coruscant as soon as possible, and don\'t tell '
+            u'anyone, even General Cracken.')
+    msg = Message()
+    msg.sender = u'Wedge Antilles <wedge.antilles@starfighter.gffa>'
+    msg.to = [u'Tycho Celchu <tycho.celchu@starfighter.gffa>']
+    msg.subject = u'Secret Rogue Squadron plans'
+    msg.text = text
+    msg.headers['Date'] = 'right now'
+    msg.headers['Message-ID'] = '382y7ghr9uqhnuf'
+    return msg
+
+
+class TestConsoleSender(object):
+    def test_basic(self):
+        sdr = ConsoleSender()
+        sdr.stream = out = StringIO()
+        msg = make_test_message()
+        sdr.send(msg)
+        assert out.getvalue() == (dump_text(msg) + '\n\n')
+    
+    def test_stdout_and_stderr(self):
+        sdr = ConsoleSender()
+        assert sdr.stream is sys.stdout
+        sdr = ConsoleSender(stream='stdout')
+        assert sdr.stream is sys.stdout
+        sdr = ConsoleSender(stream='stderr')
+        assert sdr.stream is sys.stderr
+    
+    def test_mime_format(self):
+        sdr = ConsoleSender(format='mime')
+        sdr.stream = out = StringIO()
+        msg = make_test_message()
+        sdr.send(msg)
+        assert out.getvalue() == (sdr.encode_message(msg) + '\n\n')