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

        with pytest.raises(StopIteration):
            next(stream)

    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_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)

    def test_reference_read(self):
        fp = BytesIO(b'foo')
        r = HTTPResponse(fp, preload_content=False)

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

    def test_reference_read(self):
        fp = BytesIO(b'foo')
        r = HTTPResponse(fp, preload_content=False)

        assert r.read(1) == b'f'
        assert r.read(2) == b'oo'
        assert r.read() == b''
        assert r.read() == b''

    def test_reference_read(self):
        fp = BytesIO(b"foo")
        r = HTTPResponse(fp, preload_content=False)

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

    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_streaming(self):
        fp = BytesIO(b'foo')
        resp = HTTPResponse(fp, preload_content=False)
        stream = resp.stream(2, decode_content=False)

        assert next(stream) == b'fo'
        assert next(stream) == b'o'
        with pytest.raises(StopIteration):
            next(stream)

 def test_invalid_chunks(self):
     stream = [b"foooo", b"bbbbaaaaar"]
     fp = MockChunkedInvalidEncoding(stream)
     r = httplib.HTTPResponse(MockSock)
     r.fp = fp
     r.chunked = True
     r.chunk_left = None
     resp = HTTPResponse(r, preload_content=False, headers={'transfer-encoding': 'chunked'})
     self.assertRaises(ProtocolError, next, resp.read_chunked())

 def test_chunked_response_with_extensions(self):
     stream = [b"foo", b"bar"]
     fp = MockChunkedEncodingWithExtensions(stream)
     r = httplib.HTTPResponse(MockSock)
     r.fp = fp
     r.chunked = True
     r.chunk_left = None
     resp = HTTPResponse(r, preload_content=False, headers={'transfer-encoding': 'chunked'})
     assert stream == list(resp.stream())

 def test_mock_transfer_encoding_chunked_unlmtd_read(self):
     stream = [b"foooo", b"bbbbaaaaar"]
     fp = MockChunkedEncodingResponse(stream)
     r = httplib.HTTPResponse(MockSock)
     r.fp = fp
     r.chunked = True
     r.chunk_left = None
     resp = HTTPResponse(r, preload_content=False, headers={'transfer-encoding': 'chunked'})
     assert stream == list(resp.read_chunked())

    def test_mock_transfer_encoding_chunked(self):
        stream = [b"fo", b"o", b"bar"]
        fp = MockChunkedEncodingResponse(stream)
        r = httplib.HTTPResponse(MockSock)
        r.fp = fp
        resp = HTTPResponse(r, preload_content=False, headers={'transfer-encoding': 'chunked'})

        for i, c in enumerate(resp.stream()):
            assert c == stream[i]

    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_mock_transfer_encoding_chunked(self):
        stream = [b"fo", b"o", b"bar"]
        fp = MockChunkedEncodingResponse(stream)
        r = httplib.HTTPResponse(MockSock)
        r.fp = fp
        resp = HTTPResponse(r, preload_content=False, headers={'transfer-encoding': 'chunked'})

        i = 0
        for c in resp.stream():
            self.assertEqual(c, stream[i])
            i += 1

 def test_geturl_retries(self):
     fp = BytesIO(b'')
     resp = HTTPResponse(fp, request_url='http://example.com')
     request_histories = [
         RequestHistory(method='GET', url='http://example.com', error=None,
                        status=301, redirect_location='https://example.com/'),
         RequestHistory(method='GET', url='https://example.com/', error=None,
                        status=301, redirect_location='https://www.example.com')]
     retry = Retry(history=request_histories)
     resp = HTTPResponse(fp, retries=retry)
     assert resp.geturl() == 'https://www.example.com'

    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_mock_transfer_encoding_chunked_custom_read(self):
     stream = [b"foooo", b"bbbbaaaaar"]
     fp = MockChunkedEncodingResponse(stream)
     r = httplib.HTTPResponse(MockSock)
     r.fp = fp
     r.chunked = True
     r.chunk_left = None
     resp = HTTPResponse(r, preload_content=False, headers={'transfer-encoding': 'chunked'})
     expected_response = [b'fo', b'oo', b'o', b'bb', b'bb', b'aa', b'aa', b'ar']
     response = list(resp.read_chunked(2))
     assert expected_response == response

    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_deflate_streaming(self):
        data = zlib.compress(b'foo')

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

        assert next(stream) == b'f'
        assert next(stream) == b'oo'
        with pytest.raises(StopIteration):
            next(stream)

    def test_deflate_streaming_tell_intermediate_point(self):
        # Ensure that ``tell()`` returns the correct number of bytes when
        # part-way through streaming compressed content.
        import zlib

        NUMBER_OF_READS = 10

        class MockCompressedDataReading(BytesIO):
            """
            A ByteIO-like reader returning ``payload`` in ``NUMBER_OF_READS``
            calls to ``read``.
            """

            def __init__(self, payload, payload_part_size):
                self.payloads = [
                    payload[i*payload_part_size:(i+1)*payload_part_size]
                    for i in range(NUMBER_OF_READS+1)]

                assert b"".join(self.payloads) == payload

            def read(self, _):
                # Amount is unused.
                if len(self.payloads) > 0:
                    return self.payloads.pop(0)
                return b""

        uncompressed_data = zlib.decompress(ZLIB_PAYLOAD)

        payload_part_size = len(ZLIB_PAYLOAD) // NUMBER_OF_READS
        fp = MockCompressedDataReading(ZLIB_PAYLOAD, payload_part_size)
        resp = HTTPResponse(fp, headers={'content-encoding': 'deflate'},
                            preload_content=False)
        stream = resp.stream()

        parts_positions = [(part, resp.tell()) for part in stream]
        end_of_stream = resp.tell()

        with pytest.raises(StopIteration):
            next(stream)

        parts, positions = zip(*parts_positions)

        # Check that the payload is equal to the uncompressed data
        payload = b"".join(parts)
        assert uncompressed_data == payload

        # Check that the positions in the stream are correct
        expected = [(i+1)*payload_part_size for i in range(NUMBER_OF_READS)]
        assert expected == list(positions)

        # Check that the end of the stream is in the correct place
        assert len(ZLIB_PAYLOAD) == end_of_stream