source: python/trunk/Lib/test/test_ftplib.py

Last change on this file was 391, checked in by dmik, 11 years ago

python: Merge vendor 2.7.6 to trunk.
  • Property svn:eol-style set to native
File size: 25.4 KB
Line 
1"""Test script for ftplib module."""
2
3# Modified by Giampaolo Rodola' to test FTP class, IPv6 and TLS
4# environment
5
6import ftplib
7import asyncore
8import asynchat
9import socket
10import StringIO
11import errno
12import os
13try:
14 import ssl
15except ImportError:
16 ssl = None
17
18from unittest import TestCase
19from test import test_support
20from test.test_support import HOST, HOSTv6
21threading = test_support.import_module('threading')
22
23
24# the dummy data returned by server over the data channel when
25# RETR, LIST and NLST commands are issued
26RETR_DATA = 'abcde12345\r\n' * 1000
27LIST_DATA = 'foo\r\nbar\r\n'
28NLST_DATA = 'foo\r\nbar\r\n'
29
30
31class DummyDTPHandler(asynchat.async_chat):
32 dtp_conn_closed = False
33
34 def __init__(self, conn, baseclass):
35 asynchat.async_chat.__init__(self, conn)
36 self.baseclass = baseclass
37 self.baseclass.last_received_data = ''
38
39 def handle_read(self):
40 self.baseclass.last_received_data += self.recv(1024)
41
42 def handle_close(self):
43 # XXX: this method can be called many times in a row for a single
44 # connection, including in clear-text (non-TLS) mode.
45 # (behaviour witnessed with test_data_connection)
46 if not self.dtp_conn_closed:
47 self.baseclass.push('226 transfer complete')
48 self.close()
49 self.dtp_conn_closed = True
50
51 def handle_error(self):
52 raise
53
54
55class DummyFTPHandler(asynchat.async_chat):
56
57 dtp_handler = DummyDTPHandler
58
59 def __init__(self, conn):
60 asynchat.async_chat.__init__(self, conn)
61 self.set_terminator("\r\n")
62 self.in_buffer = []
63 self.dtp = None
64 self.last_received_cmd = None
65 self.last_received_data = ''
66 self.next_response = ''
67 self.rest = None
68 self.next_retr_data = RETR_DATA
69 self.push('220 welcome')
70
71 def collect_incoming_data(self, data):
72 self.in_buffer.append(data)
73
74 def found_terminator(self):
75 line = ''.join(self.in_buffer)
76 self.in_buffer = []
77 if self.next_response:
78 self.push(self.next_response)
79 self.next_response = ''
80 cmd = line.split(' ')[0].lower()
81 self.last_received_cmd = cmd
82 space = line.find(' ')
83 if space != -1:
84 arg = line[space + 1:]
85 else:
86 arg = ""
87 if hasattr(self, 'cmd_' + cmd):
88 method = getattr(self, 'cmd_' + cmd)
89 method(arg)
90 else:
91 self.push('550 command "%s" not understood.' %cmd)
92
93 def handle_error(self):
94 raise
95
96 def push(self, data):
97 asynchat.async_chat.push(self, data + '\r\n')
98
99 def cmd_port(self, arg):
100 addr = map(int, arg.split(','))
101 ip = '%d.%d.%d.%d' %tuple(addr[:4])
102 port = (addr[4] * 256) + addr[5]
103 s = socket.create_connection((ip, port), timeout=10)
104 self.dtp = self.dtp_handler(s, baseclass=self)
105 self.push('200 active data connection established')
106
107 def cmd_pasv(self, arg):
108 sock = socket.socket()
109 sock.bind((self.socket.getsockname()[0], 0))
110 sock.listen(5)
111 sock.settimeout(10)
112 ip, port = sock.getsockname()[:2]
113 ip = ip.replace('.', ',')
114 p1, p2 = divmod(port, 256)
115 self.push('227 entering passive mode (%s,%d,%d)' %(ip, p1, p2))
116 conn, addr = sock.accept()
117 self.dtp = self.dtp_handler(conn, baseclass=self)
118
119 def cmd_eprt(self, arg):
120 af, ip, port = arg.split(arg[0])[1:-1]
121 port = int(port)
122 s = socket.create_connection((ip, port), timeout=10)
123 self.dtp = self.dtp_handler(s, baseclass=self)
124 self.push('200 active data connection established')
125
126 def cmd_epsv(self, arg):
127 sock = socket.socket(socket.AF_INET6)
128 sock.bind((self.socket.getsockname()[0], 0))
129 sock.listen(5)
130 sock.settimeout(10)
131 port = sock.getsockname()[1]
132 self.push('229 entering extended passive mode (|||%d|)' %port)
133 conn, addr = sock.accept()
134 self.dtp = self.dtp_handler(conn, baseclass=self)
135
136 def cmd_echo(self, arg):
137 # sends back the received string (used by the test suite)
138 self.push(arg)
139
140 def cmd_user(self, arg):
141 self.push('331 username ok')
142
143 def cmd_pass(self, arg):
144 self.push('230 password ok')
145
146 def cmd_acct(self, arg):
147 self.push('230 acct ok')
148
149 def cmd_rnfr(self, arg):
150 self.push('350 rnfr ok')
151
152 def cmd_rnto(self, arg):
153 self.push('250 rnto ok')
154
155 def cmd_dele(self, arg):
156 self.push('250 dele ok')
157
158 def cmd_cwd(self, arg):
159 self.push('250 cwd ok')
160
161 def cmd_size(self, arg):
162 self.push('250 1000')
163
164 def cmd_mkd(self, arg):
165 self.push('257 "%s"' %arg)
166
167 def cmd_rmd(self, arg):
168 self.push('250 rmd ok')
169
170 def cmd_pwd(self, arg):
171 self.push('257 "pwd ok"')
172
173 def cmd_type(self, arg):
174 self.push('200 type ok')
175
176 def cmd_quit(self, arg):
177 self.push('221 quit ok')
178 self.close()
179
180 def cmd_stor(self, arg):
181 self.push('125 stor ok')
182
183 def cmd_rest(self, arg):
184 self.rest = arg
185 self.push('350 rest ok')
186
187 def cmd_retr(self, arg):
188 self.push('125 retr ok')
189 if self.rest is not None:
190 offset = int(self.rest)
191 else:
192 offset = 0
193 self.dtp.push(self.next_retr_data[offset:])
194 self.dtp.close_when_done()
195 self.rest = None
196
197 def cmd_list(self, arg):
198 self.push('125 list ok')
199 self.dtp.push(LIST_DATA)
200 self.dtp.close_when_done()
201
202 def cmd_nlst(self, arg):
203 self.push('125 nlst ok')
204 self.dtp.push(NLST_DATA)
205 self.dtp.close_when_done()
206
207 def cmd_setlongretr(self, arg):
208 # For testing. Next RETR will return long line.
209 self.next_retr_data = 'x' * int(arg)
210 self.push('125 setlongretr ok')
211
212
213class DummyFTPServer(asyncore.dispatcher, threading.Thread):
214
215 handler = DummyFTPHandler
216
217 def __init__(self, address, af=socket.AF_INET):
218 threading.Thread.__init__(self)
219 asyncore.dispatcher.__init__(self)
220 self.create_socket(af, socket.SOCK_STREAM)
221 self.bind(address)
222 self.listen(5)
223 self.active = False
224 self.active_lock = threading.Lock()
225 self.host, self.port = self.socket.getsockname()[:2]
226
227 def start(self):
228 assert not self.active
229 self.__flag = threading.Event()
230 threading.Thread.start(self)
231 self.__flag.wait()
232
233 def run(self):
234 self.active = True
235 self.__flag.set()
236 while self.active and asyncore.socket_map:
237 self.active_lock.acquire()
238 asyncore.loop(timeout=0.1, count=1)
239 self.active_lock.release()
240 asyncore.close_all(ignore_all=True)
241
242 def stop(self):
243 assert self.active
244 self.active = False
245 self.join()
246
247 def handle_accept(self):
248 conn, addr = self.accept()
249 self.handler = self.handler(conn)
250 self.close()
251
252 def handle_connect(self):
253 self.close()
254 handle_read = handle_connect
255
256 def writable(self):
257 return 0
258
259 def handle_error(self):
260 raise
261
262
263if ssl is not None:
264
265 CERTFILE = os.path.join(os.path.dirname(__file__), "keycert.pem")
266
267 class SSLConnection(object, asyncore.dispatcher):
268 """An asyncore.dispatcher subclass supporting TLS/SSL."""
269
270 _ssl_accepting = False
271 _ssl_closing = False
272
273 def secure_connection(self):
274 self.socket = ssl.wrap_socket(self.socket, suppress_ragged_eofs=False,
275 certfile=CERTFILE, server_side=True,
276 do_handshake_on_connect=False,
277 ssl_version=ssl.PROTOCOL_SSLv23)
278 self._ssl_accepting = True
279
280 def _do_ssl_handshake(self):
281 try:
282 self.socket.do_handshake()
283 except ssl.SSLError, err:
284 if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
285 ssl.SSL_ERROR_WANT_WRITE):
286 return
287 elif err.args[0] == ssl.SSL_ERROR_EOF:
288 return self.handle_close()
289 raise
290 except socket.error, err:
291 if err.args[0] == errno.ECONNABORTED:
292 return self.handle_close()
293 else:
294 self._ssl_accepting = False
295
296 def _do_ssl_shutdown(self):
297 self._ssl_closing = True
298 try:
299 self.socket = self.socket.unwrap()
300 except ssl.SSLError, err:
301 if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
302 ssl.SSL_ERROR_WANT_WRITE):
303 return
304 except socket.error, err:
305 # Any "socket error" corresponds to a SSL_ERROR_SYSCALL return
306 # from OpenSSL's SSL_shutdown(), corresponding to a
307 # closed socket condition. See also:
308 # http://www.mail-archive.com/openssl-users@openssl.org/msg60710.html
309 pass
310 self._ssl_closing = False
311 super(SSLConnection, self).close()
312
313 def handle_read_event(self):
314 if self._ssl_accepting:
315 self._do_ssl_handshake()
316 elif self._ssl_closing:
317 self._do_ssl_shutdown()
318 else:
319 super(SSLConnection, self).handle_read_event()
320
321 def handle_write_event(self):
322 if self._ssl_accepting:
323 self._do_ssl_handshake()
324 elif self._ssl_closing:
325 self._do_ssl_shutdown()
326 else:
327 super(SSLConnection, self).handle_write_event()
328
329 def send(self, data):
330 try:
331 return super(SSLConnection, self).send(data)
332 except ssl.SSLError, err:
333 if err.args[0] in (ssl.SSL_ERROR_EOF, ssl.SSL_ERROR_ZERO_RETURN,
334 ssl.SSL_ERROR_WANT_READ,
335 ssl.SSL_ERROR_WANT_WRITE):
336 return 0
337 raise
338
339 def recv(self, buffer_size):
340 try:
341 return super(SSLConnection, self).recv(buffer_size)
342 except ssl.SSLError, err:
343 if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
344 ssl.SSL_ERROR_WANT_WRITE):
345 return ''
346 if err.args[0] in (ssl.SSL_ERROR_EOF, ssl.SSL_ERROR_ZERO_RETURN):
347 self.handle_close()
348 return ''
349 raise
350
351 def handle_error(self):
352 raise
353
354 def close(self):
355 if (isinstance(self.socket, ssl.SSLSocket) and
356 self.socket._sslobj is not None):
357 self._do_ssl_shutdown()
358
359
360 class DummyTLS_DTPHandler(SSLConnection, DummyDTPHandler):
361 """A DummyDTPHandler subclass supporting TLS/SSL."""
362
363 def __init__(self, conn, baseclass):
364 DummyDTPHandler.__init__(self, conn, baseclass)
365 if self.baseclass.secure_data_channel:
366 self.secure_connection()
367
368
369 class DummyTLS_FTPHandler(SSLConnection, DummyFTPHandler):
370 """A DummyFTPHandler subclass supporting TLS/SSL."""
371
372 dtp_handler = DummyTLS_DTPHandler
373
374 def __init__(self, conn):
375 DummyFTPHandler.__init__(self, conn)
376 self.secure_data_channel = False
377
378 def cmd_auth(self, line):
379 """Set up secure control channel."""
380 self.push('234 AUTH TLS successful')
381 self.secure_connection()
382
383 def cmd_pbsz(self, line):
384 """Negotiate size of buffer for secure data transfer.
385 For TLS/SSL the only valid value for the parameter is '0'.
386 Any other value is accepted but ignored.
387 """
388 self.push('200 PBSZ=0 successful.')
389
390 def cmd_prot(self, line):
391 """Setup un/secure data channel."""
392 arg = line.upper()
393 if arg == 'C':
394 self.push('200 Protection set to Clear')
395 self.secure_data_channel = False
396 elif arg == 'P':
397 self.push('200 Protection set to Private')
398 self.secure_data_channel = True
399 else:
400 self.push("502 Unrecognized PROT type (use C or P).")
401
402
403 class DummyTLS_FTPServer(DummyFTPServer):
404 handler = DummyTLS_FTPHandler
405
406
407class TestFTPClass(TestCase):
408
409 def setUp(self):
410 self.server = DummyFTPServer((HOST, 0))
411 self.server.start()
412 self.client = ftplib.FTP(timeout=10)
413 self.client.connect(self.server.host, self.server.port)
414
415 def tearDown(self):
416 self.client.close()
417 self.server.stop()
418
419 def test_getwelcome(self):
420 self.assertEqual(self.client.getwelcome(), '220 welcome')
421
422 def test_sanitize(self):
423 self.assertEqual(self.client.sanitize('foo'), repr('foo'))
424 self.assertEqual(self.client.sanitize('pass 12345'), repr('pass *****'))
425 self.assertEqual(self.client.sanitize('PASS 12345'), repr('PASS *****'))
426
427 def test_exceptions(self):
428 self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400')
429 self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499')
430 self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500')
431 self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599')
432 self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999')
433
434 def test_all_errors(self):
435 exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm,
436 ftplib.error_proto, ftplib.Error, IOError, EOFError)
437 for x in exceptions:
438 try:
439 raise x('exception not included in all_errors set')
440 except ftplib.all_errors:
441 pass
442
443 def test_set_pasv(self):
444 # passive mode is supposed to be enabled by default
445 self.assertTrue(self.client.passiveserver)
446 self.client.set_pasv(True)
447 self.assertTrue(self.client.passiveserver)
448 self.client.set_pasv(False)
449 self.assertFalse(self.client.passiveserver)
450
451 def test_voidcmd(self):
452 self.client.voidcmd('echo 200')
453 self.client.voidcmd('echo 299')
454 self.assertRaises(ftplib.error_reply, self.client.voidcmd, 'echo 199')
455 self.assertRaises(ftplib.error_reply, self.client.voidcmd, 'echo 300')
456
457 def test_login(self):
458 self.client.login()
459
460 def test_acct(self):
461 self.client.acct('passwd')
462
463 def test_rename(self):
464 self.client.rename('a', 'b')
465 self.server.handler.next_response = '200'
466 self.assertRaises(ftplib.error_reply, self.client.rename, 'a', 'b')
467
468 def test_delete(self):
469 self.client.delete('foo')
470 self.server.handler.next_response = '199'
471 self.assertRaises(ftplib.error_reply, self.client.delete, 'foo')
472
473 def test_size(self):
474 self.client.size('foo')
475
476 def test_mkd(self):
477 dir = self.client.mkd('/foo')
478 self.assertEqual(dir, '/foo')
479
480 def test_rmd(self):
481 self.client.rmd('foo')
482
483 def test_cwd(self):
484 dir = self.client.cwd('/foo')
485 self.assertEqual(dir, '250 cwd ok')
486
487 def test_mkd(self):
488 dir = self.client.mkd('/foo')
489 self.assertEqual(dir, '/foo')
490
491 def test_pwd(self):
492 dir = self.client.pwd()
493 self.assertEqual(dir, 'pwd ok')
494
495 def test_quit(self):
496 self.assertEqual(self.client.quit(), '221 quit ok')
497 # Ensure the connection gets closed; sock attribute should be None
498 self.assertEqual(self.client.sock, None)
499
500 def test_retrbinary(self):
501 received = []
502 self.client.retrbinary('retr', received.append)
503 self.assertEqual(''.join(received), RETR_DATA)
504
505 def test_retrbinary_rest(self):
506 for rest in (0, 10, 20):
507 received = []
508 self.client.retrbinary('retr', received.append, rest=rest)
509 self.assertEqual(''.join(received), RETR_DATA[rest:],
510 msg='rest test case %d %d %d' % (rest,
511 len(''.join(received)),
512 len(RETR_DATA[rest:])))
513
514 def test_retrlines(self):
515 received = []
516 self.client.retrlines('retr', received.append)
517 self.assertEqual(''.join(received), RETR_DATA.replace('\r\n', ''))
518
519 def test_storbinary(self):
520 f = StringIO.StringIO(RETR_DATA)
521 self.client.storbinary('stor', f)
522 self.assertEqual(self.server.handler.last_received_data, RETR_DATA)
523 # test new callback arg
524 flag = []
525 f.seek(0)
526 self.client.storbinary('stor', f, callback=lambda x: flag.append(None))
527 self.assertTrue(flag)
528
529 def test_storbinary_rest(self):
530 f = StringIO.StringIO(RETR_DATA)
531 for r in (30, '30'):
532 f.seek(0)
533 self.client.storbinary('stor', f, rest=r)
534 self.assertEqual(self.server.handler.rest, str(r))
535
536 def test_storlines(self):
537 f = StringIO.StringIO(RETR_DATA.replace('\r\n', '\n'))
538 self.client.storlines('stor', f)
539 self.assertEqual(self.server.handler.last_received_data, RETR_DATA)
540 # test new callback arg
541 flag = []
542 f.seek(0)
543 self.client.storlines('stor foo', f, callback=lambda x: flag.append(None))
544 self.assertTrue(flag)
545
546 def test_nlst(self):
547 self.client.nlst()
548 self.assertEqual(self.client.nlst(), NLST_DATA.split('\r\n')[:-1])
549
550 def test_dir(self):
551 l = []
552 self.client.dir(lambda x: l.append(x))
553 self.assertEqual(''.join(l), LIST_DATA.replace('\r\n', ''))
554
555 def test_makeport(self):
556 self.client.makeport()
557 # IPv4 is in use, just make sure send_eprt has not been used
558 self.assertEqual(self.server.handler.last_received_cmd, 'port')
559
560 def test_makepasv(self):
561 host, port = self.client.makepasv()
562 conn = socket.create_connection((host, port), 10)
563 conn.close()
564 # IPv4 is in use, just make sure send_epsv has not been used
565 self.assertEqual(self.server.handler.last_received_cmd, 'pasv')
566
567 def test_line_too_long(self):
568 self.assertRaises(ftplib.Error, self.client.sendcmd,
569 'x' * self.client.maxline * 2)
570
571 def test_retrlines_too_long(self):
572 self.client.sendcmd('SETLONGRETR %d' % (self.client.maxline * 2))
573 received = []
574 self.assertRaises(ftplib.Error,
575 self.client.retrlines, 'retr', received.append)
576
577 def test_storlines_too_long(self):
578 f = StringIO.StringIO('x' * self.client.maxline * 2)
579 self.assertRaises(ftplib.Error, self.client.storlines, 'stor', f)
580
581
582class TestIPv6Environment(TestCase):
583
584 def setUp(self):
585 self.server = DummyFTPServer((HOSTv6, 0), af=socket.AF_INET6)
586 self.server.start()
587 self.client = ftplib.FTP()
588 self.client.connect(self.server.host, self.server.port)
589
590 def tearDown(self):
591 self.client.close()
592 self.server.stop()
593
594 def test_af(self):
595 self.assertEqual(self.client.af, socket.AF_INET6)
596
597 def test_makeport(self):
598 self.client.makeport()
599 self.assertEqual(self.server.handler.last_received_cmd, 'eprt')
600
601 def test_makepasv(self):
602 host, port = self.client.makepasv()
603 conn = socket.create_connection((host, port), 10)
604 conn.close()
605 self.assertEqual(self.server.handler.last_received_cmd, 'epsv')
606
607 def test_transfer(self):
608 def retr():
609 received = []
610 self.client.retrbinary('retr', received.append)
611 self.assertEqual(''.join(received), RETR_DATA)
612 self.client.set_pasv(True)
613 retr()
614 self.client.set_pasv(False)
615 retr()
616
617
618class TestTLS_FTPClassMixin(TestFTPClass):
619 """Repeat TestFTPClass tests starting the TLS layer for both control
620 and data connections first.
621 """
622
623 def setUp(self):
624 self.server = DummyTLS_FTPServer((HOST, 0))
625 self.server.start()
626 self.client = ftplib.FTP_TLS(timeout=10)
627 self.client.connect(self.server.host, self.server.port)
628 # enable TLS
629 self.client.auth()
630 self.client.prot_p()
631
632
633class TestTLS_FTPClass(TestCase):
634 """Specific TLS_FTP class tests."""
635
636 def setUp(self):
637 self.server = DummyTLS_FTPServer((HOST, 0))
638 self.server.start()
639 self.client = ftplib.FTP_TLS(timeout=10)
640 self.client.connect(self.server.host, self.server.port)
641
642 def tearDown(self):
643 self.client.close()
644 self.server.stop()
645
646 def test_control_connection(self):
647 self.assertNotIsInstance(self.client.sock, ssl.SSLSocket)
648 self.client.auth()
649 self.assertIsInstance(self.client.sock, ssl.SSLSocket)
650
651 def test_data_connection(self):
652 # clear text
653 sock = self.client.transfercmd('list')
654 self.assertNotIsInstance(sock, ssl.SSLSocket)
655 sock.close()
656 self.assertEqual(self.client.voidresp(), "226 transfer complete")
657
658 # secured, after PROT P
659 self.client.prot_p()
660 sock = self.client.transfercmd('list')
661 self.assertIsInstance(sock, ssl.SSLSocket)
662 sock.close()
663 self.assertEqual(self.client.voidresp(), "226 transfer complete")
664
665 # PROT C is issued, the connection must be in cleartext again
666 self.client.prot_c()
667 sock = self.client.transfercmd('list')
668 self.assertNotIsInstance(sock, ssl.SSLSocket)
669 sock.close()
670 self.assertEqual(self.client.voidresp(), "226 transfer complete")
671
672 def test_login(self):
673 # login() is supposed to implicitly secure the control connection
674 self.assertNotIsInstance(self.client.sock, ssl.SSLSocket)
675 self.client.login()
676 self.assertIsInstance(self.client.sock, ssl.SSLSocket)
677 # make sure that AUTH TLS doesn't get issued again
678 self.client.login()
679
680 def test_auth_issued_twice(self):
681 self.client.auth()
682 self.assertRaises(ValueError, self.client.auth)
683
684 def test_auth_ssl(self):
685 try:
686 self.client.ssl_version = ssl.PROTOCOL_SSLv3
687 self.client.auth()
688 self.assertRaises(ValueError, self.client.auth)
689 finally:
690 self.client.ssl_version = ssl.PROTOCOL_TLSv1
691
692
693class TestTimeouts(TestCase):
694
695 def setUp(self):
696 self.evt = threading.Event()
697 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
698 self.sock.settimeout(10)
699 self.port = test_support.bind_port(self.sock)
700 threading.Thread(target=self.server, args=(self.evt,self.sock)).start()
701 # Wait for the server to be ready.
702 self.evt.wait()
703 self.evt.clear()
704 ftplib.FTP.port = self.port
705
706 def tearDown(self):
707 self.evt.wait()
708
709 def server(self, evt, serv):
710 # This method sets the evt 3 times:
711 # 1) when the connection is ready to be accepted.
712 # 2) when it is safe for the caller to close the connection
713 # 3) when we have closed the socket
714 serv.listen(5)
715 # (1) Signal the caller that we are ready to accept the connection.
716 evt.set()
717 try:
718 conn, addr = serv.accept()
719 except socket.timeout:
720 pass
721 else:
722 conn.send("1 Hola mundo\n")
723 # (2) Signal the caller that it is safe to close the socket.
724 evt.set()
725 conn.close()
726 finally:
727 serv.close()
728 # (3) Signal the caller that we are done.
729 evt.set()
730
731 def testTimeoutDefault(self):
732 # default -- use global socket timeout
733 self.assertTrue(socket.getdefaulttimeout() is None)
734 socket.setdefaulttimeout(30)
735 try:
736 ftp = ftplib.FTP(HOST)
737 finally:
738 socket.setdefaulttimeout(None)
739 self.assertEqual(ftp.sock.gettimeout(), 30)
740 self.evt.wait()
741 ftp.close()
742
743 def testTimeoutNone(self):
744 # no timeout -- do not use global socket timeout
745 self.assertTrue(socket.getdefaulttimeout() is None)
746 socket.setdefaulttimeout(30)
747 try:
748 ftp = ftplib.FTP(HOST, timeout=None)
749 finally:
750 socket.setdefaulttimeout(None)
751 self.assertTrue(ftp.sock.gettimeout() is None)
752 self.evt.wait()
753 ftp.close()
754
755 def testTimeoutValue(self):
756 # a value
757 ftp = ftplib.FTP(HOST, timeout=30)
758 self.assertEqual(ftp.sock.gettimeout(), 30)
759 self.evt.wait()
760 ftp.close()
761
762 def testTimeoutConnect(self):
763 ftp = ftplib.FTP()
764 ftp.connect(HOST, timeout=30)
765 self.assertEqual(ftp.sock.gettimeout(), 30)
766 self.evt.wait()
767 ftp.close()
768
769 def testTimeoutDifferentOrder(self):
770 ftp = ftplib.FTP(timeout=30)
771 ftp.connect(HOST)
772 self.assertEqual(ftp.sock.gettimeout(), 30)
773 self.evt.wait()
774 ftp.close()
775
776 def testTimeoutDirectAccess(self):
777 ftp = ftplib.FTP()
778 ftp.timeout = 30
779 ftp.connect(HOST)
780 self.assertEqual(ftp.sock.gettimeout(), 30)
781 self.evt.wait()
782 ftp.close()
783
784
785def test_main():
786 tests = [TestFTPClass, TestTimeouts]
787 if socket.has_ipv6:
788 try:
789 DummyFTPServer((HOST, 0), af=socket.AF_INET6)
790 except socket.error:
791 pass
792 else:
793 tests.append(TestIPv6Environment)
794
795 if ssl is not None:
796 tests.extend([TestTLS_FTPClassMixin, TestTLS_FTPClass])
797
798 thread_info = test_support.threading_setup()
799 try:
800 test_support.run_unittest(*tests)
801 finally:
802 test_support.threading_cleanup(*thread_info)
803
804
805if __name__ == '__main__':
806 test_main()
Note: See TracBrowser for help on using the repository browser.