|
21 | 21 | import html
|
22 | 22 | import http, http.client
|
23 | 23 | import urllib.parse
|
24 |
| -import urllib.request |
25 | 24 | import tempfile
|
26 | 25 | import time
|
27 | 26 | import datetime
|
|
34 | 33 | from test.support import (
|
35 | 34 | is_apple, import_helper, os_helper, threading_helper
|
36 | 35 | )
|
37 |
| -from test.support.script_helper import kill_python, spawn_python |
38 |
| -from test.support.socket_helper import find_unused_port |
39 | 36 |
|
40 | 37 | try:
|
41 | 38 | import ssl
|
@@ -1455,86 +1452,6 @@ def test_unknown_flag(self, _):
|
1455 | 1452 | self.assertIn('error', stderr.getvalue())
|
1456 | 1453 |
|
1457 | 1454 |
|
1458 |
| -class CommandLineRunTimeTestCase(unittest.TestCase): |
1459 |
| - served_data = os.urandom(32) |
1460 |
| - served_file_name = 'served_filename' |
1461 |
| - tls_cert = certdata_file('ssl_cert.pem') |
1462 |
| - tls_key = certdata_file('ssl_key.pem') |
1463 |
| - tls_password = 'somepass' |
1464 |
| - |
1465 |
| - def setUp(self): |
1466 |
| - super().setUp() |
1467 |
| - with open(self.served_file_name, 'wb') as f: |
1468 |
| - f.write(self.served_data) |
1469 |
| - self.addCleanup(os_helper.unlink, self.served_file_name) |
1470 |
| - self.tls_password_file = tempfile.mktemp() |
1471 |
| - with open(self.tls_password_file, 'wb') as f: |
1472 |
| - f.write(self.tls_password.encode()) |
1473 |
| - self.addCleanup(os_helper.unlink, self.tls_password_file) |
1474 |
| - |
1475 |
| - def fetch_file(self, path): |
1476 |
| - context = ssl.create_default_context() |
1477 |
| - # allow self-signed certificates |
1478 |
| - context.check_hostname = False |
1479 |
| - context.verify_mode = ssl.CERT_NONE |
1480 |
| - req = urllib.request.Request(path, method='GET') |
1481 |
| - with urllib.request.urlopen(req, context=context) as res: |
1482 |
| - return res.read() |
1483 |
| - |
1484 |
| - def parse_cli_output(self, output): |
1485 |
| - matches = re.search(r'\((https?)://([^/:]+):(\d+)/?\)', output) |
1486 |
| - if matches is None: |
1487 |
| - return None, None, None |
1488 |
| - return matches.group(1), matches.group(2), int(matches.group(3)) |
1489 |
| - |
1490 |
| - def wait_for_server(self, proc, protocol, port, bind, timeout=50): |
1491 |
| - """Check the server process output. |
1492 |
| -
|
1493 |
| - Return True if the server was successfully started |
1494 |
| - and is listening on the given port and bind address. |
1495 |
| - """ |
1496 |
| - while timeout > 0: |
1497 |
| - line = proc.stdout.readline() |
1498 |
| - if not line: |
1499 |
| - time.sleep(0.1) |
1500 |
| - timeout -= 1 |
1501 |
| - continue |
1502 |
| - protocol_, host_, port_ = self.parse_cli_output(line) |
1503 |
| - if not protocol_ or not host_ or not port_: |
1504 |
| - time.sleep(0.1) |
1505 |
| - timeout -= 1 |
1506 |
| - continue |
1507 |
| - if protocol_ == protocol and host_ == bind and port_ == port: |
1508 |
| - return True |
1509 |
| - break |
1510 |
| - return False |
1511 |
| - |
1512 |
| - def test_http_client(self): |
1513 |
| - port = find_unused_port() |
1514 |
| - bind = '127.0.0.1' |
1515 |
| - proc = spawn_python('-u', '-m', 'http.server', str(port), '-b', bind, |
1516 |
| - bufsize=1, text=True) |
1517 |
| - self.addCleanup(kill_python, proc) |
1518 |
| - self.addCleanup(proc.terminate) |
1519 |
| - self.assertTrue(self.wait_for_server(proc, 'http', port, bind)) |
1520 |
| - res = self.fetch_file(f'http://{bind}:{port}/{self.served_file_name}') |
1521 |
| - self.assertEqual(res, self.served_data) |
1522 |
| - |
1523 |
| - def test_https_client(self): |
1524 |
| - port = find_unused_port() |
1525 |
| - bind = '127.0.0.1' |
1526 |
| - proc = spawn_python('-u', '-m', 'http.server', str(port), '-b', bind, |
1527 |
| - '--tls-cert', self.tls_cert, |
1528 |
| - '--tls-key', self.tls_key, |
1529 |
| - '--tls-password-file', self.tls_password_file, |
1530 |
| - bufsize=1, text=True) |
1531 |
| - self.addCleanup(kill_python, proc) |
1532 |
| - self.addCleanup(proc.terminate) |
1533 |
| - self.assertTrue(self.wait_for_server(proc, 'https', port, bind)) |
1534 |
| - res = self.fetch_file(f'https://{bind}:{port}/{self.served_file_name}') |
1535 |
| - self.assertEqual(res, self.served_data) |
1536 |
| - |
1537 |
| - |
1538 | 1455 | def setUpModule():
|
1539 | 1456 | unittest.addModuleCleanup(os.chdir, os.getcwd())
|
1540 | 1457 |
|
|
0 commit comments