← Back to branch summary

~ubuntu-branches/ubuntu/trusty/python-urllib3/trusty-proposed

« back to all changes in this revision

Viewing changes to test/test_response.py

  • Committer: Package Import Robot
  • Author(s): Daniele Tricoli
  • Date: 2013-10-17 13:28:10 UTC
  • mfrom: (4.1.4 sid)
  • Revision ID: package-import@ubuntu.com-20131017132810-dt44kuqupqqulnwc
Tags: 1.7.1-1
* New upstream release
* Switched to pybuild
* debian/clean
  - Switched to debian/clean for cleaning instead of using debian/rules
* debian/compat
  - Bumped debhelper compatibility level to 9
* debian/control
  - Added python-mock to Build-Depends
  - Bumped debhelper B-D to (>= 9)
* debian/copyright
  - Removed stanza about mimetools_choose_boundary since not shipped
    anymore
* debian/patches/01_do-not-use-embedded-python-six.patch
  - Refreshed
* debian/patches/02_require-cert-verification.patch
  - Refreshed
* debian/patches/04_relax_nosetests_options.patch
  - Refreshed
* debian/patches/05_fix_python3_syntax_error_in_ntlmpool.patch
  - Removed since fixed upstream
* debian/patches/06_fix_abuse_of_match_hostname_for_DoS.patch
  - Removed since fixed upstream
* debian/watch
  - Switched download URL to https

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/test_response.py
1
1
import unittest
2
2
 
3
 
from io import BytesIO
 
3
from io import BytesIO, BufferedReader
4
4
 
5
5
from urllib3.response import HTTPResponse
6
6
from urllib3.exceptions import DecodeError
112
112
        self.assertEqual(r.read(1), b'f')
113
113
        self.assertEqual(r.read(2), b'oo')
114
114
 
 
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
 
115
248
if __name__ == '__main__':
116
249
    unittest.main()