def setUp(self):

if os.path.lexists(support.TESTFN):

os.unlink(support.TESTFN)

tearDown = setUp

def test_access(self):

f = os.open(support.TESTFN, os.O_CREAT|os.O_RDWR)

os.close(f)

self.assertTrue(os.access(support.TESTFN, os.W_OK))

def test_closerange(self):

first = os.open(support.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

while second != first + 1:

os.close(first)

retries += 1

if retries > 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

def test_rename(self):

path = support.TESTFN

old = sys.getrefcount(path)

self.assertRaises(TypeError, os.rename, path, 0)

new = sys.getrefcount(path)

self.assertEqual(old, new)

def test_read(self):

with open(support.TESTFN, "w+b") as fobj:

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")

def test_write(self):

# os.write() accepts bytes- and buffer-like objects but not strings

fd = os.open(support.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)

with open(support.TESTFN, "rb") as fobj:

self.assertEqual(fobj.read().splitlines(),

[b"bacon", b"eggs", b"spam"])

def write_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',

def test_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)

def fdopen_helper(self, *args):

fd = os.open(support.TESTFN, os.O_RDONLY)

f = os.fdopen(fd, *args)

f.close()

def test_fdopen(self):

fd = os.open(support.TESTFN, os.O_CREAT|os.O_RDWR)

os.close(fd)

self.fdopen_helper()

self.fdopen_helper('r')

self.fdopen_helper('r', 100)

def test_replace(self):

TESTFN2 = support.TESTFN + ".2"

with open(support.TESTFN, 'w') as f:

f.write("1")

with open(TESTFN2, 'w') as f:

f.write("2")

self.addCleanup(os.unlink, TESTFN2)

os.replace(support.TESTFN, TESTFN2)

self.assertRaises(FileNotFoundError, os.stat, support.TESTFN)

with open(TESTFN2, 'r') as f:

self.assertEqual(f.read(), "1")

def test_open_keywords(self):

f = os.open(path=__file__, flags=os.O_RDONLY, mode=0o777,

dir_fd=None)

os.close(f)

def test_symlink_keywords(self):

symlink = support.get_attribute(os, "symlink")

try:

symlink(src='target', dst=support.TESTFN,

target_is_directory=False, dir_fd=None)

except (NotImplementedError, OSError):

def setUp(self):

os.mkdir(support.TESTFN)

self.fname = os.path.join(support.TESTFN, "f1")

f = open(self.fname, 'wb')

f.write(b"ABC")

f.close()

def tearDown(self):

os.unlink(self.fname)

os.rmdir(support.TESTFN)

@unittest.skipUnless(hasattr(os, 'stat'), 'test needs os.stat()')

def check_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)

for name in dir(stat):

if name[:3] == 'ST_':

attr = name.lower()

if name.endswith("TIME"):

def trunc(x): return int(x)

else:

def trunc(x): return x

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)

for name in '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)

try:

result[200]

self.fail("No exception raised")

except IndexError:

pass

# Make sure that assignment fails

try:

result.st_mode = 1

self.fail("No exception raised")

except AttributeError:

pass

try:

result.st_rdev = 1

self.fail("No exception raised")

except (AttributeError, TypeError):

pass

try:

result.parrot = 1

self.fail("No exception raised")

except AttributeError:

pass

# Use the stat_result constructor with a too-short tuple.

try:

result2 = os.stat_result((10,))

self.fail("No exception raised")

except TypeError:

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))

except TypeError:

pass

def test_stat_attributes(self):

self.check_stat_attributes(self.fname)

def test_stat_attributes_bytes(self):

try:

fname = self.fname.encode(sys.getfilesystemencoding())

except UnicodeEncodeError:

self.skipTest("cannot encode %a for the filesystem" % self.fname)

with warnings.catch_warnings():

warnings.simplefilter("ignore", DeprecationWarning)

self.check_stat_attributes(fname)

def test_stat_result_pickle(self):

result = os.stat(self.fname)

for proto in range(pickle.HIGHEST_PROTOCOL + 1):

p = pickle.dumps(result, proto)

self.assertIn(b'stat_result', p)

if proto < 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()')

def test_statvfs_attributes(self):

try:

result = os.statvfs(self.fname)

except OSError as e:

# On AtheOS, glibc always returns ENOSYS

if e.errno == errno.ENOSYS:

self.skipTest('os.statvfs() failed with ENOSYS')

# 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')

for value, member in enumerate(members):

self.assertEqual(getattr(result, 'f_' + member), result[value])

# Make sure that assignment really fails

try:

result.f_bfree = 1

self.fail("No exception raised")

except AttributeError:

pass

try:

result.parrot = 1

self.fail("No exception raised")

except AttributeError:

pass

# Use the constructor with a too-short tuple.

try:

result2 = os.statvfs_result((10,))

self.fail("No exception raised")

except TypeError:

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))

except TypeError:

pass

@unittest.skipUnless(hasattr(os, 'statvfs'),

def test_statvfs_result_pickle(self):

try:

result = os.statvfs(self.fname)

except OSError as e:

# On AtheOS, glibc always returns ENOSYS

if e.errno == errno.ENOSYS:

self.skipTest('os.statvfs() failed with ENOSYS')

for proto in range(pickle.HIGHEST_PROTOCOL + 1):

p = pickle.dumps(result, proto)

self.assertIn(b'statvfs_result', p)

if proto < 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")

def test_1686475(self):

# Verify that an open file can be stat'ed

try:

os.stat(r"c:\pagefile.sys")

except FileNotFoundError:

self.skipTest(r'c:\pagefile.sys does not exist')

except OSError as e:

self.fail("Could not stat pagefile.sys")

@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")

@unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")

def test_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)

with self.assertRaises(OSError) as ctx:

os.stat(r)

def setUp(self):

self.dirname = support.TESTFN

self.fname = os.path.join(self.dirname, "f1")

self.addCleanup(support.rmtree, self.dirname)

os.mkdir(self.dirname)

with open(self.fname, 'wb') as fp:

fp.write(b"ABC")

def restore_float_times(state):

with warnings.catch_warnings():

warnings.simplefilter("ignore", DeprecationWarning)

os.stat_float_times(state)

# ensure that st_atime and st_mtime are float

with warnings.catch_warnings():

warnings.simplefilter("ignore", DeprecationWarning)

old_float_times = os.stat_float_times(-1)

self.addCleanup(restore_float_times, old_float_times)

os.stat_float_times(True)

def support_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):

if not filename:

filename = self.fname

support_subsecond = self.support_subsecond(filename)

if support_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)

if support_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)

def test_utime(self):

def set_time(filename, ns):

# test the ns keyword parameter

os.utime(filename, ns=ns)

self._test_utime(set_time)

@staticmethod

def ns_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

def test_utime_by_indexed(self):

# pass times as floating point seconds as the second indexed parameter

def set_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)

def test_utime_by_times(self):

def set_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.utime in os.supports_follow_symlinks,

def test_utime_nofollow_symlinks(self):

def set_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.utime in os.supports_fd,

def test_utime_fd(self):

def set_time(filename, ns):

with open(filename, 'wb') as fp:

# 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.utime in os.supports_dir_fd,

def test_utime_dir_fd(self):

def set_time(filename, ns):

dirname, name = os.path.split(filename)

dirfd = os.open(dirname, os.O_RDONLY)

try:

# pass dir_fd to test utimensat(timespec) or futimesat(timeval)

os.utime(name, dir_fd=dirfd, ns=ns)

finally:

os.close(dirfd)

self._test_utime(set_time)

def test_utime_directory(self):

def set_time(filename, ns):

# test calling os.utime() on a directory

os.utime(filename, ns=ns)

self._test_utime(set_time, filename=self.dirname)

def _test_utime_current(self, set_time):

# Get the system clock

current = time.time()

# Call os.utime() to set the timestamp to the current system clock

set_time(self.fname)

if not self.support_subsecond(self.fname):

delta = 1.0

else:

# On Windows, the usual resolution of time.time() is 15.6 ms

delta = 0.020

st = os.stat(self.fname)

msg = ("st_time=%r, current=%r, dt=%r"

% (st.st_mtime, current, st.st_mtime - current))

self.assertAlmostEqual(st.st_mtime, current,

delta=delta, msg=msg)

def test_utime_current(self):

def set_time(filename):

# Set to the current time in the new way

os.utime(self.fname)

self._test_utime_current(set_time)

def test_utime_current_old(self):

def set_time(filename):

# Set to the current time in the old explicit way.

os.utime(self.fname, None)

self._test_utime_current(set_time)

def get_file_system(self, path):

if sys.platform == 'win32':

root = os.path.splitdrive(os.path.abspath(path))[0] + '\\'

import ctypes

kernel32 = ctypes.windll.kernel32

buf = ctypes.create_unicode_buffer("", 100)

ok = kernel32.GetVolumeInformationW(root, None, 0,

None, None, None,

buf, len(buf))

if ok:

return buf.value

# return None if the filesystem is unknown

def test_large_time(self):

# Many filesystems are limited to the year 2038. At least, the test

# pass with NTFS filesystem.

if self.get_file_system(self.dirname) != "NTFS":

self.skipTest("requires NTFS")

large = 5000000000 # some day in 2128

os.utime(self.fname, (large, large))

self.assertEqual(os.stat(self.fname).st_mtime, large)

def test_utime_invalid_arguments(self):

# seconds and nanoseconds parameters are mutually exclusive

with self.assertRaises(ValueError):

"""check that os.environ object conform to mapping protocol"""

type2test = None

def setUp(self):

self.__save = dict(os.environ)

if os.supports_bytes_environ:

self.__saveb = dict(os.environb)

for key, value in self._reference().items():

os.environ[key] = value

def tearDown(self):

os.environ.clear()

os.environ.update(self.__save)

if os.supports_bytes_environ:

os.environb.clear()

os.environb.update(self.__saveb)

def _reference(self):

return {"KEY1":"VALUE1", "KEY2":"VALUE2", "KEY3":"VALUE3"}

def _empty_mapping(self):

os.environ.clear()

return os.environ

# Bug 1110478

@unittest.skipUnless(os.path.exists('/bin/sh'), 'requires /bin/sh')

def test_update2(self):

os.environ.clear()

os.environ.update(HELLO="World")

with os.popen("/bin/sh -c 'echo $HELLO'") as popen:

value = popen.read().strip()

self.assertEqual(value, "World")

@unittest.skipUnless(os.path.exists('/bin/sh'), 'requires /bin/sh')

def test_os_popen_iter(self):

with os.popen(

"/bin/sh -c 'echo \"line1\nline2\nline3\"'") as popen:

it = iter(popen)

self.assertEqual(next(it), "line1\n")

self.assertEqual(next(it), "line2\n")

self.assertEqual(next(it), "line3\n")

self.assertRaises(StopIteration, next, it)

# Verify environ keys and values from the OS are of the

# correct str type.

def test_keyvalue_types(self):

for key, val in os.environ.items():

self.assertEqual(type(key), str)

self.assertEqual(type(val), str)

def test_items(self):

for key, value in self._reference().items():

self.assertEqual(os.environ.get(key), value)

# Issue 7310

def test___repr__(self):

"""Check that the repr() of os.environ looks like environ({...})."""

env = os.environ

self.assertEqual(repr(env), 'environ({{{}}})'.format(', '.join(

'{!r}: {!r}'.format(key, value)

for key, value in env.items())))

def test_get_exec_path(self):

defpath_list = os.defpath.split(os.pathsep)

test_path = ['/monty', '/python', '', '/flying/circus']

test_env = {'PATH': os.pathsep.join(test_path)}

saved_environ = os.environ

try:

os.environ = dict(test_env)

# Test that defaulting to os.environ works.

self.assertSequenceEqual(test_path, os.get_exec_path())

self.assertSequenceEqual(test_path, os.get_exec_path(env=None))

finally:

os.environ = saved_environ

# No PATH environment variable

self.assertSequenceEqual(defpath_list, os.get_exec_path({}))

# Empty PATH environment variable

self.assertSequenceEqual(('',), os.get_exec_path({'PATH':''}))

# Supplied PATH environment variable

self.assertSequenceEqual(test_path, os.get_exec_path(test_env))

if os.supports_bytes_environ:

# env cannot contain 'PATH' and b'PATH' keys

try:

# ignore BytesWarning warning

with warnings.catch_warnings(record=True):

mixed_env = {'PATH': '1', b'PATH': b'2'}

except BytesWarning:

# mixed_env cannot be created with python -bb

pass

else:

self.assertRaises(ValueError, os.get_exec_path, mixed_env)

# bytes key and/or value

self.assertSequenceEqual(os.get_exec_path({b'PATH': b'abc'}),

['abc'])

self.assertSequenceEqual(os.get_exec_path({b'PATH': 'abc'}),

['abc'])

self.assertSequenceEqual(os.get_exec_path({'PATH': b'abc'}),

['abc'])

@unittest.skipUnless(os.supports_bytes_environ,

def test_environb(self):

# os.environ -> os.environb

value = 'euro\u20ac'

try:

value_bytes = value.encode(sys.getfilesystemencoding(),

'surrogateescape')

except UnicodeEncodeError:

msg = "U+20AC character is not encodable to %s" % (

sys.getfilesystemencoding(),)

self.skipTest(msg)

os.environ['unicode'] = value

self.assertEqual(os.environ['unicode'], value)

self.assertEqual(os.environb[b'unicode'], value_bytes)

# os.environb -> os.environ

value = b'\xff'

os.environb[b'bytes'] = value

self.assertEqual(os.environb[b'bytes'], value)

value_str = value.decode(sys.getfilesystemencoding(), 'surrogateescape')

self.assertEqual(os.environ['bytes'], value_str)

# On FreeBSD < 7 and OS X < 10.6, unsetenv() doesn't return a value (issue

# #13415).

@support.requires_freebsd_version(7)

@support.requires_mac_ver(10, 6)

def test_unset_error(self):

if sys.platform == "win32":

# an environment variable is limited to 32,767 characters

key = 'x' * 50000

self.assertRaises(ValueError, os.environ.__delitem__, key)

else:

# "=" is not allowed in a variable name

key = 'key='

self.assertRaises(OSError, os.environ.__delitem__, key)

def test_key_type(self):

missing = 'missingkey'

self.assertNotIn(missing, os.environ)

with self.assertRaises(KeyError) as cm:

os.environ[missing]

self.assertIs(cm.exception.args[0], missing)

self.assertTrue(cm.exception.__suppress_context__)

with self.assertRaises(KeyError) as cm:

del os.environ[missing]

self.assertIs(cm.exception.args[0], missing)

"""Tests for os.walk()."""

# Wrapper to hide minor differences between os.walk and os.fwalk

# to tests both functions with the same code base

def walk(self, directory, topdown=True, follow_symlinks=False):

walk_it = os.walk(directory,

topdown=topdown,

followlinks=follow_symlinks)

for root, dirs, files in walk_it:

yield (root, dirs, files)

def setUp(self):

join = os.path.join

# Build:

# TESTFN/

# TEST1/ a file kid and two directory kids

# tmp1

# SUB1/ a file kid and a directory kid

# tmp2

# SUB11/ no kids

# SUB2/ a file kid and a dirsymlink kid

# tmp3

# link/ a symlink to TESTFN.2

# broken_link

# TEST2/

# tmp4 a lone file

self.walk_path = join(support.TESTFN, "TEST1")

self.sub1_path = join(self.walk_path, "SUB1")

self.sub11_path = join(self.sub1_path, "SUB11")

sub2_path = join(self.walk_path, "SUB2")

tmp1_path = join(self.walk_path, "tmp1")

tmp2_path = join(self.sub1_path, "tmp2")

tmp3_path = join(sub2_path, "tmp3")

self.link_path = join(sub2_path, "link")

t2_path = join(support.TESTFN, "TEST2")

tmp4_path = join(support.TESTFN, "TEST2", "tmp4")

broken_link_path = join(sub2_path, "broken_link")

# Create stuff.

os.makedirs(self.sub11_path)

os.makedirs(sub2_path)

os.makedirs(t2_path)

for path in tmp1_path, tmp2_path, tmp3_path, tmp4_path:

f = open(path, "w")

f.write("I'm " + path + " and proud of it. Blame test_os.\n")

f.close()

if support.can_symlink():

os.symlink(os.path.abspath(t2_path), self.link_path)

os.symlink('broken', broken_link_path, True)

if os.path.isdir(broken_link_path):

# On Windows a symlink can has the FILE_ATTRIBUTE_DIRECTORY flag.

self.sub2_tree = (sub2_path, ["broken_link", "link"], ["tmp3"])

else:

self.sub2_tree = (sub2_path, ["link"], ["broken_link", "tmp3"])

else:

self.sub2_tree = (sub2_path, [], ["tmp3"])

def test_walk_topdown(self):

# Walk top-down.

all = list(os.walk(self.walk_path))

self.assertEqual(len(all), 4)

# We can't know which order SUB1 and SUB2 will appear in.

# Not flipped: TESTFN, SUB1, SUB11, SUB2

# flipped: TESTFN, SUB2, SUB1, SUB11

flipped = all[0][1][0] != "SUB1"

all[0][1].sort()

all[3 - 2 * flipped][-1].sort()

self.assertEqual(all[0], (self.walk_path, ["SUB1", "SUB2"], ["tmp1"]))

self.assertEqual(all[1 + flipped], (self.sub1_path, ["SUB11"], ["tmp2"]))

self.assertEqual(all[2 + flipped], (self.sub11_path, [], []))

self.assertEqual(all[3 - 2 * flipped], self.sub2_tree)

def test_walk_prune(self):

# Prune the search.

all = []

for root, dirs, files in self.walk(self.walk_path):

all.append((root, dirs, files))

# Don't descend into SUB1.

if 'SUB1' in dirs:

# Note that this also mutates the dirs we appended to all!

dirs.remove('SUB1')

self.assertEqual(len(all), 2)

self.assertEqual(all[0],

(self.walk_path, ["SUB2"], ["tmp1"]))

all[1][-1].sort()

self.assertEqual(all[1], self.sub2_tree)

def test_walk_bottom_up(self):

# Walk bottom-up.

all = list(self.walk(self.walk_path, topdown=False))

self.assertEqual(len(all), 4)

# We can't know which order SUB1 and SUB2 will appear in.

# Not flipped: SUB11, SUB1, SUB2, TESTFN

# flipped: SUB2, SUB11, SUB1, TESTFN

flipped = all[3][1][0] != "SUB1"

all[3][1].sort()

all[2 - 2 * flipped][-1].sort()

self.assertEqual(all[3],

(self.walk_path, ["SUB1", "SUB2"], ["tmp1"]))

self.assertEqual(all[flipped],

(self.sub11_path, [], []))

self.assertEqual(all[flipped + 1],

(self.sub1_path, ["SUB11"], ["tmp2"]))

self.assertEqual(all[2 - 2 * flipped],

self.sub2_tree)

def test_walk_symlink(self):

if not support.can_symlink():

self.skipTest("need symlink support")

# Walk, following symlinks.

walk_it = self.walk(self.walk_path, follow_symlinks=True)

for root, dirs, files in walk_it:

if root == self.link_path:

self.assertEqual(dirs, [])

self.assertEqual(files, ["tmp4"])

break

else:

self.fail("Didn't follow symlink with followlinks=True")

def tearDown(self):

# Tear everything down. This is a decent use for bottom-up on

# Windows, which doesn't have a recursive delete command. The

# (not so) subtlety is that rmdir will fail unless the dir's

# kids are removed first, so bottom up is essential.

for root, dirs, files in os.walk(support.TESTFN, topdown=False):

for name in files:

os.remove(os.path.join(root, name))

for name in dirs:

dirname = os.path.join(root, name)

if not os.path.islink(dirname):

os.rmdir(dirname)

else:

os.remove(dirname)

"""Tests for os.fwalk()."""

def walk(self, directory, topdown=True, follow_symlinks=False):

walk_it = os.fwalk(directory,

topdown=topdown,

follow_symlinks=follow_symlinks)

for root, dirs, files, root_fd in walk_it:

yield (root, dirs, files)

def _compare_to_walk(self, walk_kwargs, fwalk_kwargs):

"""

walk_kwargs = walk_kwargs.copy()

fwalk_kwargs = fwalk_kwargs.copy()

for topdown, follow_symlinks in itertools.product((True, False), repeat=2):

walk_kwargs.update(topdown=topdown, followlinks=follow_symlinks)

fwalk_kwargs.update(topdown=topdown, follow_symlinks=follow_symlinks)

expected = {}

for root, dirs, files in os.walk(**walk_kwargs):

expected[root] = (set(dirs), set(files))

for root, dirs, files, rootfd in os.fwalk(**fwalk_kwargs):

self.assertIn(root, expected)

self.assertEqual(expected[root], (set(dirs), set(files)))

def test_compare_to_walk(self):

kwargs = {'top': support.TESTFN}

self._compare_to_walk(kwargs, kwargs)

def test_dir_fd(self):

try:

fd = os.open(".", os.O_RDONLY)

walk_kwargs = {'top': support.TESTFN}

fwalk_kwargs = walk_kwargs.copy()

fwalk_kwargs['dir_fd'] = fd

self._compare_to_walk(walk_kwargs, fwalk_kwargs)

finally:

os.close(fd)

def test_yields_correct_dir_fd(self):

# check returned file descriptors

for topdown, follow_symlinks in itertools.product((True, False), repeat=2):

args = support.TESTFN, topdown, None

for root, dirs, files, rootfd in os.fwalk(*args, follow_symlinks=follow_symlinks):

# check that the FD is valid

os.fstat(rootfd)

# redundant check

os.stat(rootfd)

# check that listdir() returns consistent information

self.assertEqual(set(os.listdir(rootfd)), set(dirs) | set(files))

def test_fd_leak(self):

# Since we're opening a lot of FDs, we must be careful to avoid leaks:

# we both check that calling fwalk() a large number of times doesn't

# yield EMFILE, and that the minimum allocated FD hasn't changed.

minfd = os.dup(1)

os.close(minfd)

for i in range(256):

for x in os.fwalk(support.TESTFN):

pass

newfd = os.dup(1)

self.addCleanup(os.close, newfd)

self.assertEqual(newfd, minfd)

def tearDown(self):

# cleanup

for root, dirs, files, rootfd in os.fwalk(support.TESTFN, topdown=False):

for name in files:

os.unlink(name, dir_fd=rootfd)

for name in dirs:

st = os.stat(name, dir_fd=rootfd, follow_symlinks=False)

if stat.S_ISDIR(st.st_mode):

os.rmdir(name, dir_fd=rootfd)

else:

os.unlink(name, dir_fd=rootfd)

def setUp(self):

os.mkdir(support.TESTFN)

def test_makedir(self):

base = support.TESTFN

path = os.path.join(base, 'dir1', 'dir2', 'dir3')

os.makedirs(path) # Should work

path = os.path.join(base, 'dir1', 'dir2', 'dir3', 'dir4')

os.makedirs(path)

# Try paths with a '.' in them

self.assertRaises(OSError, os.makedirs, os.curdir)

path = os.path.join(base, 'dir1', 'dir2', 'dir3', 'dir4', 'dir5', os.curdir)

os.makedirs(path)

path = os.path.join(base, 'dir1', os.curdir, 'dir2', 'dir3', 'dir4',

'dir5', 'dir6')

os.makedirs(path)

def test_exist_ok_existing_directory(self):

path = os.path.join(support.TESTFN, 'dir1')

mode = 0o777

old_mask = os.umask(0o022)

os.makedirs(path, mode)

self.assertRaises(OSError, os.makedirs, path, mode)

self.assertRaises(OSError, os.makedirs, path, mode, exist_ok=False)

os.makedirs(path, 0o776, exist_ok=True)

os.makedirs(path, mode=mode, exist_ok=True)

os.umask(old_mask)

# Issue #25583: A drive root could raise PermissionError on Windows

os.makedirs(os.path.abspath('/'), exist_ok=True)

@unittest.skipUnless(hasattr(os, 'chown'), 'test needs os.chown')

def test_chown_uid_gid_arguments_must_be_index(self):

stat = os.stat(support.TESTFN)

uid = stat.st_uid

gid = stat.st_gid

for value in (-1.0, -1j, decimal.Decimal(-1), fractions.Fraction(-2, 2)):

self.assertRaises(TypeError, os.chown, support.TESTFN, value, gid)

self.assertRaises(TypeError, os.chown, support.TESTFN, uid, value)

self.assertIsNone(os.chown(support.TESTFN, uid, gid))

self.assertIsNone(os.chown(support.TESTFN, -1, -1))

def test_exist_ok_s_isgid_directory(self):

path = os.path.join(support.TESTFN, 'dir1')

S_ISGID = stat.S_ISGID

mode = 0o777

old_mask = os.umask(0o022)

try:

existing_testfn_mode = stat.S_IMODE(

os.lstat(support.TESTFN).st_mode)

try: