Lines Matching refs:self
29 # Self-signed cert file for self-signed.pythontest.net
55 def __init__(self, text, fileclass=io.BytesIO, host=None, port=None):
58 self.text = text
59 self.fileclass = fileclass
60 self.data = b''
61 self.sendall_calls = 0
62 self.file_closed = False
63 self.host = host
64 self.port = port
66 def sendall(self, data):
67 self.sendall_calls += 1
68 self.data += data
70 def makefile(self, mode, bufsize=None):
74 self.file = self.fileclass(self.text)
75 self.file.close = self.file_close #nerf close ()
76 return self.file
78 def file_close(self):
79 self.file_closed = True
81 def close(self):
84 def setsockopt(self, level, optname, value):
89 def __init__(self, text, pipe_trigger):
91 FakeSocket.__init__(self, text)
92 self.pipe_trigger = pipe_trigger
94 def sendall(self, data):
95 if self.pipe_trigger in data:
97 self.data += data
99 def close(self):
108 def read(self, n=-1):
109 data = io.BytesIO.read(self, n)
114 def readline(self, length=None):
115 data = io.BytesIO.readline(self, length)
123 def __init__(self, *args):
124 self.connections = 0
126 self.fake_socket_args = args
127 self._create_connection = self.create_connection
129 def connect(self):
131 self.connections += 1
134 def create_connection(self, *pos, **kw):
135 return FakeSocket(*self.fake_socket_args)
138 def test_auto_headers(self):
143 def __init__(self):
144 self.count = {}
145 def append(self, item):
150 self.count.setdefault(lcKey, 0)
151 self.count[lcKey] += 1
152 list.append(self, item)
165 self.assertEqual(conn._buffer.count[header.lower()], 1)
167 def test_content_length_0(self):
170 def __init__(self):
171 list.__init__(self)
172 self.content_length = None
173 def append(self, item):
176 self.content_length = kv[1].strip()
177 list.append(self, item)
188 self.assertEqual(
204 self.assertEqual(
217 self.assertEqual(
228 self.assertEqual(
233 def test_putheader(self):
238 self.assertIn(b'Content-length: 42', conn._buffer)
241 self.assertIn(b'Foo: bar ', conn._buffer)
243 self.assertIn(b'Bar: \tbaz\t', conn._buffer)
245 self.assertIn(b'Authorization: Bearer mytoken', conn._buffer)
247 self.assertIn(b'IterHeader: IterA\r\n\tIterB', conn._buffer)
249 self.assertIn(b'LatinHeader: \xFF', conn._buffer)
251 self.assertIn(b'Utf8Header: \xc3\x80', conn._buffer)
253 self.assertIn(b'C1-Control: next\x85line', conn._buffer)
255 self.assertIn(b'Embedded-Fold-Space: is\r\n allowed', conn._buffer)
257 self.assertIn(b'Embedded-Fold-Tab: is\r\n\tallowed', conn._buffer)
259 self.assertIn(b'Key Space: value', conn._buffer)
261 self.assertIn(b'KeySpace : value', conn._buffer)
263 self.assertIn(b'Nonbreak\xa0Space: value', conn._buffer)
265 self.assertIn(b'\xa0NonbreakSpace: value', conn._buffer)
267 def test_ipv6host_header(self):
276 self.assertTrue(sock.data.startswith(expected))
284 self.assertTrue(sock.data.startswith(expected))
286 def test_malformed_headers_coped_with(self):
293 self.assertEqual(resp.getheader('First'), 'val')
294 self.assertEqual(resp.getheader('Second'), 'val')
296 def test_parse_all_octets(self):
312 self.assertEqual(resp.getheader('Content-Length'), '0')
313 self.assertEqual(resp.msg['Content-Length'], '0')
314 self.assertEqual(resp.getheader("!#$%&'*+-.^_`|~"), 'value')
315 self.assertEqual(resp.msg["!#$%&'*+-.^_`|~"], 'value')
317 self.assertEqual(resp.getheader('VCHAR'), vchar)
318 self.assertEqual(resp.msg['VCHAR'], vchar)
319 self.assertIsNotNone(resp.getheader('obs-text'))
320 self.assertIn('obs-text', resp.msg)
322 self.assertTrue(folded.startswith('text'))
323 self.assertIn(' folded with space', folded)
324 self.assertTrue(folded.endswith('folded with tab'))
326 def test_invalid_headers(self):
352 with self.subTest((name, value)):
353 with self.assertRaisesRegex(ValueError, 'Invalid header'):
356 def test_headers_debuglevel(self):
368 self.assertEqual(lines[0], "reply: 'HTTP/1.1 200 OK\\r\\n'")
369 self.assertEqual(lines[1], "header: First: val")
370 self.assertEqual(lines[2], "header: Second: val1")
371 self.assertEqual(lines[3], "header: Second: val2")
375 def test_invalid_method_names(self):
389 with self.assertRaisesRegex(
399 def test_endheaders_chunked(self):
403 conn.endheaders(self._make_body(), encode_chunked=True)
405 _, _, body = self._parse_request(conn.sock.data)
406 body = self._parse_chunked(body)
407 self.assertEqual(body, self.expected_body)
409 def test_explicit_headers(self):
416 'POST', '/', self._make_body(), {'Transfer-Encoding': 'chunked'})
418 _, headers, body = self._parse_request(conn.sock.data)
419 self.assertNotIn('content-length', [k.lower() for k in headers.keys()])
420 self.assertEqual(headers['Transfer-Encoding'], 'chunked')
421 self.assertEqual(body, self.expected_body)
427 'POST', '/', self.expected_body.decode('latin-1'),
430 _, headers, body = self._parse_request(conn.sock.data)
431 self.assertNotIn('content-length', [k.lower() for k in headers.keys()])
432 self.assertEqual(headers['Transfer-Encoding'], 'chunked')
433 self.assertEqual(body, self.expected_body)
441 body=self._make_body())
442 _, headers, body = self._parse_request(conn.sock.data)
443 self.assertNotIn('content-length', [k.lower() for k in headers])
444 self.assertEqual(headers['Transfer-Encoding'], 'gzip, chunked')
445 self.assertEqual(self._parse_chunked(body), self.expected_body)
447 def test_request(self):
452 'POST', '/', self._make_body(empty_lines=empty_lines))
454 _, headers, body = self._parse_request(conn.sock.data)
455 body = self._parse_chunked(body)
456 self.assertEqual(body, self.expected_body)
457 self.assertEqual(headers['Transfer-Encoding'], 'chunked')
461 self.assertNotIn('content-length', [k.lower() for k in headers])
463 def test_empty_body(self):
468 _, headers, body = self._parse_request(conn.sock.data)
469 self.assertEqual(headers['Transfer-Encoding'], 'chunked')
470 self.assertNotIn('content-length', [k.lower() for k in headers])
471 self.assertEqual(body, b"0\r\n\r\n")
473 def _make_body(self, empty_lines=False):
474 lines = self.expected_body.split(b' ')
484 def _parse_request(self, data):
497 def _parse_chunked(self, data):
511 self.assertEqual(size, len(chunk))
525 def test_dir_with_added_behavior_on_status(self):
527 self.assertTrue({'description', 'name', 'phrase', 'value'} <= set(dir(HTTPStatus(404))))
529 def test_simple_httpstatus(self):
673 def test_status_lines(self):
680 self.assertEqual(resp.read(0), b'') # Issue #20007
681 self.assertFalse(resp.isclosed())
682 self.assertFalse(resp.closed)
683 self.assertEqual(resp.read(), b"Text")
684 self.assertTrue(resp.isclosed())
685 self.assertFalse(resp.closed)
687 self.assertTrue(resp.closed)
692 self.assertRaises(client.BadStatusLine, resp.begin)
694 def test_bad_status_repr(self):
696 self.assertEqual(repr(exc), '''BadStatusLine("''")''')
698 def test_partial_reads(self):
705 self.assertEqual(resp.read(2), b'Te')
706 self.assertFalse(resp.isclosed())
707 self.assertEqual(resp.read(2), b'xt')
708 self.assertTrue(resp.isclosed())
709 self.assertFalse(resp.closed)
711 self.assertTrue(resp.closed)
713 def test_mixed_reads(self):
720 self.assertEqual(resp.readline(), b'Text\r\n')
721 self.assertFalse(resp.isclosed())
722 self.assertEqual(resp.read(), b'Another')
723 self.assertTrue(resp.isclosed())
724 self.assertFalse(resp.closed)
726 self.assertTrue(resp.closed)
728 def test_partial_readintos(self):
737 self.assertEqual(n, 2)
738 self.assertEqual(bytes(b), b'Te')
739 self.assertFalse(resp.isclosed())
741 self.assertEqual(n, 2)
742 self.assertEqual(bytes(b), b'xt')
743 self.assertTrue(resp.isclosed())
744 self.assertFalse(resp.closed)
746 self.assertTrue(resp.closed)
748 def test_partial_reads_past_end(self):
754 self.assertEqual(resp.read(10), b'Text')
755 self.assertTrue(resp.isclosed())
756 self.assertFalse(resp.closed)
758 self.assertTrue(resp.closed)
760 def test_partial_readintos_past_end(self):
768 self.assertEqual(n, 4)
769 self.assertEqual(bytes(b)[:4], b'Text')
770 self.assertTrue(resp.isclosed())
771 self.assertFalse(resp.closed)
773 self.assertTrue(resp.closed)
775 def test_partial_reads_no_content_length(self):
782 self.assertEqual(resp.read(2), b'Te')
783 self.assertFalse(resp.isclosed())
784 self.assertEqual(resp.read(2), b'xt')
785 self.assertEqual(resp.read(1), b'')
786 self.assertTrue(resp.isclosed())
787 self.assertFalse(resp.closed)
789 self.assertTrue(resp.closed)
791 def test_partial_readintos_no_content_length(self):
800 self.assertEqual(n, 2)
801 self.assertEqual(bytes(b), b'Te')
802 self.assertFalse(resp.isclosed())
804 self.assertEqual(n, 2)
805 self.assertEqual(bytes(b), b'xt')
807 self.assertEqual(n, 0)
808 self.assertTrue(resp.isclosed())
810 def test_partial_reads_incomplete_body(self):
817 self.assertEqual(resp.read(2), b'Te')
818 self.assertFalse(resp.isclosed())
819 self.assertEqual(resp.read(2), b'xt')
820 self.assertEqual(resp.read(1), b'')
821 self.assertTrue(resp.isclosed())
823 def test_partial_readintos_incomplete_body(self):
832 self.assertEqual(n, 2)
833 self.assertEqual(bytes(b), b'Te')
834 self.assertFalse(resp.isclosed())
836 self.assertEqual(n, 2)
837 self.assertEqual(bytes(b), b'xt')
839 self.assertEqual(n, 0)
840 self.assertTrue(resp.isclosed())
841 self.assertFalse(resp.closed)
843 self.assertTrue(resp.closed)
845 def test_host_port(self):
849 self.assertRaises(client.InvalidURL, client.HTTPConnection, hp)
859 self.assertEqual(h, c.host)
860 self.assertEqual(p, c.port)
862 def test_response_headers(self):
878 self.assertEqual(cookies, hdr)
880 def test_read_head(self):
891 self.fail("Did not expect response from HEAD request")
893 def test_readinto_head(self):
905 self.fail("Did not expect response from HEAD request")
906 self.assertEqual(bytes(b), b'\x00'*5)
908 def test_too_many_headers(self):
914 self.assertRaisesRegex(client.HTTPException,
917 def test_send_file(self):
928 self.assertTrue(sock.data.startswith(expected), '%r != %r' %
931 def test_send(self):
937 self.assertEqual(expected, sock.data)
940 self.assertEqual(expected, sock.data)
943 self.assertEqual(expected, sock.data)
945 def test_send_updating_file(self):
954 def read(self, blocksize=-1):
955 return next(self.d)
963 self.assertEqual(sock.data, expected)
966 def test_send_iter(self):
980 self.assertEqual(sock.data, expected)
982 def test_blocksize_request(self):
990 self.assertEqual(sock.sendall_calls, 3)
992 self.assertEqual(body, expected)
994 def test_blocksize_send(self):
1002 self.assertEqual(sock.sendall_calls, 2)
1003 self.assertEqual(sock.data, expected)
1005 def test_send_type_error(self):
1009 with self.assertRaises(TypeError):
1012 def test_chunked(self):
1017 self.assertEqual(resp.read(), expected)
1025 self.assertEqual(resp.read(n) + resp.read(n) + resp.read(), expected)
1035 self.assertEqual(i.partial, expected)
1037 self.assertEqual(repr(i), expected_message)
1038 self.assertEqual(str(i), expected_message)
1040 self.fail('IncompleteRead expected')
1044 def test_readinto_chunked(self):
1054 self.assertEqual(b[:nexpected], expected)
1055 self.assertEqual(n, nexpected)
1067 self.assertEqual(b[:nexpected], expected)
1068 self.assertEqual(i, nexpected)
1078 self.assertEqual(i.partial, expected)
1080 self.assertEqual(repr(i), expected_message)
1081 self.assertEqual(str(i), expected_message)
1083 self.fail('IncompleteRead expected')
1087 def test_chunked_head(self):
1099 self.assertEqual(resp.read(), b'')
1100 self.assertEqual(resp.status, 200)
1101 self.assertEqual(resp.reason, 'OK')
1102 self.assertTrue(resp.isclosed())
1103 self.assertFalse(resp.closed)
1105 self.assertTrue(resp.closed)
1107 def test_readinto_chunked_head(self):
1121 self.assertEqual(n, 0)
1122 self.assertEqual(bytes(b), b'\x00'*5)
1123 self.assertEqual(resp.status, 200)
1124 self.assertEqual(resp.reason, 'OK')
1125 self.assertTrue(resp.isclosed())
1126 self.assertFalse(resp.closed)
1128 self.assertTrue(resp.closed)
1130 def test_negative_content_length(self):
1135 self.assertEqual(resp.read(), b'Hello\r\n')
1136 self.assertTrue(resp.isclosed())
1138 def test_incomplete_read(self):
1145 self.assertEqual(i.partial, b'Hello\r\n')
1146 self.assertEqual(repr(i),
1148 self.assertEqual(str(i),
1150 self.assertTrue(resp.isclosed())
1152 self.fail('IncompleteRead expected')
1154 def test_epipe(self):
1162 self.assertRaises(OSError,
1165 self.assertEqual(401, resp.status)
1166 self.assertEqual("Basic realm=\"example\"",
1171 def test_overflowing_status_line(self):
1174 self.assertRaises((client.LineTooLong, client.BadStatusLine), resp.begin)
1176 def test_overflowing_header_line(self):
1182 self.assertRaises(client.LineTooLong, resp.begin)
1184 def test_overflowing_header_limit_after_100(self):
1190 with self.assertRaises(client.HTTPException) as cm:
1194 self.assertIn('got more than ', str(cm.exception))
1195 self.assertIn('headers', str(cm.exception))
1197 def test_overflowing_chunked_line(self):
1208 self.assertRaises(client.LineTooLong, resp.read)
1210 def test_early_eof(self):
1216 self.assertEqual(resp.read(), b'')
1217 self.assertTrue(resp.isclosed())
1218 self.assertFalse(resp.closed)
1220 self.assertTrue(resp.closed)
1222 def test_error_leak(self):
1227 def __init__(self, *pos, **kw):
1229 response = self # Avoid garbage collector closing the socket
1230 client.HTTPResponse.__init__(self, *pos, **kw)
1234 self.assertRaises(client.BadStatusLine, conn.getresponse)
1235 self.assertTrue(response.closed)
1236 self.assertTrue(conn.sock.file_closed)
1238 def test_chunked_extension(self):
1245 self.assertEqual(resp.read(), expected)
1248 def test_chunked_missing_end(self):
1254 self.assertEqual(resp.read(), expected)
1257 def test_chunked_trailers(self):
1263 self.assertEqual(resp.read(), expected)
1265 self.assertEqual(sock.file.read(), b"") #we read to the end
1268 def test_chunked_sync(self):
1275 self.assertEqual(resp.read(), expected)
1277 self.assertEqual(sock.file.read(), extradata.encode("ascii")) #we read to the end
1280 def test_content_length_sync(self):
1287 self.assertEqual(resp.read(), expected)
1289 self.assertEqual(sock.file.read(), extradata) #we read to the end
1292 def test_readlines_content_length(self):
1298 self.assertEqual(resp.readlines(2000), [expected])
1300 self.assertEqual(sock.file.read(), extradata) #we read to the end
1303 def test_read1_content_length(self):
1309 self.assertEqual(resp.read1(2000), expected)
1311 self.assertEqual(sock.file.read(), extradata) #we read to the end
1314 def test_readline_bound_content_length(self):
1320 self.assertEqual(resp.readline(10), expected)
1321 self.assertEqual(resp.readline(10), b"")
1323 self.assertEqual(sock.file.read(), extradata) #we read to the end
1326 def test_read1_bound_content_length(self):
1332 self.assertEqual(resp.read1(20), expected*2)
1333 self.assertEqual(resp.read(), expected)
1335 self.assertEqual(sock.file.read(), extradata) #we read to the end
1338 def test_response_fileno(self):
1341 self.addCleanup(serv.close)
1358 self.addCleanup(thread.join, float(1))
1363 self.assertEqual(response.status, client.OK)
1373 self.assertEqual(result, b"proxied data\n")
1375 def test_putrequest_override_domain_validation(self):
1381 def _validate_path(self, url):
1388 def test_putrequest_override_host_validation(self):
1390 def _validate_host(self, url):
1399 def test_putrequest_override_encoding(self):
1406 def _encode_request(self, str_url):
1444 def setUp(self):
1445 sock = FakeSocket(self.lines)
1449 self.resp = resp
1453 def test_peek(self):
1454 resp = self.resp
1469 self.assertGreater(len(p), 0)
1472 self.assertGreaterEqual(len(p2), len(p))
1473 self.assertTrue(p2.startswith(p))
1475 self.assertEqual(next, p2)
1478 self.assertFalse(next)
1482 self.assertEqual(b"".join(all), self.lines_expected)
1484 def test_readline(self):
1485 resp = self.resp
1486 self._verify_readline(self.resp.readline, self.lines_expected)
1488 def _verify_readline(self, readline, expected):
1495 self.assertTrue(line.endswith(b"\n"))
1499 self.assertEqual(b"".join(all), expected)
1501 def test_read1(self):
1502 resp = self.resp
1505 self.assertLessEqual(len(res), 4)
1508 self._verify_readline(readliner.readline, self.lines_expected)
1510 def test_read1_unbounded(self):
1511 resp = self.resp
1518 self.assertEqual(b"".join(all), self.lines_expected)
1520 def test_read1_bounded(self):
1521 resp = self.resp
1527 self.assertLessEqual(len(data), 10)
1529 self.assertEqual(b"".join(all), self.lines_expected)
1531 def test_read1_0(self):
1532 self.assertEqual(self.resp.read1(0), b"")
1534 def test_peek_0(self):
1535 p = self.resp.peek(0)
1536 self.assertLessEqual(0, len(p))
1565 def __init__(self, readfunc):
1566 self.readfunc = readfunc
1567 self.remainder = b""
1569 def readline(self, limit):
1572 read = self.remainder
1582 read = self.readfunc()
1588 self.remainder = read[idx:]
1591 self.remainder = b"".join(data)
1596 def test_all(self):
1608 self.assertCountEqual(client.__all__, expected)
1610 def test_responses(self):
1611 self.assertEqual(client.responses[client.NOT_FOUND], "Not Found")
1613 def test_client_constants(self):
1676 with self.subTest(constant=const):
1677 self.assertTrue(hasattr(client, const))
1681 def setUp(self):
1682 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1683 self.port = socket_helper.bind_port(self.serv)
1684 self.source_port = socket_helper.find_unused_port()
1685 self.serv.listen()
1686 self.conn = None
1688 def tearDown(self):
1689 if self.conn:
1690 self.conn.close()
1691 self.conn = None
1692 self.serv.close()
1693 self.serv = None
1695 def testHTTPConnectionSourceAddress(self):
1696 self.conn = client.HTTPConnection(HOST, self.port,
1697 source_address=('', self.source_port))
1698 self.conn.connect()
1699 self.assertEqual(self.conn.sock.getsockname()[1], self.source_port)
1703 def testHTTPSConnectionSourceAddress(self):
1704 self.conn = client.HTTPSConnection(HOST, self.port,
1705 source_address=('', self.source_port))
1714 def setUp(self):
1715 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1716 TimeoutTest.PORT = socket_helper.bind_port(self.serv)
1717 self.serv.listen()
1719 def tearDown(self):
1720 self.serv.close()
1721 self.serv = None
1723 def testTimeoutAttribute(self):
1728 self.assertIsNone(socket.getdefaulttimeout())
1735 self.assertEqual(httpConn.sock.gettimeout(), 30)
1739 self.assertIsNone(socket.getdefaulttimeout())
1747 self.assertEqual(httpConn.sock.gettimeout(), None)
1753 self.assertEqual(httpConn.sock.gettimeout(), 30)
1759 def test_reuse_reconnect(self):
1770 with self.subTest(version=version, header=header):
1779 self.assertIsNone(conn.sock)
1782 self.assertEqual(conn.sock is None, not reuse)
1784 self.assertEqual(conn.sock is None, not reuse)
1785 self.assertEqual(conn.connections, 1)
1787 self.assertEqual(conn.connections, 1 if reuse else 2)
1789 def test_disconnected(self):
1807 with self.subTest(exception=exception):
1810 self.assertRaises(exception, conn.getresponse)
1811 self.assertIsNone(conn.sock)
1814 self.assertEqual(conn.connections, 2)
1816 def test_100_close(self):
1823 self.assertRaises(client.RemoteDisconnected, conn.getresponse)
1824 self.assertIsNone(conn.sock)
1826 self.assertEqual(conn.connections, 2)
1831 def setUp(self):
1833 self.skipTest('ssl support required')
1835 def make_server(self, certfile):
1837 return make_https_server(self, certfile=certfile)
1839 def test_attributes(self):
1842 self.assertEqual(h.timeout, 30)
1844 def test_networked(self):
1848 with socket_helper.transient_internet('self-signed.pythontest.net'):
1849 h = client.HTTPSConnection('self-signed.pythontest.net', 443)
1850 with self.assertRaises(ssl.SSLError) as exc_info:
1852 self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
1854 def test_networked_noverification(self):
1858 with socket_helper.transient_internet('self-signed.pythontest.net'):
1860 h = client.HTTPSConnection('self-signed.pythontest.net', 443,
1865 self.assertIn('nginx', resp.getheader('server'))
1869 def test_networked_trusted_by_default_cert(self):
1879 self.assertIn('text/html', content_type)
1881 def test_networked_good_cert(self):
1885 selfsigned_pythontestdotnet = 'self-signed.pythontest.net'
1888 self.assertEqual(context.verify_mode, ssl.CERT_REQUIRED)
1889 self.assertEqual(context.check_hostname, True)
1902 # key size on self-signed.pythontest.net.
1912 self.assertIn('nginx', server_string)
1914 def test_networked_bad_cert(self):
1918 with socket_helper.transient_internet('self-signed.pythontest.net'):
1921 h = client.HTTPSConnection('self-signed.pythontest.net', 443, context=context)
1922 with self.assertRaises(ssl.SSLError) as exc_info:
1924 self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
1926 def test_local_unknown_cert(self):
1929 server = self.make_server(CERT_localhost)
1931 with self.assertRaises(ssl.SSLError) as exc_info:
1933 self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
1935 def test_local_good_hostname(self):
1938 server = self.make_server(CERT_localhost)
1942 self.addCleanup(h.close)
1945 self.addCleanup(resp.close)
1946 self.assertEqual(resp.status, 404)
1948 def test_local_bad_hostname(self):
1951 server = self.make_server(CERT_fakehostname)
1955 with self.assertRaises(ssl.CertificateError):
1961 with self.assertRaises(ssl.CertificateError):
1972 self.assertEqual(resp.status, 404)
1979 self.assertEqual(resp.status, 404)
1987 with self.assertRaises(ssl.CertificateError):
1992 def test_host_port(self):
1996 self.assertRaises(client.InvalidURL, client.HTTPSConnection, hp)
2007 self.assertEqual(h, c.host)
2008 self.assertEqual(p, c.port)
2010 def test_tls13_pha(self):
2013 self.skipTest('TLS 1.3 support required')
2016 self.assertTrue(h._context.post_handshake_auth)
2019 self.assertFalse(context.post_handshake_auth)
2021 self.assertIs(h._context, context)
2022 self.assertFalse(h._context.post_handshake_auth)
2029 self.assertTrue(h._context.post_handshake_auth)
2035 def setUp(self):
2036 self.conn = client.HTTPConnection('example.com')
2037 self.conn.sock = self.sock = FakeSocket("")
2038 self.conn.sock = self.sock
2040 def get_headers_and_fp(self):
2041 f = io.BytesIO(self.sock.data)
2046 def test_list_body(self):
2055 with self.subTest(body):
2056 self.conn = client.HTTPConnection('example.com')
2057 self.conn.sock = self.sock = FakeSocket('')
2059 self.conn.request('PUT', '/url', body)
2060 msg, f = self.get_headers_and_fp()
2061 self.assertNotIn('Content-Type', msg)
2062 self.assertNotIn('Content-Length', msg)
2063 self.assertEqual(msg.get('Transfer-Encoding'), 'chunked')
2064 self.assertEqual(expected, f.read())
2066 def test_manual_content_length(self):
2069 self.conn.request("PUT", "/url", "body",
2071 message, f = self.get_headers_and_fp()
2072 self.assertEqual("42", message.get("content-length"))
2073 self.assertEqual(4, len(f.read()))
2075 def test_ascii_body(self):
2076 self.conn.request("PUT", "/url", "body")
2077 message, f = self.get_headers_and_fp()
2078 self.assertEqual("text/plain", message.get_content_type())
2079 self.assertIsNone(message.get_charset())
2080 self.assertEqual("4", message.get("content-length"))
2081 self.assertEqual(b'body', f.read())
2083 def test_latin1_body(self):
2084 self.conn.request("PUT", "/url", "body\xc1")
2085 message, f = self.get_headers_and_fp()
2086 self.assertEqual("text/plain", message.get_content_type())
2087 self.assertIsNone(message.get_charset())
2088 self.assertEqual("5", message.get("content-length"))
2089 self.assertEqual(b'body\xc1', f.read())
2091 def test_bytes_body(self):
2092 self.conn.request("PUT", "/url", b"body\xc1")
2093 message, f = self.get_headers_and_fp()
2094 self.assertEqual("text/plain", message.get_content_type())
2095 self.assertIsNone(message.get_charset())
2096 self.assertEqual("5", message.get("content-length"))
2097 self.assertEqual(b'body\xc1', f.read())
2099 def test_text_file_body(self):
2100 self.addCleanup(os_helper.unlink, os_helper.TESTFN)
2104 self.conn.request("PUT", "/url", f)
2105 message, f = self.get_headers_and_fp()
2106 self.assertEqual("text/plain", message.get_content_type())
2107 self.assertIsNone(message.get_charset())
2110 self.assertIsNone(message.get("content-length"))
2111 self.assertEqual("chunked", message.get("transfer-encoding"))
2112 self.assertEqual(b'4\r\nbody\r\n0\r\n\r\n', f.read())
2114 def test_binary_file_body(self):
2115 self.addCleanup(os_helper.unlink, os_helper.TESTFN)
2119 self.conn.request("PUT", "/url", f)
2120 message, f = self.get_headers_and_fp()
2121 self.assertEqual("text/plain", message.get_content_type())
2122 self.assertIsNone(message.get_charset())
2123 self.assertEqual("chunked", message.get("Transfer-Encoding"))
2124 self.assertNotIn("Content-Length", message)
2125 self.assertEqual(b'5\r\nbody\xc1\r\n0\r\n\r\n', f.read())
2130 def setUp(self):
2134 self.resp = client.HTTPResponse(sock)
2135 self.resp.begin()
2137 def test_getting_header(self):
2138 header = self.resp.getheader('My-Header')
2139 self.assertEqual(header, 'first-value, second-value')
2141 header = self.resp.getheader('My-Header', 'some default')
2142 self.assertEqual(header, 'first-value, second-value')
2144 def test_getting_nonexistent_header_with_string_default(self):
2145 header = self.resp.getheader('No-Such-Header', 'default-value')
2146 self.assertEqual(header, 'default-value')
2148 def test_getting_nonexistent_header_with_iterable_default(self):
2149 header = self.resp.getheader('No-Such-Header', ['default', 'values'])
2150 self.assertEqual(header, 'default, values')
2152 header = self.resp.getheader('No-Such-Header', ('default', 'values'))
2153 self.assertEqual(header, 'default, values')
2155 def test_getting_nonexistent_header_without_default(self):
2156 header = self.resp.getheader('No-Such-Header')
2157 self.assertEqual(header, None)
2159 def test_getting_header_defaultint(self):
2160 header = self.resp.getheader('No-Such-Header',default=42)
2161 self.assertEqual(header, 42)
2164 def setUp(self):
2170 self.host = 'proxy.com'
2171 self.conn = client.HTTPConnection(self.host)
2172 self.conn._create_connection = self._create_connection(response_text)
2174 def tearDown(self):
2175 self.conn.close()
2177 def _create_connection(self, response_text):
2182 def test_set_tunnel_host_port_headers(self):
2186 self.conn.set_tunnel(tunnel_host, port=tunnel_port,
2188 self.conn.request('HEAD', '/', '')
2189 self.assertEqual(self.conn.sock.host, self.host)
2190 self.assertEqual(self.conn.sock.port, client.HTTP_PORT)
2191 self.assertEqual(self.conn._tunnel_host, tunnel_host)
2192 self.assertEqual(self.conn._tunnel_port, tunnel_port)
2193 self.assertEqual(self.conn._tunnel_headers, tunnel_headers)
2195 def test_disallow_set_tunnel_after_connect(self):
2197 self.conn.connect()
2198 self.assertRaises(RuntimeError, self.conn.set_tunnel,
2201 def test_connect_with_tunnel(self):
2202 self.conn.set_tunnel('destination.com')
2203 self.conn.request('HEAD', '/', '')
2204 self.assertEqual(self.conn.sock.host, self.host)
2205 self.assertEqual(self.conn.sock.port, client.HTTP_PORT)
2206 self.assertIn(b'CONNECT destination.com', self.conn.sock.data)
2208 self.assertNotIn(b'Host: destination.com:None', self.conn.sock.data)
2209 self.assertIn(b'Host: destination.com', self.conn.sock.data)
2212 self.assertNotIn(b'Host: proxy.com', self.conn.sock.data)
2214 def test_tunnel_connect_single_send_connection_setup(self):
2216 with mock.patch.object(self.conn, 'send') as mock_send:
2217 self.conn.set_tunnel('destination.com')
2218 self.conn.connect()
2219 self.conn.request('GET', '/')
2222 self.assertGreater(
2226 self.assertIn(b'CONNECT destination.com', proxy_setup_data_sent)
2227 self.assertTrue(
2231 def test_connect_put_request(self):
2232 self.conn.set_tunnel('destination.com')
2233 self.conn.request('PUT', '/', '')
2234 self.assertEqual(self.conn.sock.host, self.host)
2235 self.assertEqual(self.conn.sock.port, client.HTTP_PORT)
2236 self.assertIn(b'CONNECT destination.com', self.conn.sock.data)
2237 self.assertIn(b'Host: destination.com', self.conn.sock.data)
2239 def test_tunnel_debuglog(self):
2243 self.conn.set_debuglevel(1)
2244 self.conn._create_connection = self._create_connection(response_text)
2245 self.conn.set_tunnel('destination.com')
2248 self.conn.request('PUT', '/', '')
2250 self.assertIn('header: {}'.format(expected_header), lines)
2252 def test_tunnel_leak(self):
2264 self.conn._create_connection = _create_connection
2265 self.conn.set_tunnel('destination.com')
2268 self.conn.request('HEAD', '/', '')
2272 self.assertIsNotNone(exc)
2273 self.assertTrue(sock.file_closed)