diff options
Diffstat (limited to 'src/leap/mail')
| -rw-r--r-- | src/leap/mail/incoming/service.py | 5 | ||||
| -rw-r--r-- | src/leap/mail/incoming/tests/test_incoming_mail.py | 59 |
2 files changed, 51 insertions, 13 deletions
diff --git a/src/leap/mail/incoming/service.py b/src/leap/mail/incoming/service.py index 23c216c..2902141 100644 --- a/src/leap/mail/incoming/service.py +++ b/src/leap/mail/incoming/service.py @@ -215,7 +215,6 @@ class IncomingMail(Service): """ def _log_synced(result): log.msg('FETCH soledad SYNCED.') - print "Result: ", result return result try: log.msg('FETCH: syncing soledad...') @@ -404,7 +403,6 @@ class IncomingMail(Service): :param doc: the SoledadDocument to delete :type doc: SoledadDocument """ - print "DELETING INCOMING MESSAGE" log.msg("Deleting Incoming message: %s" % (doc.doc_id,)) return self._soledad.delete_doc(doc) @@ -431,7 +429,7 @@ class IncomingMail(Service): if (fromHeader is not None and (msg.get_content_type() == MULTIPART_ENCRYPTED or msg.get_content_type() == MULTIPART_SIGNED)): - senderAddress = parseaddr(fromHeader) + senderAddress = parseaddr(fromHeader)[1] def add_leap_header(ret): decrmsg, signkey = ret @@ -709,7 +707,6 @@ class IncomingMail(Service): def msgSavedCallback(result): if not empty(result): leap_events.signal(IMAP_MSG_SAVED_LOCALLY) - print "DEFERRING THE DELETION ----->" return self._delete_incoming_message(doc) # TODO add notification as a callback #leap_events.signal(IMAP_MSG_DELETED_INCOMING) diff --git a/src/leap/mail/incoming/tests/test_incoming_mail.py b/src/leap/mail/incoming/tests/test_incoming_mail.py index f8652b3..0b1e696 100644 --- a/src/leap/mail/incoming/tests/test_incoming_mail.py +++ b/src/leap/mail/incoming/tests/test_incoming_mail.py @@ -35,9 +35,11 @@ from leap.mail.adaptors import soledad_indexes as fields from leap.mail.constants import INBOX_NAME from leap.mail.imap.account import IMAPAccount from leap.mail.incoming.service import IncomingMail +from leap.mail.smtp.rfc3156 import MultipartEncrypted, PGPEncrypted from leap.mail.tests import ( TestCaseWithKeyManager, ADDRESS, + ADDRESS_2, ) from leap.soledad.common.document import SoledadDocument from leap.soledad.common.crypto import ( @@ -54,7 +56,6 @@ class IncomingMailTestCase(TestCaseWithKeyManager): Tests for the incoming mail parser """ NICKSERVER = "http://domain" - FROM_ADDRESS = "test@somedomain.com" BODY = """ Governments of the Industrial World, you weary giants of flesh and steel, I come from Cyberspace, the new home of Mind. On behalf of the future, I ask @@ -67,23 +68,25 @@ subject: independence of cyberspace %(body)s """ % { - "from": FROM_ADDRESS, + "from": ADDRESS_2, "to": ADDRESS, "body": BODY } def setUp(self): def getInbox(_): - theAccount = IMAPAccount(ADDRESS, self._soledad) - return theAccount.callWhenReady( + d = defer.Deferred() + theAccount = IMAPAccount(ADDRESS, self._soledad, d=d) + d.addCallback( lambda _: theAccount.getMailbox(INBOX_NAME)) + return d def setUpFetcher(inbox): # Soledad sync makes trial block forever. The sync it's mocked to # fix this problem. _mock_soledad_get_from_index can be used from # the tests to provide documents. # TODO ---- see here http://www.pythoneye.com/83_20424875/ - self._soledad.sync = Mock() + self._soledad.sync = Mock(return_value=defer.succeed(None)) self.fetcher = IncomingMail( self._km, @@ -91,6 +94,9 @@ subject: independence of cyberspace inbox, ADDRESS) + # The messages don't exist on soledad will fail on deletion + self.fetcher._delete_incoming_message = Mock(return_value=None) + d = super(IncomingMailTestCase, self).setUp() d.addCallback(getInbox) d.addCallback(setUpFetcher) @@ -104,7 +110,7 @@ subject: independence of cyberspace """ Test the OpenPGP header key extraction """ - KEYURL = "https://somedomain.com/key.txt" + KEYURL = "https://leap.se/key.txt" OpenPGP = "id=12345678; url=\"%s\"; preference=signencrypt" % (KEYURL,) message = Parser().parsestr(self.EMAIL) @@ -114,7 +120,7 @@ subject: independence of cyberspace def fetch_key_called(ret): self.fetcher._keymanager.fetch_key.assert_called_once_with( - self.FROM_ADDRESS, KEYURL, OpenPGPKey) + ADDRESS_2, KEYURL, OpenPGPKey) d = self._create_incoming_email(message.as_string()) d.addCallback( @@ -153,7 +159,7 @@ subject: independence of cyberspace KEY = "-----BEGIN PGP PUBLIC KEY BLOCK-----\n..." message = MIMEMultipart() - message.add_header("from", self.FROM_ADDRESS) + message.add_header("from", ADDRESS_2) key = MIMEApplication("", "pgp-keys") key.set_payload(KEY) message.attach(key) @@ -162,12 +168,47 @@ subject: independence of cyberspace def put_raw_key_called(_): self.fetcher._keymanager.put_raw_key.assert_called_once_with( - KEY, OpenPGPKey, address=self.FROM_ADDRESS) + KEY, OpenPGPKey, address=ADDRESS_2) d = self._mock_fetch(message.as_string()) d.addCallback(put_raw_key_called) return d + def testDecryptEmail(self): + self.fetcher._decryption_error = Mock() + + def create_encrypted_message(encstr): + message = Parser().parsestr(self.EMAIL) + newmsg = MultipartEncrypted('application/pgp-encrypted') + for hkey, hval in message.items(): + newmsg.add_header(hkey, hval) + + encmsg = MIMEApplication( + encstr, _subtype='octet-stream', _encoder=lambda x: x) + encmsg.add_header('content-disposition', 'attachment', + filename='msg.asc') + # create meta message + metamsg = PGPEncrypted() + metamsg.add_header('Content-Disposition', 'attachment') + # attach pgp message parts to new message + newmsg.attach(metamsg) + newmsg.attach(encmsg) + return newmsg + + def decryption_error_not_called(_): + self.assertFalse(self.fetcher._decyption_error.called, + "There was some errors with decryption") + + d = self._km.encrypt( + self.EMAIL, + ADDRESS, OpenPGPKey, sign=ADDRESS_2) + d.addCallback(create_encrypted_message) + d.addCallback( + lambda message: + self._mock_fetch(message.as_string())) + + return d + def _mock_fetch(self, message): self.fetcher._keymanager.fetch_key = Mock() d = self._create_incoming_email(message) |