mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-01 18:51:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 21

Issue #14626: Large refactoring of functions / parameters in the os module.

Browse source 
Many functions now support "dir_fd" and "follow_symlinks" parameters;
some also support accepting an open file descriptor in place of of a path
string.  Added os.support_* collections as LBYL helpers.  Removed many
functions only previously seen in 3.3 alpha releases (often starting with
"f" or "l", or ending with "at").  Originally suggested by Serhiy Storchaka;
implemented by Larry Hastings.
This commit is contained in:
Larry Hastings 2012-06-22 16:30:09 -07:00
parent f0f4742b49
commit 9cf065cfdc
9 changed files with 3413 additions and 3041 deletions
Download patch file Download diff file Expand all files Collapse all files

224
Lib/test/test_posix.py
View file

@ -129,6 +129,7 @@ class PosixTester(unittest.TestCase):
fp = open(support.TESTFN)
try:
self.assertTrue(posix.fstatvfs(fp.fileno()))
self.assertTrue(posix.statvfs(fp.fileno()))
finally:
fp.close()
@ -150,7 +151,7 @@ class PosixTester(unittest.TestCase):
fp.flush()
posix.truncate(support.TESTFN, 0)
@unittest.skipUnless(hasattr(posix, 'fexecve'), "test needs posix.fexecve()")
@unittest.skipUnless(getattr(os, 'execve', None) in os.supports_fd, "test needs execve() to support the fd parameter")
@unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()")
@unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()")
def test_fexecve(self):
@ -159,7 +160,7 @@ class PosixTester(unittest.TestCase):
pid = os.fork()
if pid == 0:
os.chdir(os.path.split(sys.executable)[0])
posix.fexecve(fp, [sys.executable, '-c', 'pass'], os.environ)
posix.execve(fp, [sys.executable, '-c', 'pass'], os.environ)
else:
self.assertEqual(os.waitpid(pid, 0), (pid, 0))
finally:
@ -234,45 +235,37 @@ class PosixTester(unittest.TestCase):
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'futimes'), "test needs posix.futimes()")
def test_futimes(self):
@unittest.skipUnless(os.utime in os.supports_fd, "test needs fd support in os.utime")
def test_utime_with_fd(self):
now = time.time()
fd = os.open(support.TESTFN, os.O_RDONLY)
try:
posix.futimes(fd, None)
posix.futimes(fd)
self.assertRaises(TypeError, posix.futimes, fd, (None, None))
self.assertRaises(TypeError, posix.futimes, fd, (now, None))
self.assertRaises(TypeError, posix.futimes, fd, (None, now))
posix.futimes(fd, (int(now), int(now)))
posix.futimes(fd, (now, now))
posix.utime(fd)
posix.utime(fd, None)
self.assertRaises(TypeError, posix.utime, fd, (None, None))
self.assertRaises(TypeError, posix.utime, fd, (now, None))
self.assertRaises(TypeError, posix.utime, fd, (None, now))
posix.utime(fd, (int(now), int(now)))
posix.utime(fd, (now, now))
self.assertRaises(ValueError, posix.utime, fd, (now, now), ns=(now, now))
self.assertRaises(ValueError, posix.utime, fd, (now, 0), ns=(None, None))
self.assertRaises(ValueError, posix.utime, fd, (None, None), ns=(now, 0))
posix.utime(fd, (int(now), int((now - int(now)) * 1e9)))
posix.utime(fd, ns=(int(now), int((now - int(now)) * 1e9)))
finally:
os.close(fd)
@unittest.skipUnless(hasattr(posix, 'lutimes'), "test needs posix.lutimes()")
def test_lutimes(self):
@unittest.skipUnless(os.utime in os.supports_follow_symlinks, "test needs follow_symlinks support in os.utime")
def test_utime_nofollow_symlinks(self):
now = time.time()
posix.lutimes(support.TESTFN, None)
self.assertRaises(TypeError, posix.lutimes, support.TESTFN, (None, None))
self.assertRaises(TypeError, posix.lutimes, support.TESTFN, (now, None))
self.assertRaises(TypeError, posix.lutimes, support.TESTFN, (None, now))
posix.lutimes(support.TESTFN, (int(now), int(now)))
posix.lutimes(support.TESTFN, (now, now))
posix.lutimes(support.TESTFN)
@unittest.skipUnless(hasattr(posix, 'futimens'), "test needs posix.futimens()")
def test_futimens(self):
now = time.time()
fd = os.open(support.TESTFN, os.O_RDONLY)
try:
self.assertRaises(TypeError, posix.futimens, fd, (None, None), (None, None))
self.assertRaises(TypeError, posix.futimens, fd, (now, 0), None)
self.assertRaises(TypeError, posix.futimens, fd, None, (now, 0))
posix.futimens(fd, (int(now), int((now - int(now)) * 1e9)),
(int(now), int((now - int(now)) * 1e9)))
posix.futimens(fd)
finally:
os.close(fd)
posix.utime(support.TESTFN, None, follow_symlinks=False)
self.assertRaises(TypeError, posix.utime, support.TESTFN, (None, None), follow_symlinks=False)
self.assertRaises(TypeError, posix.utime, support.TESTFN, (now, None), follow_symlinks=False)
self.assertRaises(TypeError, posix.utime, support.TESTFN, (None, now), follow_symlinks=False)
posix.utime(support.TESTFN, (int(now), int(now)), follow_symlinks=False)
posix.utime(support.TESTFN, (now, now), follow_symlinks=False)
posix.utime(support.TESTFN, follow_symlinks=False)
@unittest.skipUnless(hasattr(posix, 'writev'), "test needs posix.writev()")
def test_writev(self):
@ -364,6 +357,7 @@ class PosixTester(unittest.TestCase):
fp = open(support.TESTFN)
try:
self.assertTrue(posix.fstat(fp.fileno()))
self.assertTrue(posix.stat(fp.fileno()))
finally:
fp.close()
@ -462,18 +456,18 @@ class PosixTester(unittest.TestCase):
if hasattr(posix, 'listdir'):
self.assertTrue(support.TESTFN in posix.listdir())
@unittest.skipUnless(hasattr(posix, 'flistdir'), "test needs posix.flistdir()")
@unittest.skipUnless(os.listdir in os.supports_fd, "test needs fd support for os.listdir()")
def test_flistdir(self):
f = posix.open(posix.getcwd(), posix.O_RDONLY)
self.addCleanup(posix.close, f)
self.assertEqual(
sorted(posix.listdir('.')),
sorted(posix.flistdir(f))
sorted(posix.listdir(f))
)
# Check that the fd offset was reset (issue #13739)
self.assertEqual(
sorted(posix.listdir('.')),
sorted(posix.flistdir(f))
sorted(posix.listdir(f))
)
def test_access(self):
@ -532,10 +526,10 @@ class PosixTester(unittest.TestCase):
posix.utime(support.TESTFN, (int(now), int(now)))
posix.utime(support.TESTFN, (now, now))
def _test_chflags_regular_file(self, chflags_func, target_file):
def _test_chflags_regular_file(self, chflags_func, target_file, **kwargs):
st = os.stat(target_file)
self.assertTrue(hasattr(st, 'st_flags'))
chflags_func(target_file, st.st_flags | stat.UF_IMMUTABLE)
chflags_func(target_file, st.st_flags | stat.UF_IMMUTABLE, **kwargs)
try:
new_st = os.stat(target_file)
self.assertEqual(st.st_flags | stat.UF_IMMUTABLE, new_st.st_flags)
@ -553,6 +547,7 @@ class PosixTester(unittest.TestCase):
@unittest.skipUnless(hasattr(posix, 'lchflags'), 'test needs os.lchflags()')
def test_lchflags_regular_file(self):
self._test_chflags_regular_file(posix.lchflags, support.TESTFN)
self._test_chflags_regular_file(posix.chflags, support.TESTFN, follow_symlinks=False)
@unittest.skipUnless(hasattr(posix, 'lchflags'), 'test needs os.lchflags()')
def test_lchflags_symlink(self):
@ -564,17 +559,21 @@ class PosixTester(unittest.TestCase):
self.teardown_files.append(_DUMMY_SYMLINK)
dummy_symlink_st = os.lstat(_DUMMY_SYMLINK)
posix.lchflags(_DUMMY_SYMLINK,
dummy_symlink_st.st_flags | stat.UF_IMMUTABLE)
try:
new_testfn_st = os.stat(support.TESTFN)
new_dummy_symlink_st = os.lstat(_DUMMY_SYMLINK)
def chflags_nofollow(path, flags):
return posix.chflags(path, flags, follow_symlinks=False)
self.assertEqual(testfn_st.st_flags, new_testfn_st.st_flags)
self.assertEqual(dummy_symlink_st.st_flags | stat.UF_IMMUTABLE,
new_dummy_symlink_st.st_flags)
finally:
posix.lchflags(_DUMMY_SYMLINK, dummy_symlink_st.st_flags)
for fn in (posix.lchflags, chflags_nofollow):
fn(_DUMMY_SYMLINK,
dummy_symlink_st.st_flags | stat.UF_IMMUTABLE)
try:
new_testfn_st = os.stat(support.TESTFN)
new_dummy_symlink_st = os.lstat(_DUMMY_SYMLINK)
self.assertEqual(testfn_st.st_flags, new_testfn_st.st_flags)
self.assertEqual(dummy_symlink_st.st_flags | stat.UF_IMMUTABLE,
new_dummy_symlink_st.st_flags)
finally:
fn(_DUMMY_SYMLINK, dummy_symlink_st.st_flags)
def test_environ(self):
if os.name == "nt":
@ -657,40 +656,40 @@ class PosixTester(unittest.TestCase):
# tests for the posix *at functions follow
@unittest.skipUnless(hasattr(posix, 'faccessat'), "test needs posix.faccessat()")
def test_faccessat(self):
@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.faccessat(f, support.TESTFN, os.R_OK))
self.assertTrue(posix.access(support.TESTFN, os.R_OK, dir_fd=f))
finally:
posix.close(f)
@unittest.skipUnless(hasattr(posix, 'fchmodat'), "test needs posix.fchmodat()")
def test_fchmodat(self):
@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(support.TESTFN, stat.S_IRUSR)
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.fchmodat(f, support.TESTFN, stat.S_IRUSR | stat.S_IWUSR)
posix.chmod(support.TESTFN, stat.S_IRUSR | stat.S_IWUSR, dir_fd=f)
s = posix.stat(support.TESTFN)
self.assertEqual(s[0] & stat.S_IRWXU, stat.S_IRUSR | stat.S_IWUSR)
finally:
posix.close(f)
@unittest.skipUnless(hasattr(posix, 'fchownat'), "test needs posix.fchownat()")
def test_fchownat(self):
@unittest.skipUnless(os.chown in os.supports_dir_fd, "test needs dir_fd support in os.chown()")
def test_chown_dir_fd(self):
support.unlink(support.TESTFN)
support.create_empty_file(support.TESTFN)
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.fchownat(f, support.TESTFN, os.getuid(), os.getgid())
posix.chown(support.TESTFN, os.getuid(), os.getgid(), dir_fd=f)
finally:
posix.close(f)
@unittest.skipUnless(hasattr(posix, 'fstatat'), "test needs posix.fstatat()")
def test_fstatat(self):
@unittest.skipUnless(os.stat in os.supports_dir_fd, "test needs dir_fd support in os.stat()")
def test_stat_dir_fd(self):
support.unlink(support.TESTFN)
with open(support.TESTFN, 'w') as outfile:
outfile.write("testline\n")
@ -698,31 +697,41 @@ class PosixTester(unittest.TestCase):
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
s1 = posix.stat(support.TESTFN)
s2 = posix.fstatat(f, support.TESTFN)
s2 = posix.stat(support.TESTFN, dir_fd=f)
self.assertEqual(s1, s2)
finally:
posix.close(f)
@unittest.skipUnless(hasattr(posix, 'futimesat'), "test needs posix.futimesat()")
def test_futimesat(self):
@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.futimesat(f, support.TESTFN, None)
posix.futimesat(f, support.TESTFN)
self.assertRaises(TypeError, posix.futimesat, f, support.TESTFN, (None, None))
self.assertRaises(TypeError, posix.futimesat, f, support.TESTFN, (now, None))
self.assertRaises(TypeError, posix.futimesat, f, support.TESTFN, (None, now))
posix.futimesat(f, support.TESTFN, (int(now), int(now)))
posix.futimesat(f, support.TESTFN, (now, now))
posix.utime(support.TESTFN, None, dir_fd=f)
posix.utime(support.TESTFN, dir_fd=f)
self.assertRaises(TypeError, posix.utime, support.TESTFN, now, dir_fd=f)
self.assertRaises(TypeError, posix.utime, support.TESTFN, (None, None), dir_fd=f)
self.assertRaises(TypeError, posix.utime, support.TESTFN, (now, None), dir_fd=f)
self.assertRaises(TypeError, posix.utime, support.TESTFN, (None, now), dir_fd=f)
self.assertRaises(TypeError, posix.utime, support.TESTFN, (now, "x"), dir_fd=f)
posix.utime(support.TESTFN, (int(now), int(now)), dir_fd=f)
posix.utime(support.TESTFN, (now, now), dir_fd=f)
posix.utime(support.TESTFN,
(int(now), int((now - int(now)) * 1e9)), dir_fd=f)
posix.utime(support.TESTFN, dir_fd=f,
times=(int(now), int((now - int(now)) * 1e9)))
if os.utime in os.supports_follow_symlinks:
posix.utime(support.TESTFN, follow_symlinks=False, dir_fd=f)
finally:
posix.close(f)
@unittest.skipUnless(hasattr(posix, 'linkat'), "test needs posix.linkat()")
def test_linkat(self):
@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.linkat(f, support.TESTFN, f, support.TESTFN + 'link')
posix.link(support.TESTFN, support.TESTFN + 'link', src_dir_fd=f, dst_dir_fd=f)
# should have same inodes
self.assertEqual(posix.stat(support.TESTFN)[1],
posix.stat(support.TESTFN + 'link')[1])
@ -730,26 +739,26 @@ class PosixTester(unittest.TestCase):
posix.close(f)
support.unlink(support.TESTFN + 'link')
@unittest.skipUnless(hasattr(posix, 'mkdirat'), "test needs posix.mkdirat()")
def test_mkdirat(self):
@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.mkdirat(f, support.TESTFN + 'dir')
posix.mkdir(support.TESTFN + 'dir', dir_fd=f)
posix.stat(support.TESTFN + 'dir') # should not raise exception
finally:
posix.close(f)
support.rmtree(support.TESTFN + 'dir')
@unittest.skipUnless(hasattr(posix, 'mknodat') and hasattr(stat, 'S_IFIFO'),
"don't have mknodat()/S_IFIFO")
def test_mknodat(self):
@unittest.skipUnless((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).
support.unlink(support.TESTFN)
mode = stat.S_IFIFO | stat.S_IRUSR | stat.S_IWUSR
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.mknodat(f, support.TESTFN, mode, 0)
posix.mknod(support.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.
@ -759,13 +768,13 @@ class PosixTester(unittest.TestCase):
finally:
posix.close(f)
@unittest.skipUnless(hasattr(posix, 'openat'), "test needs posix.openat()")
def test_openat(self):
@unittest.skipUnless(os.open in os.supports_dir_fd, "test needs dir_fd support in os.open()")
def test_open_dir_fd(self):
support.unlink(support.TESTFN)
with open(support.TESTFN, 'w') as outfile:
outfile.write("testline\n")
a = posix.open(posix.getcwd(), posix.O_RDONLY)
b = posix.openat(a, support.TESTFN, posix.O_RDONLY)
b = posix.open(support.TESTFN, posix.O_RDONLY, dir_fd=a)
try:
res = posix.read(b, 9).decode(encoding="utf-8")
self.assertEqual("testline\n", res)
@ -773,24 +782,24 @@ class PosixTester(unittest.TestCase):
posix.close(a)
posix.close(b)
@unittest.skipUnless(hasattr(posix, 'readlinkat'), "test needs posix.readlinkat()")
def test_readlinkat(self):
@unittest.skipUnless(os.readlink in os.supports_dir_fd, "test needs dir_fd support in os.readlink()")
def test_readlink_dir_fd(self):
os.symlink(support.TESTFN, support.TESTFN + 'link')
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
self.assertEqual(posix.readlink(support.TESTFN + 'link'),
posix.readlinkat(f, support.TESTFN + 'link'))
posix.readlink(support.TESTFN + 'link', dir_fd=f))
finally:
support.unlink(support.TESTFN + 'link')
posix.close(f)
@unittest.skipUnless(hasattr(posix, 'renameat'), "test needs posix.renameat()")
def test_renameat(self):
@unittest.skipUnless(os.rename in os.supports_dir_fd, "test needs dir_fd support in os.rename()")
def test_rename_dir_fd(self):
support.unlink(support.TESTFN)
support.create_empty_file(support.TESTFN + 'ren')
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.renameat(f, support.TESTFN + 'ren', f, support.TESTFN)
posix.rename(support.TESTFN + 'ren', support.TESTFN, src_dir_fd=f, dst_dir_fd=f)
except:
posix.rename(support.TESTFN + 'ren', support.TESTFN)
raise
@ -799,23 +808,23 @@ class PosixTester(unittest.TestCase):
finally:
posix.close(f)
@unittest.skipUnless(hasattr(posix, 'symlinkat'), "test needs posix.symlinkat()")
def test_symlinkat(self):
@unittest.skipUnless(os.symlink in os.supports_dir_fd, "test needs dir_fd support in os.symlink()")
def test_symlink_dir_fd(self):
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.symlinkat(support.TESTFN, f, support.TESTFN + 'link')
posix.symlink(support.TESTFN, support.TESTFN + 'link', dir_fd=f)
self.assertEqual(posix.readlink(support.TESTFN + 'link'), support.TESTFN)
finally:
posix.close(f)
support.unlink(support.TESTFN + 'link')
@unittest.skipUnless(hasattr(posix, 'unlinkat'), "test needs posix.unlinkat()")
def test_unlinkat(self):
@unittest.skipUnless(os.unlink in os.supports_dir_fd, "test needs dir_fd support in os.unlink()")
def test_unlink_dir_fd(self):
f = posix.open(posix.getcwd(), posix.O_RDONLY)
support.create_empty_file(support.TESTFN + 'del')
posix.stat(support.TESTFN + 'del') # should not throw exception
try:
posix.unlinkat(f, support.TESTFN + 'del')
posix.unlink(support.TESTFN + 'del', dir_fd=f)
except:
support.unlink(support.TESTFN + 'del')
raise
@ -824,31 +833,12 @@ class PosixTester(unittest.TestCase):
finally:
posix.close(f)
@unittest.skipUnless(hasattr(posix, 'utimensat'), "test needs posix.utimensat()")
def test_utimensat(self):
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
now = time.time()
posix.utimensat(f, support.TESTFN, None, None)
posix.utimensat(f, support.TESTFN)
posix.utimensat(f, support.TESTFN, flags=os.AT_SYMLINK_NOFOLLOW)
self.assertRaises(TypeError, posix.utimensat, f, support.TESTFN, (None, None), (None, None))
self.assertRaises(TypeError, posix.utimensat, f, support.TESTFN, (now, 0), None)
self.assertRaises(TypeError, posix.utimensat, f, support.TESTFN, None, (now, 0))
posix.utimensat(f, support.TESTFN, (int(now), int((now - int(now)) * 1e9)),
(int(now), int((now - int(now)) * 1e9)))
posix.utimensat(dirfd=f, path=support.TESTFN,
atime=(int(now), int((now - int(now)) * 1e9)),
mtime=(int(now), int((now - int(now)) * 1e9)))
finally:
posix.close(f)
@unittest.skipUnless(hasattr(posix, 'mkfifoat'), "don't have mkfifoat()")
def test_mkfifoat(self):
@unittest.skipUnless(os.mkfifo in os.supports_dir_fd, "test needs dir_fd support in os.mkfifo()")
def test_mkfifo_dir_fd(self):
support.unlink(support.TESTFN)
f = posix.open(posix.getcwd(), posix.O_RDONLY)
try:
posix.mkfifoat(f, support.TESTFN, stat.S_IRUSR | stat.S_IWUSR)
posix.mkfifo(support.TESTFN, stat.S_IRUSR | stat.S_IWUSR, dir_fd=f)
self.assertTrue(stat.S_ISFIFO(posix.stat(support.TESTFN).st_mode))
finally:
posix.close(f)