~pythonregexp2.7/python/issue2636-12

« back to all changes in this revision

Viewing changes to Lib/test/test_urllib2net.py

  • Committer: Jeffrey C. "The TimeHorse" Jacobs
  • Date: 2008-06-09 14:52:42 UTC
  • mfrom: (39033.1.3 Regexp-2.6)
  • Revision ID: darklord@timehorse.com-20080609145242-9m268zc6u87rp1vp
Merged in changes from the core Regexp branch.

Show diffs side-by-side

added added

removed removed

Lines of Context:
11
11
import mimetools
12
12
 
13
13
 
14
 
def _urlopen_with_retry(host, *args, **kwargs):
15
 
    # Connecting to remote hosts is flaky.  Make it more robust
16
 
    # by retrying the connection several times.
 
14
def _retry_thrice(func, exc, *args, **kwargs):
17
15
    for i in range(3):
18
16
        try:
19
 
            return urllib2.urlopen(host, *args, **kwargs)
20
 
        except urllib2.URLError, last_exc:
 
17
            return func(*args, **kwargs)
 
18
        except exc, last_exc:
21
19
            continue
22
20
        except:
23
21
            raise
24
22
    raise last_exc
25
23
 
 
24
def _wrap_with_retry_thrice(func, exc):
 
25
    def wrapped(*args, **kwargs):
 
26
        return _retry_thrice(func, exc, *args, **kwargs)
 
27
    return wrapped
 
28
 
 
29
# Connecting to remote hosts is flaky.  Make it more robust by retrying
 
30
# the connection several times.
 
31
_urlopen_with_retry = _wrap_with_retry_thrice(urllib2.urlopen, urllib2.URLError)
26
32
 
27
33
 
28
34
class AuthTests(unittest.TestCase):
115
121
                'file:'+sanepathname2url(os.path.abspath(TESTFN)),
116
122
                ('file:///nonsensename/etc/passwd', None, urllib2.URLError),
117
123
                ]
118
 
            self._test_urls(urls, self._extra_handlers(), urllib2.urlopen)
 
124
            self._test_urls(urls, self._extra_handlers(), retry=True)
119
125
        finally:
120
126
            os.remove(TESTFN)
121
127
 
147
153
 
148
154
##             self._test_urls(urls, self._extra_handlers()+[bauth, dauth])
149
155
 
150
 
    def _test_urls(self, urls, handlers, urlopen=_urlopen_with_retry):
 
156
    def _test_urls(self, urls, handlers, retry=True):
151
157
        import socket
152
158
        import time
153
159
        import logging
154
160
        debug = logging.getLogger("test_urllib2").debug
155
161
 
156
 
        urllib2.install_opener(urllib2.build_opener(*handlers))
 
162
        urlopen = urllib2.build_opener(*handlers).open
 
163
        if retry:
 
164
            urlopen = _wrap_with_retry_thrice(urlopen, urllib2.URLError)
157
165
 
158
166
        for url in urls:
159
167
            if isinstance(url, tuple):
189
197
 
190
198
class TimeoutTest(unittest.TestCase):
191
199
    def test_http_basic(self):
 
200
        self.assertTrue(socket.getdefaulttimeout() is None)
192
201
        u = _urlopen_with_retry("http://www.python.org")
193
202
        self.assertTrue(u.fp._sock.fp._sock.gettimeout() is None)
194
203
 
195
 
    def test_http_NoneWithdefault(self):
196
 
        prev = socket.getdefaulttimeout()
 
204
    def test_http_default_timeout(self):
 
205
        self.assertTrue(socket.getdefaulttimeout() is None)
 
206
        socket.setdefaulttimeout(60)
 
207
        try:
 
208
            u = _urlopen_with_retry("http://www.python.org")
 
209
        finally:
 
210
            socket.setdefaulttimeout(None)
 
211
        self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 60)
 
212
 
 
213
    def test_http_no_timeout(self):
 
214
        self.assertTrue(socket.getdefaulttimeout() is None)
197
215
        socket.setdefaulttimeout(60)
198
216
        try:
199
217
            u = _urlopen_with_retry("http://www.python.org", timeout=None)
200
 
            self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 60)
201
218
        finally:
202
 
            socket.setdefaulttimeout(prev)
 
219
            socket.setdefaulttimeout(None)
 
220
        self.assertTrue(u.fp._sock.fp._sock.gettimeout() is None)
203
221
 
204
 
    def test_http_Value(self):
 
222
    def test_http_timeout(self):
205
223
        u = _urlopen_with_retry("http://www.python.org", timeout=120)
206
224
        self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 120)
207
225
 
208
 
    def test_http_NoneNodefault(self):
209
 
        u = _urlopen_with_retry("http://www.python.org", timeout=None)
210
 
        self.assertTrue(u.fp._sock.fp._sock.gettimeout() is None)
211
 
 
212
226
    FTP_HOST = "ftp://ftp.mirror.nl/pub/mirror/gnu/"
213
227
 
214
228
    def test_ftp_basic(self):
 
229
        self.assertTrue(socket.getdefaulttimeout() is None)
215
230
        u = _urlopen_with_retry(self.FTP_HOST)
216
231
        self.assertTrue(u.fp.fp._sock.gettimeout() is None)
217
232
 
218
 
    def test_ftp_NoneWithdefault(self):
219
 
        prev = socket.getdefaulttimeout()
 
233
    def test_ftp_default_timeout(self):
 
234
        self.assertTrue(socket.getdefaulttimeout() is None)
 
235
        socket.setdefaulttimeout(60)
 
236
        try:
 
237
            u = _urlopen_with_retry(self.FTP_HOST)
 
238
        finally:
 
239
            socket.setdefaulttimeout(None)
 
240
        self.assertEqual(u.fp.fp._sock.gettimeout(), 60)
 
241
 
 
242
    def test_ftp_no_timeout(self):
 
243
        self.assertTrue(socket.getdefaulttimeout() is None)
220
244
        socket.setdefaulttimeout(60)
221
245
        try:
222
246
            u = _urlopen_with_retry(self.FTP_HOST, timeout=None)
223
 
            self.assertEqual(u.fp.fp._sock.gettimeout(), 60)
224
247
        finally:
225
 
            socket.setdefaulttimeout(prev)
226
 
 
227
 
    def test_ftp_NoneNodefault(self):
228
 
        u = _urlopen_with_retry(self.FTP_HOST, timeout=None)
 
248
            socket.setdefaulttimeout(None)
229
249
        self.assertTrue(u.fp.fp._sock.gettimeout() is None)
230
250
 
231
 
    def test_ftp_Value(self):
 
251
    def test_ftp_timeout(self):
232
252
        u = _urlopen_with_retry(self.FTP_HOST, timeout=60)
233
253
        self.assertEqual(u.fp.fp._sock.gettimeout(), 60)
234
254