summaryrefslogtreecommitdiffstats
path: root/FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py
diff options
context:
space:
mode:
authoryum <yum.food.vr@gmail.com>2023-01-01 21:05:27 -0800
committeryum <yum.food.vr@gmail.com>2023-01-01 21:44:45 -0800
commite25bdba3a3a53b09be5269d8b065c13b73ab55c3 (patch)
tree1d1dc1d94cde92c2f4f8ce86017395054787515d /FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py
parent0d408cc812a094a708edbe4baf536e928731cfc3 (diff)
Embed git in package
package.ps1 fetches PortableGit and embeds it in the package. This eliminates all but one runtime dependency (MSVC++ Redistributable). * Move Python into a new FOSS folder.
Diffstat (limited to 'FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py')
-rw-r--r--FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py568
1 files changed, 568 insertions, 0 deletions
diff --git a/FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py b/FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py
new file mode 100644
index 0000000..a1e2b19
--- /dev/null
+++ b/FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_httplib.py
@@ -0,0 +1,568 @@
+"""
+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()