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