test_os.py

# As a test suite for the os module, this is woefully inadequate, but this
# does add tests for a few functions which have been determined to be more
# portable than they had been thought to be.

importasyncio
importcodecs
importcontextlib
importdecimal
importerrno
importfnmatch
importfractions
importitertools
importlocale
importos
importpickle
importplatform
importselect
importselectors
importshutil
importsignal
importsocket
importstat
importstruct
importsubprocess
importsys
importsysconfig
importtempfile
importtextwrap
importtime
importtypes
importunittest
importuuid
importwarnings
fromtestimportsupport
fromtest.supportimportimport_helper
fromtest.supportimportos_helper
fromtest.supportimportsocket_helper
fromtest.supportimportinfinite_recursion
fromtest.supportimportwarnings_helper
fromplatformimportwin32_is_iot

try:
importresource
exceptImportError:
resource=None
try:
importfcntl
exceptImportError:
fcntl=None
try:
import_winapi
exceptImportError:
_winapi=None
try:
importpwd
all_users= [u.pw_uidforuinpwd.getpwall()]
except (ImportError, AttributeError):
all_users= []
try:
import_testcapi
from_testcapiimportINT_MAX, PY_SSIZE_T_MAX
exceptImportError:
_testcapi=None
INT_MAX=PY_SSIZE_T_MAX=sys.maxsize

try:
importmmap
exceptImportError:
mmap=None

fromtest.support.script_helperimportassert_python_ok
fromtest.supportimportunix_shell
fromtest.support.os_helperimportFakePath


root_in_posix=False
ifhasattr(os, 'geteuid'):
root_in_posix= (os.geteuid() ==0)

# Detect whether we're on a Linux system that uses the (now outdated
# and unmaintained) linuxthreads threading library. There's an issue
# when combining linuxthreads with a failed execv call: see
# http://bugs.python.org/issue4970.
ifhasattr(sys, 'thread_info') andsys.thread_info.version:
USING_LINUXTHREADS=sys.thread_info.version.startswith("linuxthreads")
else:
USING_LINUXTHREADS=False

# Issue #14110: Some tests fail on FreeBSD if the user is in the wheel group.
HAVE_WHEEL_GROUP=sys.platform.startswith('freebsd') andos.getgid() ==0


defrequires_os_func(name):
returnunittest.skipUnless(hasattr(os, name), 'requires os.%s'%name)


defcreate_file(filename, content=b'content'):
withopen(filename, "xb", 0) asfp:
fp.write(content)


# bpo-41625: On AIX, splice() only works with a socket, not with a pipe.
requires_splice_pipe=unittest.skipIf(sys.platform.startswith("aix"),
'on AIX, splice() only accepts sockets')


deftearDownModule():
asyncio._set_event_loop_policy(None)


classMiscTests(unittest.TestCase):
deftest_getcwd(self):
cwd=os.getcwd()
self.assertIsInstance(cwd, str)

deftest_getcwd_long_path(self):
# bpo-37412: On Linux, PATH_MAX is usually around 4096 bytes. On
# Windows, MAX_PATH is defined as 260 characters, but Windows supports
# longer path if longer paths support is enabled. Internally, the os
# module uses MAXPATHLEN which is at least 1024.
#
# Use a directory name of 200 characters to fit into Windows MAX_PATH
# limit.
#
# On Windows, the test can stop when trying to create a path longer
# than MAX_PATH if long paths support is disabled:
# see RtlAreLongPathsEnabled().
min_len=2000# characters
# On VxWorks, PATH_MAX is defined as 1024 bytes. Creating a path
# longer than PATH_MAX will fail.
ifsys.platform=='vxworks':
min_len=1000
dirlen=200# characters
dirname='python_test_dir_'
dirname=dirname+ ('a'* (dirlen-len(dirname)))

withtempfile.TemporaryDirectory() astmpdir:
withos_helper.change_cwd(tmpdir) aspath:
expected=path

whileTrue:
cwd=os.getcwd()
self.assertEqual(cwd, expected)

need=min_len- (len(cwd) +len(os.path.sep))
ifneed<=0:
break
iflen(dirname) >needandneed>0:
dirname=dirname[:need]

path=os.path.join(path, dirname)
try:
os.mkdir(path)
# On Windows, chdir() can fail
# even if mkdir() succeeded
os.chdir(path)
exceptFileNotFoundError:
# On Windows, catch ERROR_PATH_NOT_FOUND (3) and
# ERROR_FILENAME_EXCED_RANGE (206) errors
# ("The filename or extension is too long")
break
exceptOSErrorasexc:
ifexc.errno==errno.ENAMETOOLONG:
break
else:
raise

expected=path

ifsupport.verbose:
print(f"Tested current directory length: {len(cwd)}")

deftest_getcwdb(self):
cwd=os.getcwdb()
self.assertIsInstance(cwd, bytes)
self.assertEqual(os.fsdecode(cwd), os.getcwd())


# Tests creating TESTFN
classFileTests(unittest.TestCase):
defsetUp(self):
ifos.path.lexists(os_helper.TESTFN):
os.unlink(os_helper.TESTFN)
tearDown=setUp

deftest_access(self):
f=os.open(os_helper.TESTFN, os.O_CREAT|os.O_RDWR)
os.close(f)
self.assertTrue(os.access(os_helper.TESTFN, os.W_OK))

@unittest.skipIf(
support.is_wasi, "WASI does not support dup."
 )
deftest_closerange(self):
first=os.open(os_helper.TESTFN, os.O_CREAT|os.O_RDWR)
# We must allocate two consecutive file descriptors, otherwise
# it will mess up other file descriptors (perhaps even the three
# standard ones).
second=os.dup(first)
try:
retries=0
whilesecond!=first+1:
os.close(first)
retries+=1
ifretries>10:
# XXX test skipped
self.skipTest("couldn't allocate two consecutive fds")
first, second=second, os.dup(second)
finally:
os.close(second)
# close a fd that is open, and one that isn't
os.closerange(first, first+2)
self.assertRaises(OSError, os.write, first, b"a")

@support.cpython_only
deftest_rename(self):
path=os_helper.TESTFN
old=sys.getrefcount(path)
self.assertRaises(TypeError, os.rename, path, 0)
new=sys.getrefcount(path)
self.assertEqual(old, new)

deftest_read(self):
withopen(os_helper.TESTFN, "w+b") asfobj:
fobj.write(b"spam")
fobj.flush()
fd=fobj.fileno()
os.lseek(fd, 0, 0)
s=os.read(fd, 4)
self.assertEqual(type(s), bytes)
self.assertEqual(s, b"spam")

deftest_readinto(self):
withopen(os_helper.TESTFN, "w+b") asfobj:
fobj.write(b"spam")
fobj.flush()
fd=fobj.fileno()
os.lseek(fd, 0, 0)
# Oversized so readinto without hitting end.
buffer=bytearray(7)
s=os.readinto(fd, buffer)
self.assertEqual(type(s), int)
self.assertEqual(s, 4)
# Should overwrite the first 4 bytes of the buffer.
self.assertEqual(buffer[:4], b"spam")

# Readinto at EOF should return 0 and not touch buffer.
buffer[:] =b"notspam"
s=os.readinto(fd, buffer)
self.assertEqual(type(s), int)
self.assertEqual(s, 0)
self.assertEqual(bytes(buffer), b"notspam")
s=os.readinto(fd, buffer)
self.assertEqual(s, 0)
self.assertEqual(bytes(buffer), b"notspam")

# Readinto a 0 length bytearray when at EOF should return 0
self.assertEqual(os.readinto(fd, bytearray()), 0)

# Readinto a 0 length bytearray with data available should return 0.
os.lseek(fd, 0, 0)
self.assertEqual(os.readinto(fd, bytearray()), 0)

@unittest.skipUnless(hasattr(os, 'get_blocking'),
'needs os.get_blocking() and os.set_blocking()')
@unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
@unittest.skipIf(support.is_emscripten, "set_blocking does not work correctly")
deftest_readinto_non_blocking(self):
# Verify behavior of a readinto which would block on a non-blocking fd.
r, w=os.pipe()
try:
os.set_blocking(r, False)
withself.assertRaises(BlockingIOError):
os.readinto(r, bytearray(5))

# Pass some data through
os.write(w, b"spam")
self.assertEqual(os.readinto(r, bytearray(4)), 4)

# Still don't block or return 0.
withself.assertRaises(BlockingIOError):
os.readinto(r, bytearray(5))

# At EOF should return size 0
os.close(w)
w=None
self.assertEqual(os.readinto(r, bytearray(5)), 0)
self.assertEqual(os.readinto(r, bytearray(5)), 0) # Still EOF

finally:
os.close(r)
ifwisnotNone:
os.close(w)

deftest_readinto_badarg(self):
withopen(os_helper.TESTFN, "w+b") asfobj:
fobj.write(b"spam")
fobj.flush()
fd=fobj.fileno()
os.lseek(fd, 0, 0)

forbad_argin ("test", bytes(), 14):
withself.subTest(f"bad buffer {type(bad_arg)}"):
withself.assertRaises(TypeError):
os.readinto(fd, bad_arg)

withself.subTest("doesn't work on file objects"):
withself.assertRaises(TypeError):
os.readinto(fobj, bytearray(5))

# takes two args
withself.assertRaises(TypeError):
os.readinto(fd)

# No data should have been read with the bad arguments.
buffer=bytearray(4)
s=os.readinto(fd, buffer)
self.assertEqual(s, 4)
self.assertEqual(buffer, b"spam")

@support.cpython_only
# Skip the test on 32-bit platforms: the number of bytes must fit in a
# Py_ssize_t type
@unittest.skipUnless(INT_MAX<PY_SSIZE_T_MAX,
"needs INT_MAX < PY_SSIZE_T_MAX")
@support.bigmemtest(size=INT_MAX+10, memuse=1, dry_run=False)
deftest_large_read(self, size):
self.addCleanup(os_helper.unlink, os_helper.TESTFN)
create_file(os_helper.TESTFN, b'test')

# Issue #21932: Make sure that os.read() does not raise an
# OverflowError for size larger than INT_MAX
withopen(os_helper.TESTFN, "rb") asfp:
data=os.read(fp.fileno(), size)

# The test does not try to read more than 2 GiB at once because the
# operating system is free to return less bytes than requested.
self.assertEqual(data, b'test')


@support.cpython_only
# Skip the test on 32-bit platforms: the number of bytes must fit in a
# Py_ssize_t type
@unittest.skipUnless(INT_MAX<PY_SSIZE_T_MAX,
"needs INT_MAX < PY_SSIZE_T_MAX")
@support.bigmemtest(size=INT_MAX+10, memuse=1, dry_run=False)
deftest_large_readinto(self, size):
self.addCleanup(os_helper.unlink, os_helper.TESTFN)
create_file(os_helper.TESTFN, b'test')

# Issue #21932: For readinto the buffer contains the length rather than
# a length being passed explicitly to read, should still get capped to a
# valid size / not raise an OverflowError for sizes larger than INT_MAX.
buffer=bytearray(INT_MAX+10)
withopen(os_helper.TESTFN, "rb") asfp:
length=os.readinto(fp.fileno(), buffer)

# The test does not try to read more than 2 GiB at once because the
# operating system is free to return less bytes than requested.
self.assertEqual(length, 4)
self.assertEqual(buffer[:4], b'test')

deftest_write(self):
# os.write() accepts bytes- and buffer-like objects but not strings
fd=os.open(os_helper.TESTFN, os.O_CREAT|os.O_WRONLY)
self.assertRaises(TypeError, os.write, fd, "beans")
os.write(fd, b"bacon\n")
os.write(fd, bytearray(b"eggs\n"))
os.write(fd, memoryview(b"spam\n"))
os.close(fd)
withopen(os_helper.TESTFN, "rb") asfobj:
self.assertEqual(fobj.read().splitlines(),
 [b"bacon", b"eggs", b"spam"])

defwrite_windows_console(self, *args):
retcode=subprocess.call(args,
# use a new console to not flood the test output
creationflags=subprocess.CREATE_NEW_CONSOLE,
# use a shell to hide the console window (SW_HIDE)
shell=True)
self.assertEqual(retcode, 0)

@unittest.skipUnless(sys.platform=='win32',
'test specific to the Windows console')
deftest_write_windows_console(self):
# Issue #11395: the Windows console returns an error (12: not enough
# space error) on writing into stdout if stdout mode is binary and the
# length is greater than 66,000 bytes (or less, depending on heap
# usage).
code="print('x' * 100000)"
self.write_windows_console(sys.executable, "-c", code)
self.write_windows_console(sys.executable, "-u", "-c", code)

deffdopen_helper(self, *args):
fd=os.open(os_helper.TESTFN, os.O_RDONLY)
f=os.fdopen(fd, *args, encoding="utf-8")
f.close()

deftest_fdopen(self):
fd=os.open(os_helper.TESTFN, os.O_CREAT|os.O_RDWR)
os.close(fd)

self.fdopen_helper()
self.fdopen_helper('r')
self.fdopen_helper('r', 100)

deftest_replace(self):
TESTFN2=os_helper.TESTFN+".2"
self.addCleanup(os_helper.unlink, os_helper.TESTFN)
self.addCleanup(os_helper.unlink, TESTFN2)

create_file(os_helper.TESTFN, b"1")
create_file(TESTFN2, b"2")

os.replace(os_helper.TESTFN, TESTFN2)
self.assertRaises(FileNotFoundError, os.stat, os_helper.TESTFN)
withopen(TESTFN2, 'r', encoding='utf-8') asf:
self.assertEqual(f.read(), "1")

deftest_open_keywords(self):
f=os.open(path=__file__, flags=os.O_RDONLY, mode=0o777,
dir_fd=None)
os.close(f)

deftest_symlink_keywords(self):
symlink=support.get_attribute(os, "symlink")
try:
symlink(src='target', dst=os_helper.TESTFN,
target_is_directory=False, dir_fd=None)
except (NotImplementedError, OSError):
pass# No OS support or unprivileged user

@unittest.skipUnless(hasattr(os, 'copy_file_range'), 'test needs os.copy_file_range()')
deftest_copy_file_range_invalid_values(self):
withself.assertRaises(ValueError):
os.copy_file_range(0, 1, -10)

@unittest.skipUnless(hasattr(os, 'copy_file_range'), 'test needs os.copy_file_range()')
deftest_copy_file_range(self):
TESTFN2=os_helper.TESTFN+".3"
data=b'0123456789'

create_file(os_helper.TESTFN, data)
self.addCleanup(os_helper.unlink, os_helper.TESTFN)

in_file=open(os_helper.TESTFN, 'rb')
self.addCleanup(in_file.close)
in_fd=in_file.fileno()

out_file=open(TESTFN2, 'w+b')
self.addCleanup(os_helper.unlink, TESTFN2)
self.addCleanup(out_file.close)
out_fd=out_file.fileno()

try:
i=os.copy_file_range(in_fd, out_fd, 5)
exceptOSErrorase:
# Handle the case in which Python was compiled
# in a system with the syscall but without support
# in the kernel.
ife.errno!=errno.ENOSYS:
raise
self.skipTest(e)
else:
# The number of copied bytes can be less than
# the number of bytes originally requested.
self.assertIn(i, range(0, 6));

withopen(TESTFN2, 'rb') asin_file:
self.assertEqual(in_file.read(), data[:i])

@unittest.skipUnless(hasattr(os, 'copy_file_range'), 'test needs os.copy_file_range()')
deftest_copy_file_range_offset(self):
TESTFN4=os_helper.TESTFN+".4"
data=b'0123456789'
bytes_to_copy=6
in_skip=3
out_seek=5

create_file(os_helper.TESTFN, data)
self.addCleanup(os_helper.unlink, os_helper.TESTFN)

in_file=open(os_helper.TESTFN, 'rb')
self.addCleanup(in_file.close)
in_fd=in_file.fileno()

out_file=open(TESTFN4, 'w+b')
self.addCleanup(os_helper.unlink, TESTFN4)
self.addCleanup(out_file.close)
out_fd=out_file.fileno()

try:
i=os.copy_file_range(in_fd, out_fd, bytes_to_copy,
offset_src=in_skip,
offset_dst=out_seek)
exceptOSErrorase:
# Handle the case in which Python was compiled
# in a system with the syscall but without support
# in the kernel.
ife.errno!=errno.ENOSYS:
raise
self.skipTest(e)
else:
# The number of copied bytes can be less than
# the number of bytes originally requested.
self.assertIn(i, range(0, bytes_to_copy+1));

withopen(TESTFN4, 'rb') asin_file:
read=in_file.read()
# seeked bytes (5) are zero'ed
self.assertEqual(read[:out_seek], b'\x00'*out_seek)
# 012 are skipped (in_skip)
# 345678 are copied in the file (in_skip + bytes_to_copy)
self.assertEqual(read[out_seek:],
data[in_skip:in_skip+i])

@unittest.skipUnless(hasattr(os, 'splice'), 'test needs os.splice()')
deftest_splice_invalid_values(self):
withself.assertRaises(ValueError):
os.splice(0, 1, -10)

@unittest.skipUnless(hasattr(os, 'splice'), 'test needs os.splice()')
@requires_splice_pipe
deftest_splice(self):
TESTFN2=os_helper.TESTFN+".3"
data=b'0123456789'

create_file(os_helper.TESTFN, data)
self.addCleanup(os_helper.unlink, os_helper.TESTFN)

in_file=open(os_helper.TESTFN, 'rb')
self.addCleanup(in_file.close)
in_fd=in_file.fileno()

read_fd, write_fd=os.pipe()
self.addCleanup(lambda: os.close(read_fd))
self.addCleanup(lambda: os.close(write_fd))

try:
i=os.splice(in_fd, write_fd, 5)
exceptOSErrorase:
# Handle the case in which Python was compiled
# in a system with the syscall but without support
# in the kernel.
ife.errno!=errno.ENOSYS:
raise
self.skipTest(e)
else:
# The number of copied bytes can be less than
# the number of bytes originally requested.
self.assertIn(i, range(0, 6));

self.assertEqual(os.read(read_fd, 100), data[:i])

@unittest.skipUnless(hasattr(os, 'splice'), 'test needs os.splice()')
@requires_splice_pipe
deftest_splice_offset_in(self):
TESTFN4=os_helper.TESTFN+".4"
data=b'0123456789'
bytes_to_copy=6
in_skip=3

create_file(os_helper.TESTFN, data)
self.addCleanup(os_helper.unlink, os_helper.TESTFN)

in_file=open(os_helper.TESTFN, 'rb')
self.addCleanup(in_file.close)
in_fd=in_file.fileno()

read_fd, write_fd=os.pipe()
self.addCleanup(lambda: os.close(read_fd))
self.addCleanup(lambda: os.close(write_fd))

try:
i=os.splice(in_fd, write_fd, bytes_to_copy, offset_src=in_skip)
exceptOSErrorase:
# Handle the case in which Python was compiled
# in a system with the syscall but without support
# in the kernel.
ife.errno!=errno.ENOSYS:
raise
self.skipTest(e)
else:
# The number of copied bytes can be less than
# the number of bytes originally requested.
self.assertIn(i, range(0, bytes_to_copy+1));

read=os.read(read_fd, 100)
# 012 are skipped (in_skip)
# 345678 are copied in the file (in_skip + bytes_to_copy)
self.assertEqual(read, data[in_skip:in_skip+i])

@unittest.skipUnless(hasattr(os, 'splice'), 'test needs os.splice()')
@requires_splice_pipe
deftest_splice_offset_out(self):
TESTFN4=os_helper.TESTFN+".4"
data=b'0123456789'
bytes_to_copy=6
out_seek=3

create_file(os_helper.TESTFN, data)
self.addCleanup(os_helper.unlink, os_helper.TESTFN)

read_fd, write_fd=os.pipe()
self.addCleanup(lambda: os.close(read_fd))
self.addCleanup(lambda: os.close(write_fd))
os.write(write_fd, data)

out_file=open(TESTFN4, 'w+b')
self.addCleanup(os_helper.unlink, TESTFN4)
self.addCleanup(out_file.close)
out_fd=out_file.fileno()

try:
i=os.splice(read_fd, out_fd, bytes_to_copy, offset_dst=out_seek)
exceptOSErrorase:
# Handle the case in which Python was compiled
# in a system with the syscall but without support
# in the kernel.
ife.errno!=errno.ENOSYS:
raise
self.skipTest(e)
else:
# The number of copied bytes can be less than
# the number of bytes originally requested.
self.assertIn(i, range(0, bytes_to_copy+1));

withopen(TESTFN4, 'rb') asin_file:
read=in_file.read()
# seeked bytes (5) are zero'ed
self.assertEqual(read[:out_seek], b'\x00'*out_seek)
# 012 are skipped (in_skip)
# 345678 are copied in the file (in_skip + bytes_to_copy)
self.assertEqual(read[out_seek:], data[:i])


# Test attributes on return values from os.*stat* family.
classStatAttributeTests(unittest.TestCase):
defsetUp(self):
self.fname=os_helper.TESTFN
self.addCleanup(os_helper.unlink, self.fname)
create_file(self.fname, b"ABC")

defcheck_stat_attributes(self, fname):
result=os.stat(fname)

# Make sure direct access works
self.assertEqual(result[stat.ST_SIZE], 3)
self.assertEqual(result.st_size, 3)

# Make sure all the attributes are there
members=dir(result)
fornameindir(stat):
ifname[:3] =='ST_':
attr=name.lower()
ifname.endswith("TIME"):
deftrunc(x): returnint(x)
else:
deftrunc(x): returnx
self.assertEqual(trunc(getattr(result, attr)),
result[getattr(stat, name)])
self.assertIn(attr, members)

# Make sure that the st_?time and st_?time_ns fields roughly agree
# (they should always agree up to around tens-of-microseconds)
fornamein'st_atime st_mtime st_ctime'.split():
floaty=int(getattr(result, name) *100000)
nanosecondy=getattr(result, name+"_ns") //10000
self.assertAlmostEqual(floaty, nanosecondy, delta=2)

# Ensure both birthtime and birthtime_ns roughly agree, if present
try:
floaty=int(result.st_birthtime*100000)
nanosecondy=result.st_birthtime_ns//10000
exceptAttributeError:
pass
else:
self.assertAlmostEqual(floaty, nanosecondy, delta=2)

try:
result[200]
self.fail("No exception raised")
exceptIndexError:
pass

# Make sure that assignment fails
try:
result.st_mode=1
self.fail("No exception raised")
exceptAttributeError:
pass

try:
result.st_rdev=1
self.fail("No exception raised")
except (AttributeError, TypeError):
pass

try:
result.parrot=1
self.fail("No exception raised")
exceptAttributeError:
pass

# Use the stat_result constructor with a too-short tuple.
try:
result2=os.stat_result((10,))
self.fail("No exception raised")
exceptTypeError:
pass

# Use the constructor with a too-long tuple.
try:
result2=os.stat_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
exceptTypeError:
pass

deftest_stat_attributes(self):
self.check_stat_attributes(self.fname)

deftest_stat_attributes_bytes(self):
try:
fname=self.fname.encode(sys.getfilesystemencoding())
exceptUnicodeEncodeError:
self.skipTest("cannot encode %a for the filesystem"%self.fname)
self.check_stat_attributes(fname)

deftest_stat_result_pickle(self):
result=os.stat(self.fname)
forprotoinrange(pickle.HIGHEST_PROTOCOL+1):
withself.subTest(f'protocol {proto}'):
p=pickle.dumps(result, proto)
self.assertIn(b'stat_result', p)
ifproto<4:
self.assertIn(b'cos\nstat_result\n', p)
unpickled=pickle.loads(p)
self.assertEqual(result, unpickled)

@unittest.skipUnless(hasattr(os, 'statvfs'), 'test needs os.statvfs()')
deftest_statvfs_attributes(self):
result=os.statvfs(self.fname)

# Make sure direct access works
self.assertEqual(result.f_bfree, result[3])

# Make sure all the attributes are there.
members= ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
'ffree', 'favail', 'flag', 'namemax')
forvalue, memberinenumerate(members):
self.assertEqual(getattr(result, 'f_'+member), result[value])

self.assertTrue(isinstance(result.f_fsid, int))

# Test that the size of the tuple doesn't change
self.assertEqual(len(result), 10)

# Make sure that assignment really fails
try:
result.f_bfree=1
self.fail("No exception raised")
exceptAttributeError:
pass

try:
result.parrot=1
self.fail("No exception raised")
exceptAttributeError:
pass

# Use the constructor with a too-short tuple.
try:
result2=os.statvfs_result((10,))
self.fail("No exception raised")
exceptTypeError:
pass

# Use the constructor with a too-long tuple.
try:
result2=os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
exceptTypeError:
pass

@unittest.skipUnless(hasattr(os, 'statvfs'),
"need os.statvfs()")
deftest_statvfs_result_pickle(self):
result=os.statvfs(self.fname)

forprotoinrange(pickle.HIGHEST_PROTOCOL+1):
p=pickle.dumps(result, proto)
self.assertIn(b'statvfs_result', p)
ifproto<4:
self.assertIn(b'cos\nstatvfs_result\n', p)
unpickled=pickle.loads(p)
self.assertEqual(result, unpickled)

@unittest.skipUnless(sys.platform=="win32", "Win32 specific tests")
deftest_1686475(self):
# Verify that an open file can be stat'ed
try:
os.stat(r"c:\pagefile.sys")
exceptFileNotFoundError:
self.skipTest(r'c:\pagefile.sys does not exist')
exceptOSErrorase:
self.fail("Could not stat pagefile.sys")

@unittest.skipUnless(sys.platform=="win32", "Win32 specific tests")
@unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
deftest_15261(self):
# Verify that stat'ing a closed fd does not cause crash
r, w=os.pipe()
try:
os.stat(r) # should not raise error
finally:
os.close(r)
os.close(w)
withself.assertRaises(OSError) asctx:
os.stat(r)
self.assertEqual(ctx.exception.errno, errno.EBADF)

defcheck_file_attributes(self, result):
self.assertTrue(hasattr(result, 'st_file_attributes'))
self.assertTrue(isinstance(result.st_file_attributes, int))
self.assertTrue(0<=result.st_file_attributes<=0xFFFFFFFF)

@unittest.skipUnless(sys.platform=="win32",
"st_file_attributes is Win32 specific")
deftest_file_attributes(self):
# test file st_file_attributes (FILE_ATTRIBUTE_DIRECTORY not set)
result=os.stat(self.fname)
self.check_file_attributes(result)
self.assertEqual(
result.st_file_attributes&stat.FILE_ATTRIBUTE_DIRECTORY,
0)

# test directory st_file_attributes (FILE_ATTRIBUTE_DIRECTORY set)
dirname=os_helper.TESTFN+"dir"
os.mkdir(dirname)
self.addCleanup(os.rmdir, dirname)

result=os.stat(dirname)
self.check_file_attributes(result)
self.assertEqual(
result.st_file_attributes&stat.FILE_ATTRIBUTE_DIRECTORY,
stat.FILE_ATTRIBUTE_DIRECTORY)

@unittest.skipUnless(sys.platform=="win32", "Win32 specific tests")
deftest_access_denied(self):
# Default to FindFirstFile WIN32_FIND_DATA when access is
# denied. See issue 28075.
# os.environ['TEMP'] should be located on a volume that
# supports file ACLs.
fname=os.path.join(os.environ['TEMP'], self.fname+"_access")
self.addCleanup(os_helper.unlink, fname)
create_file(fname, b'ABC')
# Deny the right to [S]YNCHRONIZE on the file to
# force CreateFile to fail with ERROR_ACCESS_DENIED.
DETACHED_PROCESS=8
subprocess.check_call(
# bpo-30584: Use security identifier *S-1-5-32-545 instead
# of localized "Users" to not depend on the locale.
 ['icacls.exe', fname, '/deny', '*S-1-5-32-545:(S)'],
creationflags=DETACHED_PROCESS
 )
result=os.stat(fname)
self.assertNotEqual(result.st_size, 0)
self.assertTrue(os.path.isfile(fname))

@unittest.skipUnless(sys.platform=="win32", "Win32 specific tests")
deftest_stat_block_device(self):
# bpo-38030: os.stat fails for block devices
# Test a filename like "//./C:"
fname="//./"+os.path.splitdrive(os.getcwd())[0]
result=os.stat(fname)
self.assertEqual(result.st_mode, stat.S_IFBLK)


classUtimeTests(unittest.TestCase):
defsetUp(self):
self.dirname=os_helper.TESTFN
self.fname=os.path.join(self.dirname, "f1")

self.addCleanup(os_helper.rmtree, self.dirname)
os.mkdir(self.dirname)
create_file(self.fname)

defsupport_subsecond(self, filename):
# Heuristic to check if the filesystem supports timestamp with
# subsecond resolution: check if float and int timestamps are different
st=os.stat(filename)
return ((st.st_atime!=st[7])
or (st.st_mtime!=st[8])
or (st.st_ctime!=st[9]))

def_test_utime(self, set_time, filename=None):
ifnotfilename:
filename=self.fname

support_subsecond=self.support_subsecond(filename)
ifsupport_subsecond:
# Timestamp with a resolution of 1 microsecond (10^-6).
#
# The resolution of the C internal function used by os.utime()
# depends on the platform: 1 sec, 1 us, 1 ns. Writing a portable
# test with a resolution of 1 ns requires more work:
# see the issue #15745.
atime_ns=1002003000# 1.002003 seconds
mtime_ns=4005006000# 4.005006 seconds
else:
# use a resolution of 1 second
atime_ns=5*10**9
mtime_ns=8*10**9

set_time(filename, (atime_ns, mtime_ns))
st=os.stat(filename)

ifsupport.is_emscripten:
# Emscripten timestamps are roundtripped through a 53 bit integer of
# nanoseconds. If we want to represent ~50 years which is an 11
# digits number of seconds:
# 2*log10(60) + log10(24) + log10(365) + log10(60) + log10(50)
# is about 11. Because 53 * log10(2) is about 16, we only have 5
# digits worth of sub-second precision.
# Some day it would be good to fix this upstream.
delta=1e-5
self.assertAlmostEqual(st.st_atime, atime_ns*1e-9, delta=1e-5)
self.assertAlmostEqual(st.st_mtime, mtime_ns*1e-9, delta=1e-5)
self.assertAlmostEqual(st.st_atime_ns, atime_ns, delta=1e9*1e-5)
self.assertAlmostEqual(st.st_mtime_ns, mtime_ns, delta=1e9*1e-5)
else:
ifsupport_subsecond:
self.assertAlmostEqual(st.st_atime, atime_ns*1e-9, delta=1e-6)
self.assertAlmostEqual(st.st_mtime, mtime_ns*1e-9, delta=1e-6)
else:
self.assertEqual(st.st_atime, atime_ns*1e-9)
self.assertEqual(st.st_mtime, mtime_ns*1e-9)
self.assertEqual(st.st_atime_ns, atime_ns)
self.assertEqual(st.st_mtime_ns, mtime_ns)

deftest_utime(self):
defset_time(filename, ns):
# test the ns keyword parameter
os.utime(filename, ns=ns)
self._test_utime(set_time)

@staticmethod
defns_to_sec(ns):
# Convert a number of nanosecond (int) to a number of seconds (float).
# Round towards infinity by adding 0.5 nanosecond to avoid rounding
# issue, os.utime() rounds towards minus infinity.
return (ns*1e-9) +0.5e-9

deftest_utime_by_indexed(self):
# pass times as floating-point seconds as the second indexed parameter
defset_time(filename, ns):
atime_ns, mtime_ns=ns
atime=self.ns_to_sec(atime_ns)
mtime=self.ns_to_sec(mtime_ns)
# test utimensat(timespec), utimes(timeval), utime(utimbuf)
# or utime(time_t)
os.utime(filename, (atime, mtime))
self._test_utime(set_time)

deftest_utime_by_times(self):
defset_time(filename, ns):
atime_ns, mtime_ns=ns
atime=self.ns_to_sec(atime_ns)
mtime=self.ns_to_sec(mtime_ns)
# test the times keyword parameter
os.utime(filename, times=(atime, mtime))
self._test_utime(set_time)

@unittest.skipUnless(os.utimeinos.supports_follow_symlinks,
"follow_symlinks support for utime required "
"for this test.")
deftest_utime_nofollow_symlinks(self):
defset_time(filename, ns):
# use follow_symlinks=False to test utimensat(timespec)
# or lutimes(timeval)
os.utime(filename, ns=ns, follow_symlinks=False)
self._test_utime(set_time)

@unittest.skipUnless(os.utimeinos.supports_fd,
"fd support for utime required for this test.")
deftest_utime_fd(self):
defset_time(filename, ns):
withopen(filename, 'wb', 0) asfp:
# use a file descriptor to test futimens(timespec)
# or futimes(timeval)
os.utime(fp.fileno(), ns=ns)
self._test_utime(set_time)

@unittest.skipUnless(os.utimeinos.supports_dir_fd,
"dir_fd support for utime required for this test.")
deftest_utime_dir_fd(self):
defset_time(filename, ns):
dirname, name=os.path.split(filename)
withos_helper.open_dir_fd(dirname) asdirfd:
# pass dir_fd to test utimensat(timespec) or futimesat(timeval)
os.utime(name, dir_fd=dirfd, ns=ns)
self._test_utime(set_time)