我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用http.client.HTTPResponse()。
def post_raw(self, relative_url: str, data: object) -> (BeautifulSoup, HTTPResponse): """ Sends data via POST to the Dualis System @param relativeUrl: the Endpoint-Url to which the data should be send, relative to /scripts/mgrqcgi @param data: a plain object representing the data to post @return: the response returned by the Dualis System, already checked for errors """ data_urlencoded = urllib.parse.urlencode(data) self.connection.request( 'POST', '/scripts/mgrqcgi%s'%( relative_url ), body=data_urlencoded, headers=self.stdHeader ) response = self.connection.getresponse() return self._initial_parse(response), response
def test_response_headers(self): # test response with multiple message headers with the same field name. text = ('HTTP/1.1 200 OK\r\n' 'Set-Cookie: Customer="WILE_E_COYOTE"; ' 'Version="1"; Path="/acme"\r\n' 'Set-Cookie: Part_Number="Rocket_Launcher_0001"; Version="1";' ' Path="/acme"\r\n' '\r\n' 'No body\r\n') hdr = ('Customer="WILE_E_COYOTE"; Version="1"; Path="/acme"' ', ' 'Part_Number="Rocket_Launcher_0001"; Version="1"; Path="/acme"') s = FakeSocket(text) r = client.HTTPResponse(s) r.begin() cookies = r.getheader("Set-Cookie") self.assertEqual(cookies, hdr)
def test_status_lines(self): # Test HTTP status lines body = "HTTP/1.1 200 Ok\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) resp.begin() self.assertEqual(resp.read(), b"Text") self.assertTrue(resp.isclosed()) self.assertFalse(resp.closed) resp.close() self.assertTrue(resp.closed) body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) self.assertRaises(client.BadStatusLine, resp.begin)
def test_partial_readintos(self): # if we have a length, the system knows when to close itself # same behaviour than when we read the whole thing with read() body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) resp.begin() b = bytearray(2) n = resp.readinto(b) self.assertEqual(n, 2) self.assertEqual(bytes(b), b'Te') self.assertFalse(resp.isclosed()) n = resp.readinto(b) self.assertEqual(n, 2) self.assertEqual(bytes(b), b'xt') self.assertTrue(resp.isclosed()) self.assertFalse(resp.closed) resp.close() self.assertTrue(resp.closed)
def test_partial_readintos_no_content_length(self): # when no length is present, the socket should be gracefully closed when # all data was read body = "HTTP/1.1 200 Ok\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) resp.begin() b = bytearray(2) n = resp.readinto(b) self.assertEqual(n, 2) self.assertEqual(bytes(b), b'Te') self.assertFalse(resp.isclosed()) n = resp.readinto(b) self.assertEqual(n, 2) self.assertEqual(bytes(b), b'xt') n = resp.readinto(b) self.assertEqual(n, 0) self.assertTrue(resp.isclosed())
def test_partial_readintos_incomplete_body(self): # if the server shuts down the connection before the whole # content-length is delivered, the socket is gracefully closed body = "HTTP/1.1 200 Ok\r\nContent-Length: 10\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) resp.begin() b = bytearray(2) n = resp.readinto(b) self.assertEqual(n, 2) self.assertEqual(bytes(b), b'Te') self.assertFalse(resp.isclosed()) n = resp.readinto(b) self.assertEqual(n, 2) self.assertEqual(bytes(b), b'xt') n = resp.readinto(b) self.assertEqual(n, 0) self.assertTrue(resp.isclosed()) self.assertFalse(resp.closed) resp.close() self.assertTrue(resp.closed)
def test_chunked_head(self): chunked_start = ( 'HTTP/1.1 200 OK\r\n' 'Transfer-Encoding: chunked\r\n\r\n' 'a\r\n' 'hello world\r\n' '1\r\n' 'd\r\n' ) sock = FakeSocket(chunked_start + '0\r\n') resp = client.HTTPResponse(sock, method="HEAD") resp.begin() self.assertEqual(resp.read(), b'') self.assertEqual(resp.status, 200) self.assertEqual(resp.reason, 'OK') self.assertTrue(resp.isclosed()) self.assertFalse(resp.closed) resp.close() self.assertTrue(resp.closed)
def test_readinto_chunked_head(self): chunked_start = ( 'HTTP/1.1 200 OK\r\n' 'Transfer-Encoding: chunked\r\n\r\n' 'a\r\n' 'hello world\r\n' '1\r\n' 'd\r\n' ) sock = FakeSocket(chunked_start + '0\r\n') resp = client.HTTPResponse(sock, method="HEAD") resp.begin() b = bytearray(5) n = resp.readinto(b) self.assertEqual(n, 0) self.assertEqual(bytes(b), b'\x00'*5) self.assertEqual(resp.status, 200) self.assertEqual(resp.reason, 'OK') self.assertTrue(resp.isclosed()) self.assertFalse(resp.closed) resp.close() self.assertTrue(resp.closed)
def test_status_lines(self): # Test HTTP status lines body = "HTTP/1.1 200 Ok\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) resp.begin() self.assertEqual(resp.read(0), b'') # Issue #20007 self.assertFalse(resp.isclosed()) self.assertFalse(resp.closed) self.assertEqual(resp.read(), b"Text") self.assertTrue(resp.isclosed()) self.assertFalse(resp.closed) resp.close() self.assertTrue(resp.closed) body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) self.assertRaises(client.BadStatusLine, resp.begin)
def __init__( self, _=None, # type: Optional[Union[AnyStr, typing.Mapping, typing.Sequence, typing.IO]] ): self._meta = None if _ is not None: if isinstance(_, HTTPResponse): meta.get(self).url = _.url _ = deserialize(_) for k, v in _.items(): try: self[k] = v except KeyError as e: if e.args and len(e.args) == 1: e.args = ( r'%s.%s: %s' % (type(self).__name__, e.args[0], json.dumps(_)), ) raise e
def tell(self): """ Obtain the number of bytes pulled over the wire so far. May differ from the amount of content returned by :meth:``HTTPResponse.read`` if bytes are encoded on the wire (e.g, compressed). """ return self._fp_bytes_read
def from_httplib(ResponseCls, r, **response_kw): """ Given an :class:`httplib.HTTPResponse` instance ``r``, return a corresponding :class:`urllib3.response.HTTPResponse` object. Remaining parameters are passed to the HTTPResponse constructor, along with ``original_response=r``. """ headers = r.msg if not isinstance(headers, HTTPHeaderDict): if PY3: # Python 3 headers = HTTPHeaderDict(headers.items()) else: # Python 2 headers = HTTPHeaderDict.from_httplib(headers) # HTTPResponse objects in Python 3 don't have a .strict attribute strict = getattr(r, 'strict', 0) resp = ResponseCls(body=r, headers=headers, status=r.status, version=r.version, reason=r.reason, strict=strict, original_response=r, **response_kw) return resp # Backwards-compatibility methods for httplib.HTTPResponse
def fileno(self): if self._fp is None: raise IOError("HTTPResponse has no file to get a fileno from") elif hasattr(self._fp, "fileno"): return self._fp.fileno() else: raise IOError("The file-like object this HTTPResponse is wrapped " "around has no file descriptor")
def test_status_lines(self): # Test HTTP status lines body = "HTTP/1.1 200 Ok\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) resp.begin() self.assertEqual(resp.read(), b"Text") self.assertTrue(resp.isclosed()) body = "HTTP/1.1 400.100 Not Ok\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) self.assertRaises(client.BadStatusLine, resp.begin)
def test_partial_reads(self): # if we have a lenght, the system knows when to close itself # same behaviour than when we read the whole thing with read() body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) resp.begin() self.assertEqual(resp.read(2), b'Te') self.assertFalse(resp.isclosed()) self.assertEqual(resp.read(2), b'xt') self.assertTrue(resp.isclosed())
def test_read_head(self): # Test that the library doesn't attempt to read any data # from a HEAD request. (Tickles SF bug #622042.) sock = FakeSocket( 'HTTP/1.1 200 OK\r\n' 'Content-Length: 14432\r\n' '\r\n', NoEOFStringIO) resp = client.HTTPResponse(sock, method="HEAD") resp.begin() if resp.read(): self.fail("Did not expect response from HEAD request")
def test_chunked(self): chunked_start = ( 'HTTP/1.1 200 OK\r\n' 'Transfer-Encoding: chunked\r\n\r\n' 'a\r\n' 'hello worl\r\n' '1\r\n' 'd\r\n' ) sock = FakeSocket(chunked_start + '0\r\n') resp = client.HTTPResponse(sock, method="GET") resp.begin() self.assertEqual(resp.read(), b'hello world') resp.close() for x in ('', 'foo\r\n'): sock = FakeSocket(chunked_start + x) resp = client.HTTPResponse(sock, method="GET") resp.begin() try: resp.read() except client.IncompleteRead as i: self.assertEqual(i.partial, b'hello world') self.assertEqual(repr(i),'IncompleteRead(11 bytes read)') self.assertEqual(str(i),'IncompleteRead(11 bytes read)') else: self.fail('IncompleteRead expected') finally: resp.close()
def test_negative_content_length(self): sock = FakeSocket( 'HTTP/1.1 200 OK\r\nContent-Length: -1\r\n\r\nHello\r\n') resp = client.HTTPResponse(sock, method="GET") resp.begin() self.assertEqual(resp.read(), b'Hello\r\n') resp.close()
def test_incomplete_read(self): sock = FakeSocket('HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nHello\r\n') resp = client.HTTPResponse(sock, method="GET") resp.begin() try: resp.read() except client.IncompleteRead as i: self.assertEqual(i.partial, b'Hello\r\n') self.assertEqual(repr(i), "IncompleteRead(7 bytes read, 3 more expected)") self.assertEqual(str(i), "IncompleteRead(7 bytes read, 3 more expected)") else: self.fail('IncompleteRead expected') finally: resp.close()
def test_overflowing_status_line(self): body = "HTTP/1.1 200 Ok" + "k" * 65536 + "\r\n" resp = client.HTTPResponse(FakeSocket(body)) self.assertRaises((client.LineTooLong, client.BadStatusLine), resp.begin)
def test_overflowing_header_line(self): body = ( 'HTTP/1.1 200 OK\r\n' 'X-Foo: bar' + 'r' * 65536 + '\r\n\r\n' ) resp = client.HTTPResponse(FakeSocket(body)) self.assertRaises(client.LineTooLong, resp.begin)
def test_overflowing_chunked_line(self): body = ( 'HTTP/1.1 200 OK\r\n' 'Transfer-Encoding: chunked\r\n\r\n' + '0' * 65536 + 'a\r\n' 'hello world\r\n' '0\r\n' ) resp = client.HTTPResponse(FakeSocket(body)) resp.begin() self.assertRaises(client.LineTooLong, resp.read)
def test_partial_reads(self): # if we have a length, the system knows when to close itself # same behaviour than when we read the whole thing with read() body = "HTTP/1.1 200 Ok\r\nContent-Length: 4\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) resp.begin() self.assertEqual(resp.read(2), b'Te') self.assertFalse(resp.isclosed()) self.assertEqual(resp.read(2), b'xt') self.assertTrue(resp.isclosed()) self.assertFalse(resp.closed) resp.close() self.assertTrue(resp.closed)
def test_partial_reads_no_content_length(self): # when no length is present, the socket should be gracefully closed when # all data was read body = "HTTP/1.1 200 Ok\r\n\r\nText" sock = FakeSocket(body) resp = client.HTTPResponse(sock) resp.begin() self.assertEqual(resp.read(2), b'Te') self.assertFalse(resp.isclosed()) self.assertEqual(resp.read(2), b'xt') self.assertEqual(resp.read(1), b'') self.assertTrue(resp.isclosed()) self.assertFalse(resp.closed) resp.close() self.assertTrue(resp.closed)
def test_read_head(self): # Test that the library doesn't attempt to read any data # from a HEAD request. (Tickles SF bug #622042.) sock = FakeSocket( 'HTTP/1.1 200 OK\r\n' 'Content-Length: 14432\r\n' '\r\n', NoEOFBytesIO) resp = client.HTTPResponse(sock, method="HEAD") resp.begin() if resp.read(): self.fail("Did not expect response from HEAD request")
