Skip to content
Snippets Groups Projects
Commit bdae25a5 authored by ale's avatar ale
Browse files

Add a test for Mailman delivery

parent 0927b101
Branches
No related tags found
No related merge requests found
Pipeline #83739 failed
...@@ -81,40 +81,44 @@ class TestMail(TestBase): ...@@ -81,40 +81,44 @@ class TestMail(TestBase):
return unique_id return unique_id
# Send an email from a to b and find it in c's IMAP account. def _wait_for_message(self, imap_user, imap_password, match_header, match_value,
def _send_email_and_read_it(self, sender, sender_password, rcpt, timeout=300):
imap_user, imap_password):
unique_id = self._send_email(sender, sender_password, rcpt)
imap = imaplib.IMAP4_SSL(self.frontend_ip()) imap = imaplib.IMAP4_SSL(self.frontend_ip())
imap.login(imap_user, imap_password) imap.login(imap_user, imap_password)
imap.select('INBOX') imap.select('INBOX')
def _check_for_msg(): def _check_for_msg():
logging.info('looking for message <%s>', unique_id) logging.info('looking for message %s', match_value)
typ, data = imap.uid( typ, data = imap.uid(
'search', None, 'HEADER', 'X-UniqueID', unique_id) 'search', None, 'HEADER', match_header, match_value)
if typ == 'OK' and data[0]: if typ == 'OK' and data[0]:
logging.info('message <%s> found, fetching uuid "%s"', logging.info('message %s found, fetching uuid "%s"',
unique_id, data[0]) match_value, data[0])
typ, msgdata = imap.uid('fetch', data[0], '(RFC822)') typ, msgdata = imap.uid('fetch', data[0], '(RFC822)')
if typ != 'OK': if typ != 'OK':
raise Exception('error fetching message: %s' % typ) raise Exception('error fetching message: %s' % typ)
msg = msgdata[0][1] msg = msgdata[0][1]
imap.uid('store', data[0], '+FLAGS', r'(\Deleted)') imap.uid('store', data[0], '+FLAGS', r'(\Deleted)')
imap.expunge() imap.expunge()
logging.info('message <%s> deleted successfully', unique_id) logging.info('message %s deleted successfully', match_value)
return msg return msg
return None return None
deadline = time.time() + 300 deadline = time.time() + timeout
delivered = False
while time.time() < deadline: while time.time() < deadline:
msg = _check_for_msg() msg = _check_for_msg()
if msg: if msg:
delivered = True return True
break
time.sleep(1) time.sleep(1)
return False
# Send an email from a to b and find it in c's IMAP account.
def _send_email_and_read_it(self, sender, sender_password, rcpt,
imap_user, imap_password):
unique_id = self._send_email(sender, sender_password, rcpt)
delivered = self._wait_for_message(imap_user, imap_password,
'X-UniqueID', unique_id)
self.assertTrue(delivered) self.assertTrue(delivered)
return msg return msg
...@@ -230,3 +234,12 @@ class TestMail(TestBase): ...@@ -230,3 +234,12 @@ class TestMail(TestBase):
'password', 'password',
'uno@investici.org', 'uno@investici.org',
from_addr='uno@investici.org') from_addr='uno@investici.org')
def test_mailman_send_message(self):
# Test user sends an email to a list it is subscribed to.
self._send_email_and_read_it(
self.users['nonpriv']['name'],
self.users['nonpriv']['password'],
'lista@investici.org',
self.users['nonpriv']['name'],
self.users['nonpriv']['password'])
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment