Lines Matching refs:self

39     def log_message(self, *args):
43 def read(self, n=None):
48 def __init__(self, test_object, request_handler):
49 threading.Thread.__init__(self)
50 self.request_handler = request_handler
51 self.test_object = test_object
53 def run(self):
54 self.server = HTTPServer(('localhost', 0), self.request_handler)
55 self.test_object.HOST, self.test_object.PORT = self.server.socket.getsockname()
56 self.test_object.server_started.set()
57 self.test_object = None
59 self.server.serve_forever(0.05)
61 self.server.server_close()
63 def stop(self):
64 self.server.shutdown()
65 self.join()
69 def setUp(self):
70 self._threads = threading_helper.threading_setup()
72 self.server_started = threading.Event()
73 self.thread = TestServerThread(self, self.request_handler)
74 self.thread.start()
75 self.server_started.wait()
77 def tearDown(self):
78 self.thread.stop()
79 self.thread = None
81 threading_helper.threading_cleanup(*self._threads)
83 def request(self, uri, method='GET', body=None, headers={}):
84 self.connection = http.client.HTTPConnection(self.HOST, self.PORT)
85 self.connection.request(method, uri, body, headers)
86 return self.connection.getresponse()
94 def do_TEST(self):
95 self.send_response(HTTPStatus.NO_CONTENT)
96 self.send_header('Content-Type', 'text/html')
97 self.send_header('Connection', 'close')
98 self.end_headers()
100 def do_KEEP(self):
101 self.send_response(HTTPStatus.NO_CONTENT)
102 self.send_header('Content-Type', 'text/html')
103 self.send_header('Connection', 'keep-alive')
104 self.end_headers()
106 def do_KEYERROR(self):
107 self.send_error(999)
109 def do_NOTFOUND(self):
110 self.send_error(HTTPStatus.NOT_FOUND)
112 def do_EXPLAINERROR(self):
113 self.send_error(999, "Short Message",
116 def do_CUSTOM(self):
117 self.send_response(999)
118 self.send_header('Content-Type', 'text/html')
119 self.send_header('Connection', 'close')
120 self.end_headers()
122 def do_LATINONEHEADER(self):
123 self.send_response(999)
124 self.send_header('X-Special', 'Dängerous Mind')
125 self.send_header('Connection', 'close')
126 self.end_headers()
127 body = self.headers['x-special-incoming'].encode('utf-8')
128 self.wfile.write(body)
130 def do_SEND_ERROR(self):
131 self.send_error(int(self.path[1:]))
133 def do_HEAD(self):
134 self.send_error(int(self.path[1:]))
136 def setUp(self):
137 BaseTestCase.setUp(self)
138 self.con = http.client.HTTPConnection(self.HOST, self.PORT)
139 self.con.connect()
141 def test_command(self):
142 self.con.request('GET', '/')
143 res = self.con.getresponse()
144 self.assertEqual(res.status, HTTPStatus.NOT_IMPLEMENTED)
146 def test_request_line_trimming(self):
147 self.con._http_vsn_str = 'HTTP/1.1\n'
148 self.con.putrequest('XYZBOGUS', '/')
149 self.con.endheaders()
150 res = self.con.getresponse()
151 self.assertEqual(res.status, HTTPStatus.NOT_IMPLEMENTED)
153 def test_version_bogus(self):
154 self.con._http_vsn_str = 'FUBAR'
155 self.con.putrequest('GET', '/')
156 self.con.endheaders()
157 res = self.con.getresponse()
158 self.assertEqual(res.status, HTTPStatus.BAD_REQUEST)
160 def test_version_digits(self):
161 self.con._http_vsn_str = 'HTTP/9.9.9'
162 self.con.putrequest('GET', '/')
163 self.con.endheaders()
164 res = self.con.getresponse()
165 self.assertEqual(res.status, HTTPStatus.BAD_REQUEST)
167 def test_version_signs_and_underscores(self):
168 self.con._http_vsn_str = 'HTTP/-9_9_9.+9_9_9'
169 self.con.putrequest('GET', '/')
170 self.con.endheaders()
171 res = self.con.getresponse()
172 self.assertEqual(res.status, HTTPStatus.BAD_REQUEST)
174 def test_major_version_number_too_long(self):
175 self.con._http_vsn_str = 'HTTP/909876543210.0'
176 self.con.putrequest('GET', '/')
177 self.con.endheaders()
178 res = self.con.getresponse()
179 self.assertEqual(res.status, HTTPStatus.BAD_REQUEST)
181 def test_minor_version_number_too_long(self):
182 self.con._http_vsn_str = 'HTTP/1.909876543210'
183 self.con.putrequest('GET', '/')
184 self.con.endheaders()
185 res = self.con.getresponse()
186 self.assertEqual(res.status, HTTPStatus.BAD_REQUEST)
188 def test_version_none_get(self):
189 self.con._http_vsn_str = ''
190 self.con.putrequest('GET', '/')
191 self.con.endheaders()
192 res = self.con.getresponse()
193 self.assertEqual(res.status, HTTPStatus.NOT_IMPLEMENTED)
195 def test_version_none(self):
197 self.con._http_vsn_str = ''
198 self.con.putrequest('CUSTOM', '/')
199 self.con.endheaders()
200 res = self.con.getresponse()
201 self.assertEqual(res.status, HTTPStatus.BAD_REQUEST)
203 def test_version_invalid(self):
204 self.con._http_vsn = 99
205 self.con._http_vsn_str = 'HTTP/9.9'
206 self.con.putrequest('GET', '/')
207 self.con.endheaders()
208 res = self.con.getresponse()
209 self.assertEqual(res.status, HTTPStatus.HTTP_VERSION_NOT_SUPPORTED)
211 def test_send_blank(self):
212 self.con._http_vsn_str = ''
213 self.con.putrequest('', '')
214 self.con.endheaders()
215 res = self.con.getresponse()
216 self.assertEqual(res.status, HTTPStatus.BAD_REQUEST)
218 def test_header_close(self):
219 self.con.putrequest('GET', '/')
220 self.con.putheader('Connection', 'close')
221 self.con.endheaders()
222 res = self.con.getresponse()
223 self.assertEqual(res.status, HTTPStatus.NOT_IMPLEMENTED)
225 def test_header_keep_alive(self):
226 self.con._http_vsn_str = 'HTTP/1.1'
227 self.con.putrequest('GET', '/')
228 self.con.putheader('Connection', 'keep-alive')
229 self.con.endheaders()
230 res = self.con.getresponse()
231 self.assertEqual(res.status, HTTPStatus.NOT_IMPLEMENTED)
233 def test_handler(self):
234 self.con.request('TEST', '/')
235 res = self.con.getresponse()
236 self.assertEqual(res.status, HTTPStatus.NO_CONTENT)
238 def test_return_header_keep_alive(self):
239 self.con.request('KEEP', '/')
240 res = self.con.getresponse()
241 self.assertEqual(res.getheader('Connection'), 'keep-alive')
242 self.con.request('TEST', '/')
243 self.addCleanup(self.con.close)
245 def test_internal_key_error(self):
246 self.con.request('KEYERROR', '/')
247 res = self.con.getresponse()
248 self.assertEqual(res.status, 999)
250 def test_return_custom_status(self):
251 self.con.request('CUSTOM', '/')
252 res = self.con.getresponse()
253 self.assertEqual(res.status, 999)
255 def test_return_explain_error(self):
256 self.con.request('EXPLAINERROR', '/')
257 res = self.con.getresponse()
258 self.assertEqual(res.status, 999)
259 self.assertTrue(int(res.getheader('Content-Length')))
261 def test_latin1_header(self):
262 self.con.request('LATINONEHEADER', '/', headers={
265 res = self.con.getresponse()
266 self.assertEqual(res.getheader('X-Special'), 'Dängerous Mind')
267 self.assertEqual(res.read(), 'Ärger mit Unicode'.encode('utf-8'))
269 def test_error_content_length(self):
271 self.con.request('NOTFOUND', '/')
272 res = self.con.getresponse()
273 self.assertEqual(res.status, HTTPStatus.NOT_FOUND)
276 self.assertEqual(int(res.getheader('Content-Length')), len(data))
278 def test_send_error(self):
284 self.con.request('SEND_ERROR', '/{}'.format(code))
285 res = self.con.getresponse()
286 self.assertEqual(code, res.status)
287 self.assertEqual(None, res.getheader('Content-Length'))
288 self.assertEqual(None, res.getheader('Content-Type'))
290 self.assertEqual(None, res.getheader('Transfer-Encoding'))
293 self.assertEqual(b'', data)
295 def test_head_via_send_error(self):
301 self.con.request('HEAD', '/{}'.format(code))
302 res = self.con.getresponse()
303 self.assertEqual(code, res.status)
305 self.assertTrue(int(res.getheader('Content-Length')) > 0)
306 self.assertIn('text/html', res.getheader('Content-Type'))
308 self.assertEqual(None, res.getheader('Content-Length'))
309 self.assertEqual(None, res.getheader('Content-Type'))
311 self.assertEqual(None, res.getheader('Transfer-Encoding'))
314 self.assertEqual(b'', data)
322 def do_GET(self):
323 self.send_response(HTTPStatus.OK)
324 self.end_headers()
326 def do_ERROR(self):
327 self.send_error(HTTPStatus.NOT_FOUND, 'File not found')
329 def test_get(self):
330 self.con = http.client.HTTPConnection(self.HOST, self.PORT)
331 self.con.connect()
334 self.con.request('GET', '/')
335 self.con.getresponse()
337 self.assertTrue(
340 def test_err(self):
341 self.con = http.client.HTTPConnection(self.HOST, self.PORT)
342 self.con.connect()
345 self.con.request('ERROR', '/')
346 self.con.getresponse()
349 self.assertTrue(lines[0].endswith('code 404, message File not found'))
350 self.assertTrue(lines[1].endswith('"ERROR / HTTP/1.1" 404 -'))
357 def setUp(self):
359 self.cwd = os.getcwd()
362 self.data = b'We are the knights who say Ni!'
363 self.tempdir = tempfile.mkdtemp(dir=basetempdir)
364 self.tempdir_name = os.path.basename(self.tempdir)
365 self.base_url = '/' + self.tempdir_name
366 tempname = os.path.join(self.tempdir, 'test')
368 temp.write(self.data)
374 self.last_modif_datetime = last_modif.replace(microsecond=0)
375 self.last_modif_header = email.utils.formatdate(
378 def tearDown(self):
380 os.chdir(self.cwd)
382 shutil.rmtree(self.tempdir)
388 def check_status_and_reason(self, response, status, data=None):
399 self.assertTrue(response)
400 self.assertEqual(response.status, status)
401 self.assertIsNotNone(response.reason)
403 self.assertEqual(data, body)
406 self.assertEqual(response.version, 10)
407 self.assertEqual(response.msg.get("Connection", "close"), "close")
408 self.assertEqual(reader.read(30), b'', 'Connection should be closed')
419 def test_undecodable_filename(self):
422 with open(os.path.join(self.tempdir, filename), 'wb') as f:
424 response = self.request(self.base_url + '/')
428 for name in os.listdir(self.tempdir):
432 body = self.check_status_and_reason(response, HTTPStatus.OK)
434 self.assertIn(('href="%s"' % quotedname)
436 self.assertIn(('>%s<' % html.escape(filename, quote=False))
438 response = self.request(self.base_url + '/' + quotedname)
439 self.check_status_and_reason(response, HTTPStatus.OK,
442 def test_undecodable_parameter(self):
444 response = self.request(self.base_url + '/?x=123').read()
445 self.assertRegex(response, f'listing for {self.base_url}/\?x=123'.encode('latin1'))
447 response = self.request(self.base_url + '/?x=%bb').read()
448 self.assertRegex(response, f'listing for {self.base_url}/\?x=\xef\xbf\xbd'.encode('latin1'))
450 def test_get_dir_redirect_location_domain_injection_bug(self):
459 os.mkdir(os.path.join(self.tempdir, 'existing_directory'))
460 url = f'/python.org/..%2f..%2f..%2f..%2f..%2f../%0a%0d/../{self.tempdir_name}/existing_directory'
464 response = self.request(url)
465 self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
467 self.assertEqual(location, expected_location, msg='non-attack failed!')
471 response = self.request(attack_url)
472 self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
474 self.assertFalse(location.startswith('//'), msg=location)
475 self.assertEqual(location, expected_location,
481 response = self.request(attack3_url)
482 self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
483 self.assertEqual(response.getheader('Location'), expected_location)
491 response = self.request(attack_scheme_netloc_2slash_url)
492 self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
497 self.assertTrue(location.startswith('https://pypi.org/'), msg=location)
499 def test_get(self):
501 response = self.request(self.base_url + '/test')
502 self.check_status_and_reason(response, HTTPStatus.OK, data=self.data)
504 response = self.request(self.base_url + '/test/')
505 self.check_status_and_reason(response, HTTPStatus.NOT_FOUND)
506 response = self.request(self.base_url + '/')
507 self.check_status_and_reason(response, HTTPStatus.OK)
508 response = self.request(self.base_url)
509 self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
510 self.assertEqual(response.getheader("Content-Length"), "0")
511 response = self.request(self.base_url + '/?hi=2')
512 self.check_status_and_reason(response, HTTPStatus.OK)
513 response = self.request(self.base_url + '?hi=1')
514 self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
515 self.assertEqual(response.getheader("Location"),
516 self.base_url + "/?hi=1")
517 response = self.request('/ThisDoesNotExist')
518 self.check_status_and_reason(response, HTTPStatus.NOT_FOUND)
519 response = self.request('/' + 'ThisDoesNotExist' + '/')
520 self.check_status_and_reason(response, HTTPStatus.NOT_FOUND)
521 os.makedirs(os.path.join(self.tempdir, 'spam', 'index.html'))
522 response = self.request(self.base_url + '/spam/')
523 self.check_status_and_reason(response, HTTPStatus.OK)
526 with open(os.path.join(self.tempdir_name, 'index.html'), 'wb') as f:
528 response = self.request(self.base_url + '/')
529 self.check_status_and_reason(response, HTTPStatus.OK, data)
534 os.chmod(self.tempdir, 0)
536 response = self.request(self.base_url + '/')
537 self.check_status_and_reason(response, HTTPStatus.NOT_FOUND)
539 os.chmod(self.tempdir, 0o755)
541 def test_head(self):
542 response = self.request(
543 self.base_url + '/test', method='HEAD')
544 self.check_status_and_reason(response, HTTPStatus.OK)
545 self.assertEqual(response.getheader('content-length'),
546 str(len(self.data)))
547 self.assertEqual(response.getheader('content-type'),
550 def test_browser_cache(self):
556 headers['If-Modified-Since'] = self.last_modif_header
557 response = self.request(self.base_url + '/test', headers=headers)
558 self.check_status_and_reason(response, HTTPStatus.NOT_MODIFIED)
561 new_dt = self.last_modif_datetime + datetime.timedelta(hours=1)
565 response = self.request(self.base_url + '/test', headers=headers)
566 self.check_status_and_reason(response, HTTPStatus.NOT_MODIFIED)
568 def test_browser_cache_file_changed(self):
570 dt = self.last_modif_datetime
576 response = self.request(self.base_url + '/test', headers=headers)
577 self.check_status_and_reason(response, HTTPStatus.OK)
579 def test_browser_cache_with_If_None_Match_header(self):
583 headers['If-Modified-Since'] = self.last_modif_header
585 response = self.request(self.base_url + '/test', headers=headers)
586 self.check_status_and_reason(response, HTTPStatus.OK)
588 def test_invalid_requests(self):
589 response = self.request('/', method='FOO')
590 self.check_status_and_reason(response, HTTPStatus.NOT_IMPLEMENTED)
592 response = self.request('/', method='custom')
593 self.check_status_and_reason(response, HTTPStatus.NOT_IMPLEMENTED)
594 response = self.request('/', method='GETs')
595 self.check_status_and_reason(response, HTTPStatus.NOT_IMPLEMENTED)
597 def test_last_modified(self):
601 response = self.request(self.base_url + '/test')
602 self.check_status_and_reason(response, HTTPStatus.OK, data=self.data)
604 self.assertEqual(last_modif_header, self.last_modif_header)
606 def test_path_without_leading_slash(self):
607 response = self.request(self.tempdir_name + '/test')
608 self.check_status_and_reason(response, HTTPStatus.OK, data=self.data)
609 response = self.request(self.tempdir_name + '/test/')
610 self.check_status_and_reason(response, HTTPStatus.NOT_FOUND)
611 response = self.request(self.tempdir_name + '/')
612 self.check_status_and_reason(response, HTTPStatus.OK)
613 response = self.request(self.tempdir_name)
614 self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
615 response = self.request(self.tempdir_name + '/?hi=2')
616 self.check_status_and_reason(response, HTTPStatus.OK)
617 response = self.request(self.tempdir_name + '?hi=1')
618 self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)
619 self.assertEqual(response.getheader("Location"),
620 self.tempdir_name + "/?hi=1")
622 def test_html_escape_filename(self):
624 fullpath = os.path.join(self.tempdir, filename)
633 response = self.request(self.base_url + '/')
634 body = self.check_status_and_reason(response, HTTPStatus.OK)
639 self.assertIsNotNone(enc)
641 self.assertIn(html_text.encode(enc), body)
706 def setUp(self):
707 BaseTestCase.setUp(self)
708 self.cwd = os.getcwd()
709 self.parent_dir = tempfile.mkdtemp()
710 self.cgi_dir = os.path.join(self.parent_dir, 'cgi-bin')
711 self.cgi_child_dir = os.path.join(self.cgi_dir, 'child-dir')
712 self.sub_dir_1 = os.path.join(self.parent_dir, 'sub')
713 self.sub_dir_2 = os.path.join(self.sub_dir_1, 'dir')
714 self.cgi_dir_in_sub_dir = os.path.join(self.sub_dir_2, 'cgi-bin')
715 os.mkdir(self.cgi_dir)
716 os.mkdir(self.cgi_child_dir)
717 os.mkdir(self.sub_dir_1)
718 os.mkdir(self.sub_dir_2)
719 os.mkdir(self.cgi_dir_in_sub_dir)
720 self.nocgi_path = None
721 self.file1_path = None
722 self.file2_path = None
723 self.file3_path = None
724 self.file4_path = None
725 self.file5_path = None
729 self._pythonexe_symlink = None
731 self.pythonexe = os.path.join(self.parent_dir, 'python')
732 self._pythonexe_symlink = support.PythonSymlink(self.pythonexe).__enter__()
734 self.pythonexe = sys.executable
740 self.pythonexe.encode('utf-8')
742 self.tearDown()
743 self.skipTest("Python executable path is not encodable to utf-8")
745 self.nocgi_path = os.path.join(self.parent_dir, 'nocgi.py')
746 with open(self.nocgi_path, 'w', encoding='utf-8') as fp:
747 fp.write(cgi_file1 % self.pythonexe)
748 os.chmod(self.nocgi_path, 0o777)
750 self.file1_path = os.path.join(self.cgi_dir, 'file1.py')
751 with open(self.file1_path, 'w', encoding='utf-8') as file1:
752 file1.write(cgi_file1 % self.pythonexe)
753 os.chmod(self.file1_path, 0o777)
755 self.file2_path = os.path.join(self.cgi_dir, 'file2.py')
756 with open(self.file2_path, 'w', encoding='utf-8') as file2:
757 file2.write(cgi_file2 % self.pythonexe)
758 os.chmod(self.file2_path, 0o777)
760 self.file3_path = os.path.join(self.cgi_child_dir, 'file3.py')
761 with open(self.file3_path, 'w', encoding='utf-8') as file3:
762 file3.write(cgi_file1 % self.pythonexe)
763 os.chmod(self.file3_path, 0o777)
765 self.file4_path = os.path.join(self.cgi_dir, 'file4.py')
766 with open(self.file4_path, 'w', encoding='utf-8') as file4:
767 file4.write(cgi_file4 % (self.pythonexe, 'QUERY_STRING'))
768 os.chmod(self.file4_path, 0o777)
770 self.file5_path = os.path.join(self.cgi_dir_in_sub_dir, 'file5.py')
771 with open(self.file5_path, 'w', encoding='utf-8') as file5:
772 file5.write(cgi_file1 % self.pythonexe)
773 os.chmod(self.file5_path, 0o777)
775 self.file6_path = os.path.join(self.cgi_dir, 'file6.py')
776 with open(self.file6_path, 'w', encoding='utf-8') as file6:
777 file6.write(cgi_file6 % self.pythonexe)
778 os.chmod(self.file6_path, 0o777)
780 os.chdir(self.parent_dir)
782 def tearDown(self):
784 os.chdir(self.cwd)
785 if self._pythonexe_symlink:
786 self._pythonexe_symlink.__exit__(None, None, None)
787 if self.nocgi_path:
788 os.remove(self.nocgi_path)
789 if self.file1_path:
790 os.remove(self.file1_path)
791 if self.file2_path:
792 os.remove(self.file2_path)
793 if self.file3_path:
794 os.remove(self.file3_path)
795 if self.file4_path:
796 os.remove(self.file4_path)
797 if self.file5_path:
798 os.remove(self.file5_path)
799 if self.file6_path:
800 os.remove(self.file6_path)
801 os.rmdir(self.cgi_child_dir)
802 os.rmdir(self.cgi_dir)
803 os.rmdir(self.cgi_dir_in_sub_dir)
804 os.rmdir(self.sub_dir_2)
805 os.rmdir(self.sub_dir_1)
806 os.rmdir(self.parent_dir)
808 BaseTestCase.tearDown(self)
810 def test_url_collapse_path(self):
844 self.assertRaises(expected,
848 self.assertEqual(expected, actual,
852 def test_headers_and_content(self):
853 res = self.request('/cgi-bin/file1.py')
854 self.assertEqual(
856 (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK))
858 def test_issue19435(self):
859 res = self.request('///////////nocgi.py/../cgi-bin/nothere.sh')
860 self.assertEqual(res.status, HTTPStatus.NOT_FOUND)
862 def test_post(self):
866 res = self.request('/cgi-bin/file2.py', 'POST', params, headers)
868 self.assertEqual(res.read(), b'1, python, 123456' + self.linesep)
870 def test_invaliduri(self):
871 res = self.request('/cgi-bin/invalid')
873 self.assertEqual(res.status, HTTPStatus.NOT_FOUND)
875 def test_authorization(self):
878 res = self.request('/cgi-bin/file1.py', 'GET', headers=headers)
879 self.assertEqual(
880 (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
883 def test_no_leading_slash(self):
885 res = self.request('cgi-bin/file1.py')
886 self.assertEqual(
887 (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
890 def test_os_environ_is_not_altered(self):
893 res = self.request('/cgi-bin/file1.py')
894 self.assertEqual(
895 (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
897 self.assertEqual(os.environ['SERVER_SOFTWARE'], signature)
899 def test_urlquote_decoding_in_cgi_check(self):
900 res = self.request('/cgi-bin%2ffile1.py')
901 self.assertEqual(
902 (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
905 def test_nested_cgi_path_issue21323(self):
906 res = self.request('/cgi-bin/child-dir/file3.py')
907 self.assertEqual(
908 (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
911 def test_query_with_multiple_question_mark(self):
912 res = self.request('/cgi-bin/file4.py?a=b?c=d')
913 self.assertEqual(
914 (b'a=b?c=d' + self.linesep, 'text/html', HTTPStatus.OK),
917 def test_query_with_continuous_slashes(self):
918 res = self.request('/cgi-bin/file4.py?k=aa%2F%2Fbb&//q//p//=//a//b//')
919 self.assertEqual(
920 (b'k=aa%2F%2Fbb&//q//p//=//a//b//' + self.linesep,
924 def test_cgi_path_in_sub_directories(self):
927 res = self.request('/sub/dir/cgi-bin/file5.py')
928 self.assertEqual(
929 (b'Hello World' + self.linesep, 'text/html', HTTPStatus.OK),
934 def test_accept(self):
946 with self.subTest(headers):
947 res = self.request('/cgi-bin/file6.py', 'GET', headers=headers)
948 self.assertEqual(http.HTTPStatus.OK, res.status)
950 self.assertIn(expected, res.read())
954 def __init__(self, directory=None):
959 self.get_called = False
960 self.protocol_version = "HTTP/1.1"
962 def do_GET(self):
963 self.get_called = True
964 self.send_response(HTTPStatus.OK)
965 self.send_header('Content-Type', 'text/html')
966 self.end_headers()
967 self.wfile.write(b'<html><body>Data</body></html>\r\n')
969 def log_message(self, format, *args):
973 def handle_expect_100(self):
974 self.send_error(HTTPStatus.EXPECTATION_FAILED)
980 def __init__(self):
981 self.datas = []
983 def write(self, data):
984 self.datas.append(data)
986 def getData(self):
987 return b''.join(self.datas)
990 def numWrites(self):
991 return len(self.datas)
1002 def setUp (self):
1003 self.handler = SocketlessRequestHandler()
1005 def send_typical_request(self, message):
1008 self.handler.rfile = input
1009 self.handler.wfile = output
1010 self.handler.handle_one_request()
1014 def verify_get_called(self):
1015 self.assertTrue(self.handler.get_called)
1017 def verify_expected_headers(self, headers):
1019 self.assertEqual(sum(h.startswith(fieldName) for h in headers), 1)
1021 def verify_http_server_response(self, response):
1022 match = self.HTTPResponseMatch.search(response)
1023 self.assertIsNotNone(match)
1025 def test_unprintable_not_logged(self):
1028 self.handler.client_address = ('127.0.0.1', 1337)
1031 log_message(self.handler, '/foo')
1032 log_message(self.handler, '/\033bar\000\033')
1033 log_message(self.handler, '/spam %s.', 'a')
1034 log_message(self.handler, '/spam %s.', '\033\x7f\x9f\xa0beans')
1035 log_message(self.handler, '"GET /foo\\b"ar\007 HTTP/1.0"')
1037 self.assertNotIn('\033', stderr) # non-printable chars are caught.
1038 self.assertNotIn('\000', stderr) # non-printable chars are caught.
1040 self.assertIn('/foo', lines[0])
1041 self.assertIn(r'/\x1bbar\x00\x1b', lines[1])
1042 self.assertIn('/spam a.', lines[2])
1043 self.assertIn('/spam \\x1b\\x7f\\x9f\xa0beans.', lines[3])
1044 self.assertIn(r'"GET /foo\\b"ar\x07 HTTP/1.0"', lines[4])
1046 def test_http_1_1(self):
1047 result = self.send_typical_request(b'GET / HTTP/1.1\r\n\r\n')
1048 self.verify_http_server_response(result[0])
1049 self.verify_expected_headers(result[1:-1])
1050 self.verify_get_called()
1051 self.assertEqual(result[-1], b'<html><body>Data</body></html>\r\n')
1052 self.assertEqual(self.handler.requestline, 'GET / HTTP/1.1')
1053 self.assertEqual(self.handler.command, 'GET')
1054 self.assertEqual(self.handler.path, '/')
1055 self.assertEqual(self.handler.request_version, 'HTTP/1.1')
1056 self.assertSequenceEqual(self.handler.headers.items(), ())
1058 def test_http_1_0(self):
1059 result = self.send_typical_request(b'GET / HTTP/1.0\r\n\r\n')
1060 self.verify_http_server_response(result[0])
1061 self.verify_expected_headers(result[1:-1])
1062 self.verify_get_called()
1063 self.assertEqual(result[-1], b'<html><body>Data</body></html>\r\n')
1064 self.assertEqual(self.handler.requestline, 'GET / HTTP/1.0')
1065 self.assertEqual(self.handler.command, 'GET')
1066 self.assertEqual(self.handler.path, '/')
1067 self.assertEqual(self.handler.request_version, 'HTTP/1.0')
1068 self.assertSequenceEqual(self.handler.headers.items(), ())
1070 def test_http_0_9(self):
1071 result = self.send_typical_request(b'GET / HTTP/0.9\r\n\r\n')
1072 self.assertEqual(len(result), 1)
1073 self.assertEqual(result[0], b'<html><body>Data</body></html>\r\n')
1074 self.verify_get_called()
1076 def test_extra_space(self):
1077 result = self.send_typical_request(
1082 self.assertTrue(result[0].startswith(b'HTTP/1.1 400 '))
1083 self.verify_expected_headers(result[1:result.index(b'\r\n')])
1084 self.assertFalse(self.handler.get_called)
1086 def test_with_continue_1_0(self):
1087 result = self.send_typical_request(b'GET / HTTP/1.0\r\nExpect: 100-continue\r\n\r\n')
1088 self.verify_http_server_response(result[0])
1089 self.verify_expected_headers(result[1:-1])
1090 self.verify_get_called()
1091 self.assertEqual(result[-1], b'<html><body>Data</body></html>\r\n')
1092 self.assertEqual(self.handler.requestline, 'GET / HTTP/1.0')
1093 self.assertEqual(self.handler.command, 'GET')
1094 self.assertEqual(self.handler.path, '/')
1095 self.assertEqual(self.handler.request_version, 'HTTP/1.0')
1097 self.assertSequenceEqual(self.handler.headers.items(), headers)
1099 def test_with_continue_1_1(self):
1100 result = self.send_typical_request(b'GET / HTTP/1.1\r\nExpect: 100-continue\r\n\r\n')
1101 self.assertEqual(result[0], b'HTTP/1.1 100 Continue\r\n')
1102 self.assertEqual(result[1], b'\r\n')
1103 self.assertEqual(result[2], b'HTTP/1.1 200 OK\r\n')
1104 self.verify_expected_headers(result[2:-1])
1105 self.verify_get_called()
1106 self.assertEqual(result[-1], b'<html><body>Data</body></html>\r\n')
1107 self.assertEqual(self.handler.requestline, 'GET / HTTP/1.1')
1108 self.assertEqual(self.handler.command, 'GET')
1109 self.assertEqual(self.handler.path, '/')
1110 self.assertEqual(self.handler.request_version, 'HTTP/1.1')
1112 self.assertSequenceEqual(self.handler.headers.items(), headers)
1114 def test_header_buffering_of_send_error(self):
1126 self.assertEqual(output.numWrites, 2)
1128 def test_header_buffering_of_send_response_only(self):
1138 self.assertEqual(output.numWrites, 0)
1140 self.assertEqual(output.numWrites, 1)
1142 def test_header_buffering_of_send_header(self):
1153 self.assertEqual(output.numWrites, 0)
1155 self.assertEqual(output.getData(), b'Foo: foo\r\nbar: bar\r\n\r\n')
1156 self.assertEqual(output.numWrites, 1)
1158 def test_header_unbuffered_when_continue(self):
1169 self.handler.rfile = input
1170 self.handler.wfile = output
1171 self.handler.request_version = 'HTTP/1.1'
1173 self.handler.handle_one_request()
1174 self.assertNotEqual(_readAndReseek(output), b'')
1176 self.assertEqual(result[0], b'HTTP/1.1 100 Continue')
1177 self.assertEqual(result[1], b'')
1178 self.assertEqual(result[2], b'HTTP/1.1 200 OK')
1180 def test_with_continue_rejected(self):
1181 usual_handler = self.handler # Save to avoid breaking any subsequent tests.
1182 self.handler = RejectingSocketlessRequestHandler()
1183 result = self.send_typical_request(b'GET / HTTP/1.1\r\nExpect: 100-continue\r\n\r\n')
1184 self.assertEqual(result[0], b'HTTP/1.1 417 Expectation Failed\r\n')
1185 self.verify_expected_headers(result[1:-1])
1188 self.assertFalse(self.handler.get_called)
1189 self.assertEqual(sum(r == b'Connection: close\r\n' for r in result[1:-1]), 1)
1190 self.handler = usual_handler # Restore to avoid breaking any subsequent tests.
1192 def test_request_length(self):
1195 result = self.send_typical_request(b'GET ' + b'x' * 65537)
1196 self.assertEqual(result[0], b'HTTP/1.1 414 Request-URI Too Long\r\n')
1197 self.assertFalse(self.handler.get_called)
1198 self.assertIsInstance(self.handler.requestline, str)
1200 def test_header_length(self):
1202 result = self.send_typical_request(
1204 self.assertEqual(result[0], b'HTTP/1.1 431 Line too long\r\n')
1205 self.assertFalse(self.handler.get_called)
1206 self.assertEqual(self.handler.requestline, 'GET / HTTP/1.1')
1208 def test_too_many_headers(self):
1209 result = self.send_typical_request(
1211 self.assertEqual(result[0], b'HTTP/1.1 431 Too many headers\r\n')
1212 self.assertFalse(self.handler.get_called)
1213 self.assertEqual(self.handler.requestline, 'GET / HTTP/1.1')
1215 def test_html_escape_on_error(self):
1216 result = self.send_typical_request(
1220 self.assertIn(html.escape(text, quote=False).encode('ascii'), result)
1222 def test_close_connection(self):
1226 self.handler.close_connection = next(close_values)
1227 self.handler.handle_one_request = handle_one_request
1230 self.handler.handle()
1231 self.assertRaises(StopIteration, next, close_values)
1234 self.handler.handle()
1235 self.assertRaises(StopIteration, next, close_values)
1237 def test_date_time_string(self):
1242 self.handler.weekdayname[wd],
1244 self.handler.monthname[month],
1247 self.assertEqual(self.handler.date_time_string(timestamp=now), expected)
1252 def setUp(self):
1253 self.translated_1 = os.path.join(os.getcwd(), 'filename')
1254 self.translated_2 = os.path.join('foo', 'filename')
1255 self.translated_3 = os.path.join('bar', 'filename')
1256 self.handler_1 = SocketlessRequestHandler()
1257 self.handler_2 = SocketlessRequestHandler(directory='foo')
1258 self.handler_3 = SocketlessRequestHandler(directory=pathlib.PurePath('bar'))
1260 def test_query_arguments(self):
1261 path = self.handler_1.translate_path('/filename')
1262 self.assertEqual(path, self.translated_1)
1263 path = self.handler_2.translate_path('/filename')
1264 self.assertEqual(path, self.translated_2)
1265 path = self.handler_3.translate_path('/filename')
1266 self.assertEqual(path, self.translated_3)
1268 path = self.handler_1.translate_path('/filename?foo=bar')
1269 self.assertEqual(path, self.translated_1)
1270 path = self.handler_2.translate_path('/filename?foo=bar')
1271 self.assertEqual(path, self.translated_2)
1272 path = self.handler_3.translate_path('/filename?foo=bar')
1273 self.assertEqual(path, self.translated_3)
1275 path = self.handler_1.translate_path('/filename?a=b&spam=eggs#zot')
1276 self.assertEqual(path, self.translated_1)
1277 path = self.handler_2.translate_path('/filename?a=b&spam=eggs#zot')
1278 self.assertEqual(path, self.translated_2)
1279 path = self.handler_3.translate_path('/filename?a=b&spam=eggs#zot')
1280 self.assertEqual(path, self.translated_3)
1282 def test_start_with_double_slash(self):
1283 path = self.handler_1.translate_path('//filename')
1284 self.assertEqual(path, self.translated_1)
1285 path = self.handler_2.translate_path('//filename')
1286 self.assertEqual(path, self.translated_2)
1287 path = self.handler_3.translate_path('//filename')
1288 self.assertEqual(path, self.translated_3)
1290 path = self.handler_1.translate_path('//filename?foo=bar')
1291 self.assertEqual(path, self.translated_1)
1292 path = self.handler_2.translate_path('//filename?foo=bar')
1293 self.assertEqual(path, self.translated_2)
1294 path = self.handler_3.translate_path('//filename?foo=bar')
1295 self.assertEqual(path, self.translated_3)
1297 def test_windows_colon(self):
1299 path = self.handler_1.translate_path('c:c:c:foo/filename')
1301 self.assertEqual(path, self.translated_1)
1302 path = self.handler_2.translate_path('c:c:c:foo/filename')
1304 self.assertEqual(path, self.translated_2)
1305 path = self.handler_3.translate_path('c:c:c:foo/filename')
1307 self.assertEqual(path, self.translated_3)
1309 path = self.handler_1.translate_path('\\c:../filename')
1311 self.assertEqual(path, self.translated_1)
1312 path = self.handler_2.translate_path('\\c:../filename')
1314 self.assertEqual(path, self.translated_2)
1315 path = self.handler_3.translate_path('\\c:../filename')
1317 self.assertEqual(path, self.translated_3)
1319 path = self.handler_1.translate_path('c:\\c:..\\foo/filename')
1321 self.assertEqual(path, self.translated_1)
1322 path = self.handler_2.translate_path('c:\\c:..\\foo/filename')
1324 self.assertEqual(path, self.translated_2)
1325 path = self.handler_3.translate_path('c:\\c:..\\foo/filename')
1327 self.assertEqual(path, self.translated_3)
1329 path = self.handler_1.translate_path('c:c:foo\\c:c:bar/filename')
1331 self.assertEqual(path, self.translated_1)
1332 path = self.handler_2.translate_path('c:c:foo\\c:c:bar/filename')
1334 self.assertEqual(path, self.translated_2)
1335 path = self.handler_3.translate_path('c:c:foo\\c:c:bar/filename')
1337 self.assertEqual(path, self.translated_3)
1341 def test_all(self):
1350 self.assertCountEqual(server.__all__, expected)
1355 def mock_server_class(self):
1369 def test_server_test_unspec(self, _):
1370 mock_server = self.mock_server_class()
1372 self.assertIn(
1378 def test_server_test_localhost(self, _):
1379 mock_server = self.mock_server_class()
1381 self.assertIn(
1399 def test_server_test_ipv6(self, _):
1400 for bind in self.ipv6_addrs:
1401 mock_server = self.mock_server_class()
1403 self.assertEqual(mock_server.address_family, socket.AF_INET6)
1406 def test_server_test_ipv4(self, _):
1407 for bind in self.ipv4_addrs:
1408 mock_server = self.mock_server_class()
1410 self.assertEqual(mock_server.address_family, socket.AF_INET)