# HG changeset patch
# User Sandro Knauß <knauss@netzguerilla.net>
# Date 1329251271 -3600
# Node ID b634a8fed51f6254848232fcc08a31963a926433
# Parent  2f9c439013b0904cc8c3fb40d1999d8898905c11
adding tests for smtp backend

diff -r 2f9c439013b0 -r b634a8fed51f tests/smtp.py
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/tests/smtp.py	Tue Feb 14 21:27:51 2012 +0100
@@ -0,0 +1,96 @@
+from twisted.trial import unittest
+
+import email
+from email.header import decode_header
+import base64
+from smtp_helper import TestSMTPServer
+
+
+from iro.model.message import Mail
+from iro.offer.smtp import SMTP
+
+HOST = "localhost"
+PORT = 9999
+
+class TestSMTPProvider(unittest.TestCase):
+    def setUp(self):
+        self.smtp_server = TestSMTPServer((HOST, PORT))
+        self.smtp_server.start()
+
+    def tearDown(self):
+        self.smtp_server.close()
+
+    def testSendMail(self):
+        c=[("send_from","send@t.de"),
+            ("host",HOST),
+            ("port",PORT),
+            ("user",""),
+            ("password",""),
+            ]
+
+        p=SMTP("test",c)
+        content = "sadfadfgwertsdgsdf\n\nsdfgaerasdfsad\nadfasdf"
+        status = p.send(Mail("sub", content, None), "t@t.de")
+
+        self.assertEqual(status.status, status.GOOD)
+
+        self.assertEqual(len(self.smtp_server.rcvd), 1)
+        fromaddr, toaddrs, message = self.smtp_server.rcvd[0]
+        msg = email.message_from_string(message)
+        
+        self.assertEqual(fromaddr,"send@t.de")
+        self.assertEqual(msg.get_all("From"),["send@t.de"])
+        self.assertEqual(toaddrs,["t@t.de"])
+        self.assertEqual(msg.get_all("To"),["t@t.de"])
+        self.assertEqual(decode_header(msg.get("Subject")),[("sub","utf-8")])
+        self.assertEqual(base64.b64decode(msg.get_payload()),content)
+
+    def testSendMailExtraFrm(self):
+        c=[("send_from","send@t.de"),
+            ("host",HOST),
+            ("port",PORT),
+            ("user",""),
+            ("password",""),
+            ]
+
+        p=SMTP("test",c)
+        content = ""
+        status = p.send(Mail("sub", content, "f@t.de"), "t@t.de")
+
+        self.assertEqual(status.status, status.GOOD)
+
+        self.assertEqual(len(self.smtp_server.rcvd), 1)
+        fromaddr, toaddrs, message = self.smtp_server.rcvd[0]
+        msg = email.message_from_string(message)
+        
+        self.assertEqual(fromaddr,"f@t.de")
+        self.assertEqual(msg.get_all("From"),["f@t.de"])
+
+    def testSendMailException(self):
+        c=[("send_from","send@t.de"),
+            ("host",HOST),
+            ("port",PORT-1),
+            ("user",""),
+            ("password",""),
+            ]
+
+        p=SMTP("test",c)
+        content = ""
+        status = p.send(Mail("sub", content, "f@t.de"), "t@t.de")
+        self.assertEqual(status.status, status.ERROR)
+        self.assertIsInstance(status.error,IOError)
+        self.assertEqual(status.error.errno,111)
+
+        self.assertEqual(len(self.smtp_server.rcvd), 0)
+
+    def testNeededOption(self):
+        pass
+    testNeededOption.todo = "test NeededOption is not implemented"
+
+    def testUnknownOption(self):
+        pass
+    testUnknownOption.todo = "test UnknownOption is not implemented"
+
+    def testSendFunc(self):
+        pass
+    testSendFunc.todo = "test getSendFunc is not implemented"
diff -r 2f9c439013b0 -r b634a8fed51f tests/smtp_helper.py
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/tests/smtp_helper.py	Tue Feb 14 21:27:51 2012 +0100
@@ -0,0 +1,28 @@
+"""
+A helper class which allows test code to intercept and store sent email.
+(from: http://lakin.weckers.net/thoughts/twisted/part1/threaded/)
+"""
+import smtpd
+import asyncore
+
+class TestSMTPServer(smtpd.SMTPServer):
+    """
+    An SMTP Server used to allow integration tests which ensure email is sent.
+    """
+
+    def __init__(self, localaddr):
+        self.rcvd = []
+        smtpd.SMTPServer.__init__(self, localaddr, None)
+
+    def start(self):
+        import threading
+        self.poller = threading.Thread(target=asyncore.loop,
+                kwargs={'timeout':0.1, 'use_poll':True})
+        self.poller.start()
+
+    def process_message(self, peer, mailfrom, rcpttos, data):
+        self.rcvd.append((mailfrom, rcpttos, data))
+
+    def close(self):
+        smtpd.SMTPServer.close(self)
+        self.poller.join()