Ticket #18: testmode.patch

turbomail_tests/test_testmode.py

@@ +1 @@
+#!/usr/bin/env python
+# -*- coding: UTF-8 -*-
+
+# Copyright (c) 2007 Felix Schwarz <felix.schwarz@schwarz.eu>
+#
+# This code is placed under the MIT license:
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy 
+# of this software and associated documentation files (the "Software"), to deal 
+# in the Software without restriction, including without limitation the rights 
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 
+# copies of the Software, and to permit persons to whom the Software is 
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in 
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE 
+# SOFTWARE.
+
+import email
+import logging
+import time
+import unittest
+
+logging.disable(logging.INFO)
+
+import turbogears
+import turbomail
+
+from turbomail_tests.lib.smtp_mailsink import SMTPMailsink
+from turbomail_tests.lib.utils import get_received_mail, save_config
+
+
+class TestTestmode(unittest.TestCase):
+        "Test cases for enabling the TurboGears 'testmode'."
+
+        def setUp(self):
+                server_port = 42042
+                test_config = {'mail.on': True, 'mail.timeout': 1,
+                                           'mail.server': 'localhost:%d' % server_port,
+                                           'mail.testmode': True}
+                self._original_config = save_config(test_config.keys())
+                turbogears.config.update(test_config)
+                self.sink = SMTPMailsink(host='localhost', port=server_port)
+                self.sink.start()
+                turbogears.startup.startTurboGears()
+
+
+        def tearDown(self):
+                turbogears.startup.stopTurboGears()
+                turbogears.config.update(self._original_config)
+                self.sink.stop()
+
+
+        def test_enable_testmode(self):
+                """Test that a message is not lost if the mail hub was unavailable for
+                a short period of time."""
+                message = turbomail.Message('sender@foo.example', 
+                                                                        'recipient@foo.example', 'foo bar')
+                message.plain = 'Hello World!'          
+                turbomail.enqueue(message)
+                time.sleep(1)
+                msg = turbomail._queue._queue.get(block=False)
+                assert 'Hello World' in msg['message']
+

turbomail/turbomail/pool.py

@@ -25 +25 @@
         method.
         """
         
-        def __init__(self, interval=10, threads=4, jobs=10, timeout=60, polling=False, **kw):
+        def __init__(self, interval=10, threads=4, jobs=10, timeout=60, polling=False, testmode=False, **kw):
                 """Initialize the thread pool.
                 
                 @param interval: A delay, in seconds, between spawn runs.
@@ -46 +46 @@
                                 mechanism.  Disabled, threads will be created,
                                 as required, when work is enqueued.
                 @type polling: bool
+
+                @param testmode: Enable or disable the unit testing mode. 
+                                Disabled, no mail is actually sent but only
+                                stored in the queue for later retrieval.
+                @type testmode: bool
                 """
                 
                 super(Pool, self).__init__()
@@ -58 +63 @@
                 self._jobs = jobs
                 self._timeout = timeout
                 self._polling = polling
+                self._testmode = testmode
                 self._kw = kw
                 
                 log.debug("Thread pool created.")
@@ -87 +93 @@
                         self._queue.put(work(), block=block, timeout=timeout)
                 else:
                         self._queue.put(work, block=block, timeout=timeout)
+
+                if self._testmode:
+                        log.debug("Testmode enabled, not sending mail.")
+                        return
                 
                 optimum_threads = min(self._threads, math.ceil(self._queue.qsize() / float(self._jobs)))
 

turbomail/turbomail/startup.py

@@ -46 +46 @@
                         password=turbogears.config.get("mail.password", None),
                         debug=turbogears.config.get("mail.debug", False),
                         tls=turbogears.config.get("mail.tls", None),
-                        polling=turbogears.config.get("mail.polling", None)
+                        polling=turbogears.config.get("mail.polling", None),
+                        testmode=turbogears.config.get("mail.testmode", False)
                 )
         
         if turbomail._queue._polling: