[Python-checkins] python/dist/src/Lib/test test_urllib.py, 1.16, 1.16.4.1

[birkenfeld@users.sourceforge.net]

Update of /cvsroot/python/python/dist/src/Lib/test
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv11552/Lib/test

Modified Files:
      Tag: release24-maint
	test_urllib.py 
Log Message:
backport patch [ 810023 ] Fix for off-by-one bug in urllib.URLopener.retrieve



Index: test_urllib.py
===================================================================
RCS file: /cvsroot/python/python/dist/src/Lib/test/test_urllib.py,v
retrieving revision 1.16
retrieving revision 1.16.4.1
diff -u -d -r1.16 -r1.16.4.1
--- test_urllib.py	29 Jun 2004 13:07:53 -0000	1.16
+++ test_urllib.py	26 Aug 2005 08:51:39 -0000	1.16.4.1
@@ -6,6 +6,7 @@
 from test import test_support
 import os
 import mimetools
+import tempfile
 import StringIO
 
 def hexescape(char):
@@ -125,15 +126,53 @@
     """Test urllib.urlretrieve() on local files"""
 
     def setUp(self):
+        # Create a list of temporary files. Each item in the list is a file
+        # name (absolute path or relative to the current working directory).
+        # All files in this list will be deleted in the tearDown method. Note,
+        # this only helps to makes sure temporary files get deleted, but it
+        # does nothing about trying to close files that may still be open. It
+        # is the responsibility of the developer to properly close files even
+        # when exceptional conditions occur.
+        self.tempFiles = []
+
         # Create a temporary file.
+        self.registerFileForCleanUp(test_support.TESTFN)
         self.text = 'testing urllib.urlretrieve'
-        FILE = file(test_support.TESTFN, 'wb')
-        FILE.write(self.text)
-        FILE.close()
+        try:
+            FILE = file(test_support.TESTFN, 'wb')
+            FILE.write(self.text)
+            FILE.close()
+        finally:
+            try: FILE.close()
+            except: pass
 
     def tearDown(self):
-        # Delete the temporary file.
-        os.remove(test_support.TESTFN)
+        # Delete the temporary files.
+        for each in self.tempFiles:
+            try: os.remove(each)
+            except: pass
+
+    def constructLocalFileUrl(self, filePath):
+        return "file://%s" % urllib.pathname2url(os.path.abspath(filePath))
+
+    def createNewTempFile(self, data=""):
+        """Creates a new temporary file containing the specified data,
+        registers the file for deletion during the test fixture tear down, and
+        returns the absolute path of the file."""
+
+        newFd, newFilePath = tempfile.mkstemp()
+        try:
+            self.registerFileForCleanUp(newFilePath)
+            newFile = os.fdopen(newFd, "wb")
+            newFile.write(data)
+            newFile.close()
+        finally:
+            try: newFile.close()
+            except: pass
+        return newFilePath
+
+    def registerFileForCleanUp(self, fileName):
+        self.tempFiles.append(fileName)
 
     def test_basic(self):
         # Make sure that a local file just gets its own location returned and
@@ -147,15 +186,19 @@
     def test_copy(self):
         # Test that setting the filename argument works.
         second_temp = "%s.2" % test_support.TESTFN
-        result = urllib.urlretrieve("file:%s" % test_support.TESTFN, second_temp)
+        self.registerFileForCleanUp(second_temp)
+        result = urllib.urlretrieve(self.constructLocalFileUrl(
+            test_support.TESTFN), second_temp)
         self.assertEqual(second_temp, result[0])
         self.assert_(os.path.exists(second_temp), "copy of the file was not "
                                                   "made")
         FILE = file(second_temp, 'rb')
         try:
             text = FILE.read()
-        finally:
             FILE.close()
+        finally:
+            try: FILE.close()
+            except: pass
         self.assertEqual(self.text, text)
 
     def test_reporthook(self):
@@ -167,8 +210,49 @@
             self.assertEqual(count, count_holder[0])
             count_holder[0] = count_holder[0] + 1
         second_temp = "%s.2" % test_support.TESTFN
-        urllib.urlretrieve(test_support.TESTFN, second_temp, hooktester)
-        os.remove(second_temp)
+        self.registerFileForCleanUp(second_temp)
+        urllib.urlretrieve(self.constructLocalFileUrl(test_support.TESTFN),
+            second_temp, hooktester)
+
+    def test_reporthook_0_bytes(self):
+        # Test on zero length file. Should call reporthook only 1 time.
+        report = []
+        def hooktester(count, block_size, total_size, _report=report):
+            _report.append((count, block_size, total_size))
+        srcFileName = self.createNewTempFile()
+        urllib.urlretrieve(self.constructLocalFileUrl(srcFileName),
+            test_support.TESTFN, hooktester)
+        self.assertEqual(len(report), 1)
+        self.assertEqual(report[0][2], 0)
+
+    def test_reporthook_5_bytes(self):
+        # Test on 5 byte file. Should call reporthook only 2 times (once when
+        # the "network connection" is established and once when the block is
+        # read). Since the block size is 8192 bytes, only one block read is
+        # required to read the entire file.
+        report = []
+        def hooktester(count, block_size, total_size, _report=report):
+            _report.append((count, block_size, total_size))
+        srcFileName = self.createNewTempFile("x" * 5)
+        urllib.urlretrieve(self.constructLocalFileUrl(srcFileName),
+            test_support.TESTFN, hooktester)
+        self.assertEqual(len(report), 2)
+        self.assertEqual(report[0][1], 8192)
+        self.assertEqual(report[0][2], 5)
+
+    def test_reporthook_8193_bytes(self):
+        # Test on 8193 byte file. Should call reporthook only 3 times (once
+        # when the "network connection" is established, once for the next 8192
+        # bytes, and once for the last byte).
+        report = []
+        def hooktester(count, block_size, total_size, _report=report):
+            _report.append((count, block_size, total_size))
+        srcFileName = self.createNewTempFile("x" * 8193)
+        urllib.urlretrieve(self.constructLocalFileUrl(srcFileName),
+            test_support.TESTFN, hooktester)
+        self.assertEqual(len(report), 3)
+        self.assertEqual(report[0][1], 8192)
+        self.assertEqual(report[0][2], 8193)
 
 class QuotingTests(unittest.TestCase):
     """Tests for urllib.quote() and urllib.quote_plus()