ยปCore Development>Code coverage>Lib/test/test_urllib2net.py

Python code coverage for Lib/test/test_urllib2net.py

#countcontent
1n/aimport unittest
2n/afrom test import support
3n/afrom test.test_urllib2 import sanepathname2url
4n/a
5n/aimport os
6n/aimport socket
7n/aimport urllib.error
8n/aimport urllib.request
9n/aimport sys
10n/a
11n/asupport.requires("network")
12n/a
13n/aTIMEOUT = 60 # seconds
14n/a
15n/a
16n/adef _retry_thrice(func, exc, *args, **kwargs):
17n/a for i in range(3):
18n/a try:
19n/a return func(*args, **kwargs)
20n/a except exc as e:
21n/a last_exc = e
22n/a continue
23n/a raise last_exc
24n/a
25n/adef _wrap_with_retry_thrice(func, exc):
26n/a def wrapped(*args, **kwargs):
27n/a return _retry_thrice(func, exc, *args, **kwargs)
28n/a return wrapped
29n/a
30n/a# Connecting to remote hosts is flaky. Make it more robust by retrying
31n/a# the connection several times.
32n/a_urlopen_with_retry = _wrap_with_retry_thrice(urllib.request.urlopen,
33n/a urllib.error.URLError)
34n/a
35n/a
36n/aclass AuthTests(unittest.TestCase):
37n/a """Tests urllib2 authentication features."""
38n/a
39n/a## Disabled at the moment since there is no page under python.org which
40n/a## could be used to HTTP authentication.
41n/a#
42n/a# def test_basic_auth(self):
43n/a# import http.client
44n/a#
45n/a# test_url = "http://www.python.org/test/test_urllib2/basic_auth"
46n/a# test_hostport = "www.python.org"
47n/a# test_realm = 'Test Realm'
48n/a# test_user = 'test.test_urllib2net'
49n/a# test_password = 'blah'
50n/a#
51n/a# # failure
52n/a# try:
53n/a# _urlopen_with_retry(test_url)
54n/a# except urllib2.HTTPError, exc:
55n/a# self.assertEqual(exc.code, 401)
56n/a# else:
57n/a# self.fail("urlopen() should have failed with 401")
58n/a#
59n/a# # success
60n/a# auth_handler = urllib2.HTTPBasicAuthHandler()
61n/a# auth_handler.add_password(test_realm, test_hostport,
62n/a# test_user, test_password)
63n/a# opener = urllib2.build_opener(auth_handler)
64n/a# f = opener.open('http://localhost/')
65n/a# response = _urlopen_with_retry("http://www.python.org/")
66n/a#
67n/a# # The 'userinfo' URL component is deprecated by RFC 3986 for security
68n/a# # reasons, let's not implement it! (it's already implemented for proxy
69n/a# # specification strings (that is, URLs or authorities specifying a
70n/a# # proxy), so we must keep that)
71n/a# self.assertRaises(http.client.InvalidURL,
72n/a# urllib2.urlopen, "http://evil:thing@example.com")
73n/a
74n/a
75n/aclass CloseSocketTest(unittest.TestCase):
76n/a
77n/a def test_close(self):
78n/a # calling .close() on urllib2's response objects should close the
79n/a # underlying socket
80n/a url = "http://www.example.com/"
81n/a with support.transient_internet(url):
82n/a response = _urlopen_with_retry(url)
83n/a sock = response.fp
84n/a self.assertFalse(sock.closed)
85n/a response.close()
86n/a self.assertTrue(sock.closed)
87n/a
88n/aclass OtherNetworkTests(unittest.TestCase):
89n/a def setUp(self):
90n/a if 0: # for debugging
91n/a import logging
92n/a logger = logging.getLogger("test_urllib2net")
93n/a logger.addHandler(logging.StreamHandler())
94n/a
95n/a # XXX The rest of these tests aren't very good -- they don't check much.
96n/a # They do sometimes catch some major disasters, though.
97n/a
98n/a def test_ftp(self):
99n/a urls = [
100n/a 'ftp://ftp.debian.org/debian/README',
101n/a ('ftp://ftp.debian.org/debian/non-existent-file',
102n/a None, urllib.error.URLError),
103n/a ]
104n/a self._test_urls(urls, self._extra_handlers())
105n/a
106n/a def test_file(self):
107n/a TESTFN = support.TESTFN
108n/a f = open(TESTFN, 'w')
109n/a try:
110n/a f.write('hi there\n')
111n/a f.close()
112n/a urls = [
113n/a 'file:' + sanepathname2url(os.path.abspath(TESTFN)),
114n/a ('file:///nonsensename/etc/passwd', None,
115n/a urllib.error.URLError),
116n/a ]
117n/a self._test_urls(urls, self._extra_handlers(), retry=True)
118n/a finally:
119n/a os.remove(TESTFN)
120n/a
121n/a self.assertRaises(ValueError, urllib.request.urlopen,'./relative_path/to/file')
122n/a
123n/a # XXX Following test depends on machine configurations that are internal
124n/a # to CNRI. Need to set up a public server with the right authentication
125n/a # configuration for test purposes.
126n/a
127n/a## def test_cnri(self):
128n/a## if socket.gethostname() == 'bitdiddle':
129n/a## localhost = 'bitdiddle.cnri.reston.va.us'
130n/a## elif socket.gethostname() == 'bitdiddle.concentric.net':
131n/a## localhost = 'localhost'
132n/a## else:
133n/a## localhost = None
134n/a## if localhost is not None:
135n/a## urls = [
136n/a## 'file://%s/etc/passwd' % localhost,
137n/a## 'http://%s/simple/' % localhost,
138n/a## 'http://%s/digest/' % localhost,
139n/a## 'http://%s/not/found.h' % localhost,
140n/a## ]
141n/a
142n/a## bauth = HTTPBasicAuthHandler()
143n/a## bauth.add_password('basic_test_realm', localhost, 'jhylton',
144n/a## 'password')
145n/a## dauth = HTTPDigestAuthHandler()
146n/a## dauth.add_password('digest_test_realm', localhost, 'jhylton',
147n/a## 'password')
148n/a
149n/a## self._test_urls(urls, self._extra_handlers()+[bauth, dauth])
150n/a
151n/a def test_urlwithfrag(self):
152n/a urlwith_frag = "http://www.pythontest.net/index.html#frag"
153n/a with support.transient_internet(urlwith_frag):
154n/a req = urllib.request.Request(urlwith_frag)
155n/a res = urllib.request.urlopen(req)
156n/a self.assertEqual(res.geturl(),
157n/a "http://www.pythontest.net/index.html#frag")
158n/a
159n/a def test_redirect_url_withfrag(self):
160n/a redirect_url_with_frag = "http://www.pythontest.net/redir/with_frag/"
161n/a with support.transient_internet(redirect_url_with_frag):
162n/a req = urllib.request.Request(redirect_url_with_frag)
163n/a res = urllib.request.urlopen(req)
164n/a self.assertEqual(res.geturl(),
165n/a "http://www.pythontest.net/elsewhere/#frag")
166n/a
167n/a def test_custom_headers(self):
168n/a url = "http://www.example.com"
169n/a with support.transient_internet(url):
170n/a opener = urllib.request.build_opener()
171n/a request = urllib.request.Request(url)
172n/a self.assertFalse(request.header_items())
173n/a opener.open(request)
174n/a self.assertTrue(request.header_items())
175n/a self.assertTrue(request.has_header('User-agent'))
176n/a request.add_header('User-Agent','Test-Agent')
177n/a opener.open(request)
178n/a self.assertEqual(request.get_header('User-agent'),'Test-Agent')
179n/a
180n/a def test_sites_no_connection_close(self):
181n/a # Some sites do not send Connection: close header.
182n/a # Verify that those work properly. (#issue12576)
183n/a
184n/a URL = 'http://www.imdb.com' # mangles Connection:close
185n/a
186n/a with support.transient_internet(URL):
187n/a try:
188n/a with urllib.request.urlopen(URL) as res:
189n/a pass
190n/a except ValueError as e:
191n/a self.fail("urlopen failed for site not sending \
192n/a Connection:close")
193n/a else:
194n/a self.assertTrue(res)
195n/a
196n/a req = urllib.request.urlopen(URL)
197n/a res = req.read()
198n/a self.assertTrue(res)
199n/a
200n/a def _test_urls(self, urls, handlers, retry=True):
201n/a import time
202n/a import logging
203n/a debug = logging.getLogger("test_urllib2").debug
204n/a
205n/a urlopen = urllib.request.build_opener(*handlers).open
206n/a if retry:
207n/a urlopen = _wrap_with_retry_thrice(urlopen, urllib.error.URLError)
208n/a
209n/a for url in urls:
210n/a with self.subTest(url=url):
211n/a if isinstance(url, tuple):
212n/a url, req, expected_err = url
213n/a else:
214n/a req = expected_err = None
215n/a
216n/a with support.transient_internet(url):
217n/a try:
218n/a f = urlopen(url, req, TIMEOUT)
219n/a # urllib.error.URLError is a subclass of OSError
220n/a except OSError as err:
221n/a if expected_err:
222n/a msg = ("Didn't get expected error(s) %s for %s %s, got %s: %s" %
223n/a (expected_err, url, req, type(err), err))
224n/a self.assertIsInstance(err, expected_err, msg)
225n/a else:
226n/a raise
227n/a else:
228n/a try:
229n/a with support.time_out, \
230n/a support.socket_peer_reset, \
231n/a support.ioerror_peer_reset:
232n/a buf = f.read()
233n/a debug("read %d bytes" % len(buf))
234n/a except socket.timeout:
235n/a print("<timeout: %s>" % url, file=sys.stderr)
236n/a f.close()
237n/a time.sleep(0.1)
238n/a
239n/a def _extra_handlers(self):
240n/a handlers = []
241n/a
242n/a cfh = urllib.request.CacheFTPHandler()
243n/a self.addCleanup(cfh.clear_cache)
244n/a cfh.setTimeout(1)
245n/a handlers.append(cfh)
246n/a
247n/a return handlers
248n/a
249n/a
250n/aclass TimeoutTest(unittest.TestCase):
251n/a def test_http_basic(self):
252n/a self.assertIsNone(socket.getdefaulttimeout())
253n/a url = "http://www.example.com"
254n/a with support.transient_internet(url, timeout=None):
255n/a u = _urlopen_with_retry(url)
256n/a self.addCleanup(u.close)
257n/a self.assertIsNone(u.fp.raw._sock.gettimeout())
258n/a
259n/a def test_http_default_timeout(self):
260n/a self.assertIsNone(socket.getdefaulttimeout())
261n/a url = "http://www.example.com"
262n/a with support.transient_internet(url):
263n/a socket.setdefaulttimeout(60)
264n/a try:
265n/a u = _urlopen_with_retry(url)
266n/a self.addCleanup(u.close)
267n/a finally:
268n/a socket.setdefaulttimeout(None)
269n/a self.assertEqual(u.fp.raw._sock.gettimeout(), 60)
270n/a
271n/a def test_http_no_timeout(self):
272n/a self.assertIsNone(socket.getdefaulttimeout())
273n/a url = "http://www.example.com"
274n/a with support.transient_internet(url):
275n/a socket.setdefaulttimeout(60)
276n/a try:
277n/a u = _urlopen_with_retry(url, timeout=None)
278n/a self.addCleanup(u.close)
279n/a finally:
280n/a socket.setdefaulttimeout(None)
281n/a self.assertIsNone(u.fp.raw._sock.gettimeout())
282n/a
283n/a def test_http_timeout(self):
284n/a url = "http://www.example.com"
285n/a with support.transient_internet(url):
286n/a u = _urlopen_with_retry(url, timeout=120)
287n/a self.addCleanup(u.close)
288n/a self.assertEqual(u.fp.raw._sock.gettimeout(), 120)
289n/a
290n/a FTP_HOST = 'ftp://ftp.debian.org/debian/'
291n/a
292n/a def test_ftp_basic(self):
293n/a self.assertIsNone(socket.getdefaulttimeout())
294n/a with support.transient_internet(self.FTP_HOST, timeout=None):
295n/a u = _urlopen_with_retry(self.FTP_HOST)
296n/a self.addCleanup(u.close)
297n/a self.assertIsNone(u.fp.fp.raw._sock.gettimeout())
298n/a
299n/a def test_ftp_default_timeout(self):
300n/a self.assertIsNone(socket.getdefaulttimeout())
301n/a with support.transient_internet(self.FTP_HOST):
302n/a socket.setdefaulttimeout(60)
303n/a try:
304n/a u = _urlopen_with_retry(self.FTP_HOST)
305n/a self.addCleanup(u.close)
306n/a finally:
307n/a socket.setdefaulttimeout(None)
308n/a self.assertEqual(u.fp.fp.raw._sock.gettimeout(), 60)
309n/a
310n/a def test_ftp_no_timeout(self):
311n/a self.assertIsNone(socket.getdefaulttimeout())
312n/a with support.transient_internet(self.FTP_HOST):
313n/a socket.setdefaulttimeout(60)
314n/a try:
315n/a u = _urlopen_with_retry(self.FTP_HOST, timeout=None)
316n/a self.addCleanup(u.close)
317n/a finally:
318n/a socket.setdefaulttimeout(None)
319n/a self.assertIsNone(u.fp.fp.raw._sock.gettimeout())
320n/a
321n/a def test_ftp_timeout(self):
322n/a with support.transient_internet(self.FTP_HOST):
323n/a u = _urlopen_with_retry(self.FTP_HOST, timeout=60)
324n/a self.addCleanup(u.close)
325n/a self.assertEqual(u.fp.fp.raw._sock.gettimeout(), 60)
326n/a
327n/a
328n/aif __name__ == "__main__":
329n/a unittest.main()