diff options
Diffstat (limited to 'Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py')
| -rw-r--r-- | Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py | 568 |
1 files changed, 0 insertions, 568 deletions
diff --git a/Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py b/Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py deleted file mode 100644 index a1e2b19..0000000 --- a/Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py +++ /dev/null @@ -1,568 +0,0 @@ -""" -Tests for the http.client module - -Adapted for the python-future module from the Python 2.7 standard -library tests. -""" - -from __future__ import (absolute_import, division, - print_function, unicode_literals) -from future.builtins import * -from future import utils -from future.tests.base import unittest, skip26 - -from future.backports.http import client -from future.backports.test import support -import array -import io -import socket -import errno -import sys - -TestCase = unittest.TestCase -HOST = support.HOST - - -class FakeSocket(object): - def __init__(self, text, fileclass=io.BytesIO): - if isinstance(text, str): - text = str(text).encode('ascii') - self.text = text - self.fileclass = fileclass - self.data = bytes(b'') - - def sendall(self, data): - # self.data += bytes(data) - olddata = self.data - assert isinstance(olddata, bytes) - if utils.PY3: - self.data += data - else: - if isinstance(data, type(u'')): # i.e. unicode - newdata = data.encode('ascii') - elif isinstance(data, type(b'')): # native string type. FIXME! - newdata = bytes(data) - elif isinstance(data, bytes): - newdata = data - elif isinstance(data, array.array): - newdata = data.tostring() - else: - newdata = bytes(b'').join(chr(d) for d in bytes(data)) - self.data += newdata - - def makefile(self, mode, bufsize=None): - if mode != 'r' and mode != 'rb': - raise client.UnimplementedFileMode() - return self.fileclass(self.text) - -class EPipeSocket(FakeSocket): - - def __init__(self, text, pipe_trigger): - # When sendall() is called with pipe_trigger, raise EPIPE. - FakeSocket.__init__(self, text) - self.pipe_trigger = pipe_trigger - - def sendall(self, data): - if self.pipe_trigger in data: - raise socket.error(errno.EPIPE, "gotcha") - self.data += data - - def close(self): - pass - -class NoEOFBytesIO(io.BytesIO): - """Like BytesIO, but raises AssertionError on EOF. - - This is used below to test that http.client doesn't try to read - more from the underlying file than it should. - """ - def read(self, n=-1): - data = io.BytesIO.read(self, n) - if data == b'': - raise AssertionError('caller tried to read past EOF') - return data - - def readline(self, length=None): - data = io.BytesIO.readline(self, length) - if data == b'': - raise AssertionError('caller tried to read past EOF') - return data - - -@skip26 -class HeaderTests(TestCase): - def test_auto_headers(self): - # Some headers are added automatically, but should not be added by - # .request() if they are explicitly set. - - class HeaderCountingBuffer(list): - def __init__(self): - self.count = {} - def append(self, item): - kv = item.split(b':') - if len(kv) > 1: - # item is a 'Key: Value' header string - lcKey = kv[0].decode('ascii').lower() - self.count.setdefault(lcKey, 0) - self.count[lcKey] += 1 - list.append(self, item) - - for explicit_header in True, False: - for header in 'Content-length', 'Host', 'Accept-encoding': - conn = client.HTTPConnection('example.com') - conn.sock = FakeSocket('blahblahblah') - conn._buffer = HeaderCountingBuffer() - - body = 'spamspamspam' - headers = {} - if explicit_header: - headers[header] = str(len(body)) - conn.request('POST', '/', body, headers) - self.assertEqual(conn._buffer.count[header.lower()], 1) - - def test_content_length_0(self): - - class ContentLengthChecker(list): - def __init__(self): - list.__init__(self) - self.content_length = None - def append(self, item): - kv = item.split(b':', 1) - if len(kv) > 1 and kv[0].lower() == b'content-length': - self.content_length = kv[1].strip() - list.append(self, item) - - # POST with empty body - conn = client.HTTPConnection('example.com') - conn.sock = FakeSocket(None) - conn._buffer = ContentLengthChecker() - conn.request('POST', '/', '') - self.assertEqual(conn._buffer.content_length, b'0', - 'Header Content-Length not set') - - # PUT request with empty body - conn = client.HTTPConnection('example.com') - conn.sock = FakeSocket(None) - conn._buffer = ContentLengthChecker() - conn.request('PUT', '/', '') - self.assertEqual(conn._buffer.content_length, b'0', - 'Header Content-Length not set') - - def test_putheader(self): - conn = client.HTTPConnection('example.com') - conn.sock = FakeSocket(None) - conn.putrequest('GET','/') - conn.putheader('Content-length', 42) - self.assertTrue(b'Content-length: 42' in conn._buffer) - - def test_ipv6host_header(self): - # Default host header on IPv6 transaction should wrapped by [] if - # its actual IPv6 address - expected = bytes(b'GET /foo HTTP/1.1\r\nHost: [2001::]:81\r\n') + \ - bytes(b'Accept-Encoding: identity\r\n\r\n') - conn = client.HTTPConnection('[2001::]:81') - sock = FakeSocket('') - conn.sock = sock - conn.request('GET', '/foo') - self.assertTrue(sock.data.startswith(expected)) - - expected = bytes(b'GET /foo HTTP/1.1\r\nHost: [2001:102A::]\r\n') + \ - bytes(b'Accept-Encoding: identity\r\n\r\n') - conn = client.HTTPConnection('[2001:102A::]') - sock = FakeSocket('') - conn.sock = sock - conn.request('GET', '/foo') - self.assertTrue(sock.data.startswith(expected)) - - -@skip26 -class BasicTest(TestCase): - 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_bad_status_repr(self): - exc = client.BadStatusLine('') - if not utils.PY3: - self.assertEqual(repr(exc), '''BadStatusLine("u\'\'",)''') - else: - self.assertIn(repr(exc), ('''BadStatusLine("''",)''', '''BadStatusLine("''")''')) - - 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()) - - 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()) - - def test_partial_reads_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() - 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()) - - def test_host_port(self): - # Check invalid host_port - - # Note that http.client does not accept user:password@ in the host-port. - for hp in ("www.python.org:abc", "user:password@www.python.org"): - self.assertRaises(client.InvalidURL, client.HTTPConnection, hp) - - for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", "fe80::207:e9ff:fe9b", - 8000), - ("www.python.org:80", "www.python.org", 80), - ("www.python.org", "www.python.org", 80), - ("www.python.org:", "www.python.org", 80), - ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 80)): - c = client.HTTPConnection(hp) - self.assertEqual(h, c.host) - self.assertEqual(p, c.port) - - 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_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") - - def test_send_file(self): - expected = (bytes(b'GET /foo HTTP/1.1\r\nHost: example.com\r\n') + - bytes(b'Accept-Encoding: identity\r\nContent-Length:')) - - # __file__ will usually be the .pyc, i.e. binary data - with open(__file__, 'rb') as body: - conn = client.HTTPConnection('example.com') - sock = FakeSocket(body) - conn.sock = sock - conn.request('GET', '/foo', body) - self.assertTrue(sock.data.startswith(expected), '%r != %r' % - (sock.data[:len(expected)], expected)) - - def test_send(self): - expected = bytes(b'this is a test this is only a test') - conn = client.HTTPConnection('example.com') - sock = FakeSocket(None) - conn.sock = sock - conn.send(expected) - self.assertEqual(expected, sock.data) - sock.data = bytes(b'') - if utils.PY3: - mydata = array.array('b', expected) - else: - mydata = array.array(b'b', expected) - conn.send(mydata) - self.assertEqual(expected, sock.data) - sock.data = bytes(b'') - conn.send(io.BytesIO(expected)) - self.assertEqual(expected, sock.data) - - 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_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()) - - def test_negative_content_length(self): - sock = FakeSocket('HTTP/1.1 200 OK\r\n' - 'Content-Length: -1\r\n\r\nHello\r\n') - resp = client.HTTPResponse(sock, method="GET") - resp.begin() - self.assertEqual(resp.read(), b'Hello\r\n') - self.assertTrue(resp.isclosed()) - - 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)") - self.assertTrue(resp.isclosed()) - else: - self.fail('IncompleteRead expected') - - def test_epipe(self): - sock = EPipeSocket( - "HTTP/1.0 401 Authorization Required\r\n" - "Content-type: text/html\r\n" - "WWW-Authenticate: Basic realm=\"example\"\r\n", - b"Content-Length") - conn = client.HTTPConnection("example.com") - conn.sock = sock - self.assertRaises(socket.error, - lambda: conn.request("PUT", "/url", "body")) - resp = conn.getresponse() - self.assertEqual(401, resp.status) - self.assertEqual("Basic realm=\"example\"", - resp.getheader("www-authenticate")) - - def test_filenoattr(self): - # Just test the fileno attribute in the HTTPResponse Object. - body = "HTTP/1.1 200 Ok\r\n\r\nText" - sock = FakeSocket(body) - resp = client.HTTPResponse(sock) - self.assertTrue(hasattr(resp,'fileno'), - 'HTTPResponse should expose a fileno attribute') - - # Test lines overflowing the max line size (_MAXLINE in http.client) - - def test_overflowing_status_line(self): - self.skipTest("disabled for HTTP 0.9 support") - 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_early_eof(self): - # Test httpresponse with no \r\n termination, - body = "HTTP/1.1 200 Ok" - sock = FakeSocket(body) - resp = client.HTTPResponse(sock) - resp.begin() - self.assertEqual(resp.read(), b'') - self.assertTrue(resp.isclosed()) - -@skip26 -class OfflineTest(TestCase): - def test_responses(self): - self.assertEqual(client.responses[client.NOT_FOUND], "Not Found") - - -@skip26 -class SourceAddressTest(TestCase): - def setUp(self): - self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - self.port = support.bind_port(self.serv) - self.source_port = support.find_unused_port() - self.serv.listen(5) - self.conn = None - - def tearDown(self): - if self.conn: - self.conn.close() - self.conn = None - self.serv.close() - self.serv = None - - def testHTTPConnectionSourceAddress(self): - self.conn = client.HTTPConnection(HOST, self.port, - source_address=('', self.source_port)) - self.conn.connect() - self.assertEqual(self.conn.sock.getsockname()[1], self.source_port) - - @unittest.skipIf(not hasattr(client, 'HTTPSConnection'), - 'http.client.HTTPSConnection not defined') - def testHTTPSConnectionSourceAddress(self): - self.conn = client.HTTPSConnection(HOST, self.port, - source_address=('', self.source_port)) - # We don't test anything here other the constructor not barfing as - # this code doesn't deal with setting up an active running SSL server - # for an ssl_wrapped connect() to actually return from. - - -@skip26 -class TimeoutTest(TestCase): - PORT = None - - def setUp(self): - self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - TimeoutTest.PORT = support.bind_port(self.serv) - self.serv.listen(5) - - def tearDown(self): - self.serv.close() - self.serv = None - - def testTimeoutAttribute(self): - '''This will prove that the timeout gets through - HTTPConnection and into the socket. - ''' - # default -- use global socket timeout - self.assertTrue(socket.getdefaulttimeout() is None) - socket.setdefaulttimeout(30) - try: - httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT) - httpConn.connect() - finally: - socket.setdefaulttimeout(None) - self.assertEqual(httpConn.sock.gettimeout(), 30) - httpConn.close() - - # no timeout -- do not use global socket default - self.assertTrue(socket.getdefaulttimeout() is None) - socket.setdefaulttimeout(30) - try: - httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT, - timeout=None) - httpConn.connect() - finally: - socket.setdefaulttimeout(None) - self.assertEqual(httpConn.sock.gettimeout(), None) - httpConn.close() - - # a value - httpConn = client.HTTPConnection(HOST, TimeoutTest.PORT, timeout=30) - httpConn.connect() - self.assertEqual(httpConn.sock.gettimeout(), 30) - httpConn.close() - - -@skip26 -class HTTPSTest(TestCase): - - def test_attributes(self): - # simple test to check it's storing it - if hasattr(client, 'HTTPSConnection'): - h = client.HTTPSConnection(HOST, TimeoutTest.PORT, timeout=30) - self.assertEqual(h.timeout, 30) - - @unittest.skipIf(not hasattr(client, 'HTTPSConnection'), 'http.client.HTTPSConnection not available') - def test_host_port(self): - # Check invalid host_port - - # Note that httplib does not accept user:password@ in the host-port. - for hp in ("www.python.org:abc", "user:password@www.python.org"): - self.assertRaises(client.InvalidURL, client.HTTPSConnection, hp) - - for hp, h, p in (("[fe80::207:e9ff:fe9b]:8000", "fe80::207:e9ff:fe9b", - 8000), - ("pypi.python.org:443", "pypi.python.org", 443), - ("pypi.python.org", "pypi.python.org", 443), - ("pypi.python.org:", "pypi.python.org", 443), - ("[fe80::207:e9ff:fe9b]", "fe80::207:e9ff:fe9b", 443)): - c = client.HTTPSConnection(hp) - self.assertEqual(h, c.host) - self.assertEqual(p, c.port) - - -# def test_main(verbose=None): -# support.run_unittest(HeaderTests, OfflineTest, BasicTest, TimeoutTest, -# HTTPSTest, SourceAddressTest) - -if __name__ == '__main__': - unittest.main() |