~ubuntu-branches/ubuntu/trusty/python-urllib3/trusty-proposed : contents of test/test

~ubuntu-branches/ubuntu/trusty/python-urllib3/trusty-proposed : (revision 10)

import unittest

from io import BytesIO, BufferedReader

from urllib3.response import HTTPResponse
from urllib3.exceptions import DecodeError

class TestLegacyResponse(unittest.TestCase):
    def test_getheaders(self):
        headers = {'host': 'example.com'}
        r = HTTPResponse(headers=headers)
        self.assertEqual(r.getheaders(), headers)

    def test_getheader(self):
        headers = {'host': 'example.com'}
        r = HTTPResponse(headers=headers)
        self.assertEqual(r.getheader('host'), 'example.com')


class TestResponse(unittest.TestCase):
    def test_cache_content(self):
        r = HTTPResponse('foo')
        self.assertEqual(r.data, 'foo')
        self.assertEqual(r._body, 'foo')

    def test_default(self):
        r = HTTPResponse()
        self.assertEqual(r.data, None)

    def test_none(self):
        r = HTTPResponse(None)
        self.assertEqual(r.data, None)

    def test_preload(self):
        fp = BytesIO(b'foo')

        r = HTTPResponse(fp, preload_content=True)

        self.assertEqual(fp.tell(), len(b'foo'))
        self.assertEqual(r.data, b'foo')

    def test_no_preload(self):
        fp = BytesIO(b'foo')

        r = HTTPResponse(fp, preload_content=False)

        self.assertEqual(fp.tell(), 0)
        self.assertEqual(r.data, b'foo')
        self.assertEqual(fp.tell(), len(b'foo'))

    def test_decode_bad_data(self):
        fp = BytesIO(b'\x00' * 10)
        self.assertRaises(DecodeError, HTTPResponse, fp, headers={
            'content-encoding': 'deflate'
        })

    def test_decode_deflate(self):
        import zlib
        data = zlib.compress(b'foo')

        fp = BytesIO(data)
        r = HTTPResponse(fp, headers={'content-encoding': 'deflate'})

        self.assertEqual(r.data, b'foo')

    def test_decode_deflate_case_insensitve(self):
        import zlib
        data = zlib.compress(b'foo')

        fp = BytesIO(data)
        r = HTTPResponse(fp, headers={'content-encoding': 'DeFlAtE'})

        self.assertEqual(r.data, b'foo')

    def test_chunked_decoding_deflate(self):
        import zlib
        data = zlib.compress(b'foo')

        fp = BytesIO(data)
        r = HTTPResponse(fp, headers={'content-encoding': 'deflate'},
                         preload_content=False)

        self.assertEqual(r.read(3), b'')
        self.assertEqual(r.read(1), b'f')
        self.assertEqual(r.read(2), b'oo')

    def test_chunked_decoding_deflate2(self):
        import zlib
        compress = zlib.compressobj(6, zlib.DEFLATED, -zlib.MAX_WBITS)
        data = compress.compress(b'foo')
        data += compress.flush()

        fp = BytesIO(data)
        r = HTTPResponse(fp, headers={'content-encoding': 'deflate'},
                         preload_content=False)

        self.assertEqual(r.read(1), b'')
        self.assertEqual(r.read(1), b'f')
        self.assertEqual(r.read(2), b'oo')

    def test_chunked_decoding_gzip(self):
        import zlib
        compress = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS)
        data = compress.compress(b'foo')
        data += compress.flush()

        fp = BytesIO(data)
        r = HTTPResponse(fp, headers={'content-encoding': 'gzip'},
                         preload_content=False)

        self.assertEqual(r.read(11), b'')
        self.assertEqual(r.read(1), b'f')
        self.assertEqual(r.read(2), b'oo')

    def test_io(self):
        import socket
        try:
            from http.client import HTTPResponse as OldHTTPResponse
        except:
            from httplib import HTTPResponse as OldHTTPResponse

        fp = BytesIO(b'foo')
        resp = HTTPResponse(fp, preload_content=False)

        self.assertEqual(resp.closed, False)
        self.assertEqual(resp.readable(), True)
        self.assertEqual(resp.writable(), False)
        self.assertRaises(IOError, resp.fileno)

        resp.close()
        self.assertEqual(resp.closed, True)

        # Try closing with an `httplib.HTTPResponse`, because it has an
        # `isclosed` method.
        hlr = OldHTTPResponse(socket.socket())
        resp2 = HTTPResponse(hlr, preload_content=False)
        self.assertEqual(resp2.closed, False)
        resp2.close()
        self.assertEqual(resp2.closed, True)

        #also try when only data is present.
        resp3 = HTTPResponse('foodata')
        self.assertRaises(IOError, resp3.fileno)

        resp3._fp = 2
        # A corner case where _fp is present but doesn't have `closed`,
        # `isclosed`, or `fileno`.  Unlikely, but possible.
        self.assertEqual(resp3.closed, True)
        self.assertRaises(IOError, resp3.fileno)

    def test_io_bufferedreader(self):
        fp = BytesIO(b'foo')
        resp = HTTPResponse(fp, preload_content=False)
        br = BufferedReader(resp)

        self.assertEqual(br.read(), b'foo')

        br.close()
        self.assertEqual(resp.closed, True)

    def test_streaming(self):
        fp = BytesIO(b'foo')
        resp = HTTPResponse(fp, preload_content=False)
        stream = resp.stream(2, decode_content=False)

        self.assertEqual(next(stream), b'fo')
        self.assertEqual(next(stream), b'o')
        self.assertRaises(StopIteration, next, stream)

    def test_gzipped_streaming(self):
        import zlib
        compress = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS)
        data = compress.compress(b'foo')
        data += compress.flush()

        fp = BytesIO(data)
        resp = HTTPResponse(fp, headers={'content-encoding': 'gzip'},
                         preload_content=False)
        stream = resp.stream(2)

        self.assertEqual(next(stream), b'f')
        self.assertEqual(next(stream), b'oo')
        self.assertRaises(StopIteration, next, stream)

    def test_deflate_streaming(self):
        import zlib
        data = zlib.compress(b'foo')

        fp = BytesIO(data)
        resp = HTTPResponse(fp, headers={'content-encoding': 'deflate'},
                         preload_content=False)
        stream = resp.stream(2)

        self.assertEqual(next(stream), b'f')
        self.assertEqual(next(stream), b'oo')
        self.assertRaises(StopIteration, next, stream)

    def test_deflate2_streaming(self):
        import zlib
        compress = zlib.compressobj(6, zlib.DEFLATED, -zlib.MAX_WBITS)
        data = compress.compress(b'foo')
        data += compress.flush()

        fp = BytesIO(data)
        resp = HTTPResponse(fp, headers={'content-encoding': 'deflate'},
                         preload_content=False)
        stream = resp.stream(2)

        self.assertEqual(next(stream), b'f')
        self.assertEqual(next(stream), b'oo')
        self.assertRaises(StopIteration, next, stream)

    def test_empty_stream(self):
        fp = BytesIO(b'')
        resp = HTTPResponse(fp, preload_content=False)
        stream = resp.stream(2, decode_content=False)

        self.assertRaises(StopIteration, next, stream)

    def test_mock_httpresponse_stream(self):
        # Mock out a HTTP Request that does enough to make it through urllib3's
        # read() and close() calls, and also exhausts and underlying file
        # object.
        class MockHTTPRequest(object):
            self.fp = None

            def read(self, amt):
                data = self.fp.read(amt)
                if not data:
                    self.fp = None

                return data

            def close(self):
                self.fp = None

        bio = BytesIO(b'foo')
        fp = MockHTTPRequest()
        fp.fp = bio
        resp = HTTPResponse(fp, preload_content=False)
        stream = resp.stream(2)

        self.assertEqual(next(stream), b'fo')
        self.assertEqual(next(stream), b'o')
        self.assertRaises(StopIteration, next, stream)


if __name__ == '__main__':
    unittest.main()

1 by Daniele Tricoli Import upstream version 1.2.2	1	import unittest
1 by Daniele Tricoli Import upstream version 1.2.2	2
1.2.2 by Daniele Tricoli Import upstream version 1.7.1	3	from io import BytesIO, BufferedReader
1 by Daniele Tricoli Import upstream version 1.2.2	4
1 by Daniele Tricoli Import upstream version 1.2.2	5	from urllib3.response import HTTPResponse
1.1.2 by Barry Warsaw Import upstream version 1.5	6	from urllib3.exceptions import DecodeError
1 by Daniele Tricoli Import upstream version 1.2.2	7
	8	class TestLegacyResponse(unittest.TestCase):
	9	def test_getheaders(self):
	10	headers = {'host': 'example.com'}
	11	r = HTTPResponse(headers=headers)
	12	self.assertEqual(r.getheaders(), headers)
	13
	14	def test_getheader(self):
	15	headers = {'host': 'example.com'}
	16	r = HTTPResponse(headers=headers)
	17	self.assertEqual(r.getheader('host'), 'example.com')
	18
	19
	20	class TestResponse(unittest.TestCase):
	21	def test_cache_content(self):
	22	r = HTTPResponse('foo')
	23	self.assertEqual(r.data, 'foo')
	24	self.assertEqual(r._body, 'foo')
	25
	26	def test_default(self):
	27	r = HTTPResponse()
	28	self.assertEqual(r.data, None)
	29
	30	def test_none(self):
	31	r = HTTPResponse(None)
	32	self.assertEqual(r.data, None)
	33
	34	def test_preload(self):
	35	fp = BytesIO(b'foo')
	36
	37	r = HTTPResponse(fp, preload_content=True)
	38
	39	self.assertEqual(fp.tell(), len(b'foo'))
	40	self.assertEqual(r.data, b'foo')
	41
	42	def test_no_preload(self):
	43	fp = BytesIO(b'foo')
	44
	45	r = HTTPResponse(fp, preload_content=False)
	46
	47	self.assertEqual(fp.tell(), 0)
	48	self.assertEqual(r.data, b'foo')
	49	self.assertEqual(fp.tell(), len(b'foo'))
	50
	51	def test_decode_bad_data(self):
	52	fp = BytesIO(b'\x00' * 10)
1.1.2 by Barry Warsaw Import upstream version 1.5	53	self.assertRaises(DecodeError, HTTPResponse, fp, headers={
1 by Daniele Tricoli Import upstream version 1.2.2	54	'content-encoding': 'deflate'
	55	})
	56
	57	def test_decode_deflate(self):
	58	import zlib
	59	data = zlib.compress(b'foo')
	60
	61	fp = BytesIO(data)
	62	r = HTTPResponse(fp, headers={'content-encoding': 'deflate'})
	63
	64	self.assertEqual(r.data, b'foo')
	65
1.2.1 by Daniele Tricoli Import upstream version 1.6	66	def test_decode_deflate_case_insensitve(self):
	67	import zlib
	68	data = zlib.compress(b'foo')
	69
	70	fp = BytesIO(data)
	71	r = HTTPResponse(fp, headers={'content-encoding': 'DeFlAtE'})
	72
	73	self.assertEqual(r.data, b'foo')
	74
	75	def test_chunked_decoding_deflate(self):
	76	import zlib
	77	data = zlib.compress(b'foo')
	78
	79	fp = BytesIO(data)
	80	r = HTTPResponse(fp, headers={'content-encoding': 'deflate'},
	81	preload_content=False)
	82
	83	self.assertEqual(r.read(3), b'')
	84	self.assertEqual(r.read(1), b'f')
	85	self.assertEqual(r.read(2), b'oo')
	86
	87	def test_chunked_decoding_deflate2(self):
	88	import zlib
	89	compress = zlib.compressobj(6, zlib.DEFLATED, -zlib.MAX_WBITS)
	90	data = compress.compress(b'foo')
	91	data += compress.flush()
	92
	93	fp = BytesIO(data)
	94	r = HTTPResponse(fp, headers={'content-encoding': 'deflate'},
	95	preload_content=False)
	96
	97	self.assertEqual(r.read(1), b'')
	98	self.assertEqual(r.read(1), b'f')
	99	self.assertEqual(r.read(2), b'oo')
	100
	101	def test_chunked_decoding_gzip(self):
	102	import zlib
	103	compress = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS)
	104	data = compress.compress(b'foo')
	105	data += compress.flush()
	106
	107	fp = BytesIO(data)
	108	r = HTTPResponse(fp, headers={'content-encoding': 'gzip'},
	109	preload_content=False)
	110
	111	self.assertEqual(r.read(11), b'')
	112	self.assertEqual(r.read(1), b'f')
	113	self.assertEqual(r.read(2), b'oo')
1 by Daniele Tricoli Import upstream version 1.2.2	114
1.2.2 by Daniele Tricoli Import upstream version 1.7.1	115	def test_io(self):
	116	import socket
	117	try:
	118	from http.client import HTTPResponse as OldHTTPResponse
	119	except:
	120	from httplib import HTTPResponse as OldHTTPResponse
	121
	122	fp = BytesIO(b'foo')
	123	resp = HTTPResponse(fp, preload_content=False)
	124
	125	self.assertEqual(resp.closed, False)
	126	self.assertEqual(resp.readable(), True)
	127	self.assertEqual(resp.writable(), False)
	128	self.assertRaises(IOError, resp.fileno)
	129
	130	resp.close()
	131	self.assertEqual(resp.closed, True)
	132
	133	# Try closing with an `httplib.HTTPResponse`, because it has an
	134	# `isclosed` method.
	135	hlr = OldHTTPResponse(socket.socket())
	136	resp2 = HTTPResponse(hlr, preload_content=False)
	137	self.assertEqual(resp2.closed, False)
	138	resp2.close()
	139	self.assertEqual(resp2.closed, True)
	140
	141	#also try when only data is present.
	142	resp3 = HTTPResponse('foodata')
	143	self.assertRaises(IOError, resp3.fileno)
	144
	145	resp3._fp = 2
	146	# A corner case where _fp is present but doesn't have `closed`,
	147	# `isclosed`, or `fileno`. Unlikely, but possible.
	148	self.assertEqual(resp3.closed, True)
	149	self.assertRaises(IOError, resp3.fileno)
	150
	151	def test_io_bufferedreader(self):
	152	fp = BytesIO(b'foo')
	153	resp = HTTPResponse(fp, preload_content=False)
	154	br = BufferedReader(resp)
	155
	156	self.assertEqual(br.read(), b'foo')
	157
	158	br.close()
	159	self.assertEqual(resp.closed, True)
	160
	161	def test_streaming(self):
	162	fp = BytesIO(b'foo')
	163	resp = HTTPResponse(fp, preload_content=False)
	164	stream = resp.stream(2, decode_content=False)
	165
	166	self.assertEqual(next(stream), b'fo')
	167	self.assertEqual(next(stream), b'o')
	168	self.assertRaises(StopIteration, next, stream)
	169
	170	def test_gzipped_streaming(self):
	171	import zlib
	172	compress = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS)
	173	data = compress.compress(b'foo')
	174	data += compress.flush()
	175
	176	fp = BytesIO(data)
	177	resp = HTTPResponse(fp, headers={'content-encoding': 'gzip'},
	178	preload_content=False)
179	stream = resp.stream(2)
180
181	self.assertEqual(next(stream), b'f')
182	self.assertEqual(next(stream), b'oo')
183	self.assertRaises(StopIteration, next, stream)
184
185	def test_deflate_streaming(self):
186	import zlib
187	data = zlib.compress(b'foo')
188
189	fp = BytesIO(data)
190	resp = HTTPResponse(fp, headers={'content-encoding': 'deflate'},
191	preload_content=False)
192	stream = resp.stream(2)
193
194	self.assertEqual(next(stream), b'f')
195	self.assertEqual(next(stream), b'oo')
196	self.assertRaises(StopIteration, next, stream)
197
198	def test_deflate2_streaming(self):
199	import zlib
200	compress = zlib.compressobj(6, zlib.DEFLATED, -zlib.MAX_WBITS)
201	data = compress.compress(b'foo')
202	data += compress.flush()
203
204	fp = BytesIO(data)
205	resp = HTTPResponse(fp, headers={'content-encoding': 'deflate'},
206	preload_content=False)
207	stream = resp.stream(2)
208
209	self.assertEqual(next(stream), b'f')
210	self.assertEqual(next(stream), b'oo')
211	self.assertRaises(StopIteration, next, stream)
212
213	def test_empty_stream(self):
214	fp = BytesIO(b'')
215	resp = HTTPResponse(fp, preload_content=False)
216	stream = resp.stream(2, decode_content=False)
217
218	self.assertRaises(StopIteration, next, stream)
219
220	def test_mock_httpresponse_stream(self):
221	# Mock out a HTTP Request that does enough to make it through urllib3's
222	# read() and close() calls, and also exhausts and underlying file
223	# object.
224	class MockHTTPRequest(object):
225	self.fp = None
226
227	def read(self, amt):
228	data = self.fp.read(amt)
229	if not data:
230	self.fp = None
231
232	return data
233
234	def close(self):
235	self.fp = None
236
237	bio = BytesIO(b'foo')
238	fp = MockHTTPRequest()
239	fp.fp = bio
240	resp = HTTPResponse(fp, preload_content=False)
241	stream = resp.stream(2)
242
243	self.assertEqual(next(stream), b'fo')
244	self.assertEqual(next(stream), b'o')
245	self.assertRaises(StopIteration, next, stream)
246
247
1 by Daniele Tricoli Import upstream version 1.2.2	248	if __name__ == '__main__':
1 by Daniele Tricoli Import upstream version 1.2.2	249	unittest.main()