Changeset 391 for python/trunk/Lib/test/test_urllib2net.py

Timestamp:

Mar 19, 2014, 11:31:01 PM (11 years ago)

Author:

dmik

Message:

python: Merge vendor 2.7.6 to trunk.

Location:

python/trunk

Files:

• . (modified) (1 prop)
• Lib/test/test_urllib2net.py (modified) (8 diffs)

python/trunk/Lib/test/test_urllib2net.py

@@ -7 +7 @@
 import socket
 import urllib2
+import os
 import sys
-import os
-import mimetools
+
+TIMEOUT = 60  # seconds
 
 
@@ -74 +75 @@
 
     def test_close(self):
-        import socket, httplib, gc
+        import httplib
 
         # calling .close() on urllib2's response objects should close the
@@ -82 +83 @@
         response = _urlopen_with_retry("http://www.python.org/")
         abused_fileobject = response.fp
-        self.assert_(abused_fileobject.__class__ is socket._fileobject)
+        self.assertTrue(abused_fileobject.__class__ is socket._fileobject)
         httpresponse = abused_fileobject._sock
-        self.assert_(httpresponse.__class__ is httplib.HTTPResponse)
+        self.assertTrue(httpresponse.__class__ is httplib.HTTPResponse)
         fileobject = httpresponse.fp
-        self.assert_(fileobject.__class__ is socket._fileobject)
-
-        self.assert_(not fileobject.closed)
+        self.assertTrue(fileobject.__class__ is socket._fileobject)
+
+        self.assertTrue(not fileobject.closed)
         response.close()
-        self.assert_(fileobject.closed)
+        self.assertTrue(fileobject.closed)
 
 class OtherNetworkTests(unittest.TestCase):
@@ -125 +126 @@
         finally:
             os.remove(TESTFN)
+
+        self.assertRaises(ValueError, urllib2.urlopen,'./relative_path/to/file')
 
     # XXX Following test depends on machine configurations that are internal
@@ -154 +157 @@
 ##             self._test_urls(urls, self._extra_handlers()+[bauth, dauth])
 
+    def test_urlwithfrag(self):
+        urlwith_frag = "http://docs.python.org/2/glossary.html#glossary"
+        with test_support.transient_internet(urlwith_frag):
+            req = urllib2.Request(urlwith_frag)
+            res = urllib2.urlopen(req)
+            self.assertEqual(res.geturl(),
+                    "http://docs.python.org/2/glossary.html#glossary")
+
+    def test_fileno(self):
+        req = urllib2.Request("http://www.python.org")
+        opener = urllib2.build_opener()
+        res = opener.open(req)
+        try:
+            res.fileno()
+        except AttributeError:
+            self.fail("HTTPResponse object should return a valid fileno")
+        finally:
+            res.close()
+
+    def test_custom_headers(self):
+        url = "http://www.example.com"
+        with test_support.transient_internet(url):
+            opener = urllib2.build_opener()
+            request = urllib2.Request(url)
+            self.assertFalse(request.header_items())
+            opener.open(request)
+            self.assertTrue(request.header_items())
+            self.assertTrue(request.has_header('User-agent'))
+            request.add_header('User-Agent','Test-Agent')
+            opener.open(request)
+            self.assertEqual(request.get_header('User-agent'),'Test-Agent')
+
+    def test_sites_no_connection_close(self):
+        # Some sites do not send Connection: close header.
+        # Verify that those work properly. (#issue12576)
+
+        URL = 'http://www.imdb.com' # No Connection:close
+        with test_support.transient_internet(URL):
+            req = urllib2.urlopen(URL)
+            res = req.read()
+            self.assertTrue(res)
+
     def _test_urls(self, urls, handlers, retry=True):
-        import socket
         import time
         import logging
@@ -169 +213 @@
             else:
                 req = expected_err = None
-            debug(url)
-            try:
-                f = urlopen(url, req)
-            except EnvironmentError, err:
-                debug(err)
-                if expected_err:
-                    msg = ("Didn't get expected error(s) %s for %s %s, got %s: %s" %
-                           (expected_err, url, req, type(err), err))
-                    self.assert_(isinstance(err, expected_err), msg)
-            else:
-                with test_support.transient_internet():
-                    buf = f.read()
-                f.close()
-                debug("read %d bytes" % len(buf))
+            with test_support.transient_internet(url):
+                debug(url)
+                try:
+                    f = urlopen(url, req, TIMEOUT)
+                except EnvironmentError as err:
+                    debug(err)
+                    if expected_err:
+                        msg = ("Didn't get expected error(s) %s for %s %s, got %s: %s" %
+                               (expected_err, url, req, type(err), err))
+                        self.assertIsInstance(err, expected_err, msg)
+                except urllib2.URLError as err:
+                    if isinstance(err[0], socket.timeout):
+                        print >>sys.stderr, "<timeout: %s>" % url
+                        continue
+                    else:
+                        raise
+                else:
+                    try:
+                        with test_support.transient_internet(url):
+                            buf = f.read()
+                            debug("read %d bytes" % len(buf))
+                    except socket.timeout:
+                        print >>sys.stderr, "<timeout: %s>" % url
+                    f.close()
             debug("******** next url coming up...")
             time.sleep(0.1)
@@ -190 +244 @@
 
         cfh = urllib2.CacheFTPHandler()
+        self.addCleanup(cfh.clear_cache)
         cfh.setTimeout(1)
         handlers.append(cfh)
@@ -199 +254 @@
     def test_http_basic(self):
         self.assertTrue(socket.getdefaulttimeout() is None)
-        u = _urlopen_with_retry("http://www.python.org")
-        self.assertTrue(u.fp._sock.fp._sock.gettimeout() is None)
+        url = "http://www.python.org"
+        with test_support.transient_internet(url, timeout=None):
+            u = _urlopen_with_retry(url)
+            self.assertTrue(u.fp._sock.fp._sock.gettimeout() is None)
 
     def test_http_default_timeout(self):
         self.assertTrue(socket.getdefaulttimeout() is None)
-        socket.setdefaulttimeout(60)
-        try:
-            u = _urlopen_with_retry("http://www.python.org")
-        finally:
-            socket.setdefaulttimeout(None)
-        self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 60)
+        url = "http://www.python.org"
+        with test_support.transient_internet(url):
+            socket.setdefaulttimeout(60)
+            try:
+                u = _urlopen_with_retry(url)
+            finally:
+                socket.setdefaulttimeout(None)
+            self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 60)
 
     def test_http_no_timeout(self):
         self.assertTrue(socket.getdefaulttimeout() is None)
-        socket.setdefaulttimeout(60)
-        try:
-            u = _urlopen_with_retry("http://www.python.org", timeout=None)
-        finally:
-            socket.setdefaulttimeout(None)
-        self.assertTrue(u.fp._sock.fp._sock.gettimeout() is None)
+        url = "http://www.python.org"
+        with test_support.transient_internet(url):
+            socket.setdefaulttimeout(60)
+            try:
+                u = _urlopen_with_retry(url, timeout=None)
+            finally:
+                socket.setdefaulttimeout(None)
+            self.assertTrue(u.fp._sock.fp._sock.gettimeout() is None)
 
     def test_http_timeout(self):
-        u = _urlopen_with_retry("http://www.python.org", timeout=120)
-        self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 120)
-
-    FTP_HOST = "ftp://ftp.mirror.nl/pub/mirror/gnu/"
+        url = "http://www.python.org"
+        with test_support.transient_internet(url):
+            u = _urlopen_with_retry(url, timeout=120)
+            self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 120)
+
+    FTP_HOST = "ftp://ftp.mirror.nl/pub/gnu/"
 
     def test_ftp_basic(self):
         self.assertTrue(socket.getdefaulttimeout() is None)
-        u = _urlopen_with_retry(self.FTP_HOST)
-        self.assertTrue(u.fp.fp._sock.gettimeout() is None)
+        with test_support.transient_internet(self.FTP_HOST, timeout=None):
+            u = _urlopen_with_retry(self.FTP_HOST)
+            self.assertTrue(u.fp.fp._sock.gettimeout() is None)
 
     def test_ftp_default_timeout(self):
         self.assertTrue(socket.getdefaulttimeout() is None)
-        socket.setdefaulttimeout(60)
-        try:
-            u = _urlopen_with_retry(self.FTP_HOST)
-        finally:
-            socket.setdefaulttimeout(None)
-        self.assertEqual(u.fp.fp._sock.gettimeout(), 60)
+        with test_support.transient_internet(self.FTP_HOST):
+            socket.setdefaulttimeout(60)
+            try:
+                u = _urlopen_with_retry(self.FTP_HOST)
+            finally:
+                socket.setdefaulttimeout(None)
+            self.assertEqual(u.fp.fp._sock.gettimeout(), 60)
 
     def test_ftp_no_timeout(self):
         self.assertTrue(socket.getdefaulttimeout() is None)
-        socket.setdefaulttimeout(60)
-        try:
-            u = _urlopen_with_retry(self.FTP_HOST, timeout=None)
-        finally:
-            socket.setdefaulttimeout(None)
-        self.assertTrue(u.fp.fp._sock.gettimeout() is None)
+        with test_support.transient_internet(self.FTP_HOST):
+            socket.setdefaulttimeout(60)
+            try:
+                u = _urlopen_with_retry(self.FTP_HOST, timeout=None)
+            finally:
+                socket.setdefaulttimeout(None)
+            self.assertTrue(u.fp.fp._sock.gettimeout() is None)
 
     def test_ftp_timeout(self):
-        u = _urlopen_with_retry(self.FTP_HOST, timeout=60)
-        self.assertEqual(u.fp.fp._sock.gettimeout(), 60)
+        with test_support.transient_internet(self.FTP_HOST):
+            u = _urlopen_with_retry(self.FTP_HOST, timeout=60)
+            self.assertEqual(u.fp.fp._sock.gettimeout(), 60)