merge

Lib/test/test_os.py

@@ -28,6 +28,11 @@ try:
     import threading
 except ImportError:
     threading = None
+try:
+    import resource
+except ImportError:
+    resource = None
+
 from test.script_helper import assert_python_ok
 
 with warnings.catch_warnings():
@@ -997,6 +1002,21 @@ class URandomTests(unittest.TestCase):
         data2 = self.get_urandom_subprocess(16)
         self.assertNotEqual(data1, data2)
 
+    @unittest.skipUnless(resource, "test requires the resource module")
+    def test_urandom_failure(self):
+        soft_limit, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE)
+        resource.setrlimit(resource.RLIMIT_NOFILE, (1, hard_limit))
+        try:
+            with self.assertRaises(OSError) as cm:
+                os.urandom(16)
+            self.assertEqual(cm.exception.errno, errno.EMFILE)
+        finally:
+            # We restore the old limit as soon as possible.  If doing it
+            # using addCleanup(), code running in between would fail
+            # creating any file descriptor.
+            resource.setrlimit(resource.RLIMIT_NOFILE, (soft_limit, hard_limit))
+
+
 @contextlib.contextmanager
 def _execvpe_mockup(defpath=None):
     """

Lib/test/test_shutil.py

@@ -726,6 +726,32 @@ class TestShutil(unittest.TestCase):
             shutil.rmtree(src_dir)
             shutil.rmtree(os.path.dirname(dst_dir))
 
+    def test_copytree_retains_permissions(self):
+        tmp_dir = tempfile.mkdtemp()
+        src_dir = os.path.join(tmp_dir, 'source')
+        os.mkdir(src_dir)
+        dst_dir = os.path.join(tmp_dir, 'destination')
+        self.addCleanup(shutil.rmtree, tmp_dir)
+
+        os.chmod(src_dir, 0o777)
+        write_file((src_dir, 'permissive.txt'), '123')
+        os.chmod(os.path.join(src_dir, 'permissive.txt'), 0o777)
+        write_file((src_dir, 'restrictive.txt'), '456')
+        os.chmod(os.path.join(src_dir, 'restrictive.txt'), 0o600)
+        restrictive_subdir = tempfile.mkdtemp(dir=src_dir)
+        os.chmod(restrictive_subdir, 0o600)
+
+        shutil.copytree(src_dir, dst_dir)
+        self.assertEquals(os.stat(src_dir).st_mode, os.stat(dst_dir).st_mode)
+        self.assertEquals(os.stat(os.path.join(src_dir, 'permissive.txt')).st_mode,
+                          os.stat(os.path.join(dst_dir, 'permissive.txt')).st_mode)
+        self.assertEquals(os.stat(os.path.join(src_dir, 'restrictive.txt')).st_mode,
+                          os.stat(os.path.join(dst_dir, 'restrictive.txt')).st_mode)
+        restrictive_subdir_dst = os.path.join(dst_dir,
+                                              os.path.split(restrictive_subdir)[1])
+        self.assertEquals(os.stat(restrictive_subdir).st_mode,
+                          os.stat(restrictive_subdir_dst).st_mode)
+
     @unittest.skipUnless(hasattr(os, 'link'), 'requires os.link')
     def test_dont_copy_file_onto_link_to_itself(self):
         # Temporarily disable test on Windows.

Lib/test/test_timeout.py

@@ -1,5 +1,6 @@
 """Unit tests for socket timeout feature."""
 
+import functools
 import unittest
 from test import support
 
@@ -11,6 +12,18 @@ import errno
 import socket
 
 
+@functools.lru_cache()
+def resolve_address(host, port):
+    """Resolve an (host, port) to an address.
+
+    We must perform name resolution before timeout tests, otherwise it will be
+    performed by connect().
+    """
+    with support.transient_internet(host):
+        return socket.getaddrinfo(host, port, socket.AF_INET,
+                                  socket.SOCK_STREAM)[0][4]
+
+
 class CreationTestCase(unittest.TestCase):
     """Test case for socket.gettimeout() and socket.settimeout()"""
 
@@ -132,7 +145,7 @@ class TCPTimeoutTestCase(TimeoutTestCase):
 
     def setUp(self):
         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.addr_remote = ('www.python.org.', 80)
+        self.addr_remote = resolve_address('www.python.org.', 80)
 
     def tearDown(self):
         self.sock.close()
@@ -142,7 +155,7 @@ class TCPTimeoutTestCase(TimeoutTestCase):
         # to a host that silently drops our packets.  We can't simulate this
         # from Python because it's a function of the underlying TCP/IP stack.
         # So, the following Snakebite host has been defined:
-        blackhole = ('blackhole.snakebite.net', 56666)
+        blackhole = resolve_address('blackhole.snakebite.net', 56666)
 
         # Blackhole has been configured to silently drop any incoming packets.
         # No RSTs (for TCP) or ICMP UNREACH (for UDP/ICMP) will be sent back
@@ -154,7 +167,7 @@ class TCPTimeoutTestCase(TimeoutTestCase):
         # to firewalling or general network configuration.  In order to improve
         # our confidence in testing the blackhole, a corresponding 'whitehole'
         # has also been set up using one port higher:
-        whitehole = ('whitehole.snakebite.net', 56667)
+        whitehole = resolve_address('whitehole.snakebite.net', 56667)
 
         # This address has been configured to immediately drop any incoming
         # packets as well, but it does it respectfully with regards to the