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

Support more email header in Jabber message sent to JMC

Browse Source 
darcs-hash:20071107180613-86b55-86d24b561b19d12378f5469be31c328940d0fd88.gz
This commit is contained in:
David Rousselie
2007-11-07 19:06:13 +01:00
parent 6a8afaedd0
commit a901a23fb4
5 changed files with 113 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
src/jmc/jabber/message.py
View File

@@ -34,6 +34,10 @@ class SendMailMessageHandler(MailHandler):
MailHandler.__init__(self, component)
self.__logger = logging.getLogger(\
"jmc.jabber.component.SendMailMessageHandler")
self.to_regexp = re.compile("^\s*(?i)to\s*:\s*(?P<to_email>.*)")
self.cc_regexp = re.compile("^\s*(?i)cc\s*:\s*(?P<cc_email>.*)")
self.bcc_regexp = re.compile("^\s*(?i)bcc\s*:\s*(?P<bcc_email>.*)")
self.subject_regexp = re.compile("^\s*(?i)subject\s*:\s*(?P<subject_email>.*)")
def send_mail_result(self, message, lang_class, to_email):
return [Message(from_jid=message.get_to(),
@@ -41,16 +45,75 @@ class SendMailMessageHandler(MailHandler):
subject=lang_class.send_mail_ok_subject,
body=lang_class.send_mail_ok_body % (to_email))]
def get_email_headers_from_message(self, text, regexps, groups):
"""
This method extract line from message body with given regexps
and associated matching groups names.
If a line match a regexp, this line is removed from the returned message
and matching group is extracted and returned in 'headers'.
"""
headers = []
for i in xrange(len(regexps)):
headers.append(None)
lines = text.split('\n')
message_body = []
regexps_len = len(regexps)
matched = 0
while matched < regexps_len \
and lines:
line = lines.pop(0)
has_matched = False
i = 0
while i < regexps_len \
and not has_matched:
if regexps[i] is not None:
match = regexps[i].match(line)
if match:
headers[i] = match.group(groups[i])
regexps[i] = None
has_matched = True
i += 1
if has_matched:
matched += 1
else:
message_body.append(line)
message_body.extend(lines)
return ("\n".join(message_body), headers)
def handle(self, stanza, lang_class, data):
message = stanza
accounts = data
to_node = message.get_to().node
to_email = to_node.replace('%', '@', 1)
(message_body,
(more_to_email,
subject,
cc_email,
bcc_email)) = self.get_email_headers_from_message(\
message.get_body(),
[self.to_regexp,
self.subject_regexp,
self.cc_regexp,
self.bcc_regexp],
["to_email",
"subject_email",
"cc_email",
"bcc_email"])
other_headers = {}
other_headers["Cc"] = cc_email
other_headers["Bcc"] = bcc_email
message_subject = message.get_subject()
if message_subject is not None \
and len(message_subject) > 0:
subject = message_subject
if more_to_email is not None:
to_email += ", " + more_to_email
accounts[0].send_email(\
accounts[0].create_email(accounts[0].default_from,
to_email,
message.get_subject(),
message.get_body()))
message_body,
other_headers))
return self.send_mail_result(message, lang_class, to_email)
class RootSendMailMessageHandler(SendMailMessageHandler):
@@ -79,24 +142,34 @@ class RootSendMailMessageHandler(SendMailMessageHandler):
def handle(self, stanza, lang_class, data):
message = stanza
accounts = data
to_email = None
lines = message.get_body().split('\n')
message_body = []
while to_email is None \
and lines:
line = lines.pop(0)
match = self.to_regexp.match(line)
if match:
to_email = match.group("to_email")
else:
message_body.append(line)
message_body.extend(lines)
(message_body,
(to_email,
subject,
cc_email,
bcc_email)) = self.get_email_headers_from_message(\
message.get_body(),
[self.to_regexp,
self.subject_regexp,
self.cc_regexp,
self.bcc_regexp],
["to_email",
"subject_email",
"cc_email",
"bcc_email"])
other_headers = {}
other_headers["Cc"] = cc_email
other_headers["Bcc"] = bcc_email
message_subject = message.get_subject()
if message_subject is not None \
and len(message_subject) > 0:
subject = message_subject
if to_email is not None:
accounts[0].send_email(\
accounts[0].create_email(accounts[0].default_from,
to_email,
message.get_subject(),
"\n".join(message_body)))
subject,
message_body,
other_headers))
return self.send_mail_result(message, lang_class, to_email)
else:
return [Message(from_jid=message.get_to(),

3
src/jmc/jabber/tests/__init__.py
View File

@@ -3,13 +3,14 @@ __revision__ = ""
import unittest
from jmc.jabber.tests import component, disco, command
from jmc.jabber.tests import component, disco, command, message
def suite():
suite = unittest.TestSuite()
suite.addTest(component.suite())
suite.addTest(disco.suite())
suite.addTest(command.suite())
suite.addTest(message.suite())
return suite
if __name__ == '__main__':

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

@@ -162,17 +162,6 @@ class MockPOP3Account(MockMailAccount, POP3Account):
POP3Account._init(self, *args, **kw)
MockMailAccount._init(self)
class MockSMTPAccount(object):
def __init__(self):
self.default_from = "user1@test.com"
self.email = None
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(JCLTestCase):
def setUp(self):
JCLTestCase.setUp(self, tables=[Account, PresenceAccount, User,
@@ -1162,137 +1151,6 @@ class MailComponent_TestCase(JCLTestCase):
self.assertEquals(field.children.name, "value")
self.assertEquals(field.children.content, "INBOX.dir1.subdir1")
class SendMailMessageHandler_TestCase(unittest.TestCase):
def setUp(self):
self.handler = SendMailMessageHandler(None)
def test_handle(self):
message = Message(from_jid="user1@test.com",
to_jid="user%test.com@jcl.test.com",
subject="message subject",
body="message body")
accounts = [MockSMTPAccount()]
result = self.handler.handle(message, Lang.en, accounts)
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")
self.assertEquals(result[0].get_to(), "user1@test.com")
self.assertEquals(result[0].get_subject(),
Lang.en.send_mail_ok_subject)
self.assertEquals(result[0].get_body(),
Lang.en.send_mail_ok_body % ("user@test.com"))
class RootSendMailMessageHandler_TestCase(JCLTestCase):
def setUp(self):
JCLTestCase.setUp(self, tables=[Account, SMTPAccount, User])
self.handler = RootSendMailMessageHandler(None)
def test_filter(self):
model.db_connect()
user1 = User(jid="user1@test.com")
account11 = SMTPAccount(user=user1,
name="account11",
jid="account11@jcl.test.com")
account11.default_account = True
account12 = SMTPAccount(user=user1,
name="account12",
jid="account12@jcl.test.com")
message = Message(from_jid="user1@test.com",
to_jid="account11@jcl.test.com",
body="message")
accounts = self.handler.filter(message, None)
self.assertEquals(accounts.count(), 1)
model.db_disconnect()
def test_filter_no_default_account(self):
model.db_connect()
user1 = User(jid="user1@test.com")
account11 = SMTPAccount(user=user1,
name="account11",
jid="account11@jcl.test.com")
account12 = SMTPAccount(user=user1,
name="account12",
jid="account12@jcl.test.com")
message = Message(from_jid="user1@test.com",
to_jid="account11@jcl.test.com",
body="message")
accounts = self.handler.filter(message, None)
self.assertEquals(accounts.count(), 2)
self.assertEquals(accounts[0].name, "account11")
model.db_disconnect()
def test_filter_wrong_dest(self):
model.db_connect()
user1 = User(jid="user1@test.com")
account11 = SMTPAccount(user=user1,
name="account11",
jid="account11@jcl.test.com")
account12 = SMTPAccount(user=user1,
name="account12",
jid="account12@jcl.test.com")
message = Message(from_jid="user1@test.com",
to_jid="user2%test.com@jcl.test.com",
body="message")
accounts = self.handler.filter(message, None)
self.assertEquals(accounts.count(), 2)
model.db_disconnect()
def test_filter_wrong_user(self):
model.db_connect()
user1 = User(jid="user1@test.com")
account11 = SMTPAccount(user=user1,
name="account11",
jid="account11@jcl.test.com")
account12 = SMTPAccount(user=user1,
name="account12",
jid="account12@jcl.test.com")
message = Message(from_jid="user2@test.com",
to_jid="account11@jcl.test.com",
body="message")
accounts = self.handler.filter(message, None)
self.assertEquals(accounts.count(), 0)
model.db_disconnect()
def test_handle_email_found_in_header(self):
message = Message(from_jid="user1@test.com",
to_jid="jcl.test.com",
subject="message subject",
body="to: user@test.com\n" \
"message body\nother line")
accounts = [MockSMTPAccount()]
result = self.handler.handle(message, Lang.en, accounts)
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")
self.assertEquals(result[0].get_to(), "user1@test.com")
self.assertEquals(result[0].get_subject(),
Lang.en.send_mail_ok_subject)
self.assertEquals(result[0].get_body(),
Lang.en.send_mail_ok_body % ("user@test.com"))
def test_handle_email_not_found_in_header(self):
message = Message(from_jid="user1@test.com",
to_jid="jcl.test.com",
subject="message subject",
body="message body")
accounts = [MockSMTPAccount()]
result = self.handler.handle(message, Lang.en, accounts)
self.assertEquals(len(result), 1)
self.assertEquals(result[0].get_type(), "error")
self.assertEquals(result[0].get_from(), "jcl.test.com")
self.assertEquals(result[0].get_to(), "user1@test.com")
self.assertEquals(result[0].get_subject(),
Lang.en.send_mail_error_no_to_header_subject)
self.assertEquals(result[0].get_body(),
Lang.en.send_mail_error_no_to_header_body)
class MailSender_TestCase(JCLTestCase):
def setUp(self):
JCLTestCase.setUp(self, tables=[Account, PresenceAccount, MailAccount,
@@ -1565,8 +1423,6 @@ class MailFeederHandler_TestCase(JCLTestCase):
def suite():
test_suite = unittest.TestSuite()
test_suite.addTest(unittest.makeSuite(MailComponent_TestCase, 'test'))
test_suite.addTest(unittest.makeSuite(SendMailMessageHandler_TestCase, 'test'))
test_suite.addTest(unittest.makeSuite(RootSendMailMessageHandler_TestCase, 'test'))
test_suite.addTest(unittest.makeSuite(MailSender_TestCase, 'test'))
test_suite.addTest(unittest.makeSuite(MailHandler_TestCase, 'test'))
test_suite.addTest(unittest.makeSuite(MailUnsubscribeHandler_TestCase, 'test'))

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

@@ -717,7 +717,7 @@ class SMTPAccount(Account):
get_register_fields = classmethod(_get_register_fields)
def create_email(self, from_email, to_email, subject, body):
def create_email(self, from_email, to_email, subject, body, other_headers=None):
"""Create new email"""
email = MIMEText(body)
if subject is None:
@@ -725,6 +725,9 @@ class SMTPAccount(Account):
email['Subject'] = Header(str(subject))
email['From'] = Header(str(from_email))
email['To'] = Header(str(to_email))
if other_headers is not None:
for header_name in other_headers.keys():
email[header_name] = Header(other_headers[header_name])
return email
def __say_hello(self, connection):

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

@@ -515,6 +515,25 @@ class SMTPAccount_TestCase(Account_TestCase):
self.assertEqual(email['Subject'], "subject")
self.assertEqual(email.get_payload(), "body")
def test_create_email_other_headers(self):
model.db_connect()
account11 = SMTPAccount(user=User(jid="user1@test.com"),
name="account11",
jid="account11@jmc.test.com")
model.db_disconnect()
email = account11.create_email("from@test.com",
"to@test.com",
"subject",
"body",
{"Bcc": "bcc@test.com",
"Cc": "cc@test.com"})
self.assertEqual(email['From'], "from@test.com")
self.assertEqual(email['To'], "to@test.com")
self.assertEqual(email['Subject'], "subject")
self.assertEqual(email['Bcc'], "bcc@test.com")
self.assertEqual(email['Cc'], "cc@test.com")
self.assertEqual(email.get_payload(), "body")
def make_test(self, responses=None, queries=None, core=None):
def inner():
self.server = server.DummyServer("localhost", 1025)