1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
# Copyright (C) 2001-2007 by the Free Software Foundation, Inc.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301,
# USA.
"""Base class for tests that email things."""
import os
import time
import errno
import smtpd
import socket
import asyncore
import subprocess
from Mailman.configuration import config
from Mailman.testing.base import TestBase
TESTPORT = 10825
MSGTEXT = None
class OneShotChannel(smtpd.SMTPChannel):
def smtp_QUIT(self, arg):
smtpd.SMTPChannel.smtp_QUIT(self, arg)
raise asyncore.ExitNow
class SinkServer(smtpd.SMTPServer):
def handle_accept(self):
conn, addr = self.accept()
channel = OneShotChannel(self, conn, addr)
def process_message(self, peer, mailfrom, rcpttos, data):
global MSGTEXT
MSGTEXT = data
class EmailBase(TestBase):
def setUp(self):
TestBase.setUp(self)
try:
# Second argument is ignored.
self._server = SinkServer(('localhost', TESTPORT), None)
except:
TestBase.tearDown(self)
raise
try:
os.system('bin/mailmanctl -C %s -q start' % config.filename)
# If any errors occur in the above, be sure to manually call
# tearDown(). unittest doesn't call tearDown() for errors in
# setUp().
except:
self.tearDown()
raise
def tearDown(self):
os.system('bin/mailmanctl -C %s -q stop' % config.filename)
self._server.close()
# Wait a while until the server actually goes away
while True:
try:
s = socket.socket()
s.connect(('localhost', TESTPORT))
s.close()
time.sleep(3)
except socket.error, e:
# IWBNI e had an errno attribute
if e[0] in (errno.ECONNREFUSED, errno.ETIMEDOUT):
break
else:
raise
TestBase.tearDown(self)
def _readmsg(self):
global MSGTEXT
# Save and unlock the list so that the qrunner process can open it and
# lock it if necessary. We'll re-lock the list in our finally clause
# since that if an invariant of the test harness.
self._mlist.Unlock()
try:
try:
# timeout is in milliseconds, see asyncore.py poll3()
asyncore.loop()
MSGTEXT = None
except asyncore.ExitNow:
pass
asyncore.close_all()
return MSGTEXT
finally:
self._mlist.Lock()
|
