|
11 | 11 | import contextlib |
12 | 12 | import os |
13 | 13 | import socket |
| 14 | +import subprocess |
14 | 15 | import sys |
15 | 16 | import re |
16 | 17 | import base64 |
|
22 | 23 | import html |
23 | 24 | import http, http.client |
24 | 25 | import urllib.parse |
| 26 | +import urllib.request |
25 | 27 | import tempfile |
26 | 28 | import time |
27 | 29 | import datetime |
|
34 | 36 | from test.support import ( |
35 | 37 | is_apple, import_helper, os_helper, requires_subprocess, threading_helper |
36 | 38 | ) |
| 39 | +from test.support.socket_helper import find_unused_port |
37 | 40 |
|
38 | 41 | try: |
39 | 42 | import ssl |
@@ -1555,13 +1558,18 @@ class CommandLineTestCase(unittest.TestCase): |
1555 | 1558 | 'tls_key': None, |
1556 | 1559 | 'tls_password': None, |
1557 | 1560 | } |
| 1561 | + random_data = os.urandom(1024) |
| 1562 | + random_file_name = 'random.bin' |
1558 | 1563 |
|
1559 | 1564 | def setUp(self): |
1560 | 1565 | super().setUp() |
1561 | 1566 | self.tls_password_file = tempfile.mktemp() |
1562 | 1567 | with open(self.tls_password_file, 'wb') as f: |
1563 | 1568 | f.write(self.tls_password.encode()) |
1564 | 1569 | self.addCleanup(os_helper.unlink, self.tls_password_file) |
| 1570 | + with open(self.random_file_name, 'wb') as f: |
| 1571 | + f.write(self.random_data) |
| 1572 | + self.addCleanup(os_helper.unlink, self.random_file_name) |
1565 | 1573 |
|
1566 | 1574 | def invoke_httpd(self, *args): |
1567 | 1575 | output = StringIO() |
@@ -1726,6 +1734,44 @@ def test_unknown_flag(self, _): |
1726 | 1734 | output = self.invoke_httpd('--unknown-flag') |
1727 | 1735 | self.assertStartsWith(output, 'usage: ') |
1728 | 1736 |
|
| 1737 | + def fetch_file(self, path, allow_self_signed_cert=True) -> bytes: |
| 1738 | + context = ssl.create_default_context() |
| 1739 | + if allow_self_signed_cert: |
| 1740 | + context.check_hostname = False |
| 1741 | + context.verify_mode = ssl.CERT_NONE |
| 1742 | + req = urllib.request.Request(path, method='GET') |
| 1743 | + res = urllib.request.urlopen(req, context=context) |
| 1744 | + return res.read() |
| 1745 | + |
| 1746 | + def test_http_client(self): |
| 1747 | + port = find_unused_port() |
| 1748 | + bind = '127.0.0.1' |
| 1749 | + proc = subprocess.Popen([sys.executable, '-m', 'http.server', |
| 1750 | + str(port), '-b', bind], |
| 1751 | + stdout=subprocess.DEVNULL, |
| 1752 | + stderr=subprocess.DEVNULL) |
| 1753 | + time.sleep(0.5) # Wait for the server to start. |
| 1754 | + # TODO: Find a better way to wait for the server to start. |
| 1755 | + res = self.fetch_file(f'http://{bind}:{port}/{self.random_file_name}') |
| 1756 | + self.assertEqual(res, self.random_data) |
| 1757 | + proc.kill() |
| 1758 | + proc.wait() |
| 1759 | + |
| 1760 | + def test_https_client(self): |
| 1761 | + port = find_unused_port() |
| 1762 | + bind = '127.0.0.1' |
| 1763 | + proc = subprocess.Popen([sys.executable, '-m', 'http.server', |
| 1764 | + str(port), '-b', bind, |
| 1765 | + '--tls-cert', self.tls_cert, |
| 1766 | + '--tls-key', self.tls_key, |
| 1767 | + '--tls-password-file', self.tls_password_file], |
| 1768 | + stdout=subprocess.DEVNULL, |
| 1769 | + stderr=subprocess.DEVNULL) |
| 1770 | + time.sleep(0.5) |
| 1771 | + res = self.fetch_file(f'https://{bind}:{port}/{self.random_file_name}') |
| 1772 | + self.assertEqual(res, self.random_data) |
| 1773 | + proc.kill() |
| 1774 | + proc.wait() |
1729 | 1775 |
|
1730 | 1776 | def setUpModule(): |
1731 | 1777 | unittest.addModuleCleanup(os.chdir, os.getcwd()) |
|
0 commit comments