diff options
Diffstat (limited to 'Lib/test/test_httpservers.py')
| -rw-r--r-- | Lib/test/test_httpservers.py | 174 |
1 files changed, 120 insertions, 54 deletions
diff --git a/Lib/test/test_httpservers.py b/Lib/test/test_httpservers.py index 74e0714..518e38e 100644 --- a/Lib/test/test_httpservers.py +++ b/Lib/test/test_httpservers.py @@ -6,7 +6,7 @@ Josip Dzolonga, and Michael Otteneder for the 2007/08 GHOP contest. from http.server import BaseHTTPRequestHandler, HTTPServer, \ SimpleHTTPRequestHandler, CGIHTTPRequestHandler -from http import server +from http import server, HTTPStatus import os import sys @@ -79,13 +79,13 @@ class BaseHTTPServerTestCase(BaseTestCase): default_request_version = 'HTTP/1.1' def do_TEST(self): - self.send_response(204) + self.send_response(HTTPStatus.NO_CONTENT) self.send_header('Content-Type', 'text/html') self.send_header('Connection', 'close') self.end_headers() def do_KEEP(self): - self.send_response(204) + self.send_response(HTTPStatus.NO_CONTENT) self.send_header('Content-Type', 'text/html') self.send_header('Connection', 'keep-alive') self.end_headers() @@ -94,7 +94,7 @@ class BaseHTTPServerTestCase(BaseTestCase): self.send_error(999) def do_NOTFOUND(self): - self.send_error(404) + self.send_error(HTTPStatus.NOT_FOUND) def do_EXPLAINERROR(self): self.send_error(999, "Short Message", @@ -122,35 +122,35 @@ class BaseHTTPServerTestCase(BaseTestCase): def test_command(self): self.con.request('GET', '/') res = self.con.getresponse() - self.assertEqual(res.status, 501) + self.assertEqual(res.status, HTTPStatus.NOT_IMPLEMENTED) def test_request_line_trimming(self): self.con._http_vsn_str = 'HTTP/1.1\n' self.con.putrequest('XYZBOGUS', '/') self.con.endheaders() res = self.con.getresponse() - self.assertEqual(res.status, 501) + self.assertEqual(res.status, HTTPStatus.NOT_IMPLEMENTED) def test_version_bogus(self): self.con._http_vsn_str = 'FUBAR' self.con.putrequest('GET', '/') self.con.endheaders() res = self.con.getresponse() - self.assertEqual(res.status, 400) + self.assertEqual(res.status, HTTPStatus.BAD_REQUEST) def test_version_digits(self): self.con._http_vsn_str = 'HTTP/9.9.9' self.con.putrequest('GET', '/') self.con.endheaders() res = self.con.getresponse() - self.assertEqual(res.status, 400) + self.assertEqual(res.status, HTTPStatus.BAD_REQUEST) def test_version_none_get(self): self.con._http_vsn_str = '' self.con.putrequest('GET', '/') self.con.endheaders() res = self.con.getresponse() - self.assertEqual(res.status, 501) + self.assertEqual(res.status, HTTPStatus.NOT_IMPLEMENTED) def test_version_none(self): # Test that a valid method is rejected when not HTTP/1.x @@ -158,7 +158,7 @@ class BaseHTTPServerTestCase(BaseTestCase): self.con.putrequest('CUSTOM', '/') self.con.endheaders() res = self.con.getresponse() - self.assertEqual(res.status, 400) + self.assertEqual(res.status, HTTPStatus.BAD_REQUEST) def test_version_invalid(self): self.con._http_vsn = 99 @@ -166,21 +166,21 @@ class BaseHTTPServerTestCase(BaseTestCase): self.con.putrequest('GET', '/') self.con.endheaders() res = self.con.getresponse() - self.assertEqual(res.status, 505) + self.assertEqual(res.status, HTTPStatus.HTTP_VERSION_NOT_SUPPORTED) def test_send_blank(self): self.con._http_vsn_str = '' self.con.putrequest('', '') self.con.endheaders() res = self.con.getresponse() - self.assertEqual(res.status, 400) + self.assertEqual(res.status, HTTPStatus.BAD_REQUEST) def test_header_close(self): self.con.putrequest('GET', '/') self.con.putheader('Connection', 'close') self.con.endheaders() res = self.con.getresponse() - self.assertEqual(res.status, 501) + self.assertEqual(res.status, HTTPStatus.NOT_IMPLEMENTED) def test_head_keep_alive(self): self.con._http_vsn_str = 'HTTP/1.1' @@ -188,12 +188,12 @@ class BaseHTTPServerTestCase(BaseTestCase): self.con.putheader('Connection', 'keep-alive') self.con.endheaders() res = self.con.getresponse() - self.assertEqual(res.status, 501) + self.assertEqual(res.status, HTTPStatus.NOT_IMPLEMENTED) def test_handler(self): self.con.request('TEST', '/') res = self.con.getresponse() - self.assertEqual(res.status, 204) + self.assertEqual(res.status, HTTPStatus.NO_CONTENT) def test_return_header_keep_alive(self): self.con.request('KEEP', '/') @@ -230,11 +230,48 @@ class BaseHTTPServerTestCase(BaseTestCase): # Issue #16088: standard error responses should have a content-length self.con.request('NOTFOUND', '/') res = self.con.getresponse() - self.assertEqual(res.status, 404) + self.assertEqual(res.status, HTTPStatus.NOT_FOUND) + data = res.read() self.assertEqual(int(res.getheader('Content-Length')), len(data)) +class RequestHandlerLoggingTestCase(BaseTestCase): + class request_handler(BaseHTTPRequestHandler): + protocol_version = 'HTTP/1.1' + default_request_version = 'HTTP/1.1' + + def do_GET(self): + self.send_response(HTTPStatus.OK) + self.end_headers() + + def do_ERROR(self): + self.send_error(HTTPStatus.NOT_FOUND, 'File not found') + + def test_get(self): + self.con = http.client.HTTPConnection(self.HOST, self.PORT) + self.con.connect() + + with support.captured_stderr() as err: + self.con.request('GET', '/') + self.con.getresponse() + + self.assertTrue( + err.getvalue().endswith('"GET / HTTP/1.1" 200 -\n')) + + def test_err(self): + self.con = http.client.HTTPConnection(self.HOST, self.PORT) + self.con.connect() + + with support.captured_stderr() as err: + self.con.request('ERROR', '/') + self.con.getresponse() + + lines = err.getvalue().split('\n') + self.assertTrue(lines[0].endswith('code 404, message File not found')) + self.assertTrue(lines[1].endswith('"ERROR / HTTP/1.1" 404 -')) + + class SimpleHTTPServerTestCase(BaseTestCase): class request_handler(NoLogRequestHandler, SimpleHTTPRequestHandler): pass @@ -261,12 +298,28 @@ class SimpleHTTPServerTestCase(BaseTestCase): BaseTestCase.tearDown(self) def check_status_and_reason(self, response, status, data=None): + def close_conn(): + """Don't close reader yet so we can check if there was leftover + buffered input""" + nonlocal reader + reader = response.fp + response.fp = None + reader = None + response._close_conn = close_conn + body = response.read() self.assertTrue(response) self.assertEqual(response.status, status) self.assertIsNotNone(response.reason) if data: self.assertEqual(data, body) + # Ensure the server has not set up a persistent connection, and has + # not sent any extra data + self.assertEqual(response.version, 10) + self.assertEqual(response.msg.get("Connection", "close"), "close") + self.assertEqual(reader.read(30), b'', 'Connection should be closed') + + reader.close() return body @support.requires_mac_ver(10, 5) @@ -285,52 +338,58 @@ class SimpleHTTPServerTestCase(BaseTestCase): if name != 'test': # Ignore a filename created in setUp(). filename = name break - body = self.check_status_and_reason(response, 200) + 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)) .encode(enc, 'surrogateescape'), body) response = self.request(self.tempdir_name + '/' + quotedname) - self.check_status_and_reason(response, 200, + self.check_status_and_reason(response, HTTPStatus.OK, data=support.TESTFN_UNDECODABLE) def test_get(self): #constructs the path relative to the root directory of the HTTPServer response = self.request(self.tempdir_name + '/test') - self.check_status_and_reason(response, 200, data=self.data) + self.check_status_and_reason(response, HTTPStatus.OK, data=self.data) # check for trailing "/" which should return 404. See Issue17324 response = self.request(self.tempdir_name + '/test/') - self.check_status_and_reason(response, 404) + self.check_status_and_reason(response, HTTPStatus.NOT_FOUND) response = self.request(self.tempdir_name + '/') - self.check_status_and_reason(response, 200) + self.check_status_and_reason(response, HTTPStatus.OK) response = self.request(self.tempdir_name) - self.check_status_and_reason(response, 301) + self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY) response = self.request(self.tempdir_name + '/?hi=2') - self.check_status_and_reason(response, 200) + self.check_status_and_reason(response, HTTPStatus.OK) response = self.request(self.tempdir_name + '?hi=1') - self.check_status_and_reason(response, 301) + self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY) self.assertEqual(response.getheader("Location"), self.tempdir_name + "/?hi=1") response = self.request('/ThisDoesNotExist') - self.check_status_and_reason(response, 404) + self.check_status_and_reason(response, HTTPStatus.NOT_FOUND) response = self.request('/' + 'ThisDoesNotExist' + '/') - self.check_status_and_reason(response, 404) - with open(os.path.join(self.tempdir_name, 'index.html'), 'w') as f: - response = self.request('/' + self.tempdir_name + '/') - self.check_status_and_reason(response, 200) - # chmod() doesn't work as expected on Windows, and filesystem - # permissions are ignored by root on Unix. - if os.name == 'posix' and os.geteuid() != 0: - os.chmod(self.tempdir, 0) + self.check_status_and_reason(response, HTTPStatus.NOT_FOUND) + + data = b"Dummy index file\r\n" + with open(os.path.join(self.tempdir_name, 'index.html'), 'wb') as f: + f.write(data) + response = self.request('/' + self.tempdir_name + '/') + self.check_status_and_reason(response, HTTPStatus.OK, data) + + # chmod() doesn't work as expected on Windows, and filesystem + # permissions are ignored by root on Unix. + if os.name == 'posix' and os.geteuid() != 0: + os.chmod(self.tempdir, 0) + try: response = self.request(self.tempdir_name + '/') - self.check_status_and_reason(response, 404) + self.check_status_and_reason(response, HTTPStatus.NOT_FOUND) + finally: os.chmod(self.tempdir, 0o755) def test_head(self): response = self.request( self.tempdir_name + '/test', method='HEAD') - self.check_status_and_reason(response, 200) + self.check_status_and_reason(response, HTTPStatus.OK) self.assertEqual(response.getheader('content-length'), str(len(self.data))) self.assertEqual(response.getheader('content-type'), @@ -338,12 +397,12 @@ class SimpleHTTPServerTestCase(BaseTestCase): def test_invalid_requests(self): response = self.request('/', method='FOO') - self.check_status_and_reason(response, 501) + self.check_status_and_reason(response, HTTPStatus.NOT_IMPLEMENTED) # requests must be case sensitive,so this should fail too response = self.request('/', method='custom') - self.check_status_and_reason(response, 501) + self.check_status_and_reason(response, HTTPStatus.NOT_IMPLEMENTED) response = self.request('/', method='GETs') - self.check_status_and_reason(response, 501) + self.check_status_and_reason(response, HTTPStatus.NOT_IMPLEMENTED) cgi_file1 = """\ @@ -490,12 +549,13 @@ class CGIHTTPServerTestCase(BaseTestCase): def test_headers_and_content(self): res = self.request('/cgi-bin/file1.py') - self.assertEqual((b'Hello World' + self.linesep, 'text/html', 200), - (res.read(), res.getheader('Content-type'), res.status)) + 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, 404) + self.assertEqual(res.status, HTTPStatus.NOT_FOUND) def test_post(self): params = urllib.parse.urlencode( @@ -508,38 +568,43 @@ class CGIHTTPServerTestCase(BaseTestCase): def test_invaliduri(self): res = self.request('/cgi-bin/invalid') res.read() - self.assertEqual(res.status, 404) + 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', 200), - (res.read(), res.getheader('Content-type'), res.status)) + 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', 200), - (res.read(), res.getheader('Content-type'), res.status)) + 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', 200), - (res.read(), res.getheader('Content-type'), res.status)) + 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', 200), - (res.read(), res.getheader('Content-type'), res.status)) + 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', 200), - (res.read(), res.getheader('Content-type'), res.status)) + self.assertEqual( + (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK), + (res.read(), res.getheader('Content-type'), res.status)) class SocketlessRequestHandler(SimpleHTTPRequestHandler): @@ -549,7 +614,7 @@ class SocketlessRequestHandler(SimpleHTTPRequestHandler): def do_GET(self): self.get_called = True - self.send_response(200) + self.send_response(HTTPStatus.OK) self.send_header('Content-Type', 'text/html') self.end_headers() self.wfile.write(b'<html><body>Data</body></html>\r\n') @@ -559,7 +624,7 @@ class SocketlessRequestHandler(SimpleHTTPRequestHandler): class RejectingSocketlessRequestHandler(SocketlessRequestHandler): def handle_expect_100(self): - self.send_error(417) + self.send_error(HTTPStatus.EXPECTATION_FAILED) return False @@ -816,6 +881,7 @@ def test_main(verbose=None): cwd = os.getcwd() try: support.run_unittest( + RequestHandlerLoggingTestCase, BaseHTTPRequestHandlerTestCase, BaseHTTPServerTestCase, SimpleHTTPServerTestCase, |