diff options
Diffstat (limited to 'tests/EmailBase.py')
| -rw-r--r-- | tests/EmailBase.py | 75 |
1 files changed, 0 insertions, 75 deletions
diff --git a/tests/EmailBase.py b/tests/EmailBase.py deleted file mode 100644 index d83ccbcf4..000000000 --- a/tests/EmailBase.py +++ /dev/null @@ -1,75 +0,0 @@ -# Copyright (C) 2001-2003 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 socket -import asyncore -import smtpd - -from Mailman import mm_cfg - -from TestBase import TestBase - - - -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) - # Second argument tuple is ignored. - self._server = SinkServer(('localhost', mm_cfg.SMTPPORT), - ('localhost', 25)) - - def tearDown(self): - self._server.close() - 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(timeout=30.0) - MSGTEXT = None - except asyncore.ExitNow: - pass - return MSGTEXT - finally: - self._mlist.Lock() |
