test script for poplib module modified by giampaolo rodola to give poplib pop3 and poplib pop3ssl a real test suite import poplib import socket import os import errno import threading import unittest from unittest import testcase skipunless from test import support as testsupport from test support import hashlibhelper from test support import sockethelper from test support import threadinghelper from test support import asynchat from test support import asyncore testsupport requiresworkingsocketmoduletrue host sockethelper host port 0 supportsssl false if hasattrpoplib pop3ssl import ssl supportsssl true certfile os path joinos path dirnamefile or os curdir certdata keycert3 pem cafile os path joinos path dirnamefile or os curdir certdata pycacert pem requiresssl skipunlesssupportsssl ssl not supported the dummy data returned by server when list and retr commands are issued listresp b 1 1rn2 2rn3 3rn4 4rn5 5rn rn retrresp bfrom postmasterpython org rncontenttype textplainrn mimeversion 1 0rn subject dummyrn rn line1rn line2rn line3rn rn class dummypop3handlerasynchat asyncchat capas uidl implementation pythontestlibpopserver enableutf8 false def initself conn asynchat asyncchat initself conn self setterminatorbrn self inbuffer self push ok dummy pop3 server ready timestamp self tlsactive false self tlsstarting false def collectincomingdataself data self inbuffer appenddata def foundterminatorself line b joinself inbuffer line strline iso88591 self inbuffer cmd line split 0 lower space line find if space 1 arg linespace 1 else arg if hasattrself cmd cmd method getattrself cmd cmd methodarg else self push err unrecognized pop3 command s cmd def handleerrorself raise def pushself data asynchat asyncchat pushself data encodeiso88591 b rn def cmdechoself arg sends back the received string used by the test suite self pusharg def cmduserself arg if arg guido self pusherr no such user self push ok password required def cmdpassself arg if arg python self pusherr wrong password self push ok 10 messages def cmdstatself arg self push ok 10 100 def cmdlistself arg if arg self push ok s s arg arg else self push ok asynchat asyncchat pushself listresp cmduidl cmdlist def cmdretrself arg self push ok s bytes lenretrresp asynchat asyncchat pushself retrresp cmdtop cmdretr def cmddeleself arg self push ok message marked for deletion def cmdnoopself arg self push ok done nothing def cmdrpopself arg self push ok done nothing def cmdapopself arg self push ok done nothing def cmdquitself arg self push ok closing self closewhendone def getcapasself capas dictself capas if not self tlsactive and supportsssl capas stls return capas def cmdcapaself arg self push ok capability list follows if self getcapas for cap params in self getcapas items ln cap if params ln extendparams self push joinln self push def cmdutf8self arg self push ok i know rfc6856 if self enableutf8 else err what is utf8 if supportsssl def cmdstlsself arg if self tlsactive is false self push ok begin tls negotiation context ssl sslcontextssl protocoltlsserver context loadcertchaincertfile tlssock context wrapsocketself socket serversidetrue dohandshakeonconnectfalse suppressraggedeofsfalse self delchannel self setsockettlssock self tlsactive true self tlsstarting true self inbuffer self dotlshandshake else self push err command not permitted when tls active def dotlshandshakeself try self socket dohandshake except ssl sslerror as err if err args0 in ssl sslerrorwantread ssl sslerrorwantwrite return elif err args0 ssl sslerroreof return self handleclose todo sslerror does not expose alert information elif sslv3alertbadcertificate in err args1 or sslv3alertcertificateunknown in err args1 return self handleclose raise except oserror as err if err args0 errno econnaborted return self handleclose else self tlsactive true self tlsstarting false def handlereadself if self tlsstarting self dotlshandshake else try asynchat asyncchat handlereadself except ssl ssleoferror self handleclose class dummypop3serverasyncore dispatcher threading thread handler dummypop3handler def initself address afsocket afinet threading thread initself asyncore dispatcher initself self daemon true self createsocketaf socket sockstream self bindaddress self listen5 self active false self activelock threading lock self host self port self socket getsockname 2 self handlerinstance none def startself assert not self active self flag threading event threading thread startself self flag wait def runself self active true self flag set try while self active and asyncore socketmap with self activelock asyncore looptimeout0 1 count1 finally asyncore closeallignorealltrue def stopself assert self active self active false self join def handleacceptedself conn addr self handlerinstance self handlerconn def handleconnectself self close handleread handleconnect def writableself return 0 def handleerrorself raise class testpop3classtestcase def assertokself resp self asserttrueresp startswithbok def setupself self server dummypop3serverhost port self server start self client poplib pop3self server host self server port timeouttestsupport loopbacktimeout def teardownself self client close self server stop explicitly clear the attribute to prevent dangling thread self server none def testgetwelcomeself self assertequalself client getwelcome b ok dummy pop3 server ready timestamp def testexceptionsself self assertraisespoplib errorproto self client shortcmd echo err def testuserself self assertokself client user guido self assertraisespoplib errorproto self client user invalid def testpassself self assertokself client pass python self assertraisespoplib errorproto self client user invalid def teststatself self assertequalself client stat 10 100 def testlistself self assertequalself client list1 b 1 1 b 2 2 b 3 3 b 4 4 b 5 5 25 self asserttrueself client list 1 endswithbok 1 1 def testretrself expected b ok 116 bytes b from postmasterpython org b contenttype textplain b mimeversion 1 0 b subject dummy b b line1 b line2 b line3 113 foo self client retr foo self assertequalfoo expected def testtoolonglinesself self assertraisespoplib errorproto self client shortcmd echo s poplib maxline 10 a def testdeleself self assertokself client dele foo def testnoopself self assertokself client noop def testrpopself self assertokself client rpop foo hashlibhelper requireshashdigest md5 openssltrue def testapopnormalself self assertokself client apop foo dummypassword hashlibhelper requireshashdigest md5 openssltrue def testapopredosself replace welcome with very long evil welcome nb the upper bound on welcome length is currently 2048 at this length evil input makes each apop call take on the order of milliseconds instead of microseconds evilwelcome b ok b 1000000 with testsupport swapattrself client welcome evilwelcome the evil welcome is invalid so apop should throw self assertraisespoplib errorproto self client apop a kb def testtopself expected b ok 116 bytes b from postmasterpython org b contenttype textplain b mimeversion 1 0 b subject dummy b b line1 b line2 b line3 113 self assertequalself client top1 1 expected def testuidlself self client uidl self client uidl foo def testutf8raisesifunsupportedself self server handler enableutf8 false self assertraisespoplib errorproto self client utf8 def testutf8self self server handler enableutf8 true expected b ok i know rfc6856 result self client utf8 self assertequalresult expected def testcapaself capa self client capa self asserttrue implementation in capa keys def testquitself resp self client quit self asserttrueresp self assertisnoneself client sock self assertisnoneself client file requiresssl def teststlscapaself capa self client capa self asserttrue stls in capa keys requiresssl def teststlsself expected b ok begin tls negotiation resp self client stls self assertequalresp expected requiresssl def teststlscontextself expected b ok begin tls negotiation ctx ssl sslcontextssl protocoltlsclient ctx loadverifylocationscafile self assertequalctx verifymode ssl certrequired self assertequalctx checkhostname true with self assertraisesssl certificateerror resp self client stlscontextctx self client poplib pop3localhost self server port timeouttestsupport loopbacktimeout resp self client stlscontextctx self assertequalresp expected if supportsssl from test testftplib import sslconnection class dummypop3sslhandlersslconnection dummypop3handler def initself conn asynchat asyncchat initself conn self secureconnection self setterminatorbrn self inbuffer self push ok dummy pop3 server ready timestamp self tlsactive true self tlsstarting false requiresssl class testpop3sslclasstestpop3class repeat previous tests by using poplib pop3ssl def setupself self server dummypop3serverhost port self server handler dummypop3sslhandler self server start self client poplib pop3sslself server host self server port def testallself self assertin pop3ssl poplib all def testcontextself ctx ssl sslcontextssl protocoltlsclient ctx checkhostname false ctx verifymode ssl certnone self client quit self client poplib pop3sslself server host self server port contextctx self assertisinstanceself client sock ssl sslsocket self assertisself client sock context ctx self asserttrueself client noop startswithb ok def teststlsself self assertraisespoplib errorproto self client stls teststlscontext teststls def teststlscapaself capa self client capa self assertfalse stls in capa keys requiresssl class testpop3tlsclasstestpop3class repeat previous tests by using poplib pop3 stls def setupself self server dummypop3serverhost port self server start self client poplib pop3self server host self server port timeouttestsupport loopbacktimeout self client stls def teardownself if self client file is not none and self client sock is not none try self client quit except poplib errorproto happens in the testtoolonglines case the overlong response will be treated as response to quit and raise this exception self client close self server stop explicitly clear the attribute to prevent dangling thread self server none def teststlsself self assertraisespoplib errorproto self client stls teststlscontext teststls def teststlscapaself capa self client capa self assertfalseb stls in capa keys class testtimeoutstestcase def setupself self evt threading event self sock socket socketsocket afinet socket sockstream self sock settimeout60 safety net look issue 11812 self port sockethelper bindportself sock self thread threading threadtargetself server argsself evt self sock self thread daemon true self thread start self evt wait def teardownself self thread join explicitly clear the attribute to prevent dangling thread self thread none def serverself evt serv serv listen evt set try conn addr serv accept conn sendb hola mundon conn close except timeouterror pass finally serv close def testtimeoutdefaultself self assertisnonesocket getdefaulttimeout socket setdefaulttimeouttestsupport loopbacktimeout try pop poplib pop3host self port finally socket setdefaulttimeoutnone self assertequalpop sock gettimeout testsupport loopbacktimeout pop close def testtimeoutnoneself self assertisnonesocket getdefaulttimeout socket setdefaulttimeout30 try pop poplib pop3host self port timeoutnone finally socket setdefaulttimeoutnone self assertisnonepop sock gettimeout pop close def testtimeoutvalueself pop poplib pop3host self port timeouttestsupport loopbacktimeout self assertequalpop sock gettimeout testsupport loopbacktimeout pop close with self assertraisesvalueerror poplib pop3host self port timeout0 def setupmodule threadinfo threadinghelper threadingsetup unittest addmodulecleanupthreadinghelper threadingcleanup threadinfo if name main unittest main modified by giampaolo rodola to give poplib pop3 and poplib pop3_ssl a real test suite the dummy data returned by server when list and retr commands are issued from postmaster python org r ncontent type text plain r n mime version 1 0 r n subject dummy r n r n line1 r n line2 r n line3 r n r n sends back the received string used by the test suite todo sslerror does not expose alert information explicitly clear the attribute to prevent dangling thread replace welcome with very long evil welcome nb the upper bound on welcome length is currently 2048 at this length evil input makes each apop call take on the order of milliseconds instead of microseconds the evil welcome is invalid so apop should throw repeat previous tests by using poplib pop3_ssl repeat previous tests by using poplib pop3 stls happens in the test_too_long_lines case the overlong response will be treated as response to quit and raise this exception explicitly clear the attribute to prevent dangling thread safety net look issue 11812 explicitly clear the attribute to prevent dangling thread
import poplib
import socket
import os
import errno
import threading

import unittest
from unittest import TestCase, skipUnless
from test import support as test_support
from test.support import hashlib_helper
from test.support import socket_helper
from test.support import threading_helper
from test.support import asynchat
from test.support import asyncore


test_support.requires_working_socket(module=True)

HOST = socket_helper.HOST
PORT = 0

SUPPORTS_SSL = False
if hasattr(poplib, 'POP3_SSL'):
    import ssl

    SUPPORTS_SSL = True
    CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "certdata", "keycert3.pem")
    CAFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "certdata", "pycacert.pem")

requires_ssl = skipUnless(SUPPORTS_SSL, 'SSL not supported')


LIST_RESP = b'1 1\r\n2 2\r\n3 3\r\n4 4\r\n5 5\r\n.\r\n'
RETR_RESP = b


class DummyPOP3Handler(asynchat.async_chat):

    CAPAS = {'UIDL': [], 'IMPLEMENTATION': ['python-testlib-pop-server']}
    enable_UTF8 = False

    def __init__(self, conn):
        asynchat.async_chat.__init__(self, conn)
        self.set_terminator(b"\r\n")
        self.in_buffer = []
        self.push('+OK dummy pop3 server ready. <timestamp>')
        self.tls_active = False
        self.tls_starting = False

    def collect_incoming_data(self, data):
        self.in_buffer.append(data)

    def found_terminator(self):
        line = b''.join(self.in_buffer)
        line = str(line, 'ISO-8859-1')
        self.in_buffer = []
        cmd = line.split(' ')[0].lower()
        space = line.find(' ')
        if space != -1:
            arg = line[space + 1:]
        else:
            arg = ""
        if hasattr(self, 'cmd_' + cmd):
            method = getattr(self, 'cmd_' + cmd)
            method(arg)
        else:
            self.push('-ERR unrecognized POP3 command "%s".' %cmd)

    def handle_error(self):
        raise

    def push(self, data):
        asynchat.async_chat.push(self, data.encode("ISO-8859-1") + b'\r\n')

    def cmd_echo(self, arg):
        
        self.push(arg)

    def cmd_user(self, arg):
        if arg != "guido":
            self.push("-ERR no such user")
        self.push('+OK password required')

    def cmd_pass(self, arg):
        if arg != "python":
            self.push("-ERR wrong password")
        self.push('+OK 10 messages')

    def cmd_stat(self, arg):
        self.push('+OK 10 100')

    def cmd_list(self, arg):
        if arg:
            self.push('+OK %s %s' % (arg, arg))
        else:
            self.push('+OK')
            asynchat.async_chat.push(self, LIST_RESP)

    cmd_uidl = cmd_list

    def cmd_retr(self, arg):
        self.push('+OK %s bytes' %len(RETR_RESP))
        asynchat.async_chat.push(self, RETR_RESP)

    cmd_top = cmd_retr

    def cmd_dele(self, arg):
        self.push('+OK message marked for deletion.')

    def cmd_noop(self, arg):
        self.push('+OK done nothing.')

    def cmd_rpop(self, arg):
        self.push('+OK done nothing.')

    def cmd_apop(self, arg):
        self.push('+OK done nothing.')

    def cmd_quit(self, arg):
        self.push('+OK closing.')
        self.close_when_done()

    def _get_capas(self):
        _capas = dict(self.CAPAS)
        if not self.tls_active and SUPPORTS_SSL:
            _capas['STLS'] = []
        return _capas

    def cmd_capa(self, arg):
        self.push('+OK Capability list follows')
        if self._get_capas():
            for cap, params in self._get_capas().items():
                _ln = [cap]
                if params:
                    _ln.extend(params)
                self.push(' '.join(_ln))
        self.push('.')

    def cmd_utf8(self, arg):
        self.push('+OK I know RFC6856'
                  if self.enable_UTF8
                  else '-ERR What is UTF8?!')

    if SUPPORTS_SSL:

        def cmd_stls(self, arg):
            if self.tls_active is False:
                self.push('+OK Begin TLS negotiation')
                context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
                context.load_cert_chain(CERTFILE)
                tls_sock = context.wrap_socket(self.socket,
                                               server_side=True,
                                               do_handshake_on_connect=False,
                                               suppress_ragged_eofs=False)
                self.del_channel()
                self.set_socket(tls_sock)
                self.tls_active = True
                self.tls_starting = True
                self.in_buffer = []
                self._do_tls_handshake()
            else:
                self.push('-ERR Command not permitted when TLS active')

        def _do_tls_handshake(self):
            try:
                self.socket.do_handshake()
            except ssl.SSLError as err:
                if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
                                   ssl.SSL_ERROR_WANT_WRITE):
                    return
                elif err.args[0] == ssl.SSL_ERROR_EOF:
                    return self.handle_close()
                
                elif ("SSLV3_ALERT_BAD_CERTIFICATE" in err.args[1] or
                      "SSLV3_ALERT_CERTIFICATE_UNKNOWN" in err.args[1]):
                    return self.handle_close()
                raise
            except OSError as err:
                if err.args[0] == errno.ECONNABORTED:
                    return self.handle_close()
            else:
                self.tls_active = True
                self.tls_starting = False

        def handle_read(self):
            if self.tls_starting:
                self._do_tls_handshake()
            else:
                try:
                    asynchat.async_chat.handle_read(self)
                except ssl.SSLEOFError:
                    self.handle_close()

class DummyPOP3Server(asyncore.dispatcher, threading.Thread):

    handler = DummyPOP3Handler

    def __init__(self, address, af=socket.AF_INET):
        threading.Thread.__init__(self)
        asyncore.dispatcher.__init__(self)
        self.daemon = True
        self.create_socket(af, socket.SOCK_STREAM)
        self.bind(address)
        self.listen(5)
        self.active = False
        self.active_lock = threading.Lock()
        self.host, self.port = self.socket.getsockname()[:2]
        self.handler_instance = None

    def start(self):
        assert not self.active
        self.__flag = threading.Event()
        threading.Thread.start(self)
        self.__flag.wait()

    def run(self):
        self.active = True
        self.__flag.set()
        try:
            while self.active and asyncore.socket_map:
                with self.active_lock:
                    asyncore.loop(timeout=0.1, count=1)
        finally:
            asyncore.close_all(ignore_all=True)

    def stop(self):
        assert self.active
        self.active = False
        self.join()

    def handle_accepted(self, conn, addr):
        self.handler_instance = self.handler(conn)

    def handle_connect(self):
        self.close()
    handle_read = handle_connect

    def writable(self):
        return 0

    def handle_error(self):
        raise


class TestPOP3Class(TestCase):
    def assertOK(self, resp):
        self.assertTrue(resp.startswith(b"+OK"))

    def setUp(self):
        self.server = DummyPOP3Server((HOST, PORT))
        self.server.start()
        self.client = poplib.POP3(self.server.host, self.server.port,
                                  timeout=test_support.LOOPBACK_TIMEOUT)

    def tearDown(self):
        self.client.close()
        self.server.stop()
        
        self.server = None

    def test_getwelcome(self):
        self.assertEqual(self.client.getwelcome(),
                         b'+OK dummy pop3 server ready. <timestamp>')

    def test_exceptions(self):
        self.assertRaises(poplib.error_proto, self.client._shortcmd, 'echo -err')

    def test_user(self):
        self.assertOK(self.client.user('guido'))
        self.assertRaises(poplib.error_proto, self.client.user, 'invalid')

    def test_pass_(self):
        self.assertOK(self.client.pass_('python'))
        self.assertRaises(poplib.error_proto, self.client.user, 'invalid')

    def test_stat(self):
        self.assertEqual(self.client.stat(), (10, 100))

    def test_list(self):
        self.assertEqual(self.client.list()[1:],
                         ([b'1 1', b'2 2', b'3 3', b'4 4', b'5 5'],
                          25))
        self.assertTrue(self.client.list('1').endswith(b"OK 1 1"))

    def test_retr(self):
        expected = (b'+OK 116 bytes',
                    [b'From: postmaster@python.org', b'Content-Type: text/plain',
                     b'MIME-Version: 1.0', b'Subject: Dummy',
                     b'', b'line1', b'line2', b'line3'],
                    113)
        foo = self.client.retr('foo')
        self.assertEqual(foo, expected)

    def test_too_long_lines(self):
        self.assertRaises(poplib.error_proto, self.client._shortcmd,
                          'echo +%s' % ((poplib._MAXLINE + 10) * 'a'))

    def test_dele(self):
        self.assertOK(self.client.dele('foo'))

    def test_noop(self):
        self.assertOK(self.client.noop())

    def test_rpop(self):
        self.assertOK(self.client.rpop('foo'))

    @hashlib_helper.requires_hashdigest('md5', openssl=True)
    def test_apop_normal(self):
        self.assertOK(self.client.apop('foo', 'dummypassword'))

    @hashlib_helper.requires_hashdigest('md5', openssl=True)
    def test_apop_REDOS(self):
        
        
        
        
        evil_welcome = b'+OK' + (b'<' * 1000000)
        with test_support.swap_attr(self.client, 'welcome', evil_welcome):
            
            self.assertRaises(poplib.error_proto, self.client.apop, 'a', 'kb')

    def test_top(self):
        expected =  (b'+OK 116 bytes',
                     [b'From: postmaster@python.org', b'Content-Type: text/plain',
                      b'MIME-Version: 1.0', b'Subject: Dummy', b'',
                      b'line1', b'line2', b'line3'],
                     113)
        self.assertEqual(self.client.top(1, 1), expected)

    def test_uidl(self):
        self.client.uidl()
        self.client.uidl('foo')

    def test_utf8_raises_if_unsupported(self):
        self.server.handler.enable_UTF8 = False
        self.assertRaises(poplib.error_proto, self.client.utf8)

    def test_utf8(self):
        self.server.handler.enable_UTF8 = True
        expected = b'+OK I know RFC6856'
        result = self.client.utf8()
        self.assertEqual(result, expected)

    def test_capa(self):
        capa = self.client.capa()
        self.assertTrue('IMPLEMENTATION' in capa.keys())

    def test_quit(self):
        resp = self.client.quit()
        self.assertTrue(resp)
        self.assertIsNone(self.client.sock)
        self.assertIsNone(self.client.file)

    @requires_ssl
    def test_stls_capa(self):
        capa = self.client.capa()
        self.assertTrue('STLS' in capa.keys())

    @requires_ssl
    def test_stls(self):
        expected = b'+OK Begin TLS negotiation'
        resp = self.client.stls()
        self.assertEqual(resp, expected)

    @requires_ssl
    def test_stls_context(self):
        expected = b'+OK Begin TLS negotiation'
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.load_verify_locations(CAFILE)
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
        self.assertEqual(ctx.check_hostname, True)
        with self.assertRaises(ssl.CertificateError):
            resp = self.client.stls(context=ctx)
        self.client = poplib.POP3("localhost", self.server.port,
                                  timeout=test_support.LOOPBACK_TIMEOUT)
        resp = self.client.stls(context=ctx)
        self.assertEqual(resp, expected)


if SUPPORTS_SSL:
    from test.test_ftplib import SSLConnection

    class DummyPOP3_SSLHandler(SSLConnection, DummyPOP3Handler):

        def __init__(self, conn):
            asynchat.async_chat.__init__(self, conn)
            self.secure_connection()
            self.set_terminator(b"\r\n")
            self.in_buffer = []
            self.push('+OK dummy pop3 server ready. <timestamp>')
            self.tls_active = True
            self.tls_starting = False


@requires_ssl
class TestPOP3_SSLClass(TestPOP3Class):
    

    def setUp(self):
        self.server = DummyPOP3Server((HOST, PORT))
        self.server.handler = DummyPOP3_SSLHandler
        self.server.start()
        self.client = poplib.POP3_SSL(self.server.host, self.server.port)

    def test__all__(self):
        self.assertIn('POP3_SSL', poplib.__all__)

    def test_context(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE

        self.client.quit()
        self.client = poplib.POP3_SSL(self.server.host, self.server.port,
                                        context=ctx)
        self.assertIsInstance(self.client.sock, ssl.SSLSocket)
        self.assertIs(self.client.sock.context, ctx)
        self.assertTrue(self.client.noop().startswith(b'+OK'))

    def test_stls(self):
        self.assertRaises(poplib.error_proto, self.client.stls)

    test_stls_context = test_stls

    def test_stls_capa(self):
        capa = self.client.capa()
        self.assertFalse('STLS' in capa.keys())


@requires_ssl
class TestPOP3_TLSClass(TestPOP3Class):
    

    def setUp(self):
        self.server = DummyPOP3Server((HOST, PORT))
        self.server.start()
        self.client = poplib.POP3(self.server.host, self.server.port,
                                  timeout=test_support.LOOPBACK_TIMEOUT)
        self.client.stls()

    def tearDown(self):
        if self.client.file is not None and self.client.sock is not None:
            try:
                self.client.quit()
            except poplib.error_proto:
                
                
                
                self.client.close()
        self.server.stop()
        
        self.server = None

    def test_stls(self):
        self.assertRaises(poplib.error_proto, self.client.stls)

    test_stls_context = test_stls

    def test_stls_capa(self):
        capa = self.client.capa()
        self.assertFalse(b'STLS' in capa.keys())


class TestTimeouts(TestCase):

    def setUp(self):
        self.evt = threading.Event()
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.settimeout(60)  
        self.port = socket_helper.bind_port(self.sock)
        self.thread = threading.Thread(target=self.server, args=(self.evt, self.sock))
        self.thread.daemon = True
        self.thread.start()
        self.evt.wait()

    def tearDown(self):
        self.thread.join()
        
        self.thread = None

    def server(self, evt, serv):
        serv.listen()
        evt.set()
        try:
            conn, addr = serv.accept()
            conn.send(b"+ Hola mundo\n")
            conn.close()
        except TimeoutError:
            pass
        finally:
            serv.close()

    def testTimeoutDefault(self):
        self.assertIsNone(socket.getdefaulttimeout())
        socket.setdefaulttimeout(test_support.LOOPBACK_TIMEOUT)
        try:
            pop = poplib.POP3(HOST, self.port)
        finally:
            socket.setdefaulttimeout(None)
        self.assertEqual(pop.sock.gettimeout(), test_support.LOOPBACK_TIMEOUT)
        pop.close()

    def testTimeoutNone(self):
        self.assertIsNone(socket.getdefaulttimeout())
        socket.setdefaulttimeout(30)
        try:
            pop = poplib.POP3(HOST, self.port, timeout=None)
        finally:
            socket.setdefaulttimeout(None)
        self.assertIsNone(pop.sock.gettimeout())
        pop.close()

    def testTimeoutValue(self):
        pop = poplib.POP3(HOST, self.port, timeout=test_support.LOOPBACK_TIMEOUT)
        self.assertEqual(pop.sock.gettimeout(), test_support.LOOPBACK_TIMEOUT)
        pop.close()
        with self.assertRaises(ValueError):
            poplib.POP3(HOST, self.port, timeout=0)


def setUpModule():
    thread_info = threading_helper.threading_setup()
    unittest.addModuleCleanup(threading_helper.threading_cleanup, *thread_info)


if __name__ == '__main__':
    unittest.main()