gh-154726: Test FIFO and socket symlinks in shutil.copyfile()

Author: lkk7

Lib/test/test_shutil.py

@@ -1574,6 +1574,47 @@ def test_copyfile_socket(self):
         self.assertRaisesRegex(shutil.SpecialFileError, 'is a socket',
                                shutil.copyfile, __file__, sock_path)
 
+    def _check_copyfile_symlink_to_special_file(self, target):
+        tmp_dir = self.mkdtemp()
+        src = os.path.join(tmp_dir, 'src')
+        dst = os.path.join(tmp_dir, 'dst')
+        os.symlink(target, src)
+
+        shutil.copyfile(src, dst, follow_symlinks=False)
+
+        self.assertTrue(os.path.islink(dst))
+        self.assertEqual(os.readlink(dst), target)
+
+    @os_helper.skip_unless_symlink
+    @unittest.skipUnless(hasattr(os, "mkfifo"), 'requires os.mkfifo()')
+    @unittest.skipIf(sys.platform == "vxworks",
+                    "fifo requires special path on VxWorks")
+    def test_copyfile_symlink_to_named_pipe(self):
+        fifo_path = os.path.join(self.mkdtemp(), 'fifo')
+        try:
+            os.mkfifo(fifo_path)
+        except PermissionError as e:
+            self.skipTest('os.mkfifo(): %s' % e)
+        self._check_copyfile_symlink_to_special_file(fifo_path)
+
+    @os_helper.skip_unless_symlink
+    @socket_helper.skip_unless_bind_unix_socket
+    def test_copyfile_symlink_to_socket(self):
+        sock_path = os.path.join(self.mkdtemp(), 'sock')
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self.addCleanup(sock.close)
+        try:
+            socket_helper.bind_unix_socket(sock, sock_path)
+        except OSError as e:
+            if str(e) == "AF_UNIX path too long":
+                self.skipTest(
+                    "Pathname {0!a} is too long to serve as an AF_UNIX path"
+                    .format(sock_path))
+            else:
+                raise
+        self.addCleanup(os_helper.unlink, sock_path)
+        self._check_copyfile_symlink_to_special_file(sock_path)
+
     @unittest.skipUnless(os.path.exists('/dev/null'), 'requires /dev/null')
     def test_copyfile_character_device(self):
         self.assertRaisesRegex(shutil.SpecialFileError, 'is a character device',
@@ -1586,15 +1627,7 @@ def test_copyfile_character_device(self):
     @os_helper.skip_unless_symlink
     @unittest.skipUnless(os.path.exists('/dev/null'), 'requires /dev/null')
     def test_copyfile_symlink_to_character_device(self):
-        tmp_dir = self.mkdtemp()
-        src = os.path.join(tmp_dir, 'src')
-        dst = os.path.join(tmp_dir, 'dst')
-        os.symlink('/dev/null', src)
-
-        shutil.copyfile(src, dst, follow_symlinks=False)
-
-        self.assertTrue(os.path.islink(dst))
-        self.assertEqual(os.readlink(dst), '/dev/null')
+        self._check_copyfile_symlink_to_special_file('/dev/null')
 
     def test_copyfile_block_device(self):
         block_dev = None