|
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