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

Python code coverage for Lib/test/test_urllibnet.py

#countcontent
1n/aimport unittest
2n/afrom test import support
3n/a
4n/aimport contextlib
5n/aimport socket
6n/aimport urllib.request
7n/aimport os
8n/aimport email.message
9n/aimport time
10n/a
11n/a
12n/asupport.requires('network')
13n/a
14n/aclass URLTimeoutTest(unittest.TestCase):
15n/a # XXX this test doesn't seem to test anything useful.
16n/a
17n/a TIMEOUT = 30.0
18n/a
19n/a def setUp(self):
20n/a socket.setdefaulttimeout(self.TIMEOUT)
21n/a
22n/a def tearDown(self):
23n/a socket.setdefaulttimeout(None)
24n/a
25n/a def testURLread(self):
26n/a with support.transient_internet("www.example.com"):
27n/a f = urllib.request.urlopen("http://www.example.com/")
28n/a x = f.read()
29n/a
30n/a
31n/aclass urlopenNetworkTests(unittest.TestCase):
32n/a """Tests urllib.reqest.urlopen using the network.
33n/a
34n/a These tests are not exhaustive. Assuming that testing using files does a
35n/a good job overall of some of the basic interface features. There are no
36n/a tests exercising the optional 'data' and 'proxies' arguments. No tests
37n/a for transparent redirection have been written.
38n/a
39n/a setUp is not used for always constructing a connection to
40n/a http://www.pythontest.net/ since there a few tests that don't use that address
41n/a and making a connection is expensive enough to warrant minimizing unneeded
42n/a connections.
43n/a
44n/a """
45n/a
46n/a url = 'http://www.pythontest.net/'
47n/a
48n/a @contextlib.contextmanager
49n/a def urlopen(self, *args, **kwargs):
50n/a resource = args[0]
51n/a with support.transient_internet(resource):
52n/a r = urllib.request.urlopen(*args, **kwargs)
53n/a try:
54n/a yield r
55n/a finally:
56n/a r.close()
57n/a
58n/a def test_basic(self):
59n/a # Simple test expected to pass.
60n/a with self.urlopen(self.url) as open_url:
61n/a for attr in ("read", "readline", "readlines", "fileno", "close",
62n/a "info", "geturl"):
63n/a self.assertTrue(hasattr(open_url, attr), "object returned from "
64n/a "urlopen lacks the %s attribute" % attr)
65n/a self.assertTrue(open_url.read(), "calling 'read' failed")
66n/a
67n/a def test_readlines(self):
68n/a # Test both readline and readlines.
69n/a with self.urlopen(self.url) as open_url:
70n/a self.assertIsInstance(open_url.readline(), bytes,
71n/a "readline did not return a string")
72n/a self.assertIsInstance(open_url.readlines(), list,
73n/a "readlines did not return a list")
74n/a
75n/a def test_info(self):
76n/a # Test 'info'.
77n/a with self.urlopen(self.url) as open_url:
78n/a info_obj = open_url.info()
79n/a self.assertIsInstance(info_obj, email.message.Message,
80n/a "object returned by 'info' is not an "
81n/a "instance of email.message.Message")
82n/a self.assertEqual(info_obj.get_content_subtype(), "html")
83n/a
84n/a def test_geturl(self):
85n/a # Make sure same URL as opened is returned by geturl.
86n/a with self.urlopen(self.url) as open_url:
87n/a gotten_url = open_url.geturl()
88n/a self.assertEqual(gotten_url, self.url)
89n/a
90n/a def test_getcode(self):
91n/a # test getcode() with the fancy opener to get 404 error codes
92n/a URL = self.url + "XXXinvalidXXX"
93n/a with support.transient_internet(URL):
94n/a with self.assertWarns(DeprecationWarning):
95n/a open_url = urllib.request.FancyURLopener().open(URL)
96n/a try:
97n/a code = open_url.getcode()
98n/a finally:
99n/a open_url.close()
100n/a self.assertEqual(code, 404)
101n/a
102n/a def test_bad_address(self):
103n/a # Make sure proper exception is raised when connecting to a bogus
104n/a # address.
105n/a
106n/a # Given that both VeriSign and various ISPs have in
107n/a # the past or are presently hijacking various invalid
108n/a # domain name requests in an attempt to boost traffic
109n/a # to their own sites, finding a domain name to use
110n/a # for this test is difficult. RFC2606 leads one to
111n/a # believe that '.invalid' should work, but experience
112n/a # seemed to indicate otherwise. Single character
113n/a # TLDs are likely to remain invalid, so this seems to
114n/a # be the best choice. The trailing '.' prevents a
115n/a # related problem: The normal DNS resolver appends
116n/a # the domain names from the search path if there is
117n/a # no '.' the end and, and if one of those domains
118n/a # implements a '*' rule a result is returned.
119n/a # However, none of this will prevent the test from
120n/a # failing if the ISP hijacks all invalid domain
121n/a # requests. The real solution would be to be able to
122n/a # parameterize the framework with a mock resolver.
123n/a bogus_domain = "sadflkjsasf.i.nvali.d."
124n/a try:
125n/a socket.gethostbyname(bogus_domain)
126n/a except OSError:
127n/a # socket.gaierror is too narrow, since getaddrinfo() may also
128n/a # fail with EAI_SYSTEM and ETIMEDOUT (seen on Ubuntu 13.04),
129n/a # i.e. Python's TimeoutError.
130n/a pass
131n/a else:
132n/a # This happens with some overzealous DNS providers such as OpenDNS
133n/a self.skipTest("%r should not resolve for test to work" % bogus_domain)
134n/a failure_explanation = ('opening an invalid URL did not raise OSError; '
135n/a 'can be caused by a broken DNS server '
136n/a '(e.g. returns 404 or hijacks page)')
137n/a with self.assertRaises(OSError, msg=failure_explanation):
138n/a urllib.request.urlopen("http://{}/".format(bogus_domain))
139n/a
140n/a
141n/aclass urlretrieveNetworkTests(unittest.TestCase):
142n/a """Tests urllib.request.urlretrieve using the network."""
143n/a
144n/a @contextlib.contextmanager
145n/a def urlretrieve(self, *args, **kwargs):
146n/a resource = args[0]
147n/a with support.transient_internet(resource):
148n/a file_location, info = urllib.request.urlretrieve(*args, **kwargs)
149n/a try:
150n/a yield file_location, info
151n/a finally:
152n/a support.unlink(file_location)
153n/a
154n/a def test_basic(self):
155n/a # Test basic functionality.
156n/a with self.urlretrieve(self.logo) as (file_location, info):
157n/a self.assertTrue(os.path.exists(file_location), "file location returned by"
158n/a " urlretrieve is not a valid path")
159n/a with open(file_location, 'rb') as f:
160n/a self.assertTrue(f.read(), "reading from the file location returned"
161n/a " by urlretrieve failed")
162n/a
163n/a def test_specified_path(self):
164n/a # Make sure that specifying the location of the file to write to works.
165n/a with self.urlretrieve(self.logo,
166n/a support.TESTFN) as (file_location, info):
167n/a self.assertEqual(file_location, support.TESTFN)
168n/a self.assertTrue(os.path.exists(file_location))
169n/a with open(file_location, 'rb') as f:
170n/a self.assertTrue(f.read(), "reading from temporary file failed")
171n/a
172n/a def test_header(self):
173n/a # Make sure header returned as 2nd value from urlretrieve is good.
174n/a with self.urlretrieve(self.logo) as (file_location, info):
175n/a self.assertIsInstance(info, email.message.Message,
176n/a "info is not an instance of email.message.Message")
177n/a
178n/a logo = "http://www.pythontest.net/"
179n/a
180n/a def test_data_header(self):
181n/a with self.urlretrieve(self.logo) as (file_location, fileheaders):
182n/a datevalue = fileheaders.get('Date')
183n/a dateformat = '%a, %d %b %Y %H:%M:%S GMT'
184n/a try:
185n/a time.strptime(datevalue, dateformat)
186n/a except ValueError:
187n/a self.fail('Date value not in %r format' % dateformat)
188n/a
189n/a def test_reporthook(self):
190n/a records = []
191n/a def recording_reporthook(blocks, block_size, total_size):
192n/a records.append((blocks, block_size, total_size))
193n/a
194n/a with self.urlretrieve(self.logo, reporthook=recording_reporthook) as (
195n/a file_location, fileheaders):
196n/a expected_size = int(fileheaders['Content-Length'])
197n/a
198n/a records_repr = repr(records) # For use in error messages.
199n/a self.assertGreater(len(records), 1, msg="There should always be two "
200n/a "calls; the first one before the transfer starts.")
201n/a self.assertEqual(records[0][0], 0)
202n/a self.assertGreater(records[0][1], 0,
203n/a msg="block size can't be 0 in %s" % records_repr)
204n/a self.assertEqual(records[0][2], expected_size)
205n/a self.assertEqual(records[-1][2], expected_size)
206n/a
207n/a block_sizes = {block_size for _, block_size, _ in records}
208n/a self.assertEqual({records[0][1]}, block_sizes,
209n/a msg="block sizes in %s must be equal" % records_repr)
210n/a self.assertGreaterEqual(records[-1][0]*records[0][1], expected_size,
211n/a msg="number of blocks * block size must be"
212n/a " >= total size in %s" % records_repr)
213n/a
214n/a
215n/aif __name__ == "__main__":
216n/a unittest.main()