Source Code for Module pyzmail.tests.test_send

 1  import threading, smtpd, asyncore, socket, smtplib, time 
 2  import unittest 
 3  import pyzmail 
 4  from pyzmail.generate import * 
 5   
 6   
 7  smtpd_addr='127.0.0.1'  
 8  smtpd_port=32525 
 9  smtp_bad_port=smtpd_port-1 
10   
11  smtp_mode='normal' 
12  smtp_login=None 
13  smtp_password=None 
14       
15   
16 -class SMTPServer(smtpd.SMTPServer): 
17 -    def __init__(self, localaddr, remoteaddr, received): 
18          smtpd.SMTPServer.__init__(self, localaddr, remoteaddr) 
19          self.set_reuse_addr() 
20          # put the received mail into received list  
21          self.received=received 
22           
23 -    def process_message(self, peer, mail_from, rcpt_to, data): 
24          ret=None 
25          if mail_from.startswith('data_error'): 
26              ret='552 Requested mail action aborted: exceeded storage allocation' 
27          self.received.append((ret, peer, mail_from, rcpt_to, data)) 
28          return ret 
29   
30 -class TestSend(unittest.TestCase): 
31   
32 -    def setUp(self): 
33          self.received=[] 
34          self.smtp_server=SMTPServer((smtpd_addr, smtpd_port), None, self.received) 
35   
36          def asyncloop(): 
37              # check every sec if all channel are close 
38              asyncore.loop(1) 
39   
40                       
41          self.payload, self.mail_from, self.rcpt_to, self.msg_id=compose_mail((u'Me', 'me@foo.com'), [(u'Him', 'him@bar.com')], u'the subject', 'iso-8859-1', ('Hello world', 'us-ascii')) 
42   
43          # start the server after having built the payload, to handle failure in  
44          # the code above 
45          self.smtpd_thread=threading.Thread(target=asyncloop) 
46          self.smtpd_thread.daemon=True 
47          self.smtpd_thread.start() 
48   
49   
50 -    def tearDown(self): 
51          self.smtp_server.close() 
52          self.smtpd_thread.join() 
53           
54 -    def test_simple_send(self): 
55          """simple send""" 
56          ret=send_mail(self.payload, self.mail_from, self.rcpt_to, smtpd_addr, smtpd_port, smtp_mode=smtp_mode, smtp_login=smtp_login, smtp_password=smtp_password) 
57          self.assertEqual(ret, dict()) 
58          (ret, peer, mail_from, rcpt_to, payload)=self.received[0] 
59          self.assertEqual(self.payload, payload) 
60          self.assertEqual(self.mail_from, mail_from) 
61          self.assertEqual(self.rcpt_to, rcpt_to) 
62          self.assertEqual('127.0.0.1', peer[0]) 
63   
64 -    def test_send_to_a_wrong_port(self): 
65          """send to a wrong port""" 
66          self.smtp_server.close() 
67          ret=send_mail(self.payload, self.mail_from, self.rcpt_to, smtpd_addr, smtpd_port, smtp_mode=smtp_mode, smtp_login=smtp_login, smtp_password=smtp_password) 
68          self.assertEqual(type(ret), str) 
69   
70 -    def test_send_data_error(self): 
71          """smtp server return error code""" 
72          ret=send_mail(self.payload, 'data_error@foo.com', self.rcpt_to, smtpd_addr, smtp_bad_port, smtp_mode=smtp_mode, smtp_login=smtp_login, smtp_password=smtp_password) 
73          self.assertEqual(type(ret), str) 
74   
75  if __name__ == '__main__': 
76      unittest.main() 
77