Skip to content

gh-81793: always call linkat() from os.link(), if available #24997

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
takluyver wants to merge 4 commits into from
Closed

gh-81793: always call linkat() from os.link(), if available #24997

Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
21b0ab7
Add failing test for os.link() with symlink
takluyver Mar 23, 2021
772b3e9
bpo-37612: always call linkat() from os.link(), if available
jo-he Mar 23, 2021
0200894
📜🤖 Added by blurb_it.
blurb-it[bot] Jul 19, 2019
beada2b
Merge branch 'main' into bpo-37612
serhiy-storchaka Apr 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Merge branch 'main' into bpo-37612
  • Loading branch information
@serhiy-storchaka
serhiy-storchaka committed Apr 12, 2025
commit beada2bba3b96d6cb4dc08ff98480cd0a9f02f6f
220 changes: 21 additions & 199 deletions Lib/test/test_posix.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1270,205 +1270,6 @@ def test_getgroups(self):
symdiff = idg_groups.symmetric_difference(posix.getgroups())
self.assertTrue(not symdiff or symdiff == {posix.getegid()})

# tests for the posix *at functions follow

@unittest.skipUnless(os.access in os.supports_dir_fd, "test needs dir_fd support for os.access()")
def test_access_dir_fd(self):
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
self.assertTrue(posix.access(os_helper.TESTFN, os.R_OK, dir_fd=f))
finally:
posix.close(f)

@unittest.skipUnless(os.chmod in os.supports_dir_fd, "test needs dir_fd support in os.chmod()")
def test_chmod_dir_fd(self):
os.chmod(os_helper.TESTFN, stat.S_IRUSR)

f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.chmod(os_helper.TESTFN, stat.S_IRUSR | stat.S_IWUSR, dir_fd=f)

s = posix.stat(os_helper.TESTFN)
self.assertEqual(s[0] & stat.S_IRWXU, stat.S_IRUSR | stat.S_IWUSR)
finally:
posix.close(f)

@unittest.skipUnless(hasattr(os, 'chown') and (os.chown in os.supports_dir_fd),
"test needs dir_fd support in os.chown()")
def test_chown_dir_fd(self):
os_helper.unlink(os_helper.TESTFN)
os_helper.create_empty_file(os_helper.TESTFN)

f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.chown(os_helper.TESTFN, os.getuid(), os.getgid(), dir_fd=f)
finally:
posix.close(f)

@unittest.skipUnless(os.stat in os.supports_dir_fd, "test needs dir_fd support in os.stat()")
def test_stat_dir_fd(self):
os_helper.unlink(os_helper.TESTFN)
with open(os_helper.TESTFN, 'w') as outfile:
outfile.write("testline\n")

f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
s1 = posix.stat(os_helper.TESTFN)
s2 = posix.stat(os_helper.TESTFN, dir_fd=f)
self.assertEqual(s1, s2)
s2 = posix.stat(os_helper.TESTFN, dir_fd=None)
self.assertEqual(s1, s2)
self.assertRaisesRegex(TypeError, 'should be integer or None, not',
posix.stat, os_helper.TESTFN, dir_fd=posix.getcwd())
self.assertRaisesRegex(TypeError, 'should be integer or None, not',
posix.stat, os_helper.TESTFN, dir_fd=float(f))
self.assertRaises(OverflowError,
posix.stat, os_helper.TESTFN, dir_fd=10**20)
finally:
posix.close(f)

@unittest.skipUnless(os.utime in os.supports_dir_fd, "test needs dir_fd support in os.utime()")
def test_utime_dir_fd(self):
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
now = time.time()
posix.utime(os_helper.TESTFN, None, dir_fd=f)
posix.utime(os_helper.TESTFN, dir_fd=f)
self.assertRaises(TypeError, posix.utime, os_helper.TESTFN,
now, dir_fd=f)
self.assertRaises(TypeError, posix.utime, os_helper.TESTFN,
(None, None), dir_fd=f)
self.assertRaises(TypeError, posix.utime, os_helper.TESTFN,
(now, None), dir_fd=f)
self.assertRaises(TypeError, posix.utime, os_helper.TESTFN,
(None, now), dir_fd=f)
self.assertRaises(TypeError, posix.utime, os_helper.TESTFN,
(now, "x"), dir_fd=f)
posix.utime(os_helper.TESTFN, (int(now), int(now)), dir_fd=f)
posix.utime(os_helper.TESTFN, (now, now), dir_fd=f)
posix.utime(os_helper.TESTFN,
(int(now), int((now - int(now)) * 1e9)), dir_fd=f)
posix.utime(os_helper.TESTFN, dir_fd=f,
times=(int(now), int((now - int(now)) * 1e9)))

# try dir_fd and follow_symlinks together
if os.utime in os.supports_follow_symlinks:
try:
posix.utime(os_helper.TESTFN, follow_symlinks=False,
dir_fd=f)
except ValueError:
# whoops! using both together not supported on this platform.
pass

finally:
posix.close(f)

@unittest.skipUnless(os.link in os.supports_dir_fd, "test needs dir_fd support in os.link()")
def test_link_dir_fd(self):
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.link(os_helper.TESTFN, os_helper.TESTFN + 'link',
src_dir_fd=f, dst_dir_fd=f)
except PermissionError as e:
self.skipTest('posix.link(): %s' % e)
else:
# should have same inodes
self.assertEqual(posix.stat(os_helper.TESTFN)[1],
posix.stat(os_helper.TESTFN + 'link')[1])
finally:
posix.close(f)
os_helper.unlink(os_helper.TESTFN + 'link')

@unittest.skipUnless(os.link in os.supports_follow_symlinks, "test needs follow_symlinks support in os.link()")
def test_link_follow_symlinks(self):
symlink_fn = os_helper.TESTFN + 'symlink'
link_following = os_helper.TESTFN + 'link_following'
link_nofollow = os_helper.TESTFN + 'link_nofollow'
posix.symlink(os_helper.TESTFN, symlink_fn)
self.teardown_files.append(symlink_fn)

# follow_symlinks=False -> duplicate the symlink itself
posix.link(symlink_fn, link_nofollow, follow_symlinks=False)
self.teardown_files.append(link_nofollow)
self.assertEqual(posix.lstat(link_nofollow), posix.lstat(symlink_fn))

# follow_symlinks=True -> duplicate the target file
posix.link(symlink_fn, link_following, follow_symlinks=True)
self.teardown_files.append(link_following)
self.assertEqual(posix.lstat(link_following), posix.lstat(os_helper.TESTFN))

@unittest.skipUnless(os.mkdir in os.supports_dir_fd, "test needs dir_fd support in os.mkdir()")
def test_mkdir_dir_fd(self):
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.mkdir(os_helper.TESTFN + 'dir', dir_fd=f)
posix.stat(os_helper.TESTFN + 'dir') # should not raise exception
finally:
posix.close(f)
os_helper.rmtree(os_helper.TESTFN + 'dir')

@unittest.skipUnless(hasattr(os, 'mknod')
and (os.mknod in os.supports_dir_fd)
and hasattr(stat, 'S_IFIFO'),
"test requires both stat.S_IFIFO and dir_fd support for os.mknod()")
def test_mknod_dir_fd(self):
# Test using mknodat() to create a FIFO (the only use specified
# by POSIX).
os_helper.unlink(os_helper.TESTFN)
mode = stat.S_IFIFO | stat.S_IRUSR | stat.S_IWUSR
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.mknod(os_helper.TESTFN, mode, 0, dir_fd=f)
except OSError as e:
# Some old systems don't allow unprivileged users to use
# mknod(), or only support creating device nodes.
self.assertIn(e.errno, (errno.EPERM, errno.EINVAL, errno.EACCES))
else:
self.assertTrue(stat.S_ISFIFO(posix.stat(os_helper.TESTFN).st_mode))
finally:
posix.close(f)

@unittest.skipUnless(os.open in os.supports_dir_fd, "test needs dir_fd support in os.open()")
def test_open_dir_fd(self):
os_helper.unlink(os_helper.TESTFN)
with open(os_helper.TESTFN, 'w') as outfile:
outfile.write("testline\n")
a = posix.open(posix.getcwd(), posix.O_RDONLY)
b = posix.open(os_helper.TESTFN, posix.O_RDONLY, dir_fd=a)
try:
res = posix.read(b, 9).decode(encoding="utf-8")
self.assertEqual("testline\n", res)
finally:
posix.close(a)
posix.close(b)

@unittest.skipUnless(hasattr(os, 'readlink') and (os.readlink in os.supports_dir_fd),
"test needs dir_fd support in os.readlink()")
def test_readlink_dir_fd(self):
os.symlink(os_helper.TESTFN, os_helper.TESTFN + 'link')
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
self.assertEqual(posix.readlink(os_helper.TESTFN + 'link'),
posix.readlink(os_helper.TESTFN + 'link', dir_fd=f))
finally:
os_helper.unlink(os_helper.TESTFN + 'link')
posix.close(f)

@unittest.skipUnless(os.rename in os.supports_dir_fd, "test needs dir_fd support in os.rename()")
def test_rename_dir_fd(self):
os_helper.unlink(os_helper.TESTFN)
os_helper.create_empty_file(os_helper.TESTFN + 'ren')
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.rename(os_helper.TESTFN + 'ren', os_helper.TESTFN, src_dir_fd=f, dst_dir_fd=f)
except:
posix.rename(os_helper.TESTFN + 'ren', os_helper.TESTFN)
raise
else:
posix.stat(os_helper.TESTFN) # should not raise exception
finally:
posix.close(f)

@unittest.skipUnless(hasattr(signal, 'SIGCHLD'), 'CLD_XXXX be placed in si_code for a SIGCHLD signal')
@unittest.skipUnless(hasattr(os, 'waitid_result'), "test needs os.waitid_result")
def test_cld_xxxx_constants(self):
Expand Down Expand Up @@ -1720,6 +1521,27 @@ def test_pidfd_open(self):
self.assertEqual(cm.exception.errno, errno.EINVAL)
os.close(os.pidfd_open(os.getpid(), 0))

@unittest.skipUnless(
hasattr(os, "link") and os.link in os.supports_follow_symlinks,
"test needs follow_symlinks support in os.link()"
)
def test_link_follow_symlinks(self):
symlink_fn = os_helper.TESTFN + 'symlink'
link_following = os_helper.TESTFN + 'link_following'
link_nofollow = os_helper.TESTFN + 'link_nofollow'
posix.symlink(os_helper.TESTFN, symlink_fn)
self.addCleanup(os_helper.unlink, symlink_fn)

# follow_symlinks=False -> duplicate the symlink itself
posix.link(symlink_fn, link_nofollow, follow_symlinks=False)
self.addCleanup(os_helper.unlink, link_nofollow)
self.assertEqual(posix.lstat(link_nofollow), posix.lstat(symlink_fn))

# follow_symlinks=True -> duplicate the target file
posix.link(symlink_fn, link_following, follow_symlinks=True)
self.addCleanup(os_helper.unlink, link_following)
self.assertEqual(posix.lstat(link_following), posix.lstat(os_helper.TESTFN))


# tests for the posix *at functions follow
class TestPosixDirFd(unittest.TestCase):
Expand Down
Loading
You are viewing a condensed version of this merge commit. You can view the full changes here.