<s> """<STR_LIT>""" <EOL> import ftplib <EOL> import asyncore <EOL> import asynchat <EOL> import socket <EOL> import io <EOL> import errno <EOL> import os <EOL> import time <EOL> try : <EOL> import ssl <EOL> except ImportError : <EOL> ssl = None <EOL> from unittest import TestCase <EOL> from test import support <EOL> from test . support import HOST , HOSTv6 <EOL> threading = support . import_module ( '<STR_LIT>' ) <EOL> RETR_DATA = '<STR_LIT>' * <NUM_LIT:1000> <EOL> LIST_DATA = '<STR_LIT>' <EOL> NLST_DATA = '<STR_LIT>' <EOL> MLSD_DATA = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class DummyDTPHandler ( asynchat . async_chat ) : <EOL> dtp_conn_closed = False <EOL> def __init__ ( self , conn , baseclass ) : <EOL> asynchat . async_chat . __init__ ( self , conn ) <EOL> self . baseclass = baseclass <EOL> self . baseclass . last_received_data = '<STR_LIT>' <EOL> def handle_read ( self ) : <EOL> self . baseclass . last_received_data += self . recv ( <NUM_LIT> ) . decode ( '<STR_LIT:ascii>' ) <EOL> def handle_close ( self ) : <EOL> if not self . dtp_conn_closed : <EOL> self . baseclass . push ( '<STR_LIT>' ) <EOL> self . close ( ) <EOL> self . dtp_conn_closed = True <EOL> def push ( self , what ) : <EOL> if self . baseclass . next_data is not None : <EOL> what = self . baseclass . next_data <EOL> self . baseclass . next_data = None <EOL> if not what : <EOL> return self . close_when_done ( ) <EOL> super ( DummyDTPHandler , self ) . push ( what . encode ( '<STR_LIT:ascii>' ) ) <EOL> def handle_error ( self ) : <EOL> raise <EOL> class DummyFTPHandler ( asynchat . async_chat ) : <EOL> dtp_handler = DummyDTPHandler <EOL> def __init__ ( self , conn ) : <EOL> asynchat . async_chat . __init__ ( self , conn ) <EOL> self . socket . setsockopt ( socket . SOL_SOCKET , socket . SO_OOBINLINE , <NUM_LIT:1> ) <EOL> self . set_terminator ( b"<STR_LIT:\r\n>" ) <EOL> self . in_buffer = [ ] <EOL> self . dtp = None <EOL> self . last_received_cmd = None <EOL> self . last_received_data = '<STR_LIT>' <EOL> self . next_response = '<STR_LIT>' <EOL> self . next_data = None <EOL> self . rest = None <EOL> self . next_retr_data = RETR_DATA <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 = b'<STR_LIT>' . join ( self . in_buffer ) . decode ( '<STR_LIT:ascii>' ) <EOL> self . in_buffer = [ ] <EOL> if self . next_response : <EOL> self . push ( self . next_response ) <EOL> self . next_response = '<STR_LIT>' <EOL> cmd = line . split ( '<STR_LIT:U+0020>' ) [ <NUM_LIT:0> ] . lower ( ) <EOL> self . last_received_cmd = cmd <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 . encode ( '<STR_LIT:ascii>' ) + b'<STR_LIT:\r\n>' ) <EOL> def cmd_port ( self , arg ) : <EOL> addr = list ( map ( int , arg . split ( '<STR_LIT:U+002C>' ) ) ) <EOL> ip = '<STR_LIT>' % tuple ( addr [ : <NUM_LIT:4> ] ) <EOL> port = ( addr [ <NUM_LIT:4> ] * <NUM_LIT> ) + addr [ <NUM_LIT:5> ] <EOL> s = socket . create_connection ( ( ip , port ) , timeout = <NUM_LIT:2> ) <EOL> self . dtp = self . dtp_handler ( s , baseclass = self ) <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_pasv ( self , arg ) : <EOL> with socket . socket ( ) as sock : <EOL> sock . bind ( ( self . socket . getsockname ( ) [ <NUM_LIT:0> ] , <NUM_LIT:0> ) ) <EOL> sock . listen ( <NUM_LIT:5> ) <EOL> sock . settimeout ( <NUM_LIT:10> ) <EOL> ip , port = sock . getsockname ( ) [ : <NUM_LIT:2> ] <EOL> ip = ip . replace ( '<STR_LIT:.>' , '<STR_LIT:U+002C>' ) ; p1 = port / <NUM_LIT> ; p2 = port % <NUM_LIT> <EOL> self . push ( '<STR_LIT>' % ( ip , p1 , p2 ) ) <EOL> conn , addr = sock . accept ( ) <EOL> self . dtp = self . dtp_handler ( conn , baseclass = self ) <EOL> def cmd_eprt ( self , arg ) : <EOL> af , ip , port = arg . split ( arg [ <NUM_LIT:0> ] ) [ <NUM_LIT:1> : - <NUM_LIT:1> ] <EOL> port = int ( port ) <EOL> s = socket . create_connection ( ( ip , port ) , timeout = <NUM_LIT:2> ) <EOL> self . dtp = self . dtp_handler ( s , baseclass = self ) <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_epsv ( self , arg ) : <EOL> with socket . socket ( socket . AF_INET6 ) as sock : <EOL> sock . bind ( ( self . socket . getsockname ( ) [ <NUM_LIT:0> ] , <NUM_LIT:0> ) ) <EOL> sock . listen ( <NUM_LIT:5> ) <EOL> sock . settimeout ( <NUM_LIT:10> ) <EOL> port = sock . getsockname ( ) [ <NUM_LIT:1> ] <EOL> self . push ( '<STR_LIT>' % port ) <EOL> conn , addr = sock . accept ( ) <EOL> self . dtp = self . dtp_handler ( conn , baseclass = self ) <EOL> def cmd_echo ( self , arg ) : <EOL> self . push ( arg ) <EOL> def cmd_noop ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_user ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_pass ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_acct ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_rnfr ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_rnto ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_dele ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_cwd ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_size ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_mkd ( self , arg ) : <EOL> self . push ( '<STR_LIT>' % arg ) <EOL> def cmd_rmd ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_pwd ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_type ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_quit ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> self . close ( ) <EOL> def cmd_abor ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_stor ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_rest ( self , arg ) : <EOL> self . rest = arg <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_retr ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> if self . rest is not None : <EOL> offset = int ( self . rest ) <EOL> else : <EOL> offset = <NUM_LIT:0> <EOL> self . dtp . push ( self . next_retr_data [ offset : ] ) <EOL> self . dtp . close_when_done ( ) <EOL> self . rest = None <EOL> def cmd_list ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> self . dtp . push ( LIST_DATA ) <EOL> self . dtp . close_when_done ( ) <EOL> def cmd_nlst ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> self . dtp . push ( NLST_DATA ) <EOL> self . dtp . close_when_done ( ) <EOL> def cmd_opts ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_mlsd ( self , arg ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> self . dtp . push ( MLSD_DATA ) <EOL> self . dtp . close_when_done ( ) <EOL> def cmd_setlongretr ( self , arg ) : <EOL> self . next_retr_data = '<STR_LIT:x>' * int ( arg ) <EOL> self . push ( '<STR_LIT>' ) <EOL> class DummyFTPServer ( asyncore . dispatcher , threading . Thread ) : <EOL> handler = DummyFTPHandler <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> self . handler_instance = None <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_accepted ( self , conn , addr ) : <EOL> self . handler_instance = self . handler ( conn ) <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> if ssl is not None : <EOL> CERTFILE = os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) <EOL> class SSLConnection ( asyncore . dispatcher ) : <EOL> """<STR_LIT>""" <EOL> _ssl_accepting = False <EOL> _ssl_closing = False <EOL> def secure_connection ( self ) : <EOL> socket = ssl . wrap_socket ( self . socket , suppress_ragged_eofs = False , <EOL> certfile = CERTFILE , server_side = True , <EOL> do_handshake_on_connect = False , <EOL> ssl_version = ssl . PROTOCOL_SSLv23 ) <EOL> self . del_channel ( ) <EOL> self . set_socket ( socket ) <EOL> self . _ssl_accepting = True <EOL> def _do_ssl_handshake ( self ) : <EOL> try : <EOL> self . socket . do_handshake ( ) <EOL> except ssl . SSLError as 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 as err : <EOL> if err . args [ <NUM_LIT:0> ] == errno . ECONNABORTED : <EOL> return self . handle_close ( ) <EOL> else : <EOL> self . _ssl_accepting = False <EOL> def _do_ssl_shutdown ( self ) : <EOL> self . _ssl_closing = True <EOL> try : <EOL> self . socket = self . socket . unwrap ( ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] in ( ssl . SSL_ERROR_WANT_READ , <EOL> ssl . SSL_ERROR_WANT_WRITE ) : <EOL> return <EOL> except socket . error as err : <EOL> pass <EOL> self . _ssl_closing = False <EOL> if getattr ( self , '<STR_LIT>' , False ) is False : <EOL> super ( SSLConnection , self ) . close ( ) <EOL> else : <EOL> pass <EOL> def handle_read_event ( self ) : <EOL> if self . _ssl_accepting : <EOL> self . _do_ssl_handshake ( ) <EOL> elif self . _ssl_closing : <EOL> self . _do_ssl_shutdown ( ) <EOL> else : <EOL> super ( SSLConnection , self ) . handle_read_event ( ) <EOL> def handle_write_event ( self ) : <EOL> if self . _ssl_accepting : <EOL> self . _do_ssl_handshake ( ) <EOL> elif self . _ssl_closing : <EOL> self . _do_ssl_shutdown ( ) <EOL> else : <EOL> super ( SSLConnection , self ) . handle_write_event ( ) <EOL> def send ( self , data ) : <EOL> try : <EOL> return super ( SSLConnection , self ) . send ( data ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] in ( ssl . SSL_ERROR_EOF , ssl . SSL_ERROR_ZERO_RETURN , <EOL> ssl . SSL_ERROR_WANT_READ , <EOL> ssl . SSL_ERROR_WANT_WRITE ) : <EOL> return <NUM_LIT:0> <EOL> raise <EOL> def recv ( self , buffer_size ) : <EOL> try : <EOL> return super ( SSLConnection , self ) . recv ( buffer_size ) <EOL> except ssl . SSLError as err : <EOL> if err . args [ <NUM_LIT:0> ] in ( ssl . SSL_ERROR_WANT_READ , <EOL> ssl . SSL_ERROR_WANT_WRITE ) : <EOL> return b'<STR_LIT>' <EOL> if err . args [ <NUM_LIT:0> ] in ( ssl . SSL_ERROR_EOF , ssl . SSL_ERROR_ZERO_RETURN ) : <EOL> self . handle_close ( ) <EOL> return b'<STR_LIT>' <EOL> raise <EOL> def handle_error ( self ) : <EOL> raise <EOL> def close ( self ) : <EOL> if ( isinstance ( self . socket , ssl . SSLSocket ) and <EOL> self . socket . _sslobj is not None ) : <EOL> self . _do_ssl_shutdown ( ) <EOL> else : <EOL> super ( SSLConnection , self ) . close ( ) <EOL> class DummyTLS_DTPHandler ( SSLConnection , DummyDTPHandler ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , conn , baseclass ) : <EOL> DummyDTPHandler . __init__ ( self , conn , baseclass ) <EOL> if self . baseclass . secure_data_channel : <EOL> self . secure_connection ( ) <EOL> class DummyTLS_FTPHandler ( SSLConnection , DummyFTPHandler ) : <EOL> """<STR_LIT>""" <EOL> dtp_handler = DummyTLS_DTPHandler <EOL> def __init__ ( self , conn ) : <EOL> DummyFTPHandler . __init__ ( self , conn ) <EOL> self . secure_data_channel = False <EOL> self . _ccc = False <EOL> def cmd_auth ( self , line ) : <EOL> """<STR_LIT>""" <EOL> self . push ( '<STR_LIT>' ) <EOL> self . secure_connection ( ) <EOL> def cmd_ccc ( self , line ) : <EOL> self . push ( '<STR_LIT>' ) <EOL> self . _ccc = True <EOL> self . _do_ssl_shutdown ( ) <EOL> def cmd_pbsz ( self , line ) : <EOL> """<STR_LIT>""" <EOL> self . push ( '<STR_LIT>' ) <EOL> def cmd_prot ( self , line ) : <EOL> """<STR_LIT>""" <EOL> arg = line . upper ( ) <EOL> if arg == '<STR_LIT:C>' : <EOL> self . push ( '<STR_LIT>' ) <EOL> self . secure_data_channel = False <EOL> elif arg == '<STR_LIT:P>' : <EOL> self . push ( '<STR_LIT>' ) <EOL> self . secure_data_channel = True <EOL> else : <EOL> self . push ( "<STR_LIT>" ) <EOL> class DummyTLS_FTPServer ( DummyFTPServer ) : <EOL> handler = DummyTLS_FTPHandler <EOL> class TestFTPClass ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . server = DummyFTPServer ( ( HOST , <NUM_LIT:0> ) ) <EOL> self . server . start ( ) <EOL> self . client = ftplib . FTP ( timeout = <NUM_LIT:2> ) <EOL> self . client . connect ( self . server . host , self . server . port ) <EOL> def tearDown ( self ) : <EOL> self . client . close ( ) <EOL> self . server . stop ( ) <EOL> def check_data ( self , received , expected ) : <EOL> self . assertEqual ( len ( received ) , len ( expected ) ) <EOL> self . assertEqual ( received , expected ) <EOL> def test_getwelcome ( self ) : <EOL> self . assertEqual ( self . client . getwelcome ( ) , '<STR_LIT>' ) <EOL> def test_sanitize ( self ) : <EOL> self . assertEqual ( self . client . sanitize ( '<STR_LIT:foo>' ) , repr ( '<STR_LIT:foo>' ) ) <EOL> self . assertEqual ( self . client . sanitize ( '<STR_LIT>' ) , repr ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( self . client . sanitize ( '<STR_LIT>' ) , repr ( '<STR_LIT>' ) ) <EOL> def test_exceptions ( self ) : <EOL> self . assertRaises ( ftplib . error_temp , self . client . sendcmd , '<STR_LIT>' ) <EOL> self . assertRaises ( ftplib . error_temp , self . client . sendcmd , '<STR_LIT>' ) <EOL> self . assertRaises ( ftplib . error_perm , self . client . sendcmd , '<STR_LIT>' ) <EOL> self . assertRaises ( ftplib . error_perm , self . client . sendcmd , '<STR_LIT>' ) <EOL> self . assertRaises ( ftplib . error_proto , self . client . sendcmd , '<STR_LIT>' ) <EOL> def test_all_errors ( self ) : <EOL> exceptions = ( ftplib . error_reply , ftplib . error_temp , ftplib . error_perm , <EOL> ftplib . error_proto , ftplib . Error , IOError , EOFError ) <EOL> for x in exceptions : <EOL> try : <EOL> raise x ( '<STR_LIT>' ) <EOL> except ftplib . all_errors : <EOL> pass <EOL> def test_set_pasv ( self ) : <EOL> self . assertTrue ( self . client . passiveserver ) <EOL> self . client . set_pasv ( True ) <EOL> self . assertTrue ( self . client . passiveserver ) <EOL> self . client . set_pasv ( False ) <EOL> self . assertFalse ( self . client . passiveserver ) <EOL> def test_voidcmd ( self ) : <EOL> self . client . voidcmd ( '<STR_LIT>' ) <EOL> self . client . voidcmd ( '<STR_LIT>' ) <EOL> self . assertRaises ( ftplib . error_reply , self . client . voidcmd , '<STR_LIT>' ) <EOL> self . assertRaises ( ftplib . error_reply , self . client . voidcmd , '<STR_LIT>' ) <EOL> def test_login ( self ) : <EOL> self . client . login ( ) <EOL> def test_acct ( self ) : <EOL> self . client . acct ( '<STR_LIT>' ) <EOL> def test_rename ( self ) : <EOL> self . client . rename ( '<STR_LIT:a>' , '<STR_LIT:b>' ) <EOL> self . server . handler_instance . next_response = '<STR_LIT>' <EOL> self . assertRaises ( ftplib . error_reply , self . client . rename , '<STR_LIT:a>' , '<STR_LIT:b>' ) <EOL> def test_delete ( self ) : <EOL> self . client . delete ( '<STR_LIT:foo>' ) <EOL> self . server . handler_instance . next_response = '<STR_LIT>' <EOL> self . assertRaises ( ftplib . error_reply , self . client . delete , '<STR_LIT:foo>' ) <EOL> def test_size ( self ) : <EOL> self . client . size ( '<STR_LIT:foo>' ) <EOL> def test_mkd ( self ) : <EOL> dir = self . client . mkd ( '<STR_LIT>' ) <EOL> self . assertEqual ( dir , '<STR_LIT>' ) <EOL> def test_rmd ( self ) : <EOL> self . client . rmd ( '<STR_LIT:foo>' ) <EOL> def test_cwd ( self ) : <EOL> dir = self . client . cwd ( '<STR_LIT>' ) <EOL> self . assertEqual ( dir , '<STR_LIT>' ) <EOL> def test_pwd ( self ) : <EOL> dir = self . client . pwd ( ) <EOL> self . assertEqual ( dir , '<STR_LIT>' ) <EOL> def test_quit ( self ) : <EOL> self . assertEqual ( self . client . quit ( ) , '<STR_LIT>' ) <EOL> self . assertEqual ( self . client . sock , None ) <EOL> def test_abort ( self ) : <EOL> self . client . abort ( ) <EOL> def test_retrbinary ( self ) : <EOL> def callback ( data ) : <EOL> received . append ( data . decode ( '<STR_LIT:ascii>' ) ) <EOL> received = [ ] <EOL> self . client . retrbinary ( '<STR_LIT>' , callback ) <EOL> self . check_data ( '<STR_LIT>' . join ( received ) , RETR_DATA ) <EOL> def test_retrbinary_rest ( self ) : <EOL> def callback ( data ) : <EOL> received . append ( data . decode ( '<STR_LIT:ascii>' ) ) <EOL> for rest in ( <NUM_LIT:0> , <NUM_LIT:10> , <NUM_LIT:20> ) : <EOL> received = [ ] <EOL> self . client . retrbinary ( '<STR_LIT>' , callback , rest = rest ) <EOL> self . check_data ( '<STR_LIT>' . join ( received ) , RETR_DATA [ rest : ] ) <EOL> def test_retrlines ( self ) : <EOL> received = [ ] <EOL> self . client . retrlines ( '<STR_LIT>' , received . append ) <EOL> self . check_data ( '<STR_LIT>' . join ( received ) , RETR_DATA . replace ( '<STR_LIT:\r\n>' , '<STR_LIT>' ) ) <EOL> def test_storbinary ( self ) : <EOL> f = io . BytesIO ( RETR_DATA . encode ( '<STR_LIT:ascii>' ) ) <EOL> self . client . storbinary ( '<STR_LIT>' , f ) <EOL> self . check_data ( self . server . handler_instance . last_received_data , RETR_DATA ) <EOL> flag = [ ] <EOL> f . seek ( <NUM_LIT:0> ) <EOL> self . client . storbinary ( '<STR_LIT>' , f , callback = lambda x : flag . append ( None ) ) <EOL> self . assertTrue ( flag ) <EOL> def test_storbinary_rest ( self ) : <EOL> f = io . BytesIO ( RETR_DATA . replace ( '<STR_LIT:\r\n>' , '<STR_LIT:\n>' ) . encode ( '<STR_LIT:ascii>' ) ) <EOL> for r in ( <NUM_LIT:30> , '<STR_LIT>' ) : <EOL> f . seek ( <NUM_LIT:0> ) <EOL> self . client . storbinary ( '<STR_LIT>' , f , rest = r ) <EOL> self . assertEqual ( self . server . handler_instance . rest , str ( r ) ) <EOL> def test_storlines ( self ) : <EOL> f = io . BytesIO ( RETR_DATA . replace ( '<STR_LIT:\r\n>' , '<STR_LIT:\n>' ) . encode ( '<STR_LIT:ascii>' ) ) <EOL> self . client . storlines ( '<STR_LIT>' , f ) <EOL> self . check_data ( self . server . handler_instance . last_received_data , RETR_DATA ) <EOL> flag = [ ] <EOL> f . seek ( <NUM_LIT:0> ) <EOL> self . client . storlines ( '<STR_LIT>' , f , callback = lambda x : flag . append ( None ) ) <EOL> self . assertTrue ( flag ) <EOL> f = io . StringIO ( RETR_DATA . replace ( '<STR_LIT:\r\n>' , '<STR_LIT:\n>' ) ) <EOL> with support . check_warnings ( ( '<STR_LIT>' , BytesWarning ) , quiet = True ) : <EOL> self . assertRaises ( TypeError , self . client . storlines , '<STR_LIT>' , f ) <EOL> def test_nlst ( self ) : <EOL> self . client . nlst ( ) <EOL> self . assertEqual ( self . client . nlst ( ) , NLST_DATA . split ( '<STR_LIT:\r\n>' ) [ : - <NUM_LIT:1> ] ) <EOL> def test_dir ( self ) : <EOL> l = [ ] <EOL> self . client . dir ( lambda x : l . append ( x ) ) <EOL> self . assertEqual ( '<STR_LIT>' . join ( l ) , LIST_DATA . replace ( '<STR_LIT:\r\n>' , '<STR_LIT>' ) ) <EOL> def test_mlsd ( self ) : <EOL> list ( self . client . mlsd ( ) ) <EOL> list ( self . client . mlsd ( path = '<STR_LIT:/>' ) ) <EOL> list ( self . client . mlsd ( path = '<STR_LIT:/>' , facts = [ '<STR_LIT:size>' , '<STR_LIT:type>' ] ) ) <EOL> ls = list ( self . client . mlsd ( ) ) <EOL> for name , facts in ls : <EOL> self . assertIsInstance ( name , str ) <EOL> self . assertIsInstance ( facts , dict ) <EOL> self . assertTrue ( name ) <EOL> self . assertIn ( '<STR_LIT:type>' , facts ) <EOL> self . assertIn ( '<STR_LIT>' , facts ) <EOL> self . assertIn ( '<STR_LIT>' , facts ) <EOL> def set_data ( data ) : <EOL> self . server . handler_instance . next_data = data <EOL> def test_entry ( line , type = None , perm = None , unique = None , name = None ) : <EOL> type = '<STR_LIT:type>' if type is None else type <EOL> perm = '<STR_LIT>' if perm is None else perm <EOL> unique = '<STR_LIT>' if unique is None else unique <EOL> name = '<STR_LIT:name>' if name is None else name <EOL> set_data ( line ) <EOL> _name , facts = next ( self . client . mlsd ( ) ) <EOL> self . assertEqual ( _name , name ) <EOL> self . assertEqual ( facts [ '<STR_LIT:type>' ] , type ) <EOL> self . assertEqual ( facts [ '<STR_LIT>' ] , perm ) <EOL> self . assertEqual ( facts [ '<STR_LIT>' ] , unique ) <EOL> test_entry ( '<STR_LIT>' ) <EOL> test_entry ( '<STR_LIT>' , type = "<STR_LIT>" ) <EOL> test_entry ( '<STR_LIT>' , type = "<STR_LIT>" ) <EOL> test_entry ( '<STR_LIT>' , type = "<STR_LIT>" ) <EOL> test_entry ( '<STR_LIT>' , type = "<STR_LIT>" ) <EOL> test_entry ( '<STR_LIT>' , name = "<STR_LIT>" ) <EOL> test_entry ( '<STR_LIT>' , name = "<STR_LIT>" ) <EOL> test_entry ( '<STR_LIT>' , name = "<STR_LIT>" ) <EOL> test_entry ( '<STR_LIT>' , name = "<STR_LIT>" ) <EOL> test_entry ( '<STR_LIT>' , name = "<STR_LIT>" ) <EOL> test_entry ( '<STR_LIT>' , name = "<STR_LIT>" ) <EOL> test_entry ( '<STR_LIT>' , name = "<STR_LIT>" ) <EOL> test_entry ( '<STR_LIT>' , name = "<STR_LIT>" ) <EOL> set_data ( '<STR_LIT>' ) <EOL> _name , facts = next ( self . client . mlsd ( ) ) <EOL> for x in facts : <EOL> self . assertTrue ( x . islower ( ) ) <EOL> set_data ( '<STR_LIT>' ) <EOL> self . assertRaises ( StopIteration , next , self . client . mlsd ( ) ) <EOL> set_data ( '<STR_LIT>' ) <EOL> for x in self . client . mlsd ( ) : <EOL> self . fail ( "<STR_LIT>" % data ) <EOL> def test_makeport ( self ) : <EOL> with self . client . makeport ( ) : <EOL> self . assertEqual ( self . server . handler_instance . last_received_cmd , <EOL> '<STR_LIT:port>' ) <EOL> def test_makepasv ( self ) : <EOL> host , port = self . client . makepasv ( ) <EOL> conn = socket . create_connection ( ( host , port ) , <NUM_LIT:10> ) <EOL> conn . close ( ) <EOL> self . assertEqual ( self . server . handler_instance . last_received_cmd , '<STR_LIT>' ) <EOL> def test_with_statement ( self ) : <EOL> self . client . quit ( ) <EOL> def is_client_connected ( ) : <EOL> if self . client . sock is None : <EOL> return False <EOL> try : <EOL> self . client . sendcmd ( '<STR_LIT>' ) <EOL> except ( socket . error , EOFError ) : <EOL> return False <EOL> return True <EOL> with ftplib . FTP ( timeout = <NUM_LIT:2> ) as self . client : <EOL> self . client . connect ( self . server . host , self . server . port ) <EOL> self . client . sendcmd ( '<STR_LIT>' ) <EOL> self . assertTrue ( is_client_connected ( ) ) <EOL> self . assertEqual ( self . server . handler_instance . last_received_cmd , '<STR_LIT>' ) <EOL> self . assertFalse ( is_client_connected ( ) ) <EOL> with ftplib . FTP ( timeout = <NUM_LIT:2> ) as self . client : <EOL> self . client . connect ( self . server . host , self . server . port ) <EOL> self . client . sendcmd ( '<STR_LIT>' ) <EOL> self . client . quit ( ) <EOL> self . assertEqual ( self . server . handler_instance . last_received_cmd , '<STR_LIT>' ) <EOL> self . assertFalse ( is_client_connected ( ) ) <EOL> try : <EOL> with ftplib . FTP ( timeout = <NUM_LIT:2> ) as self . client : <EOL> self . client . connect ( self . server . host , self . server . port ) <EOL> self . client . sendcmd ( '<STR_LIT>' ) <EOL> self . server . handler_instance . next_response = '<STR_LIT>' <EOL> except ftplib . error_perm as err : <EOL> self . assertEqual ( str ( err ) , '<STR_LIT>' ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> self . assertEqual ( self . server . handler_instance . last_received_cmd , '<STR_LIT>' ) <EOL> self . assertFalse ( is_client_connected ( ) ) <EOL> def test_source_address ( self ) : <EOL> self . client . quit ( ) <EOL> port = support . find_unused_port ( ) <EOL> try : <EOL> self . client . connect ( self . server . host , self . server . port , <EOL> source_address = ( HOST , port ) ) <EOL> self . assertEqual ( self . client . sock . getsockname ( ) [ <NUM_LIT:1> ] , port ) <EOL> self . client . quit ( ) <EOL> except IOError as e : <EOL> if e . errno == errno . EADDRINUSE : <EOL> self . skipTest ( "<STR_LIT>" % port ) <EOL> raise <EOL> def test_source_address_passive_connection ( self ) : <EOL> port = support . find_unused_port ( ) <EOL> self . client . source_address = ( HOST , port ) <EOL> try : <EOL> with self . client . transfercmd ( '<STR_LIT:list>' ) as sock : <EOL> self . assertEqual ( sock . getsockname ( ) [ <NUM_LIT:1> ] , port ) <EOL> except IOError as e : <EOL> if e . errno == errno . EADDRINUSE : <EOL> self . skipTest ( "<STR_LIT>" % port ) <EOL> raise <EOL> def test_parse257 ( self ) : <EOL> self . assertEqual ( ftplib . parse257 ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( ftplib . parse257 ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( ftplib . parse257 ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( ftplib . parse257 ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertRaises ( ftplib . error_reply , ftplib . parse257 , '<STR_LIT>' ) <EOL> self . assertEqual ( ftplib . parse257 ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( ftplib . parse257 ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> def test_line_too_long ( self ) : <EOL> self . assertRaises ( ftplib . Error , self . client . sendcmd , <EOL> '<STR_LIT:x>' * self . client . maxline * <NUM_LIT:2> ) <EOL> def test_retrlines_too_long ( self ) : <EOL> self . client . sendcmd ( '<STR_LIT>' % ( self . client . maxline * <NUM_LIT:2> ) ) <EOL> received = [ ] <EOL> self . assertRaises ( ftplib . Error , <EOL> self . client . retrlines , '<STR_LIT>' , received . append ) <EOL> def test_storlines_too_long ( self ) : <EOL> f = io . BytesIO ( b'<STR_LIT:x>' * self . client . maxline * <NUM_LIT:2> ) <EOL> self . assertRaises ( ftplib . Error , self . client . storlines , '<STR_LIT>' , f ) <EOL> class TestIPv6Environment ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . server = DummyFTPServer ( ( HOSTv6 , <NUM_LIT:0> ) , af = socket . AF_INET6 ) <EOL> self . server . start ( ) <EOL> self . client = ftplib . FTP ( ) <EOL> self . client . connect ( self . server . host , self . server . port ) <EOL> def tearDown ( self ) : <EOL> self . client . close ( ) <EOL> self . server . stop ( ) <EOL> def test_af ( self ) : <EOL> self . assertEqual ( self . client . af , socket . AF_INET6 ) <EOL> def test_makeport ( self ) : <EOL> with self . client . makeport ( ) : <EOL> self . assertEqual ( self . server . handler_instance . last_received_cmd , <EOL> '<STR_LIT>' ) <EOL> def test_makepasv ( self ) : <EOL> host , port = self . client . makepasv ( ) <EOL> conn = socket . create_connection ( ( host , port ) , <NUM_LIT:10> ) <EOL> conn . close ( ) <EOL> self . assertEqual ( self . server . handler_instance . last_received_cmd , '<STR_LIT>' ) <EOL> def test_transfer ( self ) : <EOL> def retr ( ) : <EOL> def callback ( data ) : <EOL> received . append ( data . decode ( '<STR_LIT:ascii>' ) ) <EOL> received = [ ] <EOL> self . client . retrbinary ( '<STR_LIT>' , callback ) <EOL> self . assertEqual ( len ( '<STR_LIT>' . join ( received ) ) , len ( RETR_DATA ) ) <EOL> self . assertEqual ( '<STR_LIT>' . join ( received ) , RETR_DATA ) <EOL> self . client . set_pasv ( True ) <EOL> retr ( ) <EOL> self . client . set_pasv ( False ) <EOL> retr ( ) <EOL> class TestTLS_FTPClassMixin ( TestFTPClass ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . server = DummyTLS_FTPServer ( ( HOST , <NUM_LIT:0> ) ) <EOL> self . server . start ( ) <EOL> self . client = ftplib . FTP_TLS ( timeout = <NUM_LIT:2> ) <EOL> self . client . connect ( self . server . host , self . server . port ) <EOL> self . client . auth ( ) <EOL> self . client . prot_p ( ) <EOL> class TestTLS_FTPClass ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . server = DummyTLS_FTPServer ( ( HOST , <NUM_LIT:0> ) ) <EOL> self . server . start ( ) <EOL> self . client = ftplib . FTP_TLS ( timeout = <NUM_LIT:2> ) <EOL> self . client . connect ( self . server . host , self . server . port ) <EOL> def tearDown ( self ) : <EOL> self . client . close ( ) <EOL> self . server . stop ( ) <EOL> def test_control_connection ( self ) : <EOL> self . assertNotIsInstance ( self . client . sock , ssl . SSLSocket ) <EOL> self . client . auth ( ) <EOL> self . assertIsInstance ( self . client . sock , ssl . SSLSocket ) <EOL> def test_data_connection ( self ) : <EOL> with self . client . transfercmd ( '<STR_LIT:list>' ) as sock : <EOL> self . assertNotIsInstance ( sock , ssl . SSLSocket ) <EOL> self . assertEqual ( self . client . voidresp ( ) , "<STR_LIT>" ) <EOL> self . client . prot_p ( ) <EOL> with self . client . transfercmd ( '<STR_LIT:list>' ) as sock : <EOL> self . assertIsInstance ( sock , ssl . SSLSocket ) <EOL> self . assertEqual ( self . client . voidresp ( ) , "<STR_LIT>" ) <EOL> self . client . prot_c ( ) <EOL> with self . client . transfercmd ( '<STR_LIT:list>' ) as sock : <EOL> self . assertNotIsInstance ( sock , ssl . SSLSocket ) <EOL> self . assertEqual ( self . client . voidresp ( ) , "<STR_LIT>" ) <EOL> def test_login ( self ) : <EOL> self . assertNotIsInstance ( self . client . sock , ssl . SSLSocket ) <EOL> self . client . login ( ) <EOL> self . assertIsInstance ( self . client . sock , ssl . SSLSocket ) <EOL> self . client . login ( ) <EOL> def test_auth_issued_twice ( self ) : <EOL> self . client . auth ( ) <EOL> self . assertRaises ( ValueError , self . client . auth ) <EOL> def test_auth_ssl ( self ) : <EOL> try : <EOL> self . client . ssl_version = ssl . PROTOCOL_SSLv3 <EOL> self . client . auth ( ) <EOL> self . assertRaises ( ValueError , self . client . auth ) <EOL> finally : <EOL> self . client . ssl_version = ssl . PROTOCOL_TLSv1 <EOL> def test_context ( self ) : <EOL> self . client . quit ( ) <EOL> ctx = ssl . SSLContext ( ssl . PROTOCOL_TLSv1 ) <EOL> self . assertRaises ( ValueError , ftplib . FTP_TLS , keyfile = CERTFILE , <EOL> context = ctx ) <EOL> self . assertRaises ( ValueError , ftplib . FTP_TLS , certfile = CERTFILE , <EOL> context = ctx ) <EOL> self . assertRaises ( ValueError , ftplib . FTP_TLS , certfile = CERTFILE , <EOL> keyfile = CERTFILE , context = ctx ) <EOL> self . client = ftplib . FTP_TLS ( context = ctx , timeout = <NUM_LIT:2> ) <EOL> self . client . connect ( self . server . host , self . server . port ) <EOL> self . assertNotIsInstance ( self . client . sock , ssl . SSLSocket ) <EOL> self . client . auth ( ) <EOL> self . assertIs ( self . client . sock . context , ctx ) <EOL> self . assertIsInstance ( self . client . sock , ssl . SSLSocket ) <EOL> self . client . prot_p ( ) <EOL> with self . client . transfercmd ( '<STR_LIT:list>' ) as sock : <EOL> self . assertIs ( sock . context , ctx ) <EOL> self . assertIsInstance ( sock , ssl . SSLSocket ) <EOL> def test_ccc ( self ) : <EOL> self . assertRaises ( ValueError , self . client . ccc ) <EOL> self . client . login ( secure = True ) <EOL> self . assertIsInstance ( self . client . sock , ssl . SSLSocket ) <EOL> self . client . ccc ( ) <EOL> self . assertRaises ( ValueError , self . client . sock . unwrap ) <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:20> ) <EOL> self . port = support . bind_port ( self . sock ) <EOL> self . server_thread = threading . Thread ( target = self . server ) <EOL> self . server_thread . start ( ) <EOL> self . evt . wait ( ) <EOL> self . evt . clear ( ) <EOL> self . old_port = ftplib . FTP . port <EOL> ftplib . FTP . port = self . port <EOL> def tearDown ( self ) : <EOL> ftplib . FTP . port = self . old_port <EOL> self . server_thread . join ( ) <EOL> def server ( self ) : <EOL> self . sock . listen ( <NUM_LIT:5> ) <EOL> self . evt . set ( ) <EOL> try : <EOL> conn , addr = self . sock . accept ( ) <EOL> except socket . timeout : <EOL> pass <EOL> else : <EOL> conn . sendall ( b"<STR_LIT>" ) <EOL> conn . shutdown ( socket . SHUT_WR ) <EOL> self . evt . set ( ) <EOL> conn . close ( ) <EOL> finally : <EOL> self . sock . close ( ) <EOL> def testTimeoutDefault ( self ) : <EOL> self . assertTrue ( socket . getdefaulttimeout ( ) is None ) <EOL> socket . setdefaulttimeout ( <NUM_LIT:30> ) <EOL> try : <EOL> ftp = ftplib . FTP ( HOST ) <EOL> finally : <EOL> socket . setdefaulttimeout ( None ) <EOL> self . assertEqual ( ftp . sock . gettimeout ( ) , <NUM_LIT:30> ) <EOL> self . evt . wait ( ) <EOL> ftp . close ( ) <EOL> def testTimeoutNone ( self ) : <EOL> self . assertTrue ( socket . getdefaulttimeout ( ) is None ) <EOL> socket . setdefaulttimeout ( <NUM_LIT:30> ) <EOL> try : <EOL> ftp = ftplib . FTP ( HOST , timeout = None ) <EOL> finally : <EOL> socket . setdefaulttimeout ( None ) <EOL> self . assertTrue ( ftp . sock . gettimeout ( ) is None ) <EOL> self . evt . wait ( ) <EOL> ftp . close ( ) <EOL> def testTimeoutValue ( self ) : <EOL> ftp = ftplib . FTP ( HOST , timeout = <NUM_LIT:30> ) <EOL> self . assertEqual ( ftp . sock . gettimeout ( ) , <NUM_LIT:30> ) <EOL> self . evt . wait ( ) <EOL> ftp . close ( ) <EOL> def testTimeoutConnect ( self ) : <EOL> ftp = ftplib . FTP ( ) <EOL> ftp . connect ( HOST , timeout = <NUM_LIT:30> ) <EOL> self . assertEqual ( ftp . sock . gettimeout ( ) , <NUM_LIT:30> ) <EOL> self . evt . wait ( ) <EOL> ftp . close ( ) <EOL> def testTimeoutDifferentOrder ( self ) : <EOL> ftp = ftplib . FTP ( timeout = <NUM_LIT:30> ) <EOL> ftp . connect ( HOST ) <EOL> self . assertEqual ( ftp . sock . gettimeout ( ) , <NUM_LIT:30> ) <EOL> self . evt . wait ( ) <EOL> ftp . close ( ) <EOL> def testTimeoutDirectAccess ( self ) : <EOL> ftp = ftplib . FTP ( ) <EOL> ftp . timeout = <NUM_LIT:30> <EOL> ftp . connect ( HOST ) <EOL> self . assertEqual ( ftp . sock . gettimeout ( ) , <NUM_LIT:30> ) <EOL> self . evt . wait ( ) <EOL> ftp . close ( ) <EOL> def test_main ( ) : <EOL> tests = [ TestFTPClass , TestTimeouts ] <EOL> if support . IPV6_ENABLED : <EOL> tests . append ( TestIPv6Environment ) <EOL> if ssl is not None : <EOL> tests . extend ( [ TestTLS_FTPClassMixin , TestTLS_FTPClass ] ) <EOL> thread_info = support . threading_setup ( ) <EOL> try : <EOL> support . run_unittest ( * tests ) <EOL> finally : <EOL> support . threading_cleanup ( * thread_info ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> test_main ( ) </s>
