david/jmc
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

SMTP send_email real implementation

Browse Source 
darcs-hash:20070618175813-86b55-990d6242426508e186abfc454d69fc9e0cd424b0.gz
This commit is contained in:
David Rousselie
2007-06-18 19:58:13 +02:00
parent d6b91f8d02
commit 86a1367f26
5 changed files with 314 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
src/jmc/jabber/component.py
View File

@@ -243,9 +243,11 @@ class SendMailMessageHandler(MailHandler):
def handle(self, message, lang_class, accounts):
to_node = message.get_to().node
to_email = to_node.replace('%', '@', 1)
accounts[0].send_email(to_email,
accounts[0].send_email(\
accounts[0].create_email(accounts[0].default_from,
to_email,
message.get_subject(),
message.get_body())
message.get_body()))
return self.send_mail_result(message, lang_class, to_email)
class RootSendMailMessageHandler(SendMailMessageHandler):
@@ -253,7 +255,7 @@ class RootSendMailMessageHandler(SendMailMessageHandler):
def __init__(self):
SendMailMessageHandler.__init__(self)
self.to_regexp = re.compile("^\s*(to|TO)\s*:\s*(?P<to_email>.*)")
self.to_regexp = re.compile("^\s*(to|TO|To)\s*:\s*(?P<to_email>.*)")
self.__logger = logging.getLogger(\
"jmc.jabber.component.RootSendMailMessageHandler")
@@ -285,8 +287,11 @@ class RootSendMailMessageHandler(SendMailMessageHandler):
message_body.append(line)
message_body.extend(lines)
if to_email is not None:
accounts[0].send_email(to_email, message.get_subject(),
"\n".join(message_body))
accounts[0].send_email(\
accounts[0].create_email(accounts[0].default_from,
to_email,
message.get_subject(),
"\n".join(message_body)))
return self.send_mail_result(message, lang_class, to_email)
else:
return [Message(from_jid=message.get_to(),

38
src/jmc/jabber/tests/component.py
View File

@@ -142,6 +142,16 @@ class MockPOP3Account(MockMailAccount, POP3Account):
IMAPAccount._init(self, *args, **kw)
MockMailAccount._init(self)
class MockSMTPAccount(object):
def __init__(self):
self.default_from = "user1@test.com"
def create_email(self, from_email, to_email, subject, body):
return (from_email, to_email, subject, body)
def send_email(self, email):
self.email = email
class MailComponent_TestCase(unittest.TestCase):
def setUp(self):
if os.path.exists(DB_PATH):
@@ -525,16 +535,11 @@ class SendMailMessageHandler_TestCase(unittest.TestCase):
to_jid="user%test.com@jcl.test.com",
subject="message subject",
body="message body")
class MockSMTPAccount(object):
def send_email(self, to_email, subject, body):
self.to_email = to_email
self.subject = subject
self.body = body
accounts = [MockSMTPAccount()]
result = self.handler.handle(message, Lang.en, accounts)
self.assertEquals(accounts[0].to_email, "user@test.com")
self.assertEquals(accounts[0].subject, "message subject")
self.assertEquals(accounts[0].body, "message body")
self.assertEquals(accounts[0].email[1], "user@test.com")
self.assertEquals(accounts[0].email[2], "message subject")
self.assertEquals(accounts[0].email[3], "message body")
self.assertEquals(len(result), 1)
self.assertEquals(result[0].stanza_type, "message")
self.assertEquals(result[0].get_from(), "user%test.com@jcl.test.com")
@@ -632,16 +637,12 @@ class RootSendMailMessageHandler_TestCase(unittest.TestCase):
subject="message subject",
body="to: user@test.com\n" \
"message body\nother line")
class MockSMTPAccount(object):
def send_email(self, to_email, subject, body):
self.to_email = to_email
self.subject = subject
self.body = body
accounts = [MockSMTPAccount()]
result = self.handler.handle(message, Lang.en, accounts)
self.assertEquals(accounts[0].to_email, "user@test.com")
self.assertEquals(accounts[0].subject, "message subject")
self.assertEquals(accounts[0].body, "message body\nother line")
self.assertEquals(accounts[0].email[1], "user@test.com")
self.assertEquals(accounts[0].email[2], "message subject")
self.assertEquals(accounts[0].email[3],
"message body\nother line")
self.assertEquals(len(result), 1)
self.assertEquals(result[0].get_type(), None)
self.assertEquals(result[0].get_from(), "jcl.test.com")
@@ -656,11 +657,6 @@ class RootSendMailMessageHandler_TestCase(unittest.TestCase):
to_jid="jcl.test.com",
subject="message subject",
body="message body")
class MockSMTPAccount(object):
def send_email(self, to_email, subject, body):
self.to_email = to_email
self.subject = subject
self.body = body
accounts = [MockSMTPAccount()]
result = self.handler.handle(message, Lang.en, accounts)
self.assertEquals(len(result), 1)

58
src/jmc/model/account.py
View File

@@ -24,12 +24,15 @@
import sys
import logging
import email
import email.Header
from email.Header import Header
from email.MIMEText import MIMEText
from email.MIMEMultipart import MIMEMultipart
import traceback
import poplib
import imaplib
import socket
import smtplib
from sqlobject.inheritance import InheritableSQLObject
from sqlobject.col import StringCol, IntCol, BoolCol
@@ -631,9 +634,52 @@ class SMTPAccount(Account):
get_register_fields = classmethod(_get_register_fields)
def send_email(self, to_email, subject, body):
def create_email(self, from_email, to_email, subject, body):
"""Create new email"""
email = MIMEText(body)
email['Subject'] = Header(str(subject))
email['From'] = Header(str(from_email))
email['To'] = Header(str(to_email))
return email
def __say_hello(self, connection):
if not (200 <= connection.ehlo()[0] <= 299):
(code, resp) = connection.helo()
if not (200 <= code <= 299):
raise SMTPHeloError(code, resp)
def send_email(self, email):
"""Send email according to current account parameters"""
self.__logger.debug("Sending email:\n"
"From: " + self.default_from + "\n" +
"To: " + to_email + "\n" +
"Subject: " + subject + "\n\n" +
body)
+ str(email))
smtp_connection = smtplib.SMTP()
if self.__logger.getEffectiveLevel() == logging.DEBUG:
smtp_connection.set_debuglevel(1)
smtp_connection.connect(self.host, self.port)
self.__say_hello(smtp_connection)
if self.tls:
smtp_connection.starttls()
self.__say_hello(smtp_connection)
if self.login is not None and len(self.login) > 0:
auth_methods = smtp_connection.esmtp_features["auth"].split()
auth_methods.reverse()
current_error = None
for auth_method in auth_methods:
self.__logger.debug("Trying to authenticate using "
+ auth_method + " method")
smtp_connection.esmtp_features["auth"] = auth_method
try:
smtp_connection.login(self.login, self.password)
current_error = None
self.__logger.debug("Successfuly to authenticate using "
+ auth_method + " method")
break
except smtplib.SMTPAuthenticationError, error:
self.__logger.debug("Failed to authenticate using "
+ auth_method + " method")
current_error = error
if current_error is not None:
raise current_error
smtp_connection.sendmail(str(email['From']), str(email['To']),
email.as_string())
smtp_connection.quit()

188
src/jmc/model/tests/account.py
View File

@@ -466,6 +466,194 @@ class SMTPAccount_TestCase(Account_TestCase):
self.assertTrue(account12.default_account)
del account.hub.threadConnection
def test_create_email(self):
account.hub.threadConnection = connectionForURI('sqlite://' + self.db_url)
account11 = SMTPAccount(user_jid="user1@test.com",
name="account11",
jid="account11@jmc.test.com")
del account.hub.threadConnection
email = account11.create_email("from@test.com",
"to@test.com",
"subject",
"body")
self.assertEqual(email['From'], "from@test.com")
self.assertEqual(email['To'], "to@test.com")
self.assertEqual(email['Subject'], "subject")
self.assertEqual(email.get_payload(), "body")
def make_test(self, responses=None, queries=None, core=None):
def inner():
self.server = server.DummyServer("localhost", 1025)
thread.start_new_thread(self.server.serve, ())
self.server.responses = []
if responses:
self.server.responses += responses
self.server.responses += ["221 localhost closing connection\r\n"]
self.server.queries = []
if queries:
self.server.queries += queries
self.server.queries += ["quit\r\n"]
if core:
account.hub.threadConnection = connectionForURI('sqlite://'
+ self.db_url)
core(self)
del account.hub.threadConnection
self.failUnless(self.server.verify_queries())
return inner
def test_send_email_esmtp_no_auth(self):
account.hub.threadConnection = connectionForURI('sqlite://'
+ self.db_url)
smtp_account = SMTPAccount(user_jid="user1@test.com",
name="account11",
jid="account11@jmc.test.com")
smtp_account.host = "localhost"
smtp_account.port = 1025
del account.hub.threadConnection
email = smtp_account.create_email("from@test.com",
"to@test.com",
"subject",
"body")
test_func = self.make_test(["220 localhost ESMTP\r\n",
"250-localhost Hello 127.0.0.1\r\n"
+ "250-SIZE 52428800\r\n"
+ "250-PIPELINING\r\n"
+ "250 HELP\r\n",
"250 OK\r\n",
"250 Accepted\r\n",
"354 Enter message\r\n",
None, None, None, None,
None, None, None, None,
"250 OK\r\n"],
["ehlo \[127.0.0.1\]\r\n",
"mail FROM:<" + str(email['From']) + ">.*",
"rcpt TO:<" + str(email['To']) + ">\r\n",
"data\r\n"] +
email.as_string().split("\n") + [".\r\n"],
lambda self: \
smtp_account.send_email(email))
test_func()
def test_send_email_no_auth(self):
account.hub.threadConnection = connectionForURI('sqlite://'
+ self.db_url)
smtp_account = SMTPAccount(user_jid="user1@test.com",
name="account11",
jid="account11@jmc.test.com")
smtp_account.host = "localhost"
smtp_account.port = 1025
del account.hub.threadConnection
email = smtp_account.create_email("from@test.com",
"to@test.com",
"subject",
"body")
test_func = self.make_test(["220 localhost SMTP\r\n",
"504 ESMTP not supported\r\n",
"250-localhost Hello 127.0.0.1\r\n"
+ "250-SIZE 52428800\r\n"
+ "250-PIPELINING\r\n"
+ "250 HELP\r\n",
"250 OK\r\n",
"250 Accepted\r\n",
"354 Enter message\r\n",
None, None, None, None,
None, None, None, None,
"250 OK\r\n"],
["ehlo \[127.0.0.1\]\r\n",
"helo \[127.0.0.1\]\r\n",
"mail FROM:<" + str(email['From']) + ">.*",
"rcpt TO:<" + str(email['To']) + ">\r\n",
"data\r\n"] +
email.as_string().split("\n") + [".\r\n"],
lambda self: \
smtp_account.send_email(email))
test_func()
def test_send_email_esmtp_auth(self):
account.hub.threadConnection = connectionForURI('sqlite://'
+ self.db_url)
smtp_account = SMTPAccount(user_jid="user1@test.com",
name="account11",
jid="account11@jmc.test.com")
smtp_account.host = "localhost"
smtp_account.port = 1025
smtp_account.login = "user"
smtp_account.password = "pass"
del account.hub.threadConnection
email = smtp_account.create_email("from@test.com",
"to@test.com",
"subject",
"body")
test_func = self.make_test(["220 localhost ESMTP\r\n",
"250-localhost Hello 127.0.0.1\r\n"
+ "250-SIZE 52428800\r\n"
+ "250-AUTH PLAIN LOGIN CRAM-MD5\r\n"
+ "250-PIPELINING\r\n"
+ "250 HELP\r\n",
"334 ZGF4IDNmNDM2NzY0YzBhNjgyMTQ1MzhhZGNiMjE2YTYxZjRm\r\n",
"235 Authentication succeeded\r\n",
"250 OK\r\n",
"250 Accepted\r\n",
"354 Enter message\r\n",
None, None, None, None,
None, None, None, None,
"250 OK\r\n"],
["ehlo \[127.0.0.1\]\r\n",
"AUTH CRAM-MD5\r\n",
".*\r\n",
"mail FROM:<" + str(email['From']) + ">.*",
"rcpt TO:<" + str(email['To']) + ">\r\n",
"data\r\n"] +
email.as_string().split("\n") + [".\r\n"],
lambda self: \
smtp_account.send_email(email))
test_func()
def test_send_email_esmtp_auth_method2(self):
account.hub.threadConnection = connectionForURI('sqlite://'
+ self.db_url)
smtp_account = SMTPAccount(user_jid="user1@test.com",
name="account11",
jid="account11@jmc.test.com")
smtp_account.host = "localhost"
smtp_account.port = 1025
smtp_account.login = "user"
smtp_account.password = "pass"
del account.hub.threadConnection
email = smtp_account.create_email("from@test.com",
"to@test.com",
"subject",
"body")
test_func = self.make_test(["220 localhost ESMTP\r\n",
"250-localhost Hello 127.0.0.1\r\n"
+ "250-SIZE 52428800\r\n"
+ "250-AUTH PLAIN LOGIN CRAM-MD5\r\n"
+ "250-PIPELINING\r\n"
+ "250 HELP\r\n",
"334 ZGF4IDNmNDM2NzY0YzBhNjgyMTQ1MzhhZGNiMjE2YTYxZjRm\r\n",
"535 Incorrect Authentication data\r\n",
"334 asd235r4\r\n",
"235 Authentication succeeded\r\n",
"250 OK\r\n",
"250 Accepted\r\n",
"354 Enter message\r\n",
None, None, None, None,
None, None, None, None,
"250 OK\r\n"],
["ehlo \[127.0.0.1\]\r\n",
"AUTH CRAM-MD5\r\n",
".*\r\n",
"AUTH LOGIN .*\r\n",
".*\r\n",
"mail FROM:<" + str(email['From']) + ">.*",
"rcpt TO:<" + str(email['To']) + ">\r\n",
"data\r\n"] +
email.as_string().split("\n") + [".\r\n"],
lambda self: \
smtp_account.send_email(email))
test_func()
def suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(MailAccount_TestCase, 'test'))