1 from twisted.trial import unittest

     2 

     3 import email

     4 from email.header import decode_header

     5 import base64

     6 from ..test_helpers.smtp_helper import TestSMTPServer

     7 

     8 from mock import patch, Mock

     9 import smtplib

    10 

    11 from iro.error import NoRoute, NoTyp, NeededOption

    12 from iro.model.message import Mail

    13 from iro.offer.smtp import SMTP

    14 

    15 HOST = "localhost"

    16 PORT = 9999

    17 

    18 class TestSMTPProvider(unittest.TestCase):

    19     def setUp(self):

    20         self.smtp_server = TestSMTPServer((HOST, PORT))

    21         self.smtp_server.start()

    22 

    23     def tearDown(self):

    24         self.smtp_server.close()

    25 

    26     def getSMTP(self, c=None):

    27         _c={"send_from":"send@t.de",

    28             "host":HOST,

    29             "port":PORT,

    30             "typ":"smtp",

    31             }

    32 

    33         if c:

    34             _c.update(c)

    35 

    36         ret = SMTP("test")

    37         ret.load(_c.items())

    38         return ret

    39 

    40     def testSendMail(self):

    41         p=self.getSMTP()

    42         content = "sadfadfgwertsdgsdf\n\nsdfgaerasdfsad\nadfasdf"

    43         p.send("t@t.de", Mail("sub", content, None))

    44 

    45 

    46         self.assertEqual(len(self.smtp_server.rcvd), 1)

    47         fromaddr, toaddrs, message = self.smtp_server.rcvd[0]

    48         msg = email.message_from_string(message)

    49         

    50         self.assertEqual(fromaddr,"send@t.de")

    51         self.assertEqual(msg.get_all("From"),["send@t.de"])

    52         self.assertEqual(toaddrs,["t@t.de"])

    53         self.assertEqual(msg.get_all("To"),["t@t.de"])

    54         self.assertEqual(decode_header(msg.get("Subject")),[("sub","utf-8")])

    55         self.assertEqual(base64.b64decode(msg.get_payload()),content)

    56 

    57     def testSendMailExtraFrm(self):

    58         p=self.getSMTP()

    59         content = ""

    60         p.send("t@t.de", Mail("sub", content, "f@t.de"))

    61 

    62         self.assertEqual(len(self.smtp_server.rcvd), 1)

    63         fromaddr, toaddrs, message = self.smtp_server.rcvd[0]

    64         msg = email.message_from_string(message)

    65         

    66         self.assertEqual(fromaddr,"f@t.de")

    67         self.assertEqual(msg.get_all("From"),["f@t.de"])

    68 

    69     def testSendMailException(self):

    70         p=self.getSMTP({"port":PORT-1})

    71         content = ""

    72         self.assertRaises(IOError, p.send, "t@t.de", Mail("sub", content, "f@t.de"))

    73 

    74         self.assertEqual(len(self.smtp_server.rcvd), 0)

    75 

    76     @patch("smtplib.SMTP_SSL")

    77     def testSSLSendMail(self,mock_ssl):

    78         def se(*args):

    79             return smtplib.SMTP(*args)

    80         mock_ssl.side_effect=se

    81 

    82         p=self.getSMTP({"SSL":True})

    83         content = "sadfadfgwertsdgsdf\n\nsdfgaerasdfsad\nadfasdf"

    84         p.send("t@t.de", Mail("sub", content, None))

    85 

    86         self.assertEqual(mock_ssl.call_count,1)

    87 

    88         self.assertEqual(len(self.smtp_server.rcvd), 1)

    89 

    90     @patch("smtplib.SMTP")

    91     def testTLSSendMail(self,mock_smtp):

    92         mock_s = Mock()

    93         mock_smtp.return_value = mock_s

    94 

    95         p=self.getSMTP({"TLS":True})

    96         content = "sadfadfgwertsdgsdf\n\nsdfgaerasdfsad\nadfasdf"

    97         p.send("t@t.de", Mail("sub", content, None))

    98         

    99         mock_s.starttls.assert_called_once_with()

   100         self.assertEqual(mock_s.sendmail.call_count,1)

   101         self.assertEqual([i[0] for i in mock_s.method_calls],["starttls","sendmail","quit"])

   102 

   103     @patch("smtplib.SMTP")

   104     def testLoginSendMail(self,mock_smtp):

   105         mock_s = Mock()

   106         mock_smtp.return_value = mock_s

   107 

   108         p=self.getSMTP({"user":"user","password":"pw"})

   109         content = "sadfadfgwertsdgsdf\n\nsdfgaerasdfsad\nadfasdf"

   110         p.send("t@t.de", Mail("sub", content, None))

   111         

   112         mock_s.login.assert_called_once_with("user","pw")

   113         self.assertEqual(mock_s.sendmail.call_count,1)

   114         self.assertEqual([i[0] for i in mock_s.method_calls],["login","sendmail","quit"])

   115 

   116     

   117     def testNeededOption(self):

   118         c={"send_from":"send@t.de",

   119             "host":HOST,

   120             "port":PORT,

   121             "user":"u",

   122             "password":"p",

   123             "typ":"smtp",

   124             }

   125         s = self.getSMTP(c)

   126         self.assertEqual(s.send_from, "send@t.de")

   127         self.assertEqual(s.host, HOST)

   128         self.assertEqual(s.port, PORT)

   129         self.assertEqual(s.user, "u")

   130         self.assertEqual(s.password, "p")

   131         self.assertEqual(s.SSL,False)

   132         self.assertEqual(s.TLS,False)

   133 

   134         c.update({"TLS":True, "SSL":True})

   135         s = self.getSMTP(c)

   136         self.assertEqual(s.SSL,True)

   137         self.assertEqual(s.TLS,True)

   138         

   139         del c["host"]

   140         self.assertRaises(NeededOption, s.load, c)

   141 

   142     def testSendFunc(self):

   143         s = self.getSMTP()

   144         self.assertEqual(s.getSendFunc("mail",None), s.send)

   145         self.assertRaises(NoRoute,s.getSendFunc,"mail","foo")

   146         self.assertRaises(NoTyp,s.getSendFunc,"mail2","foo")