ยปCore Development>Code coverage>Lib/test/test_httplib.py

Python code coverage for Lib/test/test_httplib.py

#countcontent
1n/aimport errno
2n/afrom http import client
3n/aimport io
4n/aimport itertools
5n/aimport os
6n/aimport array
7n/aimport socket
8n/a
9n/aimport unittest
10n/aTestCase = unittest.TestCase
11n/a
12n/afrom test import support
13n/a
14n/ahere = os.path.dirname(__file__)
15n/a# Self-signed cert file for 'localhost'
16n/aCERT_localhost = os.path.join(here, 'keycert.pem')
17n/a# Self-signed cert file for 'fakehostname'
18n/aCERT_fakehostname = os.path.join(here, 'keycert2.pem')
19n/a# Self-signed cert file for self-signed.pythontest.net
20n/aCERT_selfsigned_pythontestdotnet = os.path.join(here, 'selfsigned_pythontestdotnet.pem')
21n/a
22n/a# constants for testing chunked encoding
23n/achunked_start = (
24n/a 'HTTP/1.1 200 OK\r\n'
25n/a 'Transfer-Encoding: chunked\r\n\r\n'
26n/a 'a\r\n'
27n/a 'hello worl\r\n'
28n/a '3\r\n'
29n/a 'd! \r\n'
30n/a '8\r\n'
31n/a 'and now \r\n'
32n/a '22\r\n'
33n/a 'for something completely different\r\n'
34n/a)
35n/achunked_expected = b'hello world! and now for something completely different'
36n/achunk_extension = ";foo=bar"
37n/alast_chunk = "0\r\n"
38n/alast_chunk_extended = "0" + chunk_extension + "\r\n"
39n/atrailers = "X-Dummy: foo\r\nX-Dumm2: bar\r\n"
40n/achunked_end = "\r\n"
41n/a
42n/aHOST = support.HOST
43n/a
44n/aclass FakeSocket:
45n/a def __init__(self, text, fileclass=io.BytesIO, host=None, port=None):
46n/a if isinstance(text, str):
47n/a text = text.encode("ascii")
48n/a self.text = text
49n/a self.fileclass = fileclass
50n/a self.data = b''
51n/a self.sendall_calls = 0
52n/a self.file_closed = False
53n/a self.host = host
54n/a self.port = port
55n/a
56n/a def sendall(self, data):
57n/a self.sendall_calls += 1
58n/a self.data += data
59n/a
60n/a def makefile(self, mode, bufsize=None):
61n/a if mode != 'r' and mode != 'rb':
62n/a raise client.UnimplementedFileMode()
63n/a # keep the file around so we can check how much was read from it
64n/a self.file = self.fileclass(self.text)
65n/a self.file.close = self.file_close #nerf close ()
66n/a return self.file
67n/a
68n/a def file_close(self):
69n/a self.file_closed = True
70n/a
71n/a def close(self):
72n/a pass
73n/a
74n/a def setsockopt(self, level, optname, value):
75n/a pass
76n/a
77n/aclass EPipeSocket(FakeSocket):
78n/a
79n/a def __init__(self, text, pipe_trigger):
80n/a # When sendall() is called with pipe_trigger, raise EPIPE.
81n/a FakeSocket.__init__(self, text)
82n/a self.pipe_trigger = pipe_trigger
83n/a
84n/a def sendall(self, data):
85n/a if self.pipe_trigger in data:
86n/a raise OSError(errno.EPIPE, "gotcha")
87n/a self.data += data
88n/a
89n/a def close(self):
90n/a pass
91n/a
92n/aclass NoEOFBytesIO(io.BytesIO):
93n/a """Like BytesIO, but raises AssertionError on EOF.
94n/a
95n/a This is used below to test that http.client doesn't try to read
96n/a more from the underlying file than it should.
97n/a """
98n/a def read(self, n=-1):
99n/a data = io.BytesIO.read(self, n)
100n/a if data == b'':
101n/a raise AssertionError('caller tried to read past EOF')
102n/a return data
103n/a
104n/a def readline(self, length=None):
105n/a data = io.BytesIO.readline(self, length)
106n/a if data == b'':
107n/a raise AssertionError('caller tried to read past EOF')
108n/a return data
109n/a
110n/aclass FakeSocketHTTPConnection(client.HTTPConnection):
111n/a """HTTPConnection subclass using FakeSocket; counts connect() calls"""
112n/a
113n/a def __init__(self, *args):
114n/a self.connections = 0
115n/a super().__init__('example.com')
116n/a self.fake_socket_args = args
117n/a self._create_connection = self.create_connection
118n/a
119n/a def connect(self):
120n/a """Count the number of times connect() is invoked"""
121n/a self.connections += 1
122n/a return super().connect()
123n/a
124n/a def create_connection(self, *pos, **kw):
125n/a return FakeSocket(*self.fake_socket_args)
126n/a
127n/aclass HeaderTests(TestCase):
128n/a def test_auto_headers(self):
129n/a # Some headers are added automatically, but should not be added by
130n/a # .request() if they are explicitly set.
131n/a
132n/a class HeaderCountingBuffer(list):
133n/a def __init__(self):
134n/a self.count = {}
135n/a def append(self, item):
136n/a kv = item.split(b':')
137n/a if len(kv) > 1:
138n/a # item is a 'Key: Value' header string
139n/a lcKey = kv[0].decode('ascii').lower()
140n/a self.count.setdefault(lcKey, 0)
141n/a self.count[lcKey] += 1
142n/a list.append(self, item)
143n/a
144n/a for explicit_header in True, False:
145n/a for header in 'Content-length', 'Host', 'Accept-encoding':
146n/a conn = client.HTTPConnection('example.com')
147n/a conn.sock = FakeSocket('blahblahblah')
148n/a conn._buffer = HeaderCountingBuffer()
149n/a
150n/a body = 'spamspamspam'
151n/a headers = {}
152n/a if explicit_header:
153n/a headers[header] = str(len(body))
154n/a conn.request('POST', '/', body, headers)
155n/a self.assertEqual(conn._buffer.count[header.lower()], 1)
156n/a
157n/a def test_content_length_0(self):
158n/a
159n/a class ContentLengthChecker(list):
160n/a def __init__(self):
161n/a list.__init__(self)
162n/a self.content_length = None
163n/a def append(self, item):
164n/a kv = item.split(b':', 1)
165n/a if len(kv) > 1 and kv[0].lower() == b'content-length':
166n/a self.content_length = kv[1].strip()
167n/a list.append(self, item)
168n/a
169n/a # Here, we're testing that methods expecting a body get a
170n/a # content-length set to zero if the body is empty (either None or '')
171n/a bodies = (None, '')
172n/a methods_with_body = ('PUT', 'POST', 'PATCH')
173n/a for method, body in itertools.product(methods_with_body, bodies):
174n/a conn = client.HTTPConnection('example.com')
175n/a conn.sock = FakeSocket(None)
176n/a conn._buffer = ContentLengthChecker()
177n/a conn.request(method, '/', body)
178n/a self.assertEqual(
179n/a conn._buffer.content_length, b'0',
180n/a 'Header Content-Length incorrect on {}'.format(method)
181n/a )
182n/a
183n/a # For these methods, we make sure that content-length is not set when
184n/a # the body is None because it might cause unexpected behaviour on the
185n/a # server.
186n/a methods_without_body = (
187n/a 'GET', 'CONNECT', 'DELETE', 'HEAD', 'OPTIONS', 'TRACE',
188n/a )
189n/a for method in methods_without_body:
190n/a conn = client.HTTPConnection('example.com')
191n/a conn.sock = FakeSocket(None)
192n/a conn._buffer = ContentLengthChecker()
193n/a conn.request(method, '/', None)
194n/a self.assertEqual(
195n/a conn._buffer.content_length, None,
196n/a 'Header Content-Length set for empty body on {}'.format(method)
197n/a )
198n/a
199n/a # If the body is set to '', that's considered to be "present but
200n/a # empty" rather than "missing", so content length would be set, even
201n/a # for methods that don't expect a body.
202n/a for method in methods_without_body:
203n/a conn = client.HTTPConnection('example.com')
204n/a conn.sock = FakeSocket(None)
205n/a conn._buffer = ContentLengthChecker()
206n/a conn.request(method, '/', '')
207n/a self.assertEqual(
208n/a conn._buffer.content_length, b'0',
209n/a 'Header Content-Length incorrect on {}'.format(method)
210n/a )
211n/a
212n/a # If the body is set, make sure Content-Length is set.
213n/a for method in itertools.chain(methods_without_body, methods_with_body):
214n/a conn = client.HTTPConnection('example.com')
215n/a conn.sock = FakeSocket(None)
216n/a conn._buffer = ContentLengthChecker()
217n/a conn.request(method, '/', ' ')
218n/a self.assertEqual(
219n/a conn._buffer.content_length, b'1',
220n/a 'Header Content-Length incorrect on {}'.format(method)
221n/a )
222n/a
223n/a def test_putheader(self):
224n/a conn = client.HTTPConnection('example.com')
225n/a conn.sock = FakeSocket(None)
226n/a conn.putrequest('GET','/')
227n/a conn.putheader('Content-length', 42)
228n/a self.assertIn(b'Content-length: 42', conn._buffer)
229n/a
230n/a conn.putheader('Foo', ' bar ')
231n/a self.assertIn(b'Foo: bar ', conn._buffer)
232n/a conn.putheader('Bar', '\tbaz\t')
233n/a self.assertIn(b'Bar: \tbaz\t', conn._buffer)
234n/a conn.putheader('Authorization', 'Bearer mytoken')
235n/a self.assertIn(b'Authorization: Bearer mytoken', conn._buffer)
236n/a conn.putheader('IterHeader', 'IterA', 'IterB')
237n/a self.assertIn(b'IterHeader: IterA\r\n\tIterB', conn._buffer)
238n/a conn.putheader('LatinHeader', b'\xFF')
239n/a self.assertIn(b'LatinHeader: \xFF', conn._buffer)
240n/a conn.putheader('Utf8Header', b'\xc3\x80')
241n/a self.assertIn(b'Utf8Header: \xc3\x80', conn._buffer)
242n/a conn.putheader('C1-Control', b'next\x85line')
243n/a self.assertIn(b'C1-Control: next\x85line', conn._buffer)
244n/a conn.putheader('Embedded-Fold-Space', 'is\r\n allowed')
245n/a self.assertIn(b'Embedded-Fold-Space: is\r\n allowed', conn._buffer)
246n/a conn.putheader('Embedded-Fold-Tab', 'is\r\n\tallowed')
247n/a self.assertIn(b'Embedded-Fold-Tab: is\r\n\tallowed', conn._buffer)
248n/a conn.putheader('Key Space', 'value')
249n/a self.assertIn(b'Key Space: value', conn._buffer)
250n/a conn.putheader('KeySpace ', 'value')
251n/a self.assertIn(b'KeySpace : value', conn._buffer)
252n/a conn.putheader(b'Nonbreak\xa0Space', 'value')
253n/a self.assertIn(b'Nonbreak\xa0Space: value', conn._buffer)
254n/a conn.putheader(b'\xa0NonbreakSpace', 'value')
255n/a self.assertIn(b'\xa0NonbreakSpace: value', conn._buffer)
256n/a
257n/a def test_ipv6host_header(self):
258n/a # Default host header on IPv6 transaction should be wrapped by [] if
259n/a # it is an IPv6 address
260n/a expected = b'GET /foo HTTP/1.1\r\nHost: [2001::]:81\r\n' \
261n/a b'Accept-Encoding: identity\r\n\r\n'
262n/a conn = client.HTTPConnection('[2001::]:81')
263n/a sock = FakeSocket('')
264n/a conn.sock = sock
265n/a conn.request('GET', '/foo')
266n/a self.assertTrue(sock.data.startswith(expected))
267n/a
268n/a expected = b'GET /foo HTTP/1.1\r\nHost: [2001:102A::]\r\n' \
269n/a b'Accept-Encoding: identity\r\n\r\n'
270n/a conn = client.HTTPConnection('[2001:102A::]')
271n/a sock = FakeSocket('')
272n/a conn.sock = sock
273n/a conn.request('GET', '/foo')
274n/a self.assertTrue(sock.data.startswith(expected))
275n/a
276n/a def test_malformed_headers_coped_with(self):
277n/a # Issue 19996
278n/a body = "HTTP/1.1 200 OK\r\nFirst: val\r\n: nval\r\nSecond: val\r\n\r\n"
279n/a sock = FakeSocket(body)
280n/a resp = client.HTTPResponse(sock)
281n/a resp.begin()
282n/a
283n/a self.assertEqual(resp.getheader('First'), 'val')
284n/a self.assertEqual(resp.getheader('Second'), 'val')
285n/a
286n/a def test_parse_all_octets(self):
287n/a # Ensure no valid header field octet breaks the parser
288n/a body = (
289n/a b'HTTP/1.1 200 OK\r\n'
290n/a b"!#$%&'*+-.^_`|~: value\r\n" # Special token characters
291n/a b'VCHAR: ' + bytes(range(0x21, 0x7E + 1)) + b'\r\n'
292n/a b'obs-text: ' + bytes(range(0x80, 0xFF + 1)) + b'\r\n'
293n/a b'obs-fold: text\r\n'
294n/a b' folded with space\r\n'
295n/a b'\tfolded with tab\r\n'
296n/a b'Content-Length: 0\r\n'
297n/a b'\r\n'
298n/a )
299n/a sock = FakeSocket(body)
300n/a resp = client.HTTPResponse(sock)
301n/a resp.begin()
302n/a self.assertEqual(resp.getheader('Content-Length'), '0')
303n/a self.assertEqual(resp.msg['Content-Length'], '0')
304n/a self.assertEqual(resp.getheader("!#$%&'*+-.^_`|~"), 'value')
305n/a self.assertEqual(resp.msg["!#$%&'*+-.^_`|~"], 'value')
306n/a vchar = ''.join(map(chr, range(0x21, 0x7E + 1)))
307n/a self.assertEqual(resp.getheader('VCHAR'), vchar)
308n/a self.assertEqual(resp.msg['VCHAR'], vchar)
309n/a self.assertIsNotNone(resp.getheader('obs-text'))
310n/a self.assertIn('obs-text', resp.msg)
311n/a for folded in (resp.getheader('obs-fold'), resp.msg['obs-fold']):
312n/a self.assertTrue(folded.startswith('text'))
313n/a self.assertIn(' folded with space', folded)
314n/a self.assertTrue(folded.endswith('folded with tab'))
315n/a
316n/a def test_invalid_headers(self):
317n/a conn = client.HTTPConnection('example.com')
318n/a conn.sock = FakeSocket('')
319n/a conn.putrequest('GET', '/')
320n/a
321n/a # http://tools.ietf.org/html/rfc7230#section-3.2.4, whitespace is no
322n/a # longer allowed in header names
323n/a cases = (
324n/a (b'Invalid\r\nName', b'ValidValue'),
325n/a (b'Invalid\rName', b'ValidValue'),
326n/a (b'Invalid\nName', b'ValidValue'),
327n/a (b'\r\nInvalidName', b'ValidValue'),
328n/a (b'\rInvalidName', b'ValidValue'),
329n/a (b'\nInvalidName', b'ValidValue'),
330n/a (b' InvalidName', b'ValidValue'),
331n/a (b'\tInvalidName', b'ValidValue'),
332n/a (b'Invalid:Name', b'ValidValue'),
333n/a (b':InvalidName', b'ValidValue'),
334n/a (b'ValidName', b'Invalid\r\nValue'),
335n/a (b'ValidName', b'Invalid\rValue'),
336n/a (b'ValidName', b'Invalid\nValue'),
337n/a (b'ValidName', b'InvalidValue\r\n'),
338n/a (b'ValidName', b'InvalidValue\r'),
339n/a (b'ValidName', b'InvalidValue\n'),
340n/a )
341n/a for name, value in cases:
342n/a with self.subTest((name, value)):
343n/a with self.assertRaisesRegex(ValueError, 'Invalid header'):
344n/a conn.putheader(name, value)
345n/a
346n/a
347n/aclass TransferEncodingTest(TestCase):
348n/a expected_body = b"It's just a flesh wound"
349n/a
350n/a def test_endheaders_chunked(self):
351n/a conn = client.HTTPConnection('example.com')
352n/a conn.sock = FakeSocket(b'')
353n/a conn.putrequest('POST', '/')
354n/a conn.endheaders(self._make_body(), encode_chunked=True)
355n/a
356n/a _, _, body = self._parse_request(conn.sock.data)
357n/a body = self._parse_chunked(body)
358n/a self.assertEqual(body, self.expected_body)
359n/a
360n/a def test_explicit_headers(self):
361n/a # explicit chunked
362n/a conn = client.HTTPConnection('example.com')
363n/a conn.sock = FakeSocket(b'')
364n/a # this shouldn't actually be automatically chunk-encoded because the
365n/a # calling code has explicitly stated that it's taking care of it
366n/a conn.request(
367n/a 'POST', '/', self._make_body(), {'Transfer-Encoding': 'chunked'})
368n/a
369n/a _, headers, body = self._parse_request(conn.sock.data)
370n/a self.assertNotIn('content-length', [k.lower() for k in headers.keys()])
371n/a self.assertEqual(headers['Transfer-Encoding'], 'chunked')
372n/a self.assertEqual(body, self.expected_body)
373n/a
374n/a # explicit chunked, string body
375n/a conn = client.HTTPConnection('example.com')
376n/a conn.sock = FakeSocket(b'')
377n/a conn.request(
378n/a 'POST', '/', self.expected_body.decode('latin-1'),
379n/a {'Transfer-Encoding': 'chunked'})
380n/a
381n/a _, headers, body = self._parse_request(conn.sock.data)
382n/a self.assertNotIn('content-length', [k.lower() for k in headers.keys()])
383n/a self.assertEqual(headers['Transfer-Encoding'], 'chunked')
384n/a self.assertEqual(body, self.expected_body)
385n/a
386n/a # User-specified TE, but request() does the chunk encoding
387n/a conn = client.HTTPConnection('example.com')
388n/a conn.sock = FakeSocket(b'')
389n/a conn.request('POST', '/',
390n/a headers={'Transfer-Encoding': 'gzip, chunked'},
391n/a encode_chunked=True,
392n/a body=self._make_body())
393n/a _, headers, body = self._parse_request(conn.sock.data)
394n/a self.assertNotIn('content-length', [k.lower() for k in headers])
395n/a self.assertEqual(headers['Transfer-Encoding'], 'gzip, chunked')
396n/a self.assertEqual(self._parse_chunked(body), self.expected_body)
397n/a
398n/a def test_request(self):
399n/a for empty_lines in (False, True,):
400n/a conn = client.HTTPConnection('example.com')
401n/a conn.sock = FakeSocket(b'')
402n/a conn.request(
403n/a 'POST', '/', self._make_body(empty_lines=empty_lines))
404n/a
405n/a _, headers, body = self._parse_request(conn.sock.data)
406n/a body = self._parse_chunked(body)
407n/a self.assertEqual(body, self.expected_body)
408n/a self.assertEqual(headers['Transfer-Encoding'], 'chunked')
409n/a
410n/a # Content-Length and Transfer-Encoding SHOULD not be sent in the
411n/a # same request
412n/a self.assertNotIn('content-length', [k.lower() for k in headers])
413n/a
414n/a def test_empty_body(self):
415n/a # Zero-length iterable should be treated like any other iterable
416n/a conn = client.HTTPConnection('example.com')
417n/a conn.sock = FakeSocket(b'')
418n/a conn.request('POST', '/', ())
419n/a _, headers, body = self._parse_request(conn.sock.data)
420n/a self.assertEqual(headers['Transfer-Encoding'], 'chunked')
421n/a self.assertNotIn('content-length', [k.lower() for k in headers])
422n/a self.assertEqual(body, b"0\r\n\r\n")
423n/a
424n/a def _make_body(self, empty_lines=False):
425n/a lines = self.expected_body.split(b' ')
426n/a for idx, line in enumerate(lines):
427n/a # for testing handling empty lines
428n/a if empty_lines and idx % 2:
429n/a yield b''
430n/a if idx < len(lines) - 1:
431n/a yield line + b' '
432n/a else:
433n/a yield line
434n/a
435n/a def _parse_request(self, data):
436n/a lines = data.split(b'\r\n')
437n/a request = lines[0]
438n/a headers = {}
439n/a n = 1
440n/a while n < len(lines) and len(lines[n]) > 0:
441n/a key, val = lines[n].split(b':')
442n/a key = key.decode('latin-1').strip()
443n/a headers[key] = val.decode('latin-1').strip()
444n/a n += 1
445n/a
446n/a return request, headers, b'\r\n'.join(lines[n + 1:])
447n/a
448n/a def _parse_chunked(self, data):
449n/a body = []
450n/a trailers = {}
451n/a n = 0
452n/a lines = data.split(b'\r\n')
453n/a # parse body
454n/a while True:
455n/a size, chunk = lines[n:n+2]
456n/a size = int(size, 16)
457n/a
458n/a if size == 0:
459n/a n += 1
460n/a break
461n/a
462n/a self.assertEqual(size, len(chunk))
463n/a body.append(chunk)
464n/a
465n/a n += 2
466n/a # we /should/ hit the end chunk, but check against the size of
467n/a # lines so we're not stuck in an infinite loop should we get
468n/a # malformed data
469n/a if n > len(lines):
470n/a break
471n/a
472n/a return b''.join(body)
473n/a
474n/a
475n/aclass BasicTest(TestCase):
476n/a def test_status_lines(self):
477n/a # Test HTTP status lines
478n/a
479n/a body = "HTTP/1.1 200 Ok\r\n\r\nText"
480n/a sock = FakeSocket(body)
481n/a resp = client.HTTPResponse(sock)
482n/a resp.begin()
483n/a self.assertEqual(resp.read(0), b'') # Issue #20007
484n/a self.assertFalse(resp.isclosed())
485n/a self.assertFalse(resp.closed)
486n/a self.assertEqual(resp.read(), b"Text")
487n/a self.assertTrue(resp.isclosed())
488n/a self.assertFalse(resp.closed)
489n/a resp.close()
490n/a self.assertTrue(resp.closed)
491n/a
492n/a body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText"
493n/a sock = FakeSocket(body)
494n/a resp = client.HTTPResponse(sock)
495n/a self.assertRaises(client.BadStatusLine, resp.begin)
496n/a
497n/a def test_bad_status_repr(self):
498n/a exc = client.BadStatusLine('')
499n/a self.assertEqual(repr(exc), '''BadStatusLine("\'\'",)''')
500n/a
501n/a def test_partial_reads(self):
502n/a # if we have Content-Length, HTTPResponse knows when to close itself,
503n/a # the same behaviour as when we read the whole thing with read()
504n/a body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText"
505n/a sock = FakeSocket(body)
506n/a resp = client.HTTPResponse(sock)
507n/a resp.begin()
508n/a self.assertEqual(resp.read(2), b'Te')
509n/a self.assertFalse(resp.isclosed())
510n/a self.assertEqual(resp.read(2), b'xt')
511n/a self.assertTrue(resp.isclosed())
512n/a self.assertFalse(resp.closed)
513n/a resp.close()
514n/a self.assertTrue(resp.closed)
515n/a
516n/a def test_mixed_reads(self):
517n/a # readline() should update the remaining length, so that read() knows
518n/a # how much data is left and does not raise IncompleteRead
519n/a body = "HTTP/1.1 200 Ok\r\nContent-Length: 13\r\n\r\nText\r\nAnother"
520n/a sock = FakeSocket(body)
521n/a resp = client.HTTPResponse(sock)
522n/a resp.begin()
523n/a self.assertEqual(resp.readline(), b'Text\r\n')
524n/a self.assertFalse(resp.isclosed())
525n/a self.assertEqual(resp.read(), b'Another')
526n/a self.assertTrue(resp.isclosed())
527n/a self.assertFalse(resp.closed)
528n/a resp.close()
529n/a self.assertTrue(resp.closed)
530n/a
531n/a def test_partial_readintos(self):
532n/a # if we have Content-Length, HTTPResponse knows when to close itself,
533n/a # the same behaviour as when we read the whole thing with read()
534n/a body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText"
535n/a sock = FakeSocket(body)
536n/a resp = client.HTTPResponse(sock)
537n/a resp.begin()
538n/a b = bytearray(2)
539n/a n = resp.readinto(b)
540n/a self.assertEqual(n, 2)
541n/a self.assertEqual(bytes(b), b'Te')
542n/a self.assertFalse(resp.isclosed())
543n/a n = resp.readinto(b)
544n/a self.assertEqual(n, 2)
545n/a self.assertEqual(bytes(b), b'xt')
546n/a self.assertTrue(resp.isclosed())
547n/a self.assertFalse(resp.closed)
548n/a resp.close()
549n/a self.assertTrue(resp.closed)
550n/a
551n/a def test_partial_reads_no_content_length(self):
552n/a # when no length is present, the socket should be gracefully closed when
553n/a # all data was read
554n/a body = "HTTP/1.1 200 Ok\r\n\r\nText"
555n/a sock = FakeSocket(body)
556n/a resp = client.HTTPResponse(sock)
557n/a resp.begin()
558n/a self.assertEqual(resp.read(2), b'Te')
559n/a self.assertFalse(resp.isclosed())
560n/a self.assertEqual(resp.read(2), b'xt')
561n/a self.assertEqual(resp.read(1), b'')
562n/a self.assertTrue(resp.isclosed())
563n/a self.assertFalse(resp.closed)
564n/a resp.close()
565n/a self.assertTrue(resp.closed)
566n/a
567n/a def test_partial_readintos_no_content_length(self):
568n/a # when no length is present, the socket should be gracefully closed when
569n/a # all data was read
570n/a body = "HTTP/1.1 200 Ok\r\n\r\nText"
571n/a sock = FakeSocket(body)
572n/a resp = client.HTTPResponse(sock)
573n/a resp.begin()
574n/a b = bytearray(2)
575n/a n = resp.readinto(b)
576n/a self.assertEqual(n, 2)
577n/a self.assertEqual(bytes(b), b'Te')
578n/a self.assertFalse(resp.isclosed())
579n/a n = resp.readinto(b)
580n/a self.assertEqual(n, 2)
581n/a self.assertEqual(bytes(b), b'xt')
582n/a n = resp.readinto(b)
583n/a self.assertEqual(n, 0)
584n/a self.assertTrue(resp.isclosed())
585n/a
586n/a def test_partial_reads_incomplete_body(self):
587n/a # if the server shuts down the connection before the whole
588n/a # content-length is delivered, the socket is gracefully closed
589n/a body = "HTTP/1.1 200 Ok\r\nContent-Length: 10\r\n\r\nText"
590n/a sock = FakeSocket(body)
591n/a resp = client.HTTPResponse(sock)
592n/a resp.begin()
593n/a self.assertEqual(resp.read(2), b'Te')
594n/a self.assertFalse(resp.isclosed())
595n/a self.assertEqual(resp.read(2), b'xt')
596n/a self.assertEqual(resp.read(1), b'')
597n/a self.assertTrue(resp.isclosed())
598n/a
599n/a def test_partial_readintos_incomplete_body(self):
600n/a # if the server shuts down the connection before the whole
601n/a # content-length is delivered, the socket is gracefully closed
602n/a body = "HTTP/1.1 200 Ok\r\nContent-Length: 10\r\n\r\nText"
603n/a sock = FakeSocket(body)
604n/a resp = client.HTTPResponse(sock)
605n/a resp.begin()
606n/a b = bytearray(2)
607n/a n = resp.readinto(b)
608n/a self.assertEqual(n, 2)
609n/a self.assertEqual(bytes(b), b'Te')
610n/a self.assertFalse(resp.isclosed())
611n/a n = resp.readinto(b)
612n/a self.assertEqual(n, 2)
613n/a self.assertEqual(bytes(b), b'xt')
614n/a n = resp.readinto(b)
615n/a self.assertEqual(n, 0)
616n/a self.assertTrue(resp.isclosed())
617n/a self.assertFalse(resp.closed)
618n/a resp.close()
619n/a self.assertTrue(resp.closed)
620n/a
621n/a def test_host_port(self):
622n/a # Check invalid host_port
623n/a
624n/a for hp in ("www.python.org:abc", "user:password@www.python.org"):
625n/a self.assertRaises(client.InvalidURL, client.HTTPConnection, hp)
626n/a
627n/a for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000",
628n/a "fe80::207:e9ff:fe9b", 8000),
629n/a ("www.python.org:80", "www.python.org", 80),
630n/a ("www.python.org:", "www.python.org", 80),
631n/a ("www.python.org", "www.python.org", 80),
632n/a ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 80),
633n/a ("[fe80::207:e9ff:fe9b]:", "fe80::207:e9ff:fe9b", 80)):
634n/a c = client.HTTPConnection(hp)
635n/a self.assertEqual(h, c.host)
636n/a self.assertEqual(p, c.port)
637n/a
638n/a def test_response_headers(self):
639n/a # test response with multiple message headers with the same field name.
640n/a text = ('HTTP/1.1 200 OK\r\n'
641n/a 'Set-Cookie: Customer="WILE_E_COYOTE"; '
642n/a 'Version="1"; Path="/acme"\r\n'
643n/a 'Set-Cookie: Part_Number="Rocket_Launcher_0001"; Version="1";'
644n/a ' Path="/acme"\r\n'
645n/a '\r\n'
646n/a 'No body\r\n')
647n/a hdr = ('Customer="WILE_E_COYOTE"; Version="1"; Path="/acme"'
648n/a ', '
649n/a 'Part_Number="Rocket_Launcher_0001"; Version="1"; Path="/acme"')
650n/a s = FakeSocket(text)
651n/a r = client.HTTPResponse(s)
652n/a r.begin()
653n/a cookies = r.getheader("Set-Cookie")
654n/a self.assertEqual(cookies, hdr)
655n/a
656n/a def test_read_head(self):
657n/a # Test that the library doesn't attempt to read any data
658n/a # from a HEAD request. (Tickles SF bug #622042.)
659n/a sock = FakeSocket(
660n/a 'HTTP/1.1 200 OK\r\n'
661n/a 'Content-Length: 14432\r\n'
662n/a '\r\n',
663n/a NoEOFBytesIO)
664n/a resp = client.HTTPResponse(sock, method="HEAD")
665n/a resp.begin()
666n/a if resp.read():
667n/a self.fail("Did not expect response from HEAD request")
668n/a
669n/a def test_readinto_head(self):
670n/a # Test that the library doesn't attempt to read any data
671n/a # from a HEAD request. (Tickles SF bug #622042.)
672n/a sock = FakeSocket(
673n/a 'HTTP/1.1 200 OK\r\n'
674n/a 'Content-Length: 14432\r\n'
675n/a '\r\n',
676n/a NoEOFBytesIO)
677n/a resp = client.HTTPResponse(sock, method="HEAD")
678n/a resp.begin()
679n/a b = bytearray(5)
680n/a if resp.readinto(b) != 0:
681n/a self.fail("Did not expect response from HEAD request")
682n/a self.assertEqual(bytes(b), b'\x00'*5)
683n/a
684n/a def test_too_many_headers(self):
685n/a headers = '\r\n'.join('Header%d: foo' % i
686n/a for i in range(client._MAXHEADERS + 1)) + '\r\n'
687n/a text = ('HTTP/1.1 200 OK\r\n' + headers)
688n/a s = FakeSocket(text)
689n/a r = client.HTTPResponse(s)
690n/a self.assertRaisesRegex(client.HTTPException,
691n/a r"got more than \d+ headers", r.begin)
692n/a
693n/a def test_send_file(self):
694n/a expected = (b'GET /foo HTTP/1.1\r\nHost: example.com\r\n'
695n/a b'Accept-Encoding: identity\r\n'
696n/a b'Transfer-Encoding: chunked\r\n'
697n/a b'\r\n')
698n/a
699n/a with open(__file__, 'rb') as body:
700n/a conn = client.HTTPConnection('example.com')
701n/a sock = FakeSocket(body)
702n/a conn.sock = sock
703n/a conn.request('GET', '/foo', body)
704n/a self.assertTrue(sock.data.startswith(expected), '%r != %r' %
705n/a (sock.data[:len(expected)], expected))
706n/a
707n/a def test_send(self):
708n/a expected = b'this is a test this is only a test'
709n/a conn = client.HTTPConnection('example.com')
710n/a sock = FakeSocket(None)
711n/a conn.sock = sock
712n/a conn.send(expected)
713n/a self.assertEqual(expected, sock.data)
714n/a sock.data = b''
715n/a conn.send(array.array('b', expected))
716n/a self.assertEqual(expected, sock.data)
717n/a sock.data = b''
718n/a conn.send(io.BytesIO(expected))
719n/a self.assertEqual(expected, sock.data)
720n/a
721n/a def test_send_updating_file(self):
722n/a def data():
723n/a yield 'data'
724n/a yield None
725n/a yield 'data_two'
726n/a
727n/a class UpdatingFile(io.TextIOBase):
728n/a mode = 'r'
729n/a d = data()
730n/a def read(self, blocksize=-1):
731n/a return next(self.d)
732n/a
733n/a expected = b'data'
734n/a
735n/a conn = client.HTTPConnection('example.com')
736n/a sock = FakeSocket("")
737n/a conn.sock = sock
738n/a conn.send(UpdatingFile())
739n/a self.assertEqual(sock.data, expected)
740n/a
741n/a
742n/a def test_send_iter(self):
743n/a expected = b'GET /foo HTTP/1.1\r\nHost: example.com\r\n' \
744n/a b'Accept-Encoding: identity\r\nContent-Length: 11\r\n' \
745n/a b'\r\nonetwothree'
746n/a
747n/a def body():
748n/a yield b"one"
749n/a yield b"two"
750n/a yield b"three"
751n/a
752n/a conn = client.HTTPConnection('example.com')
753n/a sock = FakeSocket("")
754n/a conn.sock = sock
755n/a conn.request('GET', '/foo', body(), {'Content-Length': '11'})
756n/a self.assertEqual(sock.data, expected)
757n/a
758n/a def test_send_type_error(self):
759n/a # See: Issue #12676
760n/a conn = client.HTTPConnection('example.com')
761n/a conn.sock = FakeSocket('')
762n/a with self.assertRaises(TypeError):
763n/a conn.request('POST', 'test', conn)
764n/a
765n/a def test_chunked(self):
766n/a expected = chunked_expected
767n/a sock = FakeSocket(chunked_start + last_chunk + chunked_end)
768n/a resp = client.HTTPResponse(sock, method="GET")
769n/a resp.begin()
770n/a self.assertEqual(resp.read(), expected)
771n/a resp.close()
772n/a
773n/a # Various read sizes
774n/a for n in range(1, 12):
775n/a sock = FakeSocket(chunked_start + last_chunk + chunked_end)
776n/a resp = client.HTTPResponse(sock, method="GET")
777n/a resp.begin()
778n/a self.assertEqual(resp.read(n) + resp.read(n) + resp.read(), expected)
779n/a resp.close()
780n/a
781n/a for x in ('', 'foo\r\n'):
782n/a sock = FakeSocket(chunked_start + x)
783n/a resp = client.HTTPResponse(sock, method="GET")
784n/a resp.begin()
785n/a try:
786n/a resp.read()
787n/a except client.IncompleteRead as i:
788n/a self.assertEqual(i.partial, expected)
789n/a expected_message = 'IncompleteRead(%d bytes read)' % len(expected)
790n/a self.assertEqual(repr(i), expected_message)
791n/a self.assertEqual(str(i), expected_message)
792n/a else:
793n/a self.fail('IncompleteRead expected')
794n/a finally:
795n/a resp.close()
796n/a
797n/a def test_readinto_chunked(self):
798n/a
799n/a expected = chunked_expected
800n/a nexpected = len(expected)
801n/a b = bytearray(128)
802n/a
803n/a sock = FakeSocket(chunked_start + last_chunk + chunked_end)
804n/a resp = client.HTTPResponse(sock, method="GET")
805n/a resp.begin()
806n/a n = resp.readinto(b)
807n/a self.assertEqual(b[:nexpected], expected)
808n/a self.assertEqual(n, nexpected)
809n/a resp.close()
810n/a
811n/a # Various read sizes
812n/a for n in range(1, 12):
813n/a sock = FakeSocket(chunked_start + last_chunk + chunked_end)
814n/a resp = client.HTTPResponse(sock, method="GET")
815n/a resp.begin()
816n/a m = memoryview(b)
817n/a i = resp.readinto(m[0:n])
818n/a i += resp.readinto(m[i:n + i])
819n/a i += resp.readinto(m[i:])
820n/a self.assertEqual(b[:nexpected], expected)
821n/a self.assertEqual(i, nexpected)
822n/a resp.close()
823n/a
824n/a for x in ('', 'foo\r\n'):
825n/a sock = FakeSocket(chunked_start + x)
826n/a resp = client.HTTPResponse(sock, method="GET")
827n/a resp.begin()
828n/a try:
829n/a n = resp.readinto(b)
830n/a except client.IncompleteRead as i:
831n/a self.assertEqual(i.partial, expected)
832n/a expected_message = 'IncompleteRead(%d bytes read)' % len(expected)
833n/a self.assertEqual(repr(i), expected_message)
834n/a self.assertEqual(str(i), expected_message)
835n/a else:
836n/a self.fail('IncompleteRead expected')
837n/a finally:
838n/a resp.close()
839n/a
840n/a def test_chunked_head(self):
841n/a chunked_start = (
842n/a 'HTTP/1.1 200 OK\r\n'
843n/a 'Transfer-Encoding: chunked\r\n\r\n'
844n/a 'a\r\n'
845n/a 'hello world\r\n'
846n/a '1\r\n'
847n/a 'd\r\n'
848n/a )
849n/a sock = FakeSocket(chunked_start + last_chunk + chunked_end)
850n/a resp = client.HTTPResponse(sock, method="HEAD")
851n/a resp.begin()
852n/a self.assertEqual(resp.read(), b'')
853n/a self.assertEqual(resp.status, 200)
854n/a self.assertEqual(resp.reason, 'OK')
855n/a self.assertTrue(resp.isclosed())
856n/a self.assertFalse(resp.closed)
857n/a resp.close()
858n/a self.assertTrue(resp.closed)
859n/a
860n/a def test_readinto_chunked_head(self):
861n/a chunked_start = (
862n/a 'HTTP/1.1 200 OK\r\n'
863n/a 'Transfer-Encoding: chunked\r\n\r\n'
864n/a 'a\r\n'
865n/a 'hello world\r\n'
866n/a '1\r\n'
867n/a 'd\r\n'
868n/a )
869n/a sock = FakeSocket(chunked_start + last_chunk + chunked_end)
870n/a resp = client.HTTPResponse(sock, method="HEAD")
871n/a resp.begin()
872n/a b = bytearray(5)
873n/a n = resp.readinto(b)
874n/a self.assertEqual(n, 0)
875n/a self.assertEqual(bytes(b), b'\x00'*5)
876n/a self.assertEqual(resp.status, 200)
877n/a self.assertEqual(resp.reason, 'OK')
878n/a self.assertTrue(resp.isclosed())
879n/a self.assertFalse(resp.closed)
880n/a resp.close()
881n/a self.assertTrue(resp.closed)
882n/a
883n/a def test_negative_content_length(self):
884n/a sock = FakeSocket(
885n/a 'HTTP/1.1 200 OK\r\nContent-Length: -1\r\n\r\nHello\r\n')
886n/a resp = client.HTTPResponse(sock, method="GET")
887n/a resp.begin()
888n/a self.assertEqual(resp.read(), b'Hello\r\n')
889n/a self.assertTrue(resp.isclosed())
890n/a
891n/a def test_incomplete_read(self):
892n/a sock = FakeSocket('HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nHello\r\n')
893n/a resp = client.HTTPResponse(sock, method="GET")
894n/a resp.begin()
895n/a try:
896n/a resp.read()
897n/a except client.IncompleteRead as i:
898n/a self.assertEqual(i.partial, b'Hello\r\n')
899n/a self.assertEqual(repr(i),
900n/a "IncompleteRead(7 bytes read, 3 more expected)")
901n/a self.assertEqual(str(i),
902n/a "IncompleteRead(7 bytes read, 3 more expected)")
903n/a self.assertTrue(resp.isclosed())
904n/a else:
905n/a self.fail('IncompleteRead expected')
906n/a
907n/a def test_epipe(self):
908n/a sock = EPipeSocket(
909n/a "HTTP/1.0 401 Authorization Required\r\n"
910n/a "Content-type: text/html\r\n"
911n/a "WWW-Authenticate: Basic realm=\"example\"\r\n",
912n/a b"Content-Length")
913n/a conn = client.HTTPConnection("example.com")
914n/a conn.sock = sock
915n/a self.assertRaises(OSError,
916n/a lambda: conn.request("PUT", "/url", "body"))
917n/a resp = conn.getresponse()
918n/a self.assertEqual(401, resp.status)
919n/a self.assertEqual("Basic realm=\"example\"",
920n/a resp.getheader("www-authenticate"))
921n/a
922n/a # Test lines overflowing the max line size (_MAXLINE in http.client)
923n/a
924n/a def test_overflowing_status_line(self):
925n/a body = "HTTP/1.1 200 Ok" + "k" * 65536 + "\r\n"
926n/a resp = client.HTTPResponse(FakeSocket(body))
927n/a self.assertRaises((client.LineTooLong, client.BadStatusLine), resp.begin)
928n/a
929n/a def test_overflowing_header_line(self):
930n/a body = (
931n/a 'HTTP/1.1 200 OK\r\n'
932n/a 'X-Foo: bar' + 'r' * 65536 + '\r\n\r\n'
933n/a )
934n/a resp = client.HTTPResponse(FakeSocket(body))
935n/a self.assertRaises(client.LineTooLong, resp.begin)
936n/a
937n/a def test_overflowing_chunked_line(self):
938n/a body = (
939n/a 'HTTP/1.1 200 OK\r\n'
940n/a 'Transfer-Encoding: chunked\r\n\r\n'
941n/a + '0' * 65536 + 'a\r\n'
942n/a 'hello world\r\n'
943n/a '0\r\n'
944n/a '\r\n'
945n/a )
946n/a resp = client.HTTPResponse(FakeSocket(body))
947n/a resp.begin()
948n/a self.assertRaises(client.LineTooLong, resp.read)
949n/a
950n/a def test_early_eof(self):
951n/a # Test httpresponse with no \r\n termination,
952n/a body = "HTTP/1.1 200 Ok"
953n/a sock = FakeSocket(body)
954n/a resp = client.HTTPResponse(sock)
955n/a resp.begin()
956n/a self.assertEqual(resp.read(), b'')
957n/a self.assertTrue(resp.isclosed())
958n/a self.assertFalse(resp.closed)
959n/a resp.close()
960n/a self.assertTrue(resp.closed)
961n/a
962n/a def test_error_leak(self):
963n/a # Test that the socket is not leaked if getresponse() fails
964n/a conn = client.HTTPConnection('example.com')
965n/a response = None
966n/a class Response(client.HTTPResponse):
967n/a def __init__(self, *pos, **kw):
968n/a nonlocal response
969n/a response = self # Avoid garbage collector closing the socket
970n/a client.HTTPResponse.__init__(self, *pos, **kw)
971n/a conn.response_class = Response
972n/a conn.sock = FakeSocket('Invalid status line')
973n/a conn.request('GET', '/')
974n/a self.assertRaises(client.BadStatusLine, conn.getresponse)
975n/a self.assertTrue(response.closed)
976n/a self.assertTrue(conn.sock.file_closed)
977n/a
978n/a def test_chunked_extension(self):
979n/a extra = '3;foo=bar\r\n' + 'abc\r\n'
980n/a expected = chunked_expected + b'abc'
981n/a
982n/a sock = FakeSocket(chunked_start + extra + last_chunk_extended + chunked_end)
983n/a resp = client.HTTPResponse(sock, method="GET")
984n/a resp.begin()
985n/a self.assertEqual(resp.read(), expected)
986n/a resp.close()
987n/a
988n/a def test_chunked_missing_end(self):
989n/a """some servers may serve up a short chunked encoding stream"""
990n/a expected = chunked_expected
991n/a sock = FakeSocket(chunked_start + last_chunk) #no terminating crlf
992n/a resp = client.HTTPResponse(sock, method="GET")
993n/a resp.begin()
994n/a self.assertEqual(resp.read(), expected)
995n/a resp.close()
996n/a
997n/a def test_chunked_trailers(self):
998n/a """See that trailers are read and ignored"""
999n/a expected = chunked_expected
1000n/a sock = FakeSocket(chunked_start + last_chunk + trailers + chunked_end)
1001n/a resp = client.HTTPResponse(sock, method="GET")
1002n/a resp.begin()
1003n/a self.assertEqual(resp.read(), expected)
1004n/a # we should have reached the end of the file
1005n/a self.assertEqual(sock.file.read(), b"") #we read to the end
1006n/a resp.close()
1007n/a
1008n/a def test_chunked_sync(self):
1009n/a """Check that we don't read past the end of the chunked-encoding stream"""
1010n/a expected = chunked_expected
1011n/a extradata = "extradata"
1012n/a sock = FakeSocket(chunked_start + last_chunk + trailers + chunked_end + extradata)
1013n/a resp = client.HTTPResponse(sock, method="GET")
1014n/a resp.begin()
1015n/a self.assertEqual(resp.read(), expected)
1016n/a # the file should now have our extradata ready to be read
1017n/a self.assertEqual(sock.file.read(), extradata.encode("ascii")) #we read to the end
1018n/a resp.close()
1019n/a
1020n/a def test_content_length_sync(self):
1021n/a """Check that we don't read past the end of the Content-Length stream"""
1022n/a extradata = b"extradata"
1023n/a expected = b"Hello123\r\n"
1024n/a sock = FakeSocket(b'HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\n' + expected + extradata)
1025n/a resp = client.HTTPResponse(sock, method="GET")
1026n/a resp.begin()
1027n/a self.assertEqual(resp.read(), expected)
1028n/a # the file should now have our extradata ready to be read
1029n/a self.assertEqual(sock.file.read(), extradata) #we read to the end
1030n/a resp.close()
1031n/a
1032n/a def test_readlines_content_length(self):
1033n/a extradata = b"extradata"
1034n/a expected = b"Hello123\r\n"
1035n/a sock = FakeSocket(b'HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\n' + expected + extradata)
1036n/a resp = client.HTTPResponse(sock, method="GET")
1037n/a resp.begin()
1038n/a self.assertEqual(resp.readlines(2000), [expected])
1039n/a # the file should now have our extradata ready to be read
1040n/a self.assertEqual(sock.file.read(), extradata) #we read to the end
1041n/a resp.close()
1042n/a
1043n/a def test_read1_content_length(self):
1044n/a extradata = b"extradata"
1045n/a expected = b"Hello123\r\n"
1046n/a sock = FakeSocket(b'HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\n' + expected + extradata)
1047n/a resp = client.HTTPResponse(sock, method="GET")
1048n/a resp.begin()
1049n/a self.assertEqual(resp.read1(2000), expected)
1050n/a # the file should now have our extradata ready to be read
1051n/a self.assertEqual(sock.file.read(), extradata) #we read to the end
1052n/a resp.close()
1053n/a
1054n/a def test_readline_bound_content_length(self):
1055n/a extradata = b"extradata"
1056n/a expected = b"Hello123\r\n"
1057n/a sock = FakeSocket(b'HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\n' + expected + extradata)
1058n/a resp = client.HTTPResponse(sock, method="GET")
1059n/a resp.begin()
1060n/a self.assertEqual(resp.readline(10), expected)
1061n/a self.assertEqual(resp.readline(10), b"")
1062n/a # the file should now have our extradata ready to be read
1063n/a self.assertEqual(sock.file.read(), extradata) #we read to the end
1064n/a resp.close()
1065n/a
1066n/a def test_read1_bound_content_length(self):
1067n/a extradata = b"extradata"
1068n/a expected = b"Hello123\r\n"
1069n/a sock = FakeSocket(b'HTTP/1.1 200 OK\r\nContent-Length: 30\r\n\r\n' + expected*3 + extradata)
1070n/a resp = client.HTTPResponse(sock, method="GET")
1071n/a resp.begin()
1072n/a self.assertEqual(resp.read1(20), expected*2)
1073n/a self.assertEqual(resp.read(), expected)
1074n/a # the file should now have our extradata ready to be read
1075n/a self.assertEqual(sock.file.read(), extradata) #we read to the end
1076n/a resp.close()
1077n/a
1078n/a def test_response_fileno(self):
1079n/a # Make sure fd returned by fileno is valid.
1080n/a threading = support.import_module("threading")
1081n/a
1082n/a serv = socket.socket(
1083n/a socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)
1084n/a self.addCleanup(serv.close)
1085n/a serv.bind((HOST, 0))
1086n/a serv.listen()
1087n/a
1088n/a result = None
1089n/a def run_server():
1090n/a [conn, address] = serv.accept()
1091n/a with conn, conn.makefile("rb") as reader:
1092n/a # Read the request header until a blank line
1093n/a while True:
1094n/a line = reader.readline()
1095n/a if not line.rstrip(b"\r\n"):
1096n/a break
1097n/a conn.sendall(b"HTTP/1.1 200 Connection established\r\n\r\n")
1098n/a nonlocal result
1099n/a result = reader.read()
1100n/a
1101n/a thread = threading.Thread(target=run_server)
1102n/a thread.start()
1103n/a self.addCleanup(thread.join, float(1))
1104n/a conn = client.HTTPConnection(*serv.getsockname())
1105n/a conn.request("CONNECT", "dummy:1234")
1106n/a response = conn.getresponse()
1107n/a try:
1108n/a self.assertEqual(response.status, client.OK)
1109n/a s = socket.socket(fileno=response.fileno())
1110n/a try:
1111n/a s.sendall(b"proxied data\n")
1112n/a finally:
1113n/a s.detach()
1114n/a finally:
1115n/a response.close()
1116n/a conn.close()
1117n/a thread.join()
1118n/a self.assertEqual(result, b"proxied data\n")
1119n/a
1120n/aclass ExtendedReadTest(TestCase):
1121n/a """
1122n/a Test peek(), read1(), readline()
1123n/a """
1124n/a lines = (
1125n/a 'HTTP/1.1 200 OK\r\n'
1126n/a '\r\n'
1127n/a 'hello world!\n'
1128n/a 'and now \n'
1129n/a 'for something completely different\n'
1130n/a 'foo'
1131n/a )
1132n/a lines_expected = lines[lines.find('hello'):].encode("ascii")
1133n/a lines_chunked = (
1134n/a 'HTTP/1.1 200 OK\r\n'
1135n/a 'Transfer-Encoding: chunked\r\n\r\n'
1136n/a 'a\r\n'
1137n/a 'hello worl\r\n'
1138n/a '3\r\n'
1139n/a 'd!\n\r\n'
1140n/a '9\r\n'
1141n/a 'and now \n\r\n'
1142n/a '23\r\n'
1143n/a 'for something completely different\n\r\n'
1144n/a '3\r\n'
1145n/a 'foo\r\n'
1146n/a '0\r\n' # terminating chunk
1147n/a '\r\n' # end of trailers
1148n/a )
1149n/a
1150n/a def setUp(self):
1151n/a sock = FakeSocket(self.lines)
1152n/a resp = client.HTTPResponse(sock, method="GET")
1153n/a resp.begin()
1154n/a resp.fp = io.BufferedReader(resp.fp)
1155n/a self.resp = resp
1156n/a
1157n/a
1158n/a
1159n/a def test_peek(self):
1160n/a resp = self.resp
1161n/a # patch up the buffered peek so that it returns not too much stuff
1162n/a oldpeek = resp.fp.peek
1163n/a def mypeek(n=-1):
1164n/a p = oldpeek(n)
1165n/a if n >= 0:
1166n/a return p[:n]
1167n/a return p[:10]
1168n/a resp.fp.peek = mypeek
1169n/a
1170n/a all = []
1171n/a while True:
1172n/a # try a short peek
1173n/a p = resp.peek(3)
1174n/a if p:
1175n/a self.assertGreater(len(p), 0)
1176n/a # then unbounded peek
1177n/a p2 = resp.peek()
1178n/a self.assertGreaterEqual(len(p2), len(p))
1179n/a self.assertTrue(p2.startswith(p))
1180n/a next = resp.read(len(p2))
1181n/a self.assertEqual(next, p2)
1182n/a else:
1183n/a next = resp.read()
1184n/a self.assertFalse(next)
1185n/a all.append(next)
1186n/a if not next:
1187n/a break
1188n/a self.assertEqual(b"".join(all), self.lines_expected)
1189n/a
1190n/a def test_readline(self):
1191n/a resp = self.resp
1192n/a self._verify_readline(self.resp.readline, self.lines_expected)
1193n/a
1194n/a def _verify_readline(self, readline, expected):
1195n/a all = []
1196n/a while True:
1197n/a # short readlines
1198n/a line = readline(5)
1199n/a if line and line != b"foo":
1200n/a if len(line) < 5:
1201n/a self.assertTrue(line.endswith(b"\n"))
1202n/a all.append(line)
1203n/a if not line:
1204n/a break
1205n/a self.assertEqual(b"".join(all), expected)
1206n/a
1207n/a def test_read1(self):
1208n/a resp = self.resp
1209n/a def r():
1210n/a res = resp.read1(4)
1211n/a self.assertLessEqual(len(res), 4)
1212n/a return res
1213n/a readliner = Readliner(r)
1214n/a self._verify_readline(readliner.readline, self.lines_expected)
1215n/a
1216n/a def test_read1_unbounded(self):
1217n/a resp = self.resp
1218n/a all = []
1219n/a while True:
1220n/a data = resp.read1()
1221n/a if not data:
1222n/a break
1223n/a all.append(data)
1224n/a self.assertEqual(b"".join(all), self.lines_expected)
1225n/a
1226n/a def test_read1_bounded(self):
1227n/a resp = self.resp
1228n/a all = []
1229n/a while True:
1230n/a data = resp.read1(10)
1231n/a if not data:
1232n/a break
1233n/a self.assertLessEqual(len(data), 10)
1234n/a all.append(data)
1235n/a self.assertEqual(b"".join(all), self.lines_expected)
1236n/a
1237n/a def test_read1_0(self):
1238n/a self.assertEqual(self.resp.read1(0), b"")
1239n/a
1240n/a def test_peek_0(self):
1241n/a p = self.resp.peek(0)
1242n/a self.assertLessEqual(0, len(p))
1243n/a
1244n/aclass ExtendedReadTestChunked(ExtendedReadTest):
1245n/a """
1246n/a Test peek(), read1(), readline() in chunked mode
1247n/a """
1248n/a lines = (
1249n/a 'HTTP/1.1 200 OK\r\n'
1250n/a 'Transfer-Encoding: chunked\r\n\r\n'
1251n/a 'a\r\n'
1252n/a 'hello worl\r\n'
1253n/a '3\r\n'
1254n/a 'd!\n\r\n'
1255n/a '9\r\n'
1256n/a 'and now \n\r\n'
1257n/a '23\r\n'
1258n/a 'for something completely different\n\r\n'
1259n/a '3\r\n'
1260n/a 'foo\r\n'
1261n/a '0\r\n' # terminating chunk
1262n/a '\r\n' # end of trailers
1263n/a )
1264n/a
1265n/a
1266n/aclass Readliner:
1267n/a """
1268n/a a simple readline class that uses an arbitrary read function and buffering
1269n/a """
1270n/a def __init__(self, readfunc):
1271n/a self.readfunc = readfunc
1272n/a self.remainder = b""
1273n/a
1274n/a def readline(self, limit):
1275n/a data = []
1276n/a datalen = 0
1277n/a read = self.remainder
1278n/a try:
1279n/a while True:
1280n/a idx = read.find(b'\n')
1281n/a if idx != -1:
1282n/a break
1283n/a if datalen + len(read) >= limit:
1284n/a idx = limit - datalen - 1
1285n/a # read more data
1286n/a data.append(read)
1287n/a read = self.readfunc()
1288n/a if not read:
1289n/a idx = 0 #eof condition
1290n/a break
1291n/a idx += 1
1292n/a data.append(read[:idx])
1293n/a self.remainder = read[idx:]
1294n/a return b"".join(data)
1295n/a except:
1296n/a self.remainder = b"".join(data)
1297n/a raise
1298n/a
1299n/a
1300n/aclass OfflineTest(TestCase):
1301n/a def test_all(self):
1302n/a # Documented objects defined in the module should be in __all__
1303n/a expected = {"responses"} # White-list documented dict() object
1304n/a # HTTPMessage, parse_headers(), and the HTTP status code constants are
1305n/a # intentionally omitted for simplicity
1306n/a blacklist = {"HTTPMessage", "parse_headers"}
1307n/a for name in dir(client):
1308n/a if name.startswith("_") or name in blacklist:
1309n/a continue
1310n/a module_object = getattr(client, name)
1311n/a if getattr(module_object, "__module__", None) == "http.client":
1312n/a expected.add(name)
1313n/a self.assertCountEqual(client.__all__, expected)
1314n/a
1315n/a def test_responses(self):
1316n/a self.assertEqual(client.responses[client.NOT_FOUND], "Not Found")
1317n/a
1318n/a def test_client_constants(self):
1319n/a # Make sure we don't break backward compatibility with 3.4
1320n/a expected = [
1321n/a 'CONTINUE',
1322n/a 'SWITCHING_PROTOCOLS',
1323n/a 'PROCESSING',
1324n/a 'OK',
1325n/a 'CREATED',
1326n/a 'ACCEPTED',
1327n/a 'NON_AUTHORITATIVE_INFORMATION',
1328n/a 'NO_CONTENT',
1329n/a 'RESET_CONTENT',
1330n/a 'PARTIAL_CONTENT',
1331n/a 'MULTI_STATUS',
1332n/a 'IM_USED',
1333n/a 'MULTIPLE_CHOICES',
1334n/a 'MOVED_PERMANENTLY',
1335n/a 'FOUND',
1336n/a 'SEE_OTHER',
1337n/a 'NOT_MODIFIED',
1338n/a 'USE_PROXY',
1339n/a 'TEMPORARY_REDIRECT',
1340n/a 'BAD_REQUEST',
1341n/a 'UNAUTHORIZED',
1342n/a 'PAYMENT_REQUIRED',
1343n/a 'FORBIDDEN',
1344n/a 'NOT_FOUND',
1345n/a 'METHOD_NOT_ALLOWED',
1346n/a 'NOT_ACCEPTABLE',
1347n/a 'PROXY_AUTHENTICATION_REQUIRED',
1348n/a 'REQUEST_TIMEOUT',
1349n/a 'CONFLICT',
1350n/a 'GONE',
1351n/a 'LENGTH_REQUIRED',
1352n/a 'PRECONDITION_FAILED',
1353n/a 'REQUEST_ENTITY_TOO_LARGE',
1354n/a 'REQUEST_URI_TOO_LONG',
1355n/a 'UNSUPPORTED_MEDIA_TYPE',
1356n/a 'REQUESTED_RANGE_NOT_SATISFIABLE',
1357n/a 'EXPECTATION_FAILED',
1358n/a 'UNPROCESSABLE_ENTITY',
1359n/a 'LOCKED',
1360n/a 'FAILED_DEPENDENCY',
1361n/a 'UPGRADE_REQUIRED',
1362n/a 'PRECONDITION_REQUIRED',
1363n/a 'TOO_MANY_REQUESTS',
1364n/a 'REQUEST_HEADER_FIELDS_TOO_LARGE',
1365n/a 'INTERNAL_SERVER_ERROR',
1366n/a 'NOT_IMPLEMENTED',
1367n/a 'BAD_GATEWAY',
1368n/a 'SERVICE_UNAVAILABLE',
1369n/a 'GATEWAY_TIMEOUT',
1370n/a 'HTTP_VERSION_NOT_SUPPORTED',
1371n/a 'INSUFFICIENT_STORAGE',
1372n/a 'NOT_EXTENDED',
1373n/a 'NETWORK_AUTHENTICATION_REQUIRED',
1374n/a ]
1375n/a for const in expected:
1376n/a with self.subTest(constant=const):
1377n/a self.assertTrue(hasattr(client, const))
1378n/a
1379n/a
1380n/aclass SourceAddressTest(TestCase):
1381n/a def setUp(self):
1382n/a self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1383n/a self.port = support.bind_port(self.serv)
1384n/a self.source_port = support.find_unused_port()
1385n/a self.serv.listen()
1386n/a self.conn = None
1387n/a
1388n/a def tearDown(self):
1389n/a if self.conn:
1390n/a self.conn.close()
1391n/a self.conn = None
1392n/a self.serv.close()
1393n/a self.serv = None
1394n/a
1395n/a def testHTTPConnectionSourceAddress(self):
1396n/a self.conn = client.HTTPConnection(HOST, self.port,
1397n/a source_address=('', self.source_port))
1398n/a self.conn.connect()
1399n/a self.assertEqual(self.conn.sock.getsockname()[1], self.source_port)
1400n/a
1401n/a @unittest.skipIf(not hasattr(client, 'HTTPSConnection'),
1402n/a 'http.client.HTTPSConnection not defined')
1403n/a def testHTTPSConnectionSourceAddress(self):
1404n/a self.conn = client.HTTPSConnection(HOST, self.port,
1405n/a source_address=('', self.source_port))
1406n/a # We don't test anything here other than the constructor not barfing as
1407n/a # this code doesn't deal with setting up an active running SSL server
1408n/a # for an ssl_wrapped connect() to actually return from.
1409n/a
1410n/a
1411n/aclass TimeoutTest(TestCase):
1412n/a PORT = None
1413n/a
1414n/a def setUp(self):
1415n/a self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1416n/a TimeoutTest.PORT = support.bind_port(self.serv)
1417n/a self.serv.listen()
1418n/a
1419n/a def tearDown(self):
1420n/a self.serv.close()
1421n/a self.serv = None
1422n/a
1423n/a def testTimeoutAttribute(self):
1424n/a # This will prove that the timeout gets through HTTPConnection
1425n/a # and into the socket.
1426n/a
1427n/a # default -- use global socket timeout
1428n/a self.assertIsNone(socket.getdefaulttimeout())
1429n/a socket.setdefaulttimeout(30)
1430n/a try:
1431n/a httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT)
1432n/a httpConn.connect()
1433n/a finally:
1434n/a socket.setdefaulttimeout(None)
1435n/a self.assertEqual(httpConn.sock.gettimeout(), 30)
1436n/a httpConn.close()
1437n/a
1438n/a # no timeout -- do not use global socket default
1439n/a self.assertIsNone(socket.getdefaulttimeout())
1440n/a socket.setdefaulttimeout(30)
1441n/a try:
1442n/a httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT,
1443n/a timeout=None)
1444n/a httpConn.connect()
1445n/a finally:
1446n/a socket.setdefaulttimeout(None)
1447n/a self.assertEqual(httpConn.sock.gettimeout(), None)
1448n/a httpConn.close()
1449n/a
1450n/a # a value
1451n/a httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT, timeout=30)
1452n/a httpConn.connect()
1453n/a self.assertEqual(httpConn.sock.gettimeout(), 30)
1454n/a httpConn.close()
1455n/a
1456n/a
1457n/aclass PersistenceTest(TestCase):
1458n/a
1459n/a def test_reuse_reconnect(self):
1460n/a # Should reuse or reconnect depending on header from server
1461n/a tests = (
1462n/a ('1.0', '', False),
1463n/a ('1.0', 'Connection: keep-alive\r\n', True),
1464n/a ('1.1', '', True),
1465n/a ('1.1', 'Connection: close\r\n', False),
1466n/a ('1.0', 'Connection: keep-ALIVE\r\n', True),
1467n/a ('1.1', 'Connection: cloSE\r\n', False),
1468n/a )
1469n/a for version, header, reuse in tests:
1470n/a with self.subTest(version=version, header=header):
1471n/a msg = (
1472n/a 'HTTP/{} 200 OK\r\n'
1473n/a '{}'
1474n/a 'Content-Length: 12\r\n'
1475n/a '\r\n'
1476n/a 'Dummy body\r\n'
1477n/a ).format(version, header)
1478n/a conn = FakeSocketHTTPConnection(msg)
1479n/a self.assertIsNone(conn.sock)
1480n/a conn.request('GET', '/open-connection')
1481n/a with conn.getresponse() as response:
1482n/a self.assertEqual(conn.sock is None, not reuse)
1483n/a response.read()
1484n/a self.assertEqual(conn.sock is None, not reuse)
1485n/a self.assertEqual(conn.connections, 1)
1486n/a conn.request('GET', '/subsequent-request')
1487n/a self.assertEqual(conn.connections, 1 if reuse else 2)
1488n/a
1489n/a def test_disconnected(self):
1490n/a
1491n/a def make_reset_reader(text):
1492n/a """Return BufferedReader that raises ECONNRESET at EOF"""
1493n/a stream = io.BytesIO(text)
1494n/a def readinto(buffer):
1495n/a size = io.BytesIO.readinto(stream, buffer)
1496n/a if size == 0:
1497n/a raise ConnectionResetError()
1498n/a return size
1499n/a stream.readinto = readinto
1500n/a return io.BufferedReader(stream)
1501n/a
1502n/a tests = (
1503n/a (io.BytesIO, client.RemoteDisconnected),
1504n/a (make_reset_reader, ConnectionResetError),
1505n/a )
1506n/a for stream_factory, exception in tests:
1507n/a with self.subTest(exception=exception):
1508n/a conn = FakeSocketHTTPConnection(b'', stream_factory)
1509n/a conn.request('GET', '/eof-response')
1510n/a self.assertRaises(exception, conn.getresponse)
1511n/a self.assertIsNone(conn.sock)
1512n/a # HTTPConnection.connect() should be automatically invoked
1513n/a conn.request('GET', '/reconnect')
1514n/a self.assertEqual(conn.connections, 2)
1515n/a
1516n/a def test_100_close(self):
1517n/a conn = FakeSocketHTTPConnection(
1518n/a b'HTTP/1.1 100 Continue\r\n'
1519n/a b'\r\n'
1520n/a # Missing final response
1521n/a )
1522n/a conn.request('GET', '/', headers={'Expect': '100-continue'})
1523n/a self.assertRaises(client.RemoteDisconnected, conn.getresponse)
1524n/a self.assertIsNone(conn.sock)
1525n/a conn.request('GET', '/reconnect')
1526n/a self.assertEqual(conn.connections, 2)
1527n/a
1528n/a
1529n/aclass HTTPSTest(TestCase):
1530n/a
1531n/a def setUp(self):
1532n/a if not hasattr(client, 'HTTPSConnection'):
1533n/a self.skipTest('ssl support required')
1534n/a
1535n/a def make_server(self, certfile):
1536n/a from test.ssl_servers import make_https_server
1537n/a return make_https_server(self, certfile=certfile)
1538n/a
1539n/a def test_attributes(self):
1540n/a # simple test to check it's storing the timeout
1541n/a h = client.HTTPSConnection(HOST, TimeoutTest.PORT, timeout=30)
1542n/a self.assertEqual(h.timeout, 30)
1543n/a
1544n/a def test_networked(self):
1545n/a # Default settings: requires a valid cert from a trusted CA
1546n/a import ssl
1547n/a support.requires('network')
1548n/a with support.transient_internet('self-signed.pythontest.net'):
1549n/a h = client.HTTPSConnection('self-signed.pythontest.net', 443)
1550n/a with self.assertRaises(ssl.SSLError) as exc_info:
1551n/a h.request('GET', '/')
1552n/a self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
1553n/a
1554n/a def test_networked_noverification(self):
1555n/a # Switch off cert verification
1556n/a import ssl
1557n/a support.requires('network')
1558n/a with support.transient_internet('self-signed.pythontest.net'):
1559n/a context = ssl._create_unverified_context()
1560n/a h = client.HTTPSConnection('self-signed.pythontest.net', 443,
1561n/a context=context)
1562n/a h.request('GET', '/')
1563n/a resp = h.getresponse()
1564n/a h.close()
1565n/a self.assertIn('nginx', resp.getheader('server'))
1566n/a resp.close()
1567n/a
1568n/a @support.system_must_validate_cert
1569n/a def test_networked_trusted_by_default_cert(self):
1570n/a # Default settings: requires a valid cert from a trusted CA
1571n/a support.requires('network')
1572n/a with support.transient_internet('www.python.org'):
1573n/a h = client.HTTPSConnection('www.python.org', 443)
1574n/a h.request('GET', '/')
1575n/a resp = h.getresponse()
1576n/a content_type = resp.getheader('content-type')
1577n/a resp.close()
1578n/a h.close()
1579n/a self.assertIn('text/html', content_type)
1580n/a
1581n/a def test_networked_good_cert(self):
1582n/a # We feed the server's cert as a validating cert
1583n/a import ssl
1584n/a support.requires('network')
1585n/a with support.transient_internet('self-signed.pythontest.net'):
1586n/a context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
1587n/a context.verify_mode = ssl.CERT_REQUIRED
1588n/a context.load_verify_locations(CERT_selfsigned_pythontestdotnet)
1589n/a h = client.HTTPSConnection('self-signed.pythontest.net', 443, context=context)
1590n/a h.request('GET', '/')
1591n/a resp = h.getresponse()
1592n/a server_string = resp.getheader('server')
1593n/a resp.close()
1594n/a h.close()
1595n/a self.assertIn('nginx', server_string)
1596n/a
1597n/a def test_networked_bad_cert(self):
1598n/a # We feed a "CA" cert that is unrelated to the server's cert
1599n/a import ssl
1600n/a support.requires('network')
1601n/a with support.transient_internet('self-signed.pythontest.net'):
1602n/a context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
1603n/a context.verify_mode = ssl.CERT_REQUIRED
1604n/a context.load_verify_locations(CERT_localhost)
1605n/a h = client.HTTPSConnection('self-signed.pythontest.net', 443, context=context)
1606n/a with self.assertRaises(ssl.SSLError) as exc_info:
1607n/a h.request('GET', '/')
1608n/a self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
1609n/a
1610n/a def test_local_unknown_cert(self):
1611n/a # The custom cert isn't known to the default trust bundle
1612n/a import ssl
1613n/a server = self.make_server(CERT_localhost)
1614n/a h = client.HTTPSConnection('localhost', server.port)
1615n/a with self.assertRaises(ssl.SSLError) as exc_info:
1616n/a h.request('GET', '/')
1617n/a self.assertEqual(exc_info.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
1618n/a
1619n/a def test_local_good_hostname(self):
1620n/a # The (valid) cert validates the HTTP hostname
1621n/a import ssl
1622n/a server = self.make_server(CERT_localhost)
1623n/a context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
1624n/a context.verify_mode = ssl.CERT_REQUIRED
1625n/a context.load_verify_locations(CERT_localhost)
1626n/a h = client.HTTPSConnection('localhost', server.port, context=context)
1627n/a self.addCleanup(h.close)
1628n/a h.request('GET', '/nonexistent')
1629n/a resp = h.getresponse()
1630n/a self.addCleanup(resp.close)
1631n/a self.assertEqual(resp.status, 404)
1632n/a
1633n/a def test_local_bad_hostname(self):
1634n/a # The (valid) cert doesn't validate the HTTP hostname
1635n/a import ssl
1636n/a server = self.make_server(CERT_fakehostname)
1637n/a context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
1638n/a context.verify_mode = ssl.CERT_REQUIRED
1639n/a context.check_hostname = True
1640n/a context.load_verify_locations(CERT_fakehostname)
1641n/a h = client.HTTPSConnection('localhost', server.port, context=context)
1642n/a with self.assertRaises(ssl.CertificateError):
1643n/a h.request('GET', '/')
1644n/a # Same with explicit check_hostname=True
1645n/a with support.check_warnings(('', DeprecationWarning)):
1646n/a h = client.HTTPSConnection('localhost', server.port,
1647n/a context=context, check_hostname=True)
1648n/a with self.assertRaises(ssl.CertificateError):
1649n/a h.request('GET', '/')
1650n/a # With check_hostname=False, the mismatching is ignored
1651n/a context.check_hostname = False
1652n/a with support.check_warnings(('', DeprecationWarning)):
1653n/a h = client.HTTPSConnection('localhost', server.port,
1654n/a context=context, check_hostname=False)
1655n/a h.request('GET', '/nonexistent')
1656n/a resp = h.getresponse()
1657n/a resp.close()
1658n/a h.close()
1659n/a self.assertEqual(resp.status, 404)
1660n/a # The context's check_hostname setting is used if one isn't passed to
1661n/a # HTTPSConnection.
1662n/a context.check_hostname = False
1663n/a h = client.HTTPSConnection('localhost', server.port, context=context)
1664n/a h.request('GET', '/nonexistent')
1665n/a resp = h.getresponse()
1666n/a self.assertEqual(resp.status, 404)
1667n/a resp.close()
1668n/a h.close()
1669n/a # Passing check_hostname to HTTPSConnection should override the
1670n/a # context's setting.
1671n/a with support.check_warnings(('', DeprecationWarning)):
1672n/a h = client.HTTPSConnection('localhost', server.port,
1673n/a context=context, check_hostname=True)
1674n/a with self.assertRaises(ssl.CertificateError):
1675n/a h.request('GET', '/')
1676n/a
1677n/a @unittest.skipIf(not hasattr(client, 'HTTPSConnection'),
1678n/a 'http.client.HTTPSConnection not available')
1679n/a def test_host_port(self):
1680n/a # Check invalid host_port
1681n/a
1682n/a for hp in ("www.python.org:abc", "user:password@www.python.org"):
1683n/a self.assertRaises(client.InvalidURL, client.HTTPSConnection, hp)
1684n/a
1685n/a for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000",
1686n/a "fe80::207:e9ff:fe9b", 8000),
1687n/a ("www.python.org:443", "www.python.org", 443),
1688n/a ("www.python.org:", "www.python.org", 443),
1689n/a ("www.python.org", "www.python.org", 443),
1690n/a ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 443),
1691n/a ("[fe80::207:e9ff:fe9b]:", "fe80::207:e9ff:fe9b",
1692n/a 443)):
1693n/a c = client.HTTPSConnection(hp)
1694n/a self.assertEqual(h, c.host)
1695n/a self.assertEqual(p, c.port)
1696n/a
1697n/a
1698n/aclass RequestBodyTest(TestCase):
1699n/a """Test cases where a request includes a message body."""
1700n/a
1701n/a def setUp(self):
1702n/a self.conn = client.HTTPConnection('example.com')
1703n/a self.conn.sock = self.sock = FakeSocket("")
1704n/a self.conn.sock = self.sock
1705n/a
1706n/a def get_headers_and_fp(self):
1707n/a f = io.BytesIO(self.sock.data)
1708n/a f.readline() # read the request line
1709n/a message = client.parse_headers(f)
1710n/a return message, f
1711n/a
1712n/a def test_list_body(self):
1713n/a # Note that no content-length is automatically calculated for
1714n/a # an iterable. The request will fall back to send chunked
1715n/a # transfer encoding.
1716n/a cases = (
1717n/a ([b'foo', b'bar'], b'3\r\nfoo\r\n3\r\nbar\r\n0\r\n\r\n'),
1718n/a ((b'foo', b'bar'), b'3\r\nfoo\r\n3\r\nbar\r\n0\r\n\r\n'),
1719n/a )
1720n/a for body, expected in cases:
1721n/a with self.subTest(body):
1722n/a self.conn = client.HTTPConnection('example.com')
1723n/a self.conn.sock = self.sock = FakeSocket('')
1724n/a
1725n/a self.conn.request('PUT', '/url', body)
1726n/a msg, f = self.get_headers_and_fp()
1727n/a self.assertNotIn('Content-Type', msg)
1728n/a self.assertNotIn('Content-Length', msg)
1729n/a self.assertEqual(msg.get('Transfer-Encoding'), 'chunked')
1730n/a self.assertEqual(expected, f.read())
1731n/a
1732n/a def test_manual_content_length(self):
1733n/a # Set an incorrect content-length so that we can verify that
1734n/a # it will not be over-ridden by the library.
1735n/a self.conn.request("PUT", "/url", "body",
1736n/a {"Content-Length": "42"})
1737n/a message, f = self.get_headers_and_fp()
1738n/a self.assertEqual("42", message.get("content-length"))
1739n/a self.assertEqual(4, len(f.read()))
1740n/a
1741n/a def test_ascii_body(self):
1742n/a self.conn.request("PUT", "/url", "body")
1743n/a message, f = self.get_headers_and_fp()
1744n/a self.assertEqual("text/plain", message.get_content_type())
1745n/a self.assertIsNone(message.get_charset())
1746n/a self.assertEqual("4", message.get("content-length"))
1747n/a self.assertEqual(b'body', f.read())
1748n/a
1749n/a def test_latin1_body(self):
1750n/a self.conn.request("PUT", "/url", "body\xc1")
1751n/a message, f = self.get_headers_and_fp()
1752n/a self.assertEqual("text/plain", message.get_content_type())
1753n/a self.assertIsNone(message.get_charset())
1754n/a self.assertEqual("5", message.get("content-length"))
1755n/a self.assertEqual(b'body\xc1', f.read())
1756n/a
1757n/a def test_bytes_body(self):
1758n/a self.conn.request("PUT", "/url", b"body\xc1")
1759n/a message, f = self.get_headers_and_fp()
1760n/a self.assertEqual("text/plain", message.get_content_type())
1761n/a self.assertIsNone(message.get_charset())
1762n/a self.assertEqual("5", message.get("content-length"))
1763n/a self.assertEqual(b'body\xc1', f.read())
1764n/a
1765n/a def test_text_file_body(self):
1766n/a self.addCleanup(support.unlink, support.TESTFN)
1767n/a with open(support.TESTFN, "w") as f:
1768n/a f.write("body")
1769n/a with open(support.TESTFN) as f:
1770n/a self.conn.request("PUT", "/url", f)
1771n/a message, f = self.get_headers_and_fp()
1772n/a self.assertEqual("text/plain", message.get_content_type())
1773n/a self.assertIsNone(message.get_charset())
1774n/a # No content-length will be determined for files; the body
1775n/a # will be sent using chunked transfer encoding instead.
1776n/a self.assertIsNone(message.get("content-length"))
1777n/a self.assertEqual("chunked", message.get("transfer-encoding"))
1778n/a self.assertEqual(b'4\r\nbody\r\n0\r\n\r\n', f.read())
1779n/a
1780n/a def test_binary_file_body(self):
1781n/a self.addCleanup(support.unlink, support.TESTFN)
1782n/a with open(support.TESTFN, "wb") as f:
1783n/a f.write(b"body\xc1")
1784n/a with open(support.TESTFN, "rb") as f:
1785n/a self.conn.request("PUT", "/url", f)
1786n/a message, f = self.get_headers_and_fp()
1787n/a self.assertEqual("text/plain", message.get_content_type())
1788n/a self.assertIsNone(message.get_charset())
1789n/a self.assertEqual("chunked", message.get("Transfer-Encoding"))
1790n/a self.assertNotIn("Content-Length", message)
1791n/a self.assertEqual(b'5\r\nbody\xc1\r\n0\r\n\r\n', f.read())
1792n/a
1793n/a
1794n/aclass HTTPResponseTest(TestCase):
1795n/a
1796n/a def setUp(self):
1797n/a body = "HTTP/1.1 200 Ok\r\nMy-Header: first-value\r\nMy-Header: \
1798n/a second-value\r\n\r\nText"
1799n/a sock = FakeSocket(body)
1800n/a self.resp = client.HTTPResponse(sock)
1801n/a self.resp.begin()
1802n/a
1803n/a def test_getting_header(self):
1804n/a header = self.resp.getheader('My-Header')
1805n/a self.assertEqual(header, 'first-value, second-value')
1806n/a
1807n/a header = self.resp.getheader('My-Header', 'some default')
1808n/a self.assertEqual(header, 'first-value, second-value')
1809n/a
1810n/a def test_getting_nonexistent_header_with_string_default(self):
1811n/a header = self.resp.getheader('No-Such-Header', 'default-value')
1812n/a self.assertEqual(header, 'default-value')
1813n/a
1814n/a def test_getting_nonexistent_header_with_iterable_default(self):
1815n/a header = self.resp.getheader('No-Such-Header', ['default', 'values'])
1816n/a self.assertEqual(header, 'default, values')
1817n/a
1818n/a header = self.resp.getheader('No-Such-Header', ('default', 'values'))
1819n/a self.assertEqual(header, 'default, values')
1820n/a
1821n/a def test_getting_nonexistent_header_without_default(self):
1822n/a header = self.resp.getheader('No-Such-Header')
1823n/a self.assertEqual(header, None)
1824n/a
1825n/a def test_getting_header_defaultint(self):
1826n/a header = self.resp.getheader('No-Such-Header',default=42)
1827n/a self.assertEqual(header, 42)
1828n/a
1829n/aclass TunnelTests(TestCase):
1830n/a def setUp(self):
1831n/a response_text = (
1832n/a 'HTTP/1.0 200 OK\r\n\r\n' # Reply to CONNECT
1833n/a 'HTTP/1.1 200 OK\r\n' # Reply to HEAD
1834n/a 'Content-Length: 42\r\n\r\n'
1835n/a )
1836n/a self.host = 'proxy.com'
1837n/a self.conn = client.HTTPConnection(self.host)
1838n/a self.conn._create_connection = self._create_connection(response_text)
1839n/a
1840n/a def tearDown(self):
1841n/a self.conn.close()
1842n/a
1843n/a def _create_connection(self, response_text):
1844n/a def create_connection(address, timeout=None, source_address=None):
1845n/a return FakeSocket(response_text, host=address[0], port=address[1])
1846n/a return create_connection
1847n/a
1848n/a def test_set_tunnel_host_port_headers(self):
1849n/a tunnel_host = 'destination.com'
1850n/a tunnel_port = 8888
1851n/a tunnel_headers = {'User-Agent': 'Mozilla/5.0 (compatible, MSIE 11)'}
1852n/a self.conn.set_tunnel(tunnel_host, port=tunnel_port,
1853n/a headers=tunnel_headers)
1854n/a self.conn.request('HEAD', '/', '')
1855n/a self.assertEqual(self.conn.sock.host, self.host)
1856n/a self.assertEqual(self.conn.sock.port, client.HTTP_PORT)
1857n/a self.assertEqual(self.conn._tunnel_host, tunnel_host)
1858n/a self.assertEqual(self.conn._tunnel_port, tunnel_port)
1859n/a self.assertEqual(self.conn._tunnel_headers, tunnel_headers)
1860n/a
1861n/a def test_disallow_set_tunnel_after_connect(self):
1862n/a # Once connected, we shouldn't be able to tunnel anymore
1863n/a self.conn.connect()
1864n/a self.assertRaises(RuntimeError, self.conn.set_tunnel,
1865n/a 'destination.com')
1866n/a
1867n/a def test_connect_with_tunnel(self):
1868n/a self.conn.set_tunnel('destination.com')
1869n/a self.conn.request('HEAD', '/', '')
1870n/a self.assertEqual(self.conn.sock.host, self.host)
1871n/a self.assertEqual(self.conn.sock.port, client.HTTP_PORT)
1872n/a self.assertIn(b'CONNECT destination.com', self.conn.sock.data)
1873n/a # issue22095
1874n/a self.assertNotIn(b'Host: destination.com:None', self.conn.sock.data)
1875n/a self.assertIn(b'Host: destination.com', self.conn.sock.data)
1876n/a
1877n/a # This test should be removed when CONNECT gets the HTTP/1.1 blessing
1878n/a self.assertNotIn(b'Host: proxy.com', self.conn.sock.data)
1879n/a
1880n/a def test_connect_put_request(self):
1881n/a self.conn.set_tunnel('destination.com')
1882n/a self.conn.request('PUT', '/', '')
1883n/a self.assertEqual(self.conn.sock.host, self.host)
1884n/a self.assertEqual(self.conn.sock.port, client.HTTP_PORT)
1885n/a self.assertIn(b'CONNECT destination.com', self.conn.sock.data)
1886n/a self.assertIn(b'Host: destination.com', self.conn.sock.data)
1887n/a
1888n/a def test_tunnel_debuglog(self):
1889n/a expected_header = 'X-Dummy: 1'
1890n/a response_text = 'HTTP/1.0 200 OK\r\n{}\r\n\r\n'.format(expected_header)
1891n/a
1892n/a self.conn.set_debuglevel(1)
1893n/a self.conn._create_connection = self._create_connection(response_text)
1894n/a self.conn.set_tunnel('destination.com')
1895n/a
1896n/a with support.captured_stdout() as output:
1897n/a self.conn.request('PUT', '/', '')
1898n/a lines = output.getvalue().splitlines()
1899n/a self.assertIn('header: {}'.format(expected_header), lines)
1900n/a
1901n/a
1902n/aif __name__ == '__main__':
1903n/a unittest.main(verbosity=2)