aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/Lib/test/test_httpservers.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test/test_httpservers.py')
-rw-r--r--Lib/test/test_httpservers.py668
1 files changed, 292 insertions, 376 deletions
diff --git a/Lib/test/test_httpservers.py b/Lib/test/test_httpservers.py
index 2cafa4e45a1..0af1c45ecb2 100644
--- a/Lib/test/test_httpservers.py
+++ b/Lib/test/test_httpservers.py
@@ -3,16 +3,16 @@
Written by Cody A.W. Somerville <cody-somerville@ubuntu.com>,
Josip Dzolonga, and Michael Otteneder for the 2007/08 GHOP contest.
"""
-from collections import OrderedDict
+
from http.server import BaseHTTPRequestHandler, HTTPServer, HTTPSServer, \
- SimpleHTTPRequestHandler, CGIHTTPRequestHandler
+ SimpleHTTPRequestHandler
from http import server, HTTPStatus
+import contextlib
import os
import socket
import sys
import re
-import base64
import ntpath
import pathlib
import shutil
@@ -31,7 +31,7 @@ from io import BytesIO, StringIO
import unittest
from test import support
from test.support import (
- is_apple, import_helper, os_helper, requires_subprocess, threading_helper
+ is_apple, import_helper, os_helper, threading_helper
)
try:
@@ -522,42 +522,120 @@ class SimpleHTTPServerTestCase(BaseTestCase):
reader.close()
return body
+ def check_list_dir_dirname(self, dirname, quotedname=None):
+ fullpath = os.path.join(self.tempdir, dirname)
+ try:
+ os.mkdir(os.path.join(self.tempdir, dirname))
+ except (OSError, UnicodeEncodeError):
+ self.skipTest(f'Can not create directory {dirname!a} '
+ f'on current file system')
+
+ if quotedname is None:
+ quotedname = urllib.parse.quote(dirname, errors='surrogatepass')
+ response = self.request(self.base_url + '/' + quotedname + '/')
+ body = self.check_status_and_reason(response, HTTPStatus.OK)
+ displaypath = html.escape(f'{self.base_url}/{dirname}/', quote=False)
+ enc = sys.getfilesystemencoding()
+ prefix = f'listing for {displaypath}</'.encode(enc, 'surrogateescape')
+ self.assertIn(prefix + b'title>', body)
+ self.assertIn(prefix + b'h1>', body)
+
+ def check_list_dir_filename(self, filename):
+ fullpath = os.path.join(self.tempdir, filename)
+ content = ascii(fullpath).encode() + (os_helper.TESTFN_UNDECODABLE or b'\xff')
+ try:
+ with open(fullpath, 'wb') as f:
+ f.write(content)
+ except OSError:
+ self.skipTest(f'Can not create file {filename!a} '
+ f'on current file system')
+
+ response = self.request(self.base_url + '/')
+ body = self.check_status_and_reason(response, HTTPStatus.OK)
+ quotedname = urllib.parse.quote(filename, errors='surrogatepass')
+ enc = response.headers.get_content_charset()
+ self.assertIsNotNone(enc)
+ self.assertIn((f'href="{quotedname}"').encode('ascii'), body)
+ displayname = html.escape(filename, quote=False)
+ self.assertIn(f'>{displayname}<'.encode(enc, 'surrogateescape'), body)
+
+ response = self.request(self.base_url + '/' + quotedname)
+ self.check_status_and_reason(response, HTTPStatus.OK, data=content)
+
+ @unittest.skipUnless(os_helper.TESTFN_NONASCII,
+ 'need os_helper.TESTFN_NONASCII')
+ def test_list_dir_nonascii_dirname(self):
+ dirname = os_helper.TESTFN_NONASCII + '.dir'
+ self.check_list_dir_dirname(dirname)
+
+ @unittest.skipUnless(os_helper.TESTFN_NONASCII,
+ 'need os_helper.TESTFN_NONASCII')
+ def test_list_dir_nonascii_filename(self):
+ filename = os_helper.TESTFN_NONASCII + '.txt'
+ self.check_list_dir_filename(filename)
+
@unittest.skipIf(is_apple,
'undecodable name cannot always be decoded on Apple platforms')
@unittest.skipIf(sys.platform == 'win32',
'undecodable name cannot be decoded on win32')
@unittest.skipUnless(os_helper.TESTFN_UNDECODABLE,
'need os_helper.TESTFN_UNDECODABLE')
- def test_undecodable_filename(self):
- enc = sys.getfilesystemencoding()
- filename = os.fsdecode(os_helper.TESTFN_UNDECODABLE) + '.txt'
- with open(os.path.join(self.tempdir, filename), 'wb') as f:
- f.write(os_helper.TESTFN_UNDECODABLE)
- response = self.request(self.base_url + '/')
- if is_apple:
- # On Apple platforms the HFS+ filesystem replaces bytes that
- # aren't valid UTF-8 into a percent-encoded value.
- for name in os.listdir(self.tempdir):
- if name != 'test': # Ignore a filename created in setUp().
- filename = name
- break
- body = self.check_status_and_reason(response, HTTPStatus.OK)
- quotedname = urllib.parse.quote(filename, errors='surrogatepass')
- self.assertIn(('href="%s"' % quotedname)
- .encode(enc, 'surrogateescape'), body)
- self.assertIn(('>%s<' % html.escape(filename, quote=False))
- .encode(enc, 'surrogateescape'), body)
- response = self.request(self.base_url + '/' + quotedname)
- self.check_status_and_reason(response, HTTPStatus.OK,
- data=os_helper.TESTFN_UNDECODABLE)
+ def test_list_dir_undecodable_dirname(self):
+ dirname = os.fsdecode(os_helper.TESTFN_UNDECODABLE) + '.dir'
+ self.check_list_dir_dirname(dirname)
- def test_undecodable_parameter(self):
- # sanity check using a valid parameter
+ @unittest.skipIf(is_apple,
+ 'undecodable name cannot always be decoded on Apple platforms')
+ @unittest.skipIf(sys.platform == 'win32',
+ 'undecodable name cannot be decoded on win32')
+ @unittest.skipUnless(os_helper.TESTFN_UNDECODABLE,
+ 'need os_helper.TESTFN_UNDECODABLE')
+ def test_list_dir_undecodable_filename(self):
+ filename = os.fsdecode(os_helper.TESTFN_UNDECODABLE) + '.txt'
+ self.check_list_dir_filename(filename)
+
+ def test_list_dir_undecodable_dirname2(self):
+ dirname = '\ufffd.dir'
+ self.check_list_dir_dirname(dirname, quotedname='%ff.dir')
+
+ @unittest.skipUnless(os_helper.TESTFN_UNENCODABLE,
+ 'need os_helper.TESTFN_UNENCODABLE')
+ def test_list_dir_unencodable_dirname(self):
+ dirname = os_helper.TESTFN_UNENCODABLE + '.dir'
+ self.check_list_dir_dirname(dirname)
+
+ @unittest.skipUnless(os_helper.TESTFN_UNENCODABLE,
+ 'need os_helper.TESTFN_UNENCODABLE')
+ def test_list_dir_unencodable_filename(self):
+ filename = os_helper.TESTFN_UNENCODABLE + '.txt'
+ self.check_list_dir_filename(filename)
+
+ def test_list_dir_escape_dirname(self):
+ # Characters that need special treating in URL or HTML.
+ for name in ('q?', 'f#', '&amp;', '&amp', '<i>', '"dq"', "'sq'",
+ '%A4', '%E2%82%AC'):
+ with self.subTest(name=name):
+ dirname = name + '.dir'
+ self.check_list_dir_dirname(dirname,
+ quotedname=urllib.parse.quote(dirname, safe='&<>\'"'))
+
+ def test_list_dir_escape_filename(self):
+ # Characters that need special treating in URL or HTML.
+ for name in ('q?', 'f#', '&amp;', '&amp', '<i>', '"dq"', "'sq'",
+ '%A4', '%E2%82%AC'):
+ with self.subTest(name=name):
+ filename = name + '.txt'
+ self.check_list_dir_filename(filename)
+ os_helper.unlink(os.path.join(self.tempdir, filename))
+
+ def test_list_dir_with_query_and_fragment(self):
+ prefix = f'listing for {self.base_url}/</'.encode('latin1')
+ response = self.request(self.base_url + '/#123').read()
+ self.assertIn(prefix + b'title>', response)
+ self.assertIn(prefix + b'h1>', response)
response = self.request(self.base_url + '/?x=123').read()
- self.assertRegex(response, rf'listing for {self.base_url}/\?x=123'.encode('latin1'))
- # now the bogus encoding
- response = self.request(self.base_url + '/?x=%bb').read()
- self.assertRegex(response, rf'listing for {self.base_url}/\?x=\xef\xbf\xbd'.encode('latin1'))
+ self.assertIn(prefix + b'title>', response)
+ self.assertIn(prefix + b'h1>', response)
def test_get_dir_redirect_location_domain_injection_bug(self):
"""Ensure //evil.co/..%2f../../X does not put //evil.co/ in Location.
@@ -615,10 +693,19 @@ class SimpleHTTPServerTestCase(BaseTestCase):
# check for trailing "/" which should return 404. See Issue17324
response = self.request(self.base_url + '/test/')
self.check_status_and_reason(response, HTTPStatus.NOT_FOUND)
+ response = self.request(self.base_url + '/test%2f')
+ self.check_status_and_reason(response, HTTPStatus.NOT_FOUND)
+ response = self.request(self.base_url + '/test%2F')
+ self.check_status_and_reason(response, HTTPStatus.NOT_FOUND)
response = self.request(self.base_url + '/')
self.check_status_and_reason(response, HTTPStatus.OK)
+ response = self.request(self.base_url + '%2f')
+ self.check_status_and_reason(response, HTTPStatus.OK)
+ response = self.request(self.base_url + '%2F')
+ self.check_status_and_reason(response, HTTPStatus.OK)
response = self.request(self.base_url)
self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
+ self.assertEqual(response.getheader("Location"), self.base_url + "/")
self.assertEqual(response.getheader("Content-Length"), "0")
response = self.request(self.base_url + '/?hi=2')
self.check_status_and_reason(response, HTTPStatus.OK)
@@ -724,6 +811,8 @@ class SimpleHTTPServerTestCase(BaseTestCase):
self.check_status_and_reason(response, HTTPStatus.OK)
response = self.request(self.tempdir_name)
self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
+ self.assertEqual(response.getheader("Location"),
+ self.tempdir_name + "/")
response = self.request(self.tempdir_name + '/?hi=2')
self.check_status_and_reason(response, HTTPStatus.OK)
response = self.request(self.tempdir_name + '?hi=1')
@@ -731,350 +820,6 @@ class SimpleHTTPServerTestCase(BaseTestCase):
self.assertEqual(response.getheader("Location"),
self.tempdir_name + "/?hi=1")
- def test_html_escape_filename(self):
- filename = '<test&>.txt'
- fullpath = os.path.join(self.tempdir, filename)
-
- try:
- open(fullpath, 'wb').close()
- except OSError:
- raise unittest.SkipTest('Can not create file %s on current file '
- 'system' % filename)
-
- try:
- response = self.request(self.base_url + '/')
- body = self.check_status_and_reason(response, HTTPStatus.OK)
- enc = response.headers.get_content_charset()
- finally:
- os.unlink(fullpath) # avoid affecting test_undecodable_filename
-
- self.assertIsNotNone(enc)
- html_text = '>%s<' % html.escape(filename, quote=False)
- self.assertIn(html_text.encode(enc), body)
-
-
-cgi_file1 = """\
-#!%s
-
-print("Content-type: text/html")
-print()
-print("Hello World")
-"""
-
-cgi_file2 = """\
-#!%s
-import os
-import sys
-import urllib.parse
-
-print("Content-type: text/html")
-print()
-
-content_length = int(os.environ["CONTENT_LENGTH"])
-query_string = sys.stdin.buffer.read(content_length)
-params = {key.decode("utf-8"): val.decode("utf-8")
- for key, val in urllib.parse.parse_qsl(query_string)}
-
-print("%%s, %%s, %%s" %% (params["spam"], params["eggs"], params["bacon"]))
-"""
-
-cgi_file4 = """\
-#!%s
-import os
-
-print("Content-type: text/html")
-print()
-
-print(os.environ["%s"])
-"""
-
-cgi_file6 = """\
-#!%s
-import os
-
-print("X-ambv: was here")
-print("Content-type: text/html")
-print()
-print("<pre>")
-for k, v in os.environ.items():
- try:
- k.encode('ascii')
- v.encode('ascii')
- except UnicodeEncodeError:
- continue # see: BPO-44647
- print(f"{k}={v}")
-print("</pre>")
-"""
-
-
-@unittest.skipIf(hasattr(os, 'geteuid') and os.geteuid() == 0,
- "This test can't be run reliably as root (issue #13308).")
-@requires_subprocess()
-class CGIHTTPServerTestCase(BaseTestCase):
- class request_handler(NoLogRequestHandler, CGIHTTPRequestHandler):
- _test_case_self = None # populated by each setUp() method call.
-
- def __init__(self, *args, **kwargs):
- with self._test_case_self.assertWarnsRegex(
- DeprecationWarning,
- r'http\.server\.CGIHTTPRequestHandler'):
- # This context also happens to catch and silence the
- # threading DeprecationWarning from os.fork().
- super().__init__(*args, **kwargs)
-
- linesep = os.linesep.encode('ascii')
-
- def setUp(self):
- self.request_handler._test_case_self = self # practical, but yuck.
- BaseTestCase.setUp(self)
- self.cwd = os.getcwd()
- self.parent_dir = tempfile.mkdtemp()
- self.cgi_dir = os.path.join(self.parent_dir, 'cgi-bin')
- self.cgi_child_dir = os.path.join(self.cgi_dir, 'child-dir')
- self.sub_dir_1 = os.path.join(self.parent_dir, 'sub')
- self.sub_dir_2 = os.path.join(self.sub_dir_1, 'dir')
- self.cgi_dir_in_sub_dir = os.path.join(self.sub_dir_2, 'cgi-bin')
- os.mkdir(self.cgi_dir)
- os.mkdir(self.cgi_child_dir)
- os.mkdir(self.sub_dir_1)
- os.mkdir(self.sub_dir_2)
- os.mkdir(self.cgi_dir_in_sub_dir)
- self.nocgi_path = None
- self.file1_path = None
- self.file2_path = None
- self.file3_path = None
- self.file4_path = None
- self.file5_path = None
-
- # The shebang line should be pure ASCII: use symlink if possible.
- # See issue #7668.
- self._pythonexe_symlink = None
- if os_helper.can_symlink():
- self.pythonexe = os.path.join(self.parent_dir, 'python')
- self._pythonexe_symlink = support.PythonSymlink(self.pythonexe).__enter__()
- else:
- self.pythonexe = sys.executable
-
- try:
- # The python executable path is written as the first line of the
- # CGI Python script. The encoding cookie cannot be used, and so the
- # path should be encodable to the default script encoding (utf-8)
- self.pythonexe.encode('utf-8')
- except UnicodeEncodeError:
- self.tearDown()
- self.skipTest("Python executable path is not encodable to utf-8")
-
- self.nocgi_path = os.path.join(self.parent_dir, 'nocgi.py')
- with open(self.nocgi_path, 'w', encoding='utf-8') as fp:
- fp.write(cgi_file1 % self.pythonexe)
- os.chmod(self.nocgi_path, 0o777)
-
- self.file1_path = os.path.join(self.cgi_dir, 'file1.py')
- with open(self.file1_path, 'w', encoding='utf-8') as file1:
- file1.write(cgi_file1 % self.pythonexe)
- os.chmod(self.file1_path, 0o777)
-
- self.file2_path = os.path.join(self.cgi_dir, 'file2.py')
- with open(self.file2_path, 'w', encoding='utf-8') as file2:
- file2.write(cgi_file2 % self.pythonexe)
- os.chmod(self.file2_path, 0o777)
-
- self.file3_path = os.path.join(self.cgi_child_dir, 'file3.py')
- with open(self.file3_path, 'w', encoding='utf-8') as file3:
- file3.write(cgi_file1 % self.pythonexe)
- os.chmod(self.file3_path, 0o777)
-
- self.file4_path = os.path.join(self.cgi_dir, 'file4.py')
- with open(self.file4_path, 'w', encoding='utf-8') as file4:
- file4.write(cgi_file4 % (self.pythonexe, 'QUERY_STRING'))
- os.chmod(self.file4_path, 0o777)
-
- self.file5_path = os.path.join(self.cgi_dir_in_sub_dir, 'file5.py')
- with open(self.file5_path, 'w', encoding='utf-8') as file5:
- file5.write(cgi_file1 % self.pythonexe)
- os.chmod(self.file5_path, 0o777)
-
- self.file6_path = os.path.join(self.cgi_dir, 'file6.py')
- with open(self.file6_path, 'w', encoding='utf-8') as file6:
- file6.write(cgi_file6 % self.pythonexe)
- os.chmod(self.file6_path, 0o777)
-
- os.chdir(self.parent_dir)
-
- def tearDown(self):
- self.request_handler._test_case_self = None
- try:
- os.chdir(self.cwd)
- if self._pythonexe_symlink:
- self._pythonexe_symlink.__exit__(None, None, None)
- if self.nocgi_path:
- os.remove(self.nocgi_path)
- if self.file1_path:
- os.remove(self.file1_path)
- if self.file2_path:
- os.remove(self.file2_path)
- if self.file3_path:
- os.remove(self.file3_path)
- if self.file4_path:
- os.remove(self.file4_path)
- if self.file5_path:
- os.remove(self.file5_path)
- if self.file6_path:
- os.remove(self.file6_path)
- os.rmdir(self.cgi_child_dir)
- os.rmdir(self.cgi_dir)
- os.rmdir(self.cgi_dir_in_sub_dir)
- os.rmdir(self.sub_dir_2)
- os.rmdir(self.sub_dir_1)
- # The 'gmon.out' file can be written in the current working
- # directory if C-level code profiling with gprof is enabled.
- os_helper.unlink(os.path.join(self.parent_dir, 'gmon.out'))
- os.rmdir(self.parent_dir)
- finally:
- BaseTestCase.tearDown(self)
-
- def test_url_collapse_path(self):
- # verify tail is the last portion and head is the rest on proper urls
- test_vectors = {
- '': '//',
- '..': IndexError,
- '/.//..': IndexError,
- '/': '//',
- '//': '//',
- '/\\': '//\\',
- '/.//': '//',
- 'cgi-bin/file1.py': '/cgi-bin/file1.py',
- '/cgi-bin/file1.py': '/cgi-bin/file1.py',
- 'a': '//a',
- '/a': '//a',
- '//a': '//a',
- './a': '//a',
- './C:/': '/C:/',
- '/a/b': '/a/b',
- '/a/b/': '/a/b/',
- '/a/b/.': '/a/b/',
- '/a/b/c/..': '/a/b/',
- '/a/b/c/../d': '/a/b/d',
- '/a/b/c/../d/e/../f': '/a/b/d/f',
- '/a/b/c/../d/e/../../f': '/a/b/f',
- '/a/b/c/../d/e/.././././..//f': '/a/b/f',
- '../a/b/c/../d/e/.././././..//f': IndexError,
- '/a/b/c/../d/e/../../../f': '/a/f',
- '/a/b/c/../d/e/../../../../f': '//f',
- '/a/b/c/../d/e/../../../../../f': IndexError,
- '/a/b/c/../d/e/../../../../f/..': '//',
- '/a/b/c/../d/e/../../../../f/../.': '//',
- }
- for path, expected in test_vectors.items():
- if isinstance(expected, type) and issubclass(expected, Exception):
- self.assertRaises(expected,
- server._url_collapse_path, path)
- else:
- actual = server._url_collapse_path(path)
- self.assertEqual(expected, actual,
- msg='path = %r\nGot: %r\nWanted: %r' %
- (path, actual, expected))
-
- def test_headers_and_content(self):
- res = self.request('/cgi-bin/file1.py')
- self.assertEqual(
- (res.read(), res.getheader('Content-type'), res.status),
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK))
-
- def test_issue19435(self):
- res = self.request('///////////nocgi.py/../cgi-bin/nothere.sh')
- self.assertEqual(res.status, HTTPStatus.NOT_FOUND)
-
- def test_post(self):
- params = urllib.parse.urlencode(
- {'spam' : 1, 'eggs' : 'python', 'bacon' : 123456})
- headers = {'Content-type' : 'application/x-www-form-urlencoded'}
- res = self.request('/cgi-bin/file2.py', 'POST', params, headers)
-
- self.assertEqual(res.read(), b'1, python, 123456' + self.linesep)
-
- def test_invaliduri(self):
- res = self.request('/cgi-bin/invalid')
- res.read()
- self.assertEqual(res.status, HTTPStatus.NOT_FOUND)
-
- def test_authorization(self):
- headers = {b'Authorization' : b'Basic ' +
- base64.b64encode(b'username:pass')}
- res = self.request('/cgi-bin/file1.py', 'GET', headers=headers)
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_no_leading_slash(self):
- # http://bugs.python.org/issue2254
- res = self.request('cgi-bin/file1.py')
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_os_environ_is_not_altered(self):
- signature = "Test CGI Server"
- os.environ['SERVER_SOFTWARE'] = signature
- res = self.request('/cgi-bin/file1.py')
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
- self.assertEqual(os.environ['SERVER_SOFTWARE'], signature)
-
- def test_urlquote_decoding_in_cgi_check(self):
- res = self.request('/cgi-bin%2ffile1.py')
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_nested_cgi_path_issue21323(self):
- res = self.request('/cgi-bin/child-dir/file3.py')
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_query_with_multiple_question_mark(self):
- res = self.request('/cgi-bin/file4.py?a=b?c=d')
- self.assertEqual(
- (b'a=b?c=d' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_query_with_continuous_slashes(self):
- res = self.request('/cgi-bin/file4.py?k=aa%2F%2Fbb&//q//p//=//a//b//')
- self.assertEqual(
- (b'k=aa%2F%2Fbb&//q//p//=//a//b//' + self.linesep,
- 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
-
- def test_cgi_path_in_sub_directories(self):
- try:
- CGIHTTPRequestHandler.cgi_directories.append('/sub/dir/cgi-bin')
- res = self.request('/sub/dir/cgi-bin/file5.py')
- self.assertEqual(
- (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
- (res.read(), res.getheader('Content-type'), res.status))
- finally:
- CGIHTTPRequestHandler.cgi_directories.remove('/sub/dir/cgi-bin')
-
- def test_accept(self):
- browser_accept = \
- 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8'
- tests = (
- ((('Accept', browser_accept),), browser_accept),
- ((), ''),
- # Hack case to get two values for the one header
- ((('Accept', 'text/html'), ('ACCEPT', 'text/plain')),
- 'text/html,text/plain'),
- )
- for headers, expected in tests:
- headers = OrderedDict(headers)
- with self.subTest(headers):
- res = self.request('/cgi-bin/file6.py', 'GET', headers=headers)
- self.assertEqual(http.HTTPStatus.OK, res.status)
- expected = f"HTTP_ACCEPT={expected}".encode('ascii')
- self.assertIn(expected, res.read())
-
class SocketlessRequestHandler(SimpleHTTPRequestHandler):
def __init__(self, directory=None):
@@ -1095,6 +840,7 @@ class SocketlessRequestHandler(SimpleHTTPRequestHandler):
def log_message(self, format, *args):
pass
+
class RejectingSocketlessRequestHandler(SocketlessRequestHandler):
def handle_expect_100(self):
self.send_error(HTTPStatus.EXPECTATION_FAILED)
@@ -1536,6 +1282,176 @@ class ScriptTestCase(unittest.TestCase):
self.assertEqual(mock_server.address_family, socket.AF_INET)
+class CommandLineTestCase(unittest.TestCase):
+ default_port = 8000
+ default_bind = None
+ default_protocol = 'HTTP/1.0'
+ default_handler = SimpleHTTPRequestHandler
+ default_server = unittest.mock.ANY
+ tls_cert = certdata_file('ssl_cert.pem')
+ tls_key = certdata_file('ssl_key.pem')
+ tls_password = 'somepass'
+ tls_cert_options = ['--tls-cert']
+ tls_key_options = ['--tls-key']
+ tls_password_options = ['--tls-password-file']
+ args = {
+ 'HandlerClass': default_handler,
+ 'ServerClass': default_server,
+ 'protocol': default_protocol,
+ 'port': default_port,
+ 'bind': default_bind,
+ 'tls_cert': None,
+ 'tls_key': None,
+ 'tls_password': None,
+ }
+
+ def setUp(self):
+ super().setUp()
+ self.tls_password_file = tempfile.mktemp()
+ with open(self.tls_password_file, 'wb') as f:
+ f.write(self.tls_password.encode())
+ self.addCleanup(os_helper.unlink, self.tls_password_file)
+
+ def invoke_httpd(self, *args, stdout=None, stderr=None):
+ stdout = StringIO() if stdout is None else stdout
+ stderr = StringIO() if stderr is None else stderr
+ with contextlib.redirect_stdout(stdout), \
+ contextlib.redirect_stderr(stderr):
+ server._main(args)
+ return stdout.getvalue(), stderr.getvalue()
+
+ @mock.patch('http.server.test')
+ def test_port_flag(self, mock_func):
+ ports = [8000, 65535]
+ for port in ports:
+ with self.subTest(port=port):
+ self.invoke_httpd(str(port))
+ call_args = self.args | dict(port=port)
+ mock_func.assert_called_once_with(**call_args)
+ mock_func.reset_mock()
+
+ @mock.patch('http.server.test')
+ def test_directory_flag(self, mock_func):
+ options = ['-d', '--directory']
+ directories = ['.', '/foo', '\\bar', '/',
+ 'C:\\', 'C:\\foo', 'C:\\bar',
+ '/home/user', './foo/foo2', 'D:\\foo\\bar']
+ for flag in options:
+ for directory in directories:
+ with self.subTest(flag=flag, directory=directory):
+ self.invoke_httpd(flag, directory)
+ mock_func.assert_called_once_with(**self.args)
+ mock_func.reset_mock()
+
+ @mock.patch('http.server.test')
+ def test_bind_flag(self, mock_func):
+ options = ['-b', '--bind']
+ bind_addresses = ['localhost', '127.0.0.1', '::1',
+ '0.0.0.0', '8.8.8.8']
+ for flag in options:
+ for bind_address in bind_addresses:
+ with self.subTest(flag=flag, bind_address=bind_address):
+ self.invoke_httpd(flag, bind_address)
+ call_args = self.args | dict(bind=bind_address)
+ mock_func.assert_called_once_with(**call_args)
+ mock_func.reset_mock()
+
+ @mock.patch('http.server.test')
+ def test_protocol_flag(self, mock_func):
+ options = ['-p', '--protocol']
+ protocols = ['HTTP/1.0', 'HTTP/1.1', 'HTTP/2.0', 'HTTP/3.0']
+ for flag in options:
+ for protocol in protocols:
+ with self.subTest(flag=flag, protocol=protocol):
+ self.invoke_httpd(flag, protocol)
+ call_args = self.args | dict(protocol=protocol)
+ mock_func.assert_called_once_with(**call_args)
+ mock_func.reset_mock()
+
+ @unittest.skipIf(ssl is None, "requires ssl")
+ @mock.patch('http.server.test')
+ def test_tls_cert_and_key_flags(self, mock_func):
+ for tls_cert_option in self.tls_cert_options:
+ for tls_key_option in self.tls_key_options:
+ self.invoke_httpd(tls_cert_option, self.tls_cert,
+ tls_key_option, self.tls_key)
+ call_args = self.args | {
+ 'tls_cert': self.tls_cert,
+ 'tls_key': self.tls_key,
+ }
+ mock_func.assert_called_once_with(**call_args)
+ mock_func.reset_mock()
+
+ @unittest.skipIf(ssl is None, "requires ssl")
+ @mock.patch('http.server.test')
+ def test_tls_cert_and_key_and_password_flags(self, mock_func):
+ for tls_cert_option in self.tls_cert_options:
+ for tls_key_option in self.tls_key_options:
+ for tls_password_option in self.tls_password_options:
+ self.invoke_httpd(tls_cert_option,
+ self.tls_cert,
+ tls_key_option,
+ self.tls_key,
+ tls_password_option,
+ self.tls_password_file)
+ call_args = self.args | {
+ 'tls_cert': self.tls_cert,
+ 'tls_key': self.tls_key,
+ 'tls_password': self.tls_password,
+ }
+ mock_func.assert_called_once_with(**call_args)
+ mock_func.reset_mock()
+
+ @unittest.skipIf(ssl is None, "requires ssl")
+ @mock.patch('http.server.test')
+ def test_missing_tls_cert_flag(self, mock_func):
+ for tls_key_option in self.tls_key_options:
+ with self.assertRaises(SystemExit):
+ self.invoke_httpd(tls_key_option, self.tls_key)
+ mock_func.reset_mock()
+
+ for tls_password_option in self.tls_password_options:
+ with self.assertRaises(SystemExit):
+ self.invoke_httpd(tls_password_option, self.tls_password)
+ mock_func.reset_mock()
+
+ @unittest.skipIf(ssl is None, "requires ssl")
+ @mock.patch('http.server.test')
+ def test_invalid_password_file(self, mock_func):
+ non_existent_file = 'non_existent_file'
+ for tls_password_option in self.tls_password_options:
+ for tls_cert_option in self.tls_cert_options:
+ with self.assertRaises(SystemExit):
+ self.invoke_httpd(tls_cert_option,
+ self.tls_cert,
+ tls_password_option,
+ non_existent_file)
+
+ @mock.patch('http.server.test')
+ def test_no_arguments(self, mock_func):
+ self.invoke_httpd()
+ mock_func.assert_called_once_with(**self.args)
+ mock_func.reset_mock()
+
+ @mock.patch('http.server.test')
+ def test_help_flag(self, _):
+ options = ['-h', '--help']
+ for option in options:
+ stdout, stderr = StringIO(), StringIO()
+ with self.assertRaises(SystemExit):
+ self.invoke_httpd(option, stdout=stdout, stderr=stderr)
+ self.assertIn('usage', stdout.getvalue())
+ self.assertEqual(stderr.getvalue(), '')
+
+ @mock.patch('http.server.test')
+ def test_unknown_flag(self, _):
+ stdout, stderr = StringIO(), StringIO()
+ with self.assertRaises(SystemExit):
+ self.invoke_httpd('--unknown-flag', stdout=stdout, stderr=stderr)
+ self.assertEqual(stdout.getvalue(), '')
+ self.assertIn('error', stderr.getvalue())
+
+
def setUpModule():
unittest.addModuleCleanup(os.chdir, os.getcwd())