~mailman-coders/mailman/2.1

# Copyright (C) 2001-2018 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)

        if mm_cfg.SMTPPORT == 0:

            mm_cfg.SMTPPORT = 25

        # 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()