<s> """<STR_LIT>""" <EOL> import poplib <EOL> import asyncore <EOL> import asynchat <EOL> import socket <EOL> import os <EOL> import time <EOL> import errno <EOL> from unittest import TestCase <EOL> from test import test_support <EOL> from test . test_support import HOST <EOL> threading = test_support . import_module ( '<STR_LIT>' ) <EOL> LIST_RESP = '<STR_LIT>' <EOL> RETR_RESP = """<STR_LIT>""" <EOL> class DummyPOP3Handler ( asynchat . async_chat ) : <EOL> def __init__ ( self , conn ) : <EOL> asynchat . async_chat . __init__ ( self , conn ) <EOL> self . set_terminator ( "<STR_LIT:\r\n>" ) <EOL> self . in_buffer = [ ] <EOL> self . push ( '<STR_LIT>' ) <EOL> def collect_incoming_data ( self , data ) : <EOL> self . in_buffer . append ( data ) <EOL> def found_terminator ( self ) : <EOL> line = '<STR_LIT>' . join ( self . in_buffer ) <EOL> self . in_buffer = [ ] <EOL> cmd = line . split ( '<STR_LIT:U+0020>' ) [ <NUM_LIT:0> ] . lower ( ) <EOL> space = line . find ( '<STR_LIT:U+0020>' ) <EOL> if space != - <NUM_LIT:1> : <EOL> arg = line [ space + <NUM_LIT:1> : ] <EOL> else : <EOL> arg = "<STR_LIT>" <EOL> if hasattr ( self , '<STR_LIT>' + cmd ) : <EOL> method = getattr ( self , '<STR_LIT>' + cmd ) <EOL> method ( arg ) <EOL> else : <EOL> self . push ( '<STR_LIT>' % cmd ) <EOL> def handle_error ( self ) : <EOL> raise <EOL> def push ( self , data ) : <EOL> asynchat . async_chat . push ( self , data + '<STR_LIT:\r\n>' ) <EOL> def cmd_echo ( self , arg ) : <EOL> self . push ( arg ) <EOL> def cmd_user ( self , arg ) : <EOL> if arg != "<STR_LIT>" : <EOL> self . push ( "<STR_LIT>" ) <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_pass ( self , arg ) : <EOL> if arg != "<STR_LIT>" : <EOL> self . push ( "<STR_LIT>" ) <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_stat ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_list ( self , arg ) : <EOL> if arg : <EOL> self . push ( '<STR_LIT>' % ( arg , arg ) ) <EOL> else : <EOL> self . push ( '<STR_LIT>' ) <EOL> asynchat . async_chat . push ( self , LIST_RESP ) <EOL> cmd_uidl = cmd_list <EOL> def cmd_retr ( self , arg ) : <EOL> self . push ( '<STR_LIT>' % len ( RETR_RESP ) ) <EOL> asynchat . async_chat . push ( self , RETR_RESP ) <EOL> cmd_top = cmd_retr <EOL> def cmd_dele ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_noop ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_rpop ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> class DummyPOP3Server ( asyncore . dispatcher , threading . Thread ) : <EOL> handler = DummyPOP3Handler <EOL> def __init__ ( self , address , af = socket . AF_INET ) : <EOL> threading . Thread . __init__ ( self ) <EOL> asyncore . dispatcher . __init__ ( self ) <EOL> self . create_socket ( af , socket . SOCK_STREAM ) <EOL> self . bind ( address ) <EOL> self . listen ( <NUM_LIT:5> ) <EOL> self . active = False <EOL> self . active_lock = threading . Lock ( ) <EOL> self . host , self . port = self . socket . getsockname ( ) [ : <NUM_LIT:2> ] <EOL> def start ( self ) : <EOL> assert not self . active <EOL> self . __flag = threading . Event ( ) <EOL> threading . Thread . start ( self ) <EOL> self . __flag . wait ( ) <EOL> def run ( self ) : <EOL> self . active = True <EOL> self . __flag . set ( ) <EOL> while self . active and asyncore . socket_map : <EOL> self . active_lock . acquire ( ) <EOL> asyncore . loop ( timeout = <NUM_LIT:0.1> , count = <NUM_LIT:1> ) <EOL> self . active_lock . release ( ) <EOL> asyncore . close_all ( ignore_all = True ) <EOL> def stop ( self ) : <EOL> assert self . active <EOL> self . active = False <EOL> self . join ( ) <EOL> def handle_accept ( self ) : <EOL> conn , addr = self . accept ( ) <EOL> self . handler = self . handler ( conn ) <EOL> self . close ( ) <EOL> def handle_connect ( self ) : <EOL> self . close ( ) <EOL> handle_read = handle_connect <EOL> def writable ( self ) : <EOL> return <NUM_LIT:0> <EOL> def handle_error ( self ) : <EOL> raise <EOL> class TestPOP3Class ( TestCase ) : <EOL> def assertOK ( self , resp ) : <EOL> self . assertTrue ( resp . startswith ( "<STR_LIT>" ) ) <EOL> def setUp ( self ) : <EOL> self . server = DummyPOP3Server ( ( HOST , <NUM_LIT:0> ) ) <EOL> self . server . start ( ) <EOL> self . client = poplib . POP3 ( self . server . host , self . server . port ) <EOL> def tearDown ( self ) : <EOL> self . client . quit ( ) <EOL> self . server . stop ( ) <EOL> def test_getwelcome ( self ) : <EOL> self . assertEqual ( self . client . getwelcome ( ) , '<STR_LIT>' ) <EOL> def test_exceptions ( self ) : <EOL> self . assertRaises ( poplib . error_proto , self . client . _shortcmd , '<STR_LIT>' ) <EOL> def test_user ( self ) : <EOL> self . assertOK ( self . client . user ( '<STR_LIT>' ) ) <EOL> self . assertRaises ( poplib . error_proto , self . client . user , '<STR_LIT>' ) <EOL> def test_pass_ ( self ) : <EOL> self . assertOK ( self . client . pass_ ( '<STR_LIT>' ) ) <EOL> self . assertRaises ( poplib . error_proto , self . client . user , '<STR_LIT>' ) <EOL> def test_stat ( self ) : <EOL> self . assertEqual ( self . client . stat ( ) , ( <NUM_LIT:10> , <NUM_LIT:100> ) ) <EOL> def test_list ( self ) : <EOL> self . assertEqual ( self . client . list ( ) [ <NUM_LIT:1> : ] , <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <NUM_LIT> ) ) <EOL> self . assertTrue ( self . client . list ( '<STR_LIT:1>' ) . endswith ( "<STR_LIT>" ) ) <EOL> def test_retr ( self ) : <EOL> expected = ( '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> self . assertEqual ( self . client . retr ( '<STR_LIT:foo>' ) , expected ) <EOL> def test_dele ( self ) : <EOL> self . assertOK ( self . client . dele ( '<STR_LIT:foo>' ) ) <EOL> def test_noop ( self ) : <EOL> self . assertOK ( self . client . noop ( ) ) <EOL> def test_rpop ( self ) : <EOL> self . assertOK ( self . client . rpop ( '<STR_LIT:foo>' ) ) <EOL> def test_top ( self ) : <EOL> expected = ( '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> ) <EOL> self . assertEqual ( self . client . top ( <NUM_LIT:1> , <NUM_LIT:1> ) , expected ) <EOL> def test_uidl ( self ) : <EOL> self . client . uidl ( ) <EOL> self . client . uidl ( '<STR_LIT:foo>' ) <EOL> SUPPORTS_SSL = False <EOL> if hasattr ( poplib , '<STR_LIT>' ) : <EOL> import ssl <EOL> SUPPORTS_SSL = True <EOL> CERTFILE = os . path . join ( os . path . dirname ( __file__ ) or os . curdir , "<STR_LIT>" ) <EOL> class DummyPOP3_SSLHandler ( DummyPOP3Handler ) : <EOL> def __init__ ( self , conn ) : <EOL> asynchat . async_chat . __init__ ( self , conn ) <EOL> self . socket = ssl . wrap_socket ( self . socket , certfile = CERTFILE , <EOL> server_side = True , <EOL> do_handshake_on_connect = False ) <EOL> self . _ssl_accepting = True <EOL> self . _do_ssl_handshake ( ) <EOL> self . set_terminator ( "<STR_LIT:\r\n>" ) <EOL> self . in_buffer = [ ] <EOL> self . push ( '<STR_LIT>' ) <EOL> def _do_ssl_handshake ( self ) : <EOL> try : <EOL> self . socket . do_handshake ( ) <EOL> except ssl . SSLError , err : <EOL> if err . args [ <NUM_LIT:0> ] in ( ssl . SSL_ERROR_WANT_READ , <EOL> ssl . SSL_ERROR_WANT_WRITE ) : <EOL> return <EOL> elif err . args [ <NUM_LIT:0> ] == ssl . SSL_ERROR_EOF : <EOL> return self . handle_close ( ) <EOL> raise <EOL> except socket . error , err : <EOL> if err . args [ <NUM_LIT:0> ] == errno . ECONNABORTED : <EOL> return self . handle_close ( ) <EOL> else : <EOL> self . _ssl_accepting = False <EOL> def handle_read ( self ) : <EOL> if self . _ssl_accepting : <EOL> self . _do_ssl_handshake ( ) <EOL> else : <EOL> DummyPOP3Handler . handle_read ( self ) <EOL> class TestPOP3_SSLClass ( TestPOP3Class ) : <EOL> def setUp ( self ) : <EOL> self . server = DummyPOP3Server ( ( HOST , <NUM_LIT:0> ) ) <EOL> self . server . handler = DummyPOP3_SSLHandler <EOL> self . server . start ( ) <EOL> self . client = poplib . POP3_SSL ( self . server . host , self . server . port ) <EOL> def test__all__ ( self ) : <EOL> self . assertIn ( '<STR_LIT>' , poplib . __all__ ) <EOL> class TestTimeouts ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . evt = threading . Event ( ) <EOL> self . sock = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> self . sock . settimeout ( <NUM_LIT> ) <EOL> self . port = test_support . bind_port ( self . sock ) <EOL> self . thread = threading . Thread ( target = self . server , args = ( self . evt , self . sock ) ) <EOL> self . thread . setDaemon ( True ) <EOL> self . thread . start ( ) <EOL> self . evt . wait ( ) <EOL> def tearDown ( self ) : <EOL> self . thread . join ( ) <EOL> del self . thread <EOL> def server ( self , evt , serv ) : <EOL> serv . listen ( <NUM_LIT:5> ) <EOL> evt . set ( ) <EOL> try : <EOL> conn , addr = serv . accept ( ) <EOL> conn . send ( "<STR_LIT>" ) <EOL> conn . close ( ) <EOL> except socket . timeout : <EOL> pass <EOL> finally : <EOL> serv . close ( ) <EOL> def testTimeoutDefault ( self ) : <EOL> self . assertTrue ( socket . getdefaulttimeout ( ) is None ) <EOL> socket . setdefaulttimeout ( <NUM_LIT:30> ) <EOL> try : <EOL> pop = poplib . POP3 ( HOST , self . port ) <EOL> finally : <EOL> socket . setdefaulttimeout ( None ) <EOL> self . assertEqual ( pop . sock . gettimeout ( ) , <NUM_LIT:30> ) <EOL> pop . sock . close ( ) <EOL> def testTimeoutNone ( self ) : <EOL> self . assertTrue ( socket . getdefaulttimeout ( ) is None ) <EOL> socket . setdefaulttimeout ( <NUM_LIT:30> ) <EOL> try : <EOL> pop = poplib . POP3 ( HOST , self . port , timeout = None ) <EOL> finally : <EOL> socket . setdefaulttimeout ( None ) <EOL> self . assertTrue ( pop . sock . gettimeout ( ) is None ) <EOL> pop . sock . close ( ) <EOL> def testTimeoutValue ( self ) : <EOL> pop = poplib . POP3 ( HOST , self . port , timeout = <NUM_LIT:30> ) <EOL> self . assertEqual ( pop . sock . gettimeout ( ) , <NUM_LIT:30> ) <EOL> pop . sock . close ( ) <EOL> def test_main ( ) : <EOL> tests = [ TestPOP3Class , TestTimeouts ] <EOL> if SUPPORTS_SSL : <EOL> tests . append ( TestPOP3_SSLClass ) <EOL> thread_info = test_support . threading_setup ( ) <EOL> try : <EOL> test_support . run_unittest ( * tests ) <EOL> finally : <EOL> test_support . threading_cleanup ( * thread_info ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> test_main ( ) </s>
