Python code coverage for Lib/test/test_os.py

#	count	content
1	n/a	# As a test suite for the os module, this is woefully inadequate, but this
2	n/a	# does add tests for a few functions which have been determined to be more
3	n/a	# portable than they had been thought to be.
4	n/a
5	n/a	import asynchat
6	n/a	import asyncore
7	n/a	import codecs
8	n/a	import contextlib
9	n/a	import decimal
10	n/a	import errno
11	n/a	import fractions
12	n/a	import getpass
13	n/a	import itertools
14	n/a	import locale
15	n/a	import mmap
16	n/a	import os
17	n/a	import pickle
18	n/a	import shutil
19	n/a	import signal
20	n/a	import socket
21	n/a	import stat
22	n/a	import subprocess
23	n/a	import sys
24	n/a	import sysconfig
25	n/a	import time
26	n/a	import unittest
27	n/a	import uuid
28	n/a	import warnings
29	n/a	from test import support
30	n/a	try:
31	n/a	import threading
32	n/a	except ImportError:
33	n/a	threading = None
34	n/a	try:
35	n/a	import resource
36	n/a	except ImportError:
37	n/a	resource = None
38	n/a	try:
39	n/a	import fcntl
40	n/a	except ImportError:
41	n/a	fcntl = None
42	n/a	try:
43	n/a	import _winapi
44	n/a	except ImportError:
45	n/a	_winapi = None
46	n/a	try:
47	n/a	import grp
48	n/a	groups = [g.gr_gid for g in grp.getgrall() if getpass.getuser() in g.gr_mem]
49	n/a	if hasattr(os, 'getgid'):
50	n/a	process_gid = os.getgid()
51	n/a	if process_gid not in groups:
52	n/a	groups.append(process_gid)
53	n/a	except ImportError:
54	n/a	groups = []
55	n/a	try:
56	n/a	import pwd
57	n/a	all_users = [u.pw_uid for u in pwd.getpwall()]
58	n/a	except (ImportError, AttributeError):
59	n/a	all_users = []
60	n/a	try:
61	n/a	from _testcapi import INT_MAX, PY_SSIZE_T_MAX
62	n/a	except ImportError:
63	n/a	INT_MAX = PY_SSIZE_T_MAX = sys.maxsize
64	n/a
65	n/a	from test.support.script_helper import assert_python_ok
66	n/a	from test.support import unix_shell
67	n/a
68	n/a
69	n/a	root_in_posix = False
70	n/a	if hasattr(os, 'geteuid'):
71	n/a	root_in_posix = (os.geteuid() == 0)
72	n/a
73	n/a	# Detect whether we're on a Linux system that uses the (now outdated
74	n/a	# and unmaintained) linuxthreads threading library. There's an issue
75	n/a	# when combining linuxthreads with a failed execv call: see
76	n/a	# http://bugs.python.org/issue4970.
77	n/a	if hasattr(sys, 'thread_info') and sys.thread_info.version:
78	n/a	USING_LINUXTHREADS = sys.thread_info.version.startswith("linuxthreads")
79	n/a	else:
80	n/a	USING_LINUXTHREADS = False
81	n/a
82	n/a	# Issue #14110: Some tests fail on FreeBSD if the user is in the wheel group.
83	n/a	HAVE_WHEEL_GROUP = sys.platform.startswith('freebsd') and os.getgid() == 0
84	n/a
85	n/a
86	n/a	@contextlib.contextmanager
87	n/a	def ignore_deprecation_warnings(msg_regex, quiet=False):
88	n/a	with support.check_warnings((msg_regex, DeprecationWarning), quiet=quiet):
89	n/a	yield
90	n/a
91	n/a
92	n/a	def requires_os_func(name):
93	n/a	return unittest.skipUnless(hasattr(os, name), 'requires os.%s' % name)
94	n/a
95	n/a
96	n/a	class _PathLike(os.PathLike):
97	n/a
98	n/a	def __init__(self, path=""):
99	n/a	self.path = path
100	n/a
101	n/a	def __str__(self):
102	n/a	return str(self.path)
103	n/a
104	n/a	def __fspath__(self):
105	n/a	if isinstance(self.path, BaseException):
106	n/a	raise self.path
107	n/a	else:
108	n/a	return self.path
109	n/a
110	n/a
111	n/a	def create_file(filename, content=b'content'):
112	n/a	with open(filename, "xb", 0) as fp:
113	n/a	fp.write(content)
114	n/a
115	n/a
116	n/a	# Tests creating TESTFN
117	n/a	class FileTests(unittest.TestCase):
118	n/a	def setUp(self):
119	n/a	if os.path.lexists(support.TESTFN):
120	n/a	os.unlink(support.TESTFN)
121	n/a	tearDown = setUp
122	n/a
123	n/a	def test_access(self):
124	n/a	f = os.open(support.TESTFN, os.O_CREAT\|os.O_RDWR)
125	n/a	os.close(f)
126	n/a	self.assertTrue(os.access(support.TESTFN, os.W_OK))
127	n/a
128	n/a	def test_closerange(self):
129	n/a	first = os.open(support.TESTFN, os.O_CREAT\|os.O_RDWR)
130	n/a	# We must allocate two consecutive file descriptors, otherwise
131	n/a	# it will mess up other file descriptors (perhaps even the three
132	n/a	# standard ones).
133	n/a	second = os.dup(first)
134	n/a	try:
135	n/a	retries = 0
136	n/a	while second != first + 1:
137	n/a	os.close(first)
138	n/a	retries += 1
139	n/a	if retries > 10:
140	n/a	# XXX test skipped
141	n/a	self.skipTest("couldn't allocate two consecutive fds")
142	n/a	first, second = second, os.dup(second)
143	n/a	finally:
144	n/a	os.close(second)
145	n/a	# close a fd that is open, and one that isn't
146	n/a	os.closerange(first, first + 2)
147	n/a	self.assertRaises(OSError, os.write, first, b"a")
148	n/a
149	n/a	@support.cpython_only
150	n/a	def test_rename(self):
151	n/a	path = support.TESTFN
152	n/a	old = sys.getrefcount(path)
153	n/a	self.assertRaises(TypeError, os.rename, path, 0)
154	n/a	new = sys.getrefcount(path)
155	n/a	self.assertEqual(old, new)
156	n/a
157	n/a	def test_read(self):
158	n/a	with open(support.TESTFN, "w+b") as fobj:
159	n/a	fobj.write(b"spam")
160	n/a	fobj.flush()
161	n/a	fd = fobj.fileno()
162	n/a	os.lseek(fd, 0, 0)
163	n/a	s = os.read(fd, 4)
164	n/a	self.assertEqual(type(s), bytes)
165	n/a	self.assertEqual(s, b"spam")
166	n/a
167	n/a	@support.cpython_only
168	n/a	# Skip the test on 32-bit platforms: the number of bytes must fit in a
169	n/a	# Py_ssize_t type
170	n/a	@unittest.skipUnless(INT_MAX < PY_SSIZE_T_MAX,
171	n/a	"needs INT_MAX < PY_SSIZE_T_MAX")
172	n/a	@support.bigmemtest(size=INT_MAX + 10, memuse=1, dry_run=False)
173	n/a	def test_large_read(self, size):
174	n/a	self.addCleanup(support.unlink, support.TESTFN)
175	n/a	create_file(support.TESTFN, b'test')
176	n/a
177	n/a	# Issue #21932: Make sure that os.read() does not raise an
178	n/a	# OverflowError for size larger than INT_MAX
179	n/a	with open(support.TESTFN, "rb") as fp:
180	n/a	data = os.read(fp.fileno(), size)
181	n/a
182	n/a	# The test does not try to read more than 2 GB at once because the
183	n/a	# operating system is free to return less bytes than requested.
184	n/a	self.assertEqual(data, b'test')
185	n/a
186	n/a	def test_write(self):
187	n/a	# os.write() accepts bytes- and buffer-like objects but not strings
188	n/a	fd = os.open(support.TESTFN, os.O_CREAT \| os.O_WRONLY)
189	n/a	self.assertRaises(TypeError, os.write, fd, "beans")
190	n/a	os.write(fd, b"bacon\n")
191	n/a	os.write(fd, bytearray(b"eggs\n"))
192	n/a	os.write(fd, memoryview(b"spam\n"))
193	n/a	os.close(fd)
194	n/a	with open(support.TESTFN, "rb") as fobj:
195	n/a	self.assertEqual(fobj.read().splitlines(),
196	n/a	[b"bacon", b"eggs", b"spam"])
197	n/a
198	n/a	def write_windows_console(self, *args):
199	n/a	retcode = subprocess.call(args,
200	n/a	# use a new console to not flood the test output
201	n/a	creationflags=subprocess.CREATE_NEW_CONSOLE,
202	n/a	# use a shell to hide the console window (SW_HIDE)
203	n/a	shell=True)
204	n/a	self.assertEqual(retcode, 0)
205	n/a
206	n/a	@unittest.skipUnless(sys.platform == 'win32',
207	n/a	'test specific to the Windows console')
208	n/a	def test_write_windows_console(self):
209	n/a	# Issue #11395: the Windows console returns an error (12: not enough
210	n/a	# space error) on writing into stdout if stdout mode is binary and the
211	n/a	# length is greater than 66,000 bytes (or less, depending on heap
212	n/a	# usage).
213	n/a	code = "print('x' * 100000)"
214	n/a	self.write_windows_console(sys.executable, "-c", code)
215	n/a	self.write_windows_console(sys.executable, "-u", "-c", code)
216	n/a
217	n/a	def fdopen_helper(self, *args):
218	n/a	fd = os.open(support.TESTFN, os.O_RDONLY)
219	n/a	f = os.fdopen(fd, *args)
220	n/a	f.close()
221	n/a
222	n/a	def test_fdopen(self):
223	n/a	fd = os.open(support.TESTFN, os.O_CREAT\|os.O_RDWR)
224	n/a	os.close(fd)
225	n/a
226	n/a	self.fdopen_helper()
227	n/a	self.fdopen_helper('r')
228	n/a	self.fdopen_helper('r', 100)
229	n/a
230	n/a	def test_replace(self):
231	n/a	TESTFN2 = support.TESTFN + ".2"
232	n/a	self.addCleanup(support.unlink, support.TESTFN)
233	n/a	self.addCleanup(support.unlink, TESTFN2)
234	n/a
235	n/a	create_file(support.TESTFN, b"1")
236	n/a	create_file(TESTFN2, b"2")
237	n/a
238	n/a	os.replace(support.TESTFN, TESTFN2)
239	n/a	self.assertRaises(FileNotFoundError, os.stat, support.TESTFN)
240	n/a	with open(TESTFN2, 'r') as f:
241	n/a	self.assertEqual(f.read(), "1")
242	n/a
243	n/a	def test_open_keywords(self):
244	n/a	f = os.open(path=__file__, flags=os.O_RDONLY, mode=0o777,
245	n/a	dir_fd=None)
246	n/a	os.close(f)
247	n/a
248	n/a	def test_symlink_keywords(self):
249	n/a	symlink = support.get_attribute(os, "symlink")
250	n/a	try:
251	n/a	symlink(src='target', dst=support.TESTFN,
252	n/a	target_is_directory=False, dir_fd=None)
253	n/a	except (NotImplementedError, OSError):
254	n/a	pass # No OS support or unprivileged user
255	n/a
256	n/a
257	n/a	# Test attributes on return values from os.stat family.
258	n/a	class StatAttributeTests(unittest.TestCase):
259	n/a	def setUp(self):
260	n/a	self.fname = support.TESTFN
261	n/a	self.addCleanup(support.unlink, self.fname)
262	n/a	create_file(self.fname, b"ABC")
263	n/a
264	n/a	@unittest.skipUnless(hasattr(os, 'stat'), 'test needs os.stat()')
265	n/a	def check_stat_attributes(self, fname):
266	n/a	result = os.stat(fname)
267	n/a
268	n/a	# Make sure direct access works
269	n/a	self.assertEqual(result[stat.ST_SIZE], 3)
270	n/a	self.assertEqual(result.st_size, 3)
271	n/a
272	n/a	# Make sure all the attributes are there
273	n/a	members = dir(result)
274	n/a	for name in dir(stat):
275	n/a	if name[:3] == 'ST_':
276	n/a	attr = name.lower()
277	n/a	if name.endswith("TIME"):
278	n/a	def trunc(x): return int(x)
279	n/a	else:
280	n/a	def trunc(x): return x
281	n/a	self.assertEqual(trunc(getattr(result, attr)),
282	n/a	result[getattr(stat, name)])
283	n/a	self.assertIn(attr, members)
284	n/a
285	n/a	# Make sure that the st_?time and st_?time_ns fields roughly agree
286	n/a	# (they should always agree up to around tens-of-microseconds)
287	n/a	for name in 'st_atime st_mtime st_ctime'.split():
288	n/a	floaty = int(getattr(result, name) * 100000)
289	n/a	nanosecondy = getattr(result, name + "_ns") // 10000
290	n/a	self.assertAlmostEqual(floaty, nanosecondy, delta=2)
291	n/a
292	n/a	try:
293	n/a	result[200]
294	n/a	self.fail("No exception raised")
295	n/a	except IndexError:
296	n/a	pass
297	n/a
298	n/a	# Make sure that assignment fails
299	n/a	try:
300	n/a	result.st_mode = 1
301	n/a	self.fail("No exception raised")
302	n/a	except AttributeError:
303	n/a	pass
304	n/a
305	n/a	try:
306	n/a	result.st_rdev = 1
307	n/a	self.fail("No exception raised")
308	n/a	except (AttributeError, TypeError):
309	n/a	pass
310	n/a
311	n/a	try:
312	n/a	result.parrot = 1
313	n/a	self.fail("No exception raised")
314	n/a	except AttributeError:
315	n/a	pass
316	n/a
317	n/a	# Use the stat_result constructor with a too-short tuple.
318	n/a	try:
319	n/a	result2 = os.stat_result((10,))
320	n/a	self.fail("No exception raised")
321	n/a	except TypeError:
322	n/a	pass
323	n/a
324	n/a	# Use the constructor with a too-long tuple.
325	n/a	try:
326	n/a	result2 = os.stat_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
327	n/a	except TypeError:
328	n/a	pass
329	n/a
330	n/a	def test_stat_attributes(self):
331	n/a	self.check_stat_attributes(self.fname)
332	n/a
333	n/a	def test_stat_attributes_bytes(self):
334	n/a	try:
335	n/a	fname = self.fname.encode(sys.getfilesystemencoding())
336	n/a	except UnicodeEncodeError:
337	n/a	self.skipTest("cannot encode %a for the filesystem" % self.fname)
338	n/a	self.check_stat_attributes(fname)
339	n/a
340	n/a	def test_stat_result_pickle(self):
341	n/a	result = os.stat(self.fname)
342	n/a	for proto in range(pickle.HIGHEST_PROTOCOL + 1):
343	n/a	p = pickle.dumps(result, proto)
344	n/a	self.assertIn(b'stat_result', p)
345	n/a	if proto < 4:
346	n/a	self.assertIn(b'cos\nstat_result\n', p)
347	n/a	unpickled = pickle.loads(p)
348	n/a	self.assertEqual(result, unpickled)
349	n/a
350	n/a	@unittest.skipUnless(hasattr(os, 'statvfs'), 'test needs os.statvfs()')
351	n/a	def test_statvfs_attributes(self):
352	n/a	try:
353	n/a	result = os.statvfs(self.fname)
354	n/a	except OSError as e:
355	n/a	# On AtheOS, glibc always returns ENOSYS
356	n/a	if e.errno == errno.ENOSYS:
357	n/a	self.skipTest('os.statvfs() failed with ENOSYS')
358	n/a
359	n/a	# Make sure direct access works
360	n/a	self.assertEqual(result.f_bfree, result[3])
361	n/a
362	n/a	# Make sure all the attributes are there.
363	n/a	members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
364	n/a	'ffree', 'favail', 'flag', 'namemax')
365	n/a	for value, member in enumerate(members):
366	n/a	self.assertEqual(getattr(result, 'f_' + member), result[value])
367	n/a
368	n/a	# Make sure that assignment really fails
369	n/a	try:
370	n/a	result.f_bfree = 1
371	n/a	self.fail("No exception raised")
372	n/a	except AttributeError:
373	n/a	pass
374	n/a
375	n/a	try:
376	n/a	result.parrot = 1
377	n/a	self.fail("No exception raised")
378	n/a	except AttributeError:
379	n/a	pass
380	n/a
381	n/a	# Use the constructor with a too-short tuple.
382	n/a	try:
383	n/a	result2 = os.statvfs_result((10,))
384	n/a	self.fail("No exception raised")
385	n/a	except TypeError:
386	n/a	pass
387	n/a
388	n/a	# Use the constructor with a too-long tuple.
389	n/a	try:
390	n/a	result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
391	n/a	except TypeError:
392	n/a	pass
393	n/a
394	n/a	@unittest.skipUnless(hasattr(os, 'statvfs'),
395	n/a	"need os.statvfs()")
396	n/a	def test_statvfs_result_pickle(self):
397	n/a	try:
398	n/a	result = os.statvfs(self.fname)
399	n/a	except OSError as e:
400	n/a	# On AtheOS, glibc always returns ENOSYS
401	n/a	if e.errno == errno.ENOSYS:
402	n/a	self.skipTest('os.statvfs() failed with ENOSYS')
403	n/a
404	n/a	for proto in range(pickle.HIGHEST_PROTOCOL + 1):
405	n/a	p = pickle.dumps(result, proto)
406	n/a	self.assertIn(b'statvfs_result', p)
407	n/a	if proto < 4:
408	n/a	self.assertIn(b'cos\nstatvfs_result\n', p)
409	n/a	unpickled = pickle.loads(p)
410	n/a	self.assertEqual(result, unpickled)
411	n/a
412	n/a	@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
413	n/a	def test_1686475(self):
414	n/a	# Verify that an open file can be stat'ed
415	n/a	try:
416	n/a	os.stat(r"c:\pagefile.sys")
417	n/a	except FileNotFoundError:
418	n/a	self.skipTest(r'c:\pagefile.sys does not exist')
419	n/a	except OSError as e:
420	n/a	self.fail("Could not stat pagefile.sys")
421	n/a
422	n/a	@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
423	n/a	@unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
424	n/a	def test_15261(self):
425	n/a	# Verify that stat'ing a closed fd does not cause crash
426	n/a	r, w = os.pipe()
427	n/a	try:
428	n/a	os.stat(r) # should not raise error
429	n/a	finally:
430	n/a	os.close(r)
431	n/a	os.close(w)
432	n/a	with self.assertRaises(OSError) as ctx:
433	n/a	os.stat(r)
434	n/a	self.assertEqual(ctx.exception.errno, errno.EBADF)
435	n/a
436	n/a	def check_file_attributes(self, result):
437	n/a	self.assertTrue(hasattr(result, 'st_file_attributes'))
438	n/a	self.assertTrue(isinstance(result.st_file_attributes, int))
439	n/a	self.assertTrue(0 <= result.st_file_attributes <= 0xFFFFFFFF)
440	n/a
441	n/a	@unittest.skipUnless(sys.platform == "win32",
442	n/a	"st_file_attributes is Win32 specific")
443	n/a	def test_file_attributes(self):
444	n/a	# test file st_file_attributes (FILE_ATTRIBUTE_DIRECTORY not set)
445	n/a	result = os.stat(self.fname)
446	n/a	self.check_file_attributes(result)
447	n/a	self.assertEqual(
448	n/a	result.st_file_attributes & stat.FILE_ATTRIBUTE_DIRECTORY,
449	n/a	0)
450	n/a
451	n/a	# test directory st_file_attributes (FILE_ATTRIBUTE_DIRECTORY set)
452	n/a	dirname = support.TESTFN + "dir"
453	n/a	os.mkdir(dirname)
454	n/a	self.addCleanup(os.rmdir, dirname)
455	n/a
456	n/a	result = os.stat(dirname)
457	n/a	self.check_file_attributes(result)
458	n/a	self.assertEqual(
459	n/a	result.st_file_attributes & stat.FILE_ATTRIBUTE_DIRECTORY,
460	n/a	stat.FILE_ATTRIBUTE_DIRECTORY)
461	n/a
462	n/a	@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
463	n/a	def test_access_denied(self):
464	n/a	# Default to FindFirstFile WIN32_FIND_DATA when access is
465	n/a	# denied. See issue 28075.
466	n/a	# os.environ['TEMP'] should be located on a volume that
467	n/a	# supports file ACLs.
468	n/a	fname = os.path.join(os.environ['TEMP'], self.fname)
469	n/a	self.addCleanup(support.unlink, fname)
470	n/a	create_file(fname, b'ABC')
471	n/a	# Deny the right to [S]YNCHRONIZE on the file to
472	n/a	# force CreateFile to fail with ERROR_ACCESS_DENIED.
473	n/a	DETACHED_PROCESS = 8
474	n/a	subprocess.check_call(
475	n/a	['icacls.exe', fname, '/deny', 'Users:(S)'],
476	n/a	creationflags=DETACHED_PROCESS
477	n/a	)
478	n/a	result = os.stat(fname)
479	n/a	self.assertNotEqual(result.st_size, 0)
480	n/a
481	n/a
482	n/a	class UtimeTests(unittest.TestCase):
483	n/a	def setUp(self):
484	n/a	self.dirname = support.TESTFN
485	n/a	self.fname = os.path.join(self.dirname, "f1")
486	n/a
487	n/a	self.addCleanup(support.rmtree, self.dirname)
488	n/a	os.mkdir(self.dirname)
489	n/a	create_file(self.fname)
490	n/a
491	n/a	def restore_float_times(state):
492	n/a	with ignore_deprecation_warnings('stat_float_times'):
493	n/a	os.stat_float_times(state)
494	n/a
495	n/a	# ensure that st_atime and st_mtime are float
496	n/a	with ignore_deprecation_warnings('stat_float_times'):
497	n/a	old_float_times = os.stat_float_times(-1)
498	n/a	self.addCleanup(restore_float_times, old_float_times)
499	n/a
500	n/a	os.stat_float_times(True)
501	n/a
502	n/a	def support_subsecond(self, filename):
503	n/a	# Heuristic to check if the filesystem supports timestamp with
504	n/a	# subsecond resolution: check if float and int timestamps are different
505	n/a	st = os.stat(filename)
506	n/a	return ((st.st_atime != st[7])
507	n/a	or (st.st_mtime != st[8])
508	n/a	or (st.st_ctime != st[9]))
509	n/a
510	n/a	def _test_utime(self, set_time, filename=None):
511	n/a	if not filename:
512	n/a	filename = self.fname
513	n/a
514	n/a	support_subsecond = self.support_subsecond(filename)
515	n/a	if support_subsecond:
516	n/a	# Timestamp with a resolution of 1 microsecond (10^-6).
517	n/a	#
518	n/a	# The resolution of the C internal function used by os.utime()
519	n/a	# depends on the platform: 1 sec, 1 us, 1 ns. Writing a portable
520	n/a	# test with a resolution of 1 ns requires more work:
521	n/a	# see the issue #15745.
522	n/a	atime_ns = 1002003000 # 1.002003 seconds
523	n/a	mtime_ns = 4005006000 # 4.005006 seconds
524	n/a	else:
525	n/a	# use a resolution of 1 second
526	n/a	atime_ns = 5 * 10**9
527	n/a	mtime_ns = 8 * 10**9
528	n/a
529	n/a	set_time(filename, (atime_ns, mtime_ns))
530	n/a	st = os.stat(filename)
531	n/a
532	n/a	if support_subsecond:
533	n/a	self.assertAlmostEqual(st.st_atime, atime_ns * 1e-9, delta=1e-6)
534	n/a	self.assertAlmostEqual(st.st_mtime, mtime_ns * 1e-9, delta=1e-6)
535	n/a	else:
536	n/a	self.assertEqual(st.st_atime, atime_ns * 1e-9)
537	n/a	self.assertEqual(st.st_mtime, mtime_ns * 1e-9)
538	n/a	self.assertEqual(st.st_atime_ns, atime_ns)
539	n/a	self.assertEqual(st.st_mtime_ns, mtime_ns)
540	n/a
541	n/a	def test_utime(self):
542	n/a	def set_time(filename, ns):
543	n/a	# test the ns keyword parameter
544	n/a	os.utime(filename, ns=ns)
545	n/a	self._test_utime(set_time)
546	n/a
547	n/a	@staticmethod
548	n/a	def ns_to_sec(ns):
549	n/a	# Convert a number of nanosecond (int) to a number of seconds (float).
550	n/a	# Round towards infinity by adding 0.5 nanosecond to avoid rounding
551	n/a	# issue, os.utime() rounds towards minus infinity.
552	n/a	return (ns * 1e-9) + 0.5e-9
553	n/a
554	n/a	def test_utime_by_indexed(self):
555	n/a	# pass times as floating point seconds as the second indexed parameter
556	n/a	def set_time(filename, ns):
557	n/a	atime_ns, mtime_ns = ns
558	n/a	atime = self.ns_to_sec(atime_ns)
559	n/a	mtime = self.ns_to_sec(mtime_ns)
560	n/a	# test utimensat(timespec), utimes(timeval), utime(utimbuf)
561	n/a	# or utime(time_t)
562	n/a	os.utime(filename, (atime, mtime))
563	n/a	self._test_utime(set_time)
564	n/a
565	n/a	def test_utime_by_times(self):
566	n/a	def set_time(filename, ns):
567	n/a	atime_ns, mtime_ns = ns
568	n/a	atime = self.ns_to_sec(atime_ns)
569	n/a	mtime = self.ns_to_sec(mtime_ns)
570	n/a	# test the times keyword parameter
571	n/a	os.utime(filename, times=(atime, mtime))
572	n/a	self._test_utime(set_time)
573	n/a
574	n/a	@unittest.skipUnless(os.utime in os.supports_follow_symlinks,
575	n/a	"follow_symlinks support for utime required "
576	n/a	"for this test.")
577	n/a	def test_utime_nofollow_symlinks(self):
578	n/a	def set_time(filename, ns):
579	n/a	# use follow_symlinks=False to test utimensat(timespec)
580	n/a	# or lutimes(timeval)
581	n/a	os.utime(filename, ns=ns, follow_symlinks=False)
582	n/a	self._test_utime(set_time)
583	n/a
584	n/a	@unittest.skipUnless(os.utime in os.supports_fd,
585	n/a	"fd support for utime required for this test.")
586	n/a	def test_utime_fd(self):
587	n/a	def set_time(filename, ns):
588	n/a	with open(filename, 'wb', 0) as fp:
589	n/a	# use a file descriptor to test futimens(timespec)
590	n/a	# or futimes(timeval)
591	n/a	os.utime(fp.fileno(), ns=ns)
592	n/a	self._test_utime(set_time)
593	n/a
594	n/a	@unittest.skipUnless(os.utime in os.supports_dir_fd,
595	n/a	"dir_fd support for utime required for this test.")
596	n/a	def test_utime_dir_fd(self):
597	n/a	def set_time(filename, ns):
598	n/a	dirname, name = os.path.split(filename)
599	n/a	dirfd = os.open(dirname, os.O_RDONLY)
600	n/a	try:
601	n/a	# pass dir_fd to test utimensat(timespec) or futimesat(timeval)
602	n/a	os.utime(name, dir_fd=dirfd, ns=ns)
603	n/a	finally:
604	n/a	os.close(dirfd)
605	n/a	self._test_utime(set_time)
606	n/a
607	n/a	def test_utime_directory(self):
608	n/a	def set_time(filename, ns):
609	n/a	# test calling os.utime() on a directory
610	n/a	os.utime(filename, ns=ns)
611	n/a	self._test_utime(set_time, filename=self.dirname)
612	n/a
613	n/a	def _test_utime_current(self, set_time):
614	n/a	# Get the system clock
615	n/a	current = time.time()
616	n/a
617	n/a	# Call os.utime() to set the timestamp to the current system clock
618	n/a	set_time(self.fname)
619	n/a
620	n/a	if not self.support_subsecond(self.fname):
621	n/a	delta = 1.0
622	n/a	else:
623	n/a	# On Windows, the usual resolution of time.time() is 15.6 ms
624	n/a	delta = 0.020
625	n/a	st = os.stat(self.fname)
626	n/a	msg = ("st_time=%r, current=%r, dt=%r"
627	n/a	% (st.st_mtime, current, st.st_mtime - current))
628	n/a	self.assertAlmostEqual(st.st_mtime, current,
629	n/a	delta=delta, msg=msg)
630	n/a
631	n/a	def test_utime_current(self):
632	n/a	def set_time(filename):
633	n/a	# Set to the current time in the new way
634	n/a	os.utime(self.fname)
635	n/a	self._test_utime_current(set_time)
636	n/a
637	n/a	def test_utime_current_old(self):
638	n/a	def set_time(filename):
639	n/a	# Set to the current time in the old explicit way.
640	n/a	os.utime(self.fname, None)
641	n/a	self._test_utime_current(set_time)
642	n/a
643	n/a	def get_file_system(self, path):
644	n/a	if sys.platform == 'win32':
645	n/a	root = os.path.splitdrive(os.path.abspath(path))[0] + '\\'
646	n/a	import ctypes
647	n/a	kernel32 = ctypes.windll.kernel32
648	n/a	buf = ctypes.create_unicode_buffer("", 100)
649	n/a	ok = kernel32.GetVolumeInformationW(root, None, 0,
650	n/a	None, None, None,
651	n/a	buf, len(buf))
652	n/a	if ok:
653	n/a	return buf.value
654	n/a	# return None if the filesystem is unknown
655	n/a
656	n/a	def test_large_time(self):
657	n/a	# Many filesystems are limited to the year 2038. At least, the test
658	n/a	# pass with NTFS filesystem.
659	n/a	if self.get_file_system(self.dirname) != "NTFS":
660	n/a	self.skipTest("requires NTFS")
661	n/a
662	n/a	large = 5000000000 # some day in 2128
663	n/a	os.utime(self.fname, (large, large))
664	n/a	self.assertEqual(os.stat(self.fname).st_mtime, large)
665	n/a
666	n/a	def test_utime_invalid_arguments(self):
667	n/a	# seconds and nanoseconds parameters are mutually exclusive
668	n/a	with self.assertRaises(ValueError):
669	n/a	os.utime(self.fname, (5, 5), ns=(5, 5))
670	n/a
671	n/a
672	n/a	from test import mapping_tests
673	n/a
674	n/a	class EnvironTests(mapping_tests.BasicTestMappingProtocol):
675	n/a	"""check that os.environ object conform to mapping protocol"""
676	n/a	type2test = None
677	n/a
678	n/a	def setUp(self):
679	n/a	self.__save = dict(os.environ)
680	n/a	if os.supports_bytes_environ:
681	n/a	self.__saveb = dict(os.environb)
682	n/a	for key, value in self._reference().items():
683	n/a	os.environ[key] = value
684	n/a
685	n/a	def tearDown(self):
686	n/a	os.environ.clear()
687	n/a	os.environ.update(self.__save)
688	n/a	if os.supports_bytes_environ:
689	n/a	os.environb.clear()
690	n/a	os.environb.update(self.__saveb)
691	n/a
692	n/a	def _reference(self):
693	n/a	return {"KEY1":"VALUE1", "KEY2":"VALUE2", "KEY3":"VALUE3"}
694	n/a
695	n/a	def _empty_mapping(self):
696	n/a	os.environ.clear()
697	n/a	return os.environ
698	n/a
699	n/a	# Bug 1110478
700	n/a	@unittest.skipUnless(unix_shell and os.path.exists(unix_shell),
701	n/a	'requires a shell')
702	n/a	def test_update2(self):
703	n/a	os.environ.clear()
704	n/a	os.environ.update(HELLO="World")
705	n/a	with os.popen("%s -c 'echo $HELLO'" % unix_shell) as popen:
706	n/a	value = popen.read().strip()
707	n/a	self.assertEqual(value, "World")
708	n/a
709	n/a	@unittest.skipUnless(unix_shell and os.path.exists(unix_shell),
710	n/a	'requires a shell')
711	n/a	def test_os_popen_iter(self):
712	n/a	with os.popen("%s -c 'echo \"line1\nline2\nline3\"'"
713	n/a	% unix_shell) as popen:
714	n/a	it = iter(popen)
715	n/a	self.assertEqual(next(it), "line1\n")
716	n/a	self.assertEqual(next(it), "line2\n")
717	n/a	self.assertEqual(next(it), "line3\n")
718	n/a	self.assertRaises(StopIteration, next, it)
719	n/a
720	n/a	# Verify environ keys and values from the OS are of the
721	n/a	# correct str type.
722	n/a	def test_keyvalue_types(self):
723	n/a	for key, val in os.environ.items():
724	n/a	self.assertEqual(type(key), str)
725	n/a	self.assertEqual(type(val), str)
726	n/a
727	n/a	def test_items(self):
728	n/a	for key, value in self._reference().items():
729	n/a	self.assertEqual(os.environ.get(key), value)
730	n/a
731	n/a	# Issue 7310
732	n/a	def test___repr__(self):
733	n/a	"""Check that the repr() of os.environ looks like environ({...})."""
734	n/a	env = os.environ
735	n/a	self.assertEqual(repr(env), 'environ({{{}}})'.format(', '.join(
736	n/a	'{!r}: {!r}'.format(key, value)
737	n/a	for key, value in env.items())))
738	n/a
739	n/a	def test_get_exec_path(self):
740	n/a	defpath_list = os.defpath.split(os.pathsep)
741	n/a	test_path = ['/monty', '/python', '', '/flying/circus']
742	n/a	test_env = {'PATH': os.pathsep.join(test_path)}
743	n/a
744	n/a	saved_environ = os.environ
745	n/a	try:
746	n/a	os.environ = dict(test_env)
747	n/a	# Test that defaulting to os.environ works.
748	n/a	self.assertSequenceEqual(test_path, os.get_exec_path())
749	n/a	self.assertSequenceEqual(test_path, os.get_exec_path(env=None))
750	n/a	finally:
751	n/a	os.environ = saved_environ
752	n/a
753	n/a	# No PATH environment variable
754	n/a	self.assertSequenceEqual(defpath_list, os.get_exec_path({}))
755	n/a	# Empty PATH environment variable
756	n/a	self.assertSequenceEqual(('',), os.get_exec_path({'PATH':''}))
757	n/a	# Supplied PATH environment variable
758	n/a	self.assertSequenceEqual(test_path, os.get_exec_path(test_env))
759	n/a
760	n/a	if os.supports_bytes_environ:
761	n/a	# env cannot contain 'PATH' and b'PATH' keys
762	n/a	try:
763	n/a	# ignore BytesWarning warning
764	n/a	with warnings.catch_warnings(record=True):
765	n/a	mixed_env = {'PATH': '1', b'PATH': b'2'}
766	n/a	except BytesWarning:
767	n/a	# mixed_env cannot be created with python -bb
768	n/a	pass
769	n/a	else:
770	n/a	self.assertRaises(ValueError, os.get_exec_path, mixed_env)
771	n/a
772	n/a	# bytes key and/or value
773	n/a	self.assertSequenceEqual(os.get_exec_path({b'PATH': b'abc'}),
774	n/a	['abc'])
775	n/a	self.assertSequenceEqual(os.get_exec_path({b'PATH': 'abc'}),
776	n/a	['abc'])
777	n/a	self.assertSequenceEqual(os.get_exec_path({'PATH': b'abc'}),
778	n/a	['abc'])
779	n/a
780	n/a	@unittest.skipUnless(os.supports_bytes_environ,
781	n/a	"os.environb required for this test.")
782	n/a	def test_environb(self):
783	n/a	# os.environ -> os.environb
784	n/a	value = 'euro\u20ac'
785	n/a	try:
786	n/a	value_bytes = value.encode(sys.getfilesystemencoding(),
787	n/a	'surrogateescape')
788	n/a	except UnicodeEncodeError:
789	n/a	msg = "U+20AC character is not encodable to %s" % (
790	n/a	sys.getfilesystemencoding(),)
791	n/a	self.skipTest(msg)
792	n/a	os.environ['unicode'] = value
793	n/a	self.assertEqual(os.environ['unicode'], value)
794	n/a	self.assertEqual(os.environb[b'unicode'], value_bytes)
795	n/a
796	n/a	# os.environb -> os.environ
797	n/a	value = b'\xff'
798	n/a	os.environb[b'bytes'] = value
799	n/a	self.assertEqual(os.environb[b'bytes'], value)
800	n/a	value_str = value.decode(sys.getfilesystemencoding(), 'surrogateescape')
801	n/a	self.assertEqual(os.environ['bytes'], value_str)
802	n/a
803	n/a	# On FreeBSD < 7 and OS X < 10.6, unsetenv() doesn't return a value (issue
804	n/a	# #13415).
805	n/a	@support.requires_freebsd_version(7)
806	n/a	@support.requires_mac_ver(10, 6)
807	n/a	def test_unset_error(self):
808	n/a	if sys.platform == "win32":
809	n/a	# an environment variable is limited to 32,767 characters
810	n/a	key = 'x' * 50000
811	n/a	self.assertRaises(ValueError, os.environ.__delitem__, key)
812	n/a	else:
813	n/a	# "=" is not allowed in a variable name
814	n/a	key = 'key='
815	n/a	self.assertRaises(OSError, os.environ.__delitem__, key)
816	n/a
817	n/a	def test_key_type(self):
818	n/a	missing = 'missingkey'
819	n/a	self.assertNotIn(missing, os.environ)
820	n/a
821	n/a	with self.assertRaises(KeyError) as cm:
822	n/a	os.environ[missing]
823	n/a	self.assertIs(cm.exception.args[0], missing)
824	n/a	self.assertTrue(cm.exception.__suppress_context__)
825	n/a
826	n/a	with self.assertRaises(KeyError) as cm:
827	n/a	del os.environ[missing]
828	n/a	self.assertIs(cm.exception.args[0], missing)
829	n/a	self.assertTrue(cm.exception.__suppress_context__)
830	n/a
831	n/a
832	n/a	class WalkTests(unittest.TestCase):
833	n/a	"""Tests for os.walk()."""
834	n/a
835	n/a	# Wrapper to hide minor differences between os.walk and os.fwalk
836	n/a	# to tests both functions with the same code base
837	n/a	def walk(self, top, **kwargs):
838	n/a	if 'follow_symlinks' in kwargs:
839	n/a	kwargs['followlinks'] = kwargs.pop('follow_symlinks')
840	n/a	return os.walk(top, **kwargs)
841	n/a
842	n/a	def setUp(self):
843	n/a	join = os.path.join
844	n/a	self.addCleanup(support.rmtree, support.TESTFN)
845	n/a
846	n/a	# Build:
847	n/a	# TESTFN/
848	n/a	# TEST1/ a file kid and two directory kids
849	n/a	# tmp1
850	n/a	# SUB1/ a file kid and a directory kid
851	n/a	# tmp2
852	n/a	# SUB11/ no kids
853	n/a	# SUB2/ a file kid and a dirsymlink kid
854	n/a	# tmp3
855	n/a	# SUB21/ not readable
856	n/a	# tmp5
857	n/a	# link/ a symlink to TESTFN.2
858	n/a	# broken_link
859	n/a	# broken_link2
860	n/a	# broken_link3
861	n/a	# TEST2/
862	n/a	# tmp4 a lone file
863	n/a	self.walk_path = join(support.TESTFN, "TEST1")
864	n/a	self.sub1_path = join(self.walk_path, "SUB1")
865	n/a	self.sub11_path = join(self.sub1_path, "SUB11")
866	n/a	sub2_path = join(self.walk_path, "SUB2")
867	n/a	sub21_path = join(sub2_path, "SUB21")
868	n/a	tmp1_path = join(self.walk_path, "tmp1")
869	n/a	tmp2_path = join(self.sub1_path, "tmp2")
870	n/a	tmp3_path = join(sub2_path, "tmp3")
871	n/a	tmp5_path = join(sub21_path, "tmp3")
872	n/a	self.link_path = join(sub2_path, "link")
873	n/a	t2_path = join(support.TESTFN, "TEST2")
874	n/a	tmp4_path = join(support.TESTFN, "TEST2", "tmp4")
875	n/a	broken_link_path = join(sub2_path, "broken_link")
876	n/a	broken_link2_path = join(sub2_path, "broken_link2")
877	n/a	broken_link3_path = join(sub2_path, "broken_link3")
878	n/a
879	n/a	# Create stuff.
880	n/a	os.makedirs(self.sub11_path)
881	n/a	os.makedirs(sub2_path)
882	n/a	os.makedirs(sub21_path)
883	n/a	os.makedirs(t2_path)
884	n/a
885	n/a	for path in tmp1_path, tmp2_path, tmp3_path, tmp4_path, tmp5_path:
886	n/a	with open(path, "x") as f:
887	n/a	f.write("I'm " + path + " and proud of it. Blame test_os.\n")
888	n/a
889	n/a	if support.can_symlink():
890	n/a	os.symlink(os.path.abspath(t2_path), self.link_path)
891	n/a	os.symlink('broken', broken_link_path, True)
892	n/a	os.symlink(join('tmp3', 'broken'), broken_link2_path, True)
893	n/a	os.symlink(join('SUB21', 'tmp5'), broken_link3_path, True)
894	n/a	self.sub2_tree = (sub2_path, ["SUB21", "link"],
895	n/a	["broken_link", "broken_link2", "broken_link3",
896	n/a	"tmp3"])
897	n/a	else:
898	n/a	self.sub2_tree = (sub2_path, [], ["tmp3"])
899	n/a
900	n/a	os.chmod(sub21_path, 0)
901	n/a	try:
902	n/a	os.listdir(sub21_path)
903	n/a	except PermissionError:
904	n/a	self.addCleanup(os.chmod, sub21_path, stat.S_IRWXU)
905	n/a	else:
906	n/a	os.chmod(sub21_path, stat.S_IRWXU)
907	n/a	os.unlink(tmp5_path)
908	n/a	os.rmdir(sub21_path)
909	n/a	del self.sub2_tree[1][:1]
910	n/a
911	n/a	def test_walk_topdown(self):
912	n/a	# Walk top-down.
913	n/a	all = list(self.walk(self.walk_path))
914	n/a
915	n/a	self.assertEqual(len(all), 4)
916	n/a	# We can't know which order SUB1 and SUB2 will appear in.
917	n/a	# Not flipped: TESTFN, SUB1, SUB11, SUB2
918	n/a	# flipped: TESTFN, SUB2, SUB1, SUB11
919	n/a	flipped = all[0][1][0] != "SUB1"
920	n/a	all[0][1].sort()
921	n/a	all[3 - 2 * flipped][-1].sort()
922	n/a	all[3 - 2 * flipped][1].sort()
923	n/a	self.assertEqual(all[0], (self.walk_path, ["SUB1", "SUB2"], ["tmp1"]))
924	n/a	self.assertEqual(all[1 + flipped], (self.sub1_path, ["SUB11"], ["tmp2"]))
925	n/a	self.assertEqual(all[2 + flipped], (self.sub11_path, [], []))
926	n/a	self.assertEqual(all[3 - 2 * flipped], self.sub2_tree)
927	n/a
928	n/a	def test_walk_prune(self, walk_path=None):
929	n/a	if walk_path is None:
930	n/a	walk_path = self.walk_path
931	n/a	# Prune the search.
932	n/a	all = []
933	n/a	for root, dirs, files in self.walk(walk_path):
934	n/a	all.append((root, dirs, files))
935	n/a	# Don't descend into SUB1.
936	n/a	if 'SUB1' in dirs:
937	n/a	# Note that this also mutates the dirs we appended to all!
938	n/a	dirs.remove('SUB1')
939	n/a
940	n/a	self.assertEqual(len(all), 2)
941	n/a	self.assertEqual(all[0],
942	n/a	(str(walk_path), ["SUB2"], ["tmp1"]))
943	n/a
944	n/a	all[1][-1].sort()
945	n/a	all[1][1].sort()
946	n/a	self.assertEqual(all[1], self.sub2_tree)
947	n/a
948	n/a	def test_file_like_path(self):
949	n/a	self.test_walk_prune(_PathLike(self.walk_path))
950	n/a
951	n/a	def test_walk_bottom_up(self):
952	n/a	# Walk bottom-up.
953	n/a	all = list(self.walk(self.walk_path, topdown=False))
954	n/a
955	n/a	self.assertEqual(len(all), 4, all)
956	n/a	# We can't know which order SUB1 and SUB2 will appear in.
957	n/a	# Not flipped: SUB11, SUB1, SUB2, TESTFN
958	n/a	# flipped: SUB2, SUB11, SUB1, TESTFN
959	n/a	flipped = all[3][1][0] != "SUB1"
960	n/a	all[3][1].sort()
961	n/a	all[2 - 2 * flipped][-1].sort()
962	n/a	all[2 - 2 * flipped][1].sort()
963	n/a	self.assertEqual(all[3],
964	n/a	(self.walk_path, ["SUB1", "SUB2"], ["tmp1"]))
965	n/a	self.assertEqual(all[flipped],
966	n/a	(self.sub11_path, [], []))
967	n/a	self.assertEqual(all[flipped + 1],
968	n/a	(self.sub1_path, ["SUB11"], ["tmp2"]))
969	n/a	self.assertEqual(all[2 - 2 * flipped],
970	n/a	self.sub2_tree)
971	n/a
972	n/a	def test_walk_symlink(self):
973	n/a	if not support.can_symlink():
974	n/a	self.skipTest("need symlink support")
975	n/a
976	n/a	# Walk, following symlinks.
977	n/a	walk_it = self.walk(self.walk_path, follow_symlinks=True)
978	n/a	for root, dirs, files in walk_it:
979	n/a	if root == self.link_path:
980	n/a	self.assertEqual(dirs, [])
981	n/a	self.assertEqual(files, ["tmp4"])
982	n/a	break
983	n/a	else:
984	n/a	self.fail("Didn't follow symlink with followlinks=True")
985	n/a
986	n/a	def test_walk_bad_dir(self):
987	n/a	# Walk top-down.
988	n/a	errors = []
989	n/a	walk_it = self.walk(self.walk_path, onerror=errors.append)
990	n/a	root, dirs, files = next(walk_it)
991	n/a	self.assertEqual(errors, [])
992	n/a	dir1 = 'SUB1'
993	n/a	path1 = os.path.join(root, dir1)
994	n/a	path1new = os.path.join(root, dir1 + '.new')
995	n/a	os.rename(path1, path1new)
996	n/a	try:
997	n/a	roots = [r for r, d, f in walk_it]
998	n/a	self.assertTrue(errors)
999	n/a	self.assertNotIn(path1, roots)
1000	n/a	self.assertNotIn(path1new, roots)
1001	n/a	for dir2 in dirs:
1002	n/a	if dir2 != dir1:
1003	n/a	self.assertIn(os.path.join(root, dir2), roots)
1004	n/a	finally:
1005	n/a	os.rename(path1new, path1)
1006	n/a
1007	n/a
1008	n/a	@unittest.skipUnless(hasattr(os, 'fwalk'), "Test needs os.fwalk()")
1009	n/a	class FwalkTests(WalkTests):
1010	n/a	"""Tests for os.fwalk()."""
1011	n/a
1012	n/a	def walk(self, top, **kwargs):
1013	n/a	for root, dirs, files, root_fd in os.fwalk(top, **kwargs):
1014	n/a	yield (root, dirs, files)
1015	n/a
1016	n/a	def _compare_to_walk(self, walk_kwargs, fwalk_kwargs):
1017	n/a	"""
1018	n/a	compare with walk() results.
1019	n/a	"""
1020	n/a	walk_kwargs = walk_kwargs.copy()
1021	n/a	fwalk_kwargs = fwalk_kwargs.copy()
1022	n/a	for topdown, follow_symlinks in itertools.product((True, False), repeat=2):
1023	n/a	walk_kwargs.update(topdown=topdown, followlinks=follow_symlinks)
1024	n/a	fwalk_kwargs.update(topdown=topdown, follow_symlinks=follow_symlinks)
1025	n/a
1026	n/a	expected = {}
1027	n/a	for root, dirs, files in os.walk(**walk_kwargs):
1028	n/a	expected[root] = (set(dirs), set(files))
1029	n/a
1030	n/a	for root, dirs, files, rootfd in os.fwalk(**fwalk_kwargs):
1031	n/a	self.assertIn(root, expected)
1032	n/a	self.assertEqual(expected[root], (set(dirs), set(files)))
1033	n/a
1034	n/a	def test_compare_to_walk(self):
1035	n/a	kwargs = {'top': support.TESTFN}
1036	n/a	self._compare_to_walk(kwargs, kwargs)
1037	n/a
1038	n/a	def test_dir_fd(self):
1039	n/a	try:
1040	n/a	fd = os.open(".", os.O_RDONLY)
1041	n/a	walk_kwargs = {'top': support.TESTFN}
1042	n/a	fwalk_kwargs = walk_kwargs.copy()
1043	n/a	fwalk_kwargs['dir_fd'] = fd
1044	n/a	self._compare_to_walk(walk_kwargs, fwalk_kwargs)
1045	n/a	finally:
1046	n/a	os.close(fd)
1047	n/a
1048	n/a	def test_yields_correct_dir_fd(self):
1049	n/a	# check returned file descriptors
1050	n/a	for topdown, follow_symlinks in itertools.product((True, False), repeat=2):
1051	n/a	args = support.TESTFN, topdown, None
1052	n/a	for root, dirs, files, rootfd in os.fwalk(*args, follow_symlinks=follow_symlinks):
1053	n/a	# check that the FD is valid
1054	n/a	os.fstat(rootfd)
1055	n/a	# redundant check
1056	n/a	os.stat(rootfd)
1057	n/a	# check that listdir() returns consistent information
1058	n/a	self.assertEqual(set(os.listdir(rootfd)), set(dirs) \| set(files))
1059	n/a
1060	n/a	def test_fd_leak(self):
1061	n/a	# Since we're opening a lot of FDs, we must be careful to avoid leaks:
1062	n/a	# we both check that calling fwalk() a large number of times doesn't
1063	n/a	# yield EMFILE, and that the minimum allocated FD hasn't changed.
1064	n/a	minfd = os.dup(1)
1065	n/a	os.close(minfd)
1066	n/a	for i in range(256):
1067	n/a	for x in os.fwalk(support.TESTFN):
1068	n/a	pass
1069	n/a	newfd = os.dup(1)
1070	n/a	self.addCleanup(os.close, newfd)
1071	n/a	self.assertEqual(newfd, minfd)
1072	n/a
1073	n/a	class BytesWalkTests(WalkTests):
1074	n/a	"""Tests for os.walk() with bytes."""
1075	n/a	def setUp(self):
1076	n/a	super().setUp()
1077	n/a	self.stack = contextlib.ExitStack()
1078	n/a
1079	n/a	def tearDown(self):
1080	n/a	self.stack.close()
1081	n/a	super().tearDown()
1082	n/a
1083	n/a	def walk(self, top, **kwargs):
1084	n/a	if 'follow_symlinks' in kwargs:
1085	n/a	kwargs['followlinks'] = kwargs.pop('follow_symlinks')
1086	n/a	for broot, bdirs, bfiles in os.walk(os.fsencode(top), **kwargs):
1087	n/a	root = os.fsdecode(broot)
1088	n/a	dirs = list(map(os.fsdecode, bdirs))
1089	n/a	files = list(map(os.fsdecode, bfiles))
1090	n/a	yield (root, dirs, files)
1091	n/a	bdirs[:] = list(map(os.fsencode, dirs))
1092	n/a	bfiles[:] = list(map(os.fsencode, files))
1093	n/a
1094	n/a
1095	n/a	class MakedirTests(unittest.TestCase):
1096	n/a	def setUp(self):
1097	n/a	os.mkdir(support.TESTFN)
1098	n/a
1099	n/a	def test_makedir(self):
1100	n/a	base = support.TESTFN
1101	n/a	path = os.path.join(base, 'dir1', 'dir2', 'dir3')
1102	n/a	os.makedirs(path) # Should work
1103	n/a	path = os.path.join(base, 'dir1', 'dir2', 'dir3', 'dir4')
1104	n/a	os.makedirs(path)
1105	n/a
1106	n/a	# Try paths with a '.' in them
1107	n/a	self.assertRaises(OSError, os.makedirs, os.curdir)
1108	n/a	path = os.path.join(base, 'dir1', 'dir2', 'dir3', 'dir4', 'dir5', os.curdir)
1109	n/a	os.makedirs(path)
1110	n/a	path = os.path.join(base, 'dir1', os.curdir, 'dir2', 'dir3', 'dir4',
1111	n/a	'dir5', 'dir6')
1112	n/a	os.makedirs(path)
1113	n/a
1114	n/a	def test_exist_ok_existing_directory(self):
1115	n/a	path = os.path.join(support.TESTFN, 'dir1')
1116	n/a	mode = 0o777
1117	n/a	old_mask = os.umask(0o022)
1118	n/a	os.makedirs(path, mode)
1119	n/a	self.assertRaises(OSError, os.makedirs, path, mode)
1120	n/a	self.assertRaises(OSError, os.makedirs, path, mode, exist_ok=False)
1121	n/a	os.makedirs(path, 0o776, exist_ok=True)
1122	n/a	os.makedirs(path, mode=mode, exist_ok=True)
1123	n/a	os.umask(old_mask)
1124	n/a
1125	n/a	# Issue #25583: A drive root could raise PermissionError on Windows
1126	n/a	os.makedirs(os.path.abspath('/'), exist_ok=True)
1127	n/a
1128	n/a	def test_exist_ok_s_isgid_directory(self):
1129	n/a	path = os.path.join(support.TESTFN, 'dir1')
1130	n/a	S_ISGID = stat.S_ISGID
1131	n/a	mode = 0o777
1132	n/a	old_mask = os.umask(0o022)
1133	n/a	try:
1134	n/a	existing_testfn_mode = stat.S_IMODE(
1135	n/a	os.lstat(support.TESTFN).st_mode)
1136	n/a	try:
1137	n/a	os.chmod(support.TESTFN, existing_testfn_mode \| S_ISGID)
1138	n/a	except PermissionError:
1139	n/a	raise unittest.SkipTest('Cannot set S_ISGID for dir.')
1140	n/a	if (os.lstat(support.TESTFN).st_mode & S_ISGID != S_ISGID):
1141	n/a	raise unittest.SkipTest('No support for S_ISGID dir mode.')
1142	n/a	# The os should apply S_ISGID from the parent dir for us, but
1143	n/a	# this test need not depend on that behavior. Be explicit.
1144	n/a	os.makedirs(path, mode \| S_ISGID)
1145	n/a	# http://bugs.python.org/issue14992
1146	n/a	# Should not fail when the bit is already set.
1147	n/a	os.makedirs(path, mode, exist_ok=True)
1148	n/a	# remove the bit.
1149	n/a	os.chmod(path, stat.S_IMODE(os.lstat(path).st_mode) & ~S_ISGID)
1150	n/a	# May work even when the bit is not already set when demanded.
1151	n/a	os.makedirs(path, mode \| S_ISGID, exist_ok=True)
1152	n/a	finally:
1153	n/a	os.umask(old_mask)
1154	n/a
1155	n/a	def test_exist_ok_existing_regular_file(self):
1156	n/a	base = support.TESTFN
1157	n/a	path = os.path.join(support.TESTFN, 'dir1')
1158	n/a	f = open(path, 'w')
1159	n/a	f.write('abc')
1160	n/a	f.close()
1161	n/a	self.assertRaises(OSError, os.makedirs, path)
1162	n/a	self.assertRaises(OSError, os.makedirs, path, exist_ok=False)
1163	n/a	self.assertRaises(OSError, os.makedirs, path, exist_ok=True)
1164	n/a	os.remove(path)
1165	n/a
1166	n/a	def tearDown(self):
1167	n/a	path = os.path.join(support.TESTFN, 'dir1', 'dir2', 'dir3',
1168	n/a	'dir4', 'dir5', 'dir6')
1169	n/a	# If the tests failed, the bottom-most directory ('../dir6')
1170	n/a	# may not have been created, so we look for the outermost directory
1171	n/a	# that exists.
1172	n/a	while not os.path.exists(path) and path != support.TESTFN:
1173	n/a	path = os.path.dirname(path)
1174	n/a
1175	n/a	os.removedirs(path)
1176	n/a
1177	n/a
1178	n/a	@unittest.skipUnless(hasattr(os, 'chown'), "Test needs chown")
1179	n/a	class ChownFileTests(unittest.TestCase):
1180	n/a
1181	n/a	@classmethod
1182	n/a	def setUpClass(cls):
1183	n/a	os.mkdir(support.TESTFN)
1184	n/a
1185	n/a	def test_chown_uid_gid_arguments_must_be_index(self):
1186	n/a	stat = os.stat(support.TESTFN)
1187	n/a	uid = stat.st_uid
1188	n/a	gid = stat.st_gid
1189	n/a	for value in (-1.0, -1j, decimal.Decimal(-1), fractions.Fraction(-2, 2)):
1190	n/a	self.assertRaises(TypeError, os.chown, support.TESTFN, value, gid)
1191	n/a	self.assertRaises(TypeError, os.chown, support.TESTFN, uid, value)
1192	n/a	self.assertIsNone(os.chown(support.TESTFN, uid, gid))
1193	n/a	self.assertIsNone(os.chown(support.TESTFN, -1, -1))
1194	n/a
1195	n/a	@unittest.skipUnless(len(groups) > 1, "test needs more than one group")
1196	n/a	def test_chown(self):
1197	n/a	gid_1, gid_2 = groups[:2]
1198	n/a	uid = os.stat(support.TESTFN).st_uid
1199	n/a	os.chown(support.TESTFN, uid, gid_1)
1200	n/a	gid = os.stat(support.TESTFN).st_gid
1201	n/a	self.assertEqual(gid, gid_1)
1202	n/a	os.chown(support.TESTFN, uid, gid_2)
1203	n/a	gid = os.stat(support.TESTFN).st_gid
1204	n/a	self.assertEqual(gid, gid_2)
1205	n/a
1206	n/a	@unittest.skipUnless(root_in_posix and len(all_users) > 1,
1207	n/a	"test needs root privilege and more than one user")
1208	n/a	def test_chown_with_root(self):
1209	n/a	uid_1, uid_2 = all_users[:2]
1210	n/a	gid = os.stat(support.TESTFN).st_gid
1211	n/a	os.chown(support.TESTFN, uid_1, gid)
1212	n/a	uid = os.stat(support.TESTFN).st_uid
1213	n/a	self.assertEqual(uid, uid_1)
1214	n/a	os.chown(support.TESTFN, uid_2, gid)
1215	n/a	uid = os.stat(support.TESTFN).st_uid
1216	n/a	self.assertEqual(uid, uid_2)
1217	n/a
1218	n/a	@unittest.skipUnless(not root_in_posix and len(all_users) > 1,
1219	n/a	"test needs non-root account and more than one user")
1220	n/a	def test_chown_without_permission(self):
1221	n/a	uid_1, uid_2 = all_users[:2]
1222	n/a	gid = os.stat(support.TESTFN).st_gid
1223	n/a	with self.assertRaises(PermissionError):
1224	n/a	os.chown(support.TESTFN, uid_1, gid)
1225	n/a	os.chown(support.TESTFN, uid_2, gid)
1226	n/a
1227	n/a	@classmethod
1228	n/a	def tearDownClass(cls):
1229	n/a	os.rmdir(support.TESTFN)
1230	n/a
1231	n/a
1232	n/a	class RemoveDirsTests(unittest.TestCase):
1233	n/a	def setUp(self):
1234	n/a	os.makedirs(support.TESTFN)
1235	n/a
1236	n/a	def tearDown(self):
1237	n/a	support.rmtree(support.TESTFN)
1238	n/a
1239	n/a	def test_remove_all(self):
1240	n/a	dira = os.path.join(support.TESTFN, 'dira')
1241	n/a	os.mkdir(dira)
1242	n/a	dirb = os.path.join(dira, 'dirb')
1243	n/a	os.mkdir(dirb)
1244	n/a	os.removedirs(dirb)
1245	n/a	self.assertFalse(os.path.exists(dirb))
1246	n/a	self.assertFalse(os.path.exists(dira))
1247	n/a	self.assertFalse(os.path.exists(support.TESTFN))
1248	n/a
1249	n/a	def test_remove_partial(self):
1250	n/a	dira = os.path.join(support.TESTFN, 'dira')
1251	n/a	os.mkdir(dira)
1252	n/a	dirb = os.path.join(dira, 'dirb')
1253	n/a	os.mkdir(dirb)
1254	n/a	create_file(os.path.join(dira, 'file.txt'))
1255	n/a	os.removedirs(dirb)
1256	n/a	self.assertFalse(os.path.exists(dirb))
1257	n/a	self.assertTrue(os.path.exists(dira))
1258	n/a	self.assertTrue(os.path.exists(support.TESTFN))
1259	n/a
1260	n/a	def test_remove_nothing(self):
1261	n/a	dira = os.path.join(support.TESTFN, 'dira')
1262	n/a	os.mkdir(dira)
1263	n/a	dirb = os.path.join(dira, 'dirb')
1264	n/a	os.mkdir(dirb)
1265	n/a	create_file(os.path.join(dirb, 'file.txt'))
1266	n/a	with self.assertRaises(OSError):
1267	n/a	os.removedirs(dirb)
1268	n/a	self.assertTrue(os.path.exists(dirb))
1269	n/a	self.assertTrue(os.path.exists(dira))
1270	n/a	self.assertTrue(os.path.exists(support.TESTFN))
1271	n/a
1272	n/a
1273	n/a	class DevNullTests(unittest.TestCase):
1274	n/a	def test_devnull(self):
1275	n/a	with open(os.devnull, 'wb', 0) as f:
1276	n/a	f.write(b'hello')
1277	n/a	f.close()
1278	n/a	with open(os.devnull, 'rb') as f:
1279	n/a	self.assertEqual(f.read(), b'')
1280	n/a
1281	n/a
1282	n/a	class URandomTests(unittest.TestCase):
1283	n/a	def test_urandom_length(self):
1284	n/a	self.assertEqual(len(os.urandom(0)), 0)
1285	n/a	self.assertEqual(len(os.urandom(1)), 1)
1286	n/a	self.assertEqual(len(os.urandom(10)), 10)
1287	n/a	self.assertEqual(len(os.urandom(100)), 100)
1288	n/a	self.assertEqual(len(os.urandom(1000)), 1000)
1289	n/a
1290	n/a	def test_urandom_value(self):
1291	n/a	data1 = os.urandom(16)
1292	n/a	self.assertIsInstance(data1, bytes)
1293	n/a	data2 = os.urandom(16)
1294	n/a	self.assertNotEqual(data1, data2)
1295	n/a
1296	n/a	def get_urandom_subprocess(self, count):
1297	n/a	code = '\n'.join((
1298	n/a	'import os, sys',
1299	n/a	'data = os.urandom(%s)' % count,
1300	n/a	'sys.stdout.buffer.write(data)',
1301	n/a	'sys.stdout.buffer.flush()'))
1302	n/a	out = assert_python_ok('-c', code)
1303	n/a	stdout = out[1]
1304	n/a	self.assertEqual(len(stdout), 16)
1305	n/a	return stdout
1306	n/a
1307	n/a	def test_urandom_subprocess(self):
1308	n/a	data1 = self.get_urandom_subprocess(16)
1309	n/a	data2 = self.get_urandom_subprocess(16)
1310	n/a	self.assertNotEqual(data1, data2)
1311	n/a
1312	n/a
1313	n/a	@unittest.skipUnless(hasattr(os, 'getrandom'), 'need os.getrandom()')
1314	n/a	class GetRandomTests(unittest.TestCase):
1315	n/a	@classmethod
1316	n/a	def setUpClass(cls):
1317	n/a	try:
1318	n/a	os.getrandom(1)
1319	n/a	except OSError as exc:
1320	n/a	if exc.errno == errno.ENOSYS:
1321	n/a	# Python compiled on a more recent Linux version
1322	n/a	# than the current Linux kernel
1323	n/a	raise unittest.SkipTest("getrandom() syscall fails with ENOSYS")
1324	n/a	else:
1325	n/a	raise
1326	n/a
1327	n/a	def test_getrandom_type(self):
1328	n/a	data = os.getrandom(16)
1329	n/a	self.assertIsInstance(data, bytes)
1330	n/a	self.assertEqual(len(data), 16)
1331	n/a
1332	n/a	def test_getrandom0(self):
1333	n/a	empty = os.getrandom(0)
1334	n/a	self.assertEqual(empty, b'')
1335	n/a
1336	n/a	def test_getrandom_random(self):
1337	n/a	self.assertTrue(hasattr(os, 'GRND_RANDOM'))
1338	n/a
1339	n/a	# Don't test os.getrandom(1, os.GRND_RANDOM) to not consume the rare
1340	n/a	# resource /dev/random
1341	n/a
1342	n/a	def test_getrandom_nonblock(self):
1343	n/a	# The call must not fail. Check also that the flag exists
1344	n/a	try:
1345	n/a	os.getrandom(1, os.GRND_NONBLOCK)
1346	n/a	except BlockingIOError:
1347	n/a	# System urandom is not initialized yet
1348	n/a	pass
1349	n/a
1350	n/a	def test_getrandom_value(self):
1351	n/a	data1 = os.getrandom(16)
1352	n/a	data2 = os.getrandom(16)
1353	n/a	self.assertNotEqual(data1, data2)
1354	n/a
1355	n/a
1356	n/a	# os.urandom() doesn't use a file descriptor when it is implemented with the
1357	n/a	# getentropy() function, the getrandom() function or the getrandom() syscall
1358	n/a	OS_URANDOM_DONT_USE_FD = (
1359	n/a	sysconfig.get_config_var('HAVE_GETENTROPY') == 1
1360	n/a	or sysconfig.get_config_var('HAVE_GETRANDOM') == 1
1361	n/a	or sysconfig.get_config_var('HAVE_GETRANDOM_SYSCALL') == 1)
1362	n/a
1363	n/a	@unittest.skipIf(OS_URANDOM_DONT_USE_FD ,
1364	n/a	"os.random() does not use a file descriptor")
1365	n/a	class URandomFDTests(unittest.TestCase):
1366	n/a	@unittest.skipUnless(resource, "test requires the resource module")
1367	n/a	def test_urandom_failure(self):
1368	n/a	# Check urandom() failing when it is not able to open /dev/random.
1369	n/a	# We spawn a new process to make the test more robust (if getrlimit()
1370	n/a	# failed to restore the file descriptor limit after this, the whole
1371	n/a	# test suite would crash; this actually happened on the OS X Tiger
1372	n/a	# buildbot).
1373	n/a	code = """if 1:
1374	n/a	import errno
1375	n/a	import os
1376	n/a	import resource
1377	n/a
1378	n/a	soft_limit, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE)
1379	n/a	resource.setrlimit(resource.RLIMIT_NOFILE, (1, hard_limit))
1380	n/a	try:
1381	n/a	os.urandom(16)
1382	n/a	except OSError as e:
1383	n/a	assert e.errno == errno.EMFILE, e.errno
1384	n/a	else:
1385	n/a	raise AssertionError("OSError not raised")
1386	n/a	"""
1387	n/a	assert_python_ok('-c', code)
1388	n/a
1389	n/a	def test_urandom_fd_closed(self):
1390	n/a	# Issue #21207: urandom() should reopen its fd to /dev/urandom if
1391	n/a	# closed.
1392	n/a	code = """if 1:
1393	n/a	import os
1394	n/a	import sys
1395	n/a	import test.support
1396	n/a	os.urandom(4)
1397	n/a	with test.support.SuppressCrashReport():
1398	n/a	os.closerange(3, 256)
1399	n/a	sys.stdout.buffer.write(os.urandom(4))
1400	n/a	"""
1401	n/a	rc, out, err = assert_python_ok('-Sc', code)
1402	n/a
1403	n/a	def test_urandom_fd_reopened(self):
1404	n/a	# Issue #21207: urandom() should detect its fd to /dev/urandom
1405	n/a	# changed to something else, and reopen it.
1406	n/a	self.addCleanup(support.unlink, support.TESTFN)
1407	n/a	create_file(support.TESTFN, b"x" * 256)
1408	n/a
1409	n/a	code = """if 1:
1410	n/a	import os
1411	n/a	import sys
1412	n/a	import test.support
1413	n/a	os.urandom(4)
1414	n/a	with test.support.SuppressCrashReport():
1415	n/a	for fd in range(3, 256):
1416	n/a	try:
1417	n/a	os.close(fd)
1418	n/a	except OSError:
1419	n/a	pass
1420	n/a	else:
1421	n/a	# Found the urandom fd (XXX hopefully)
1422	n/a	break
1423	n/a	os.closerange(3, 256)
1424	n/a	with open({TESTFN!r}, 'rb') as f:
1425	n/a	new_fd = f.fileno()
1426	n/a	# Issue #26935: posix allows new_fd and fd to be equal but
1427	n/a	# some libc implementations have dup2 return an error in this
1428	n/a	# case.
1429	n/a	if new_fd != fd:
1430	n/a	os.dup2(new_fd, fd)
1431	n/a	sys.stdout.buffer.write(os.urandom(4))
1432	n/a	sys.stdout.buffer.write(os.urandom(4))
1433	n/a	""".format(TESTFN=support.TESTFN)
1434	n/a	rc, out, err = assert_python_ok('-Sc', code)
1435	n/a	self.assertEqual(len(out), 8)
1436	n/a	self.assertNotEqual(out[0:4], out[4:8])
1437	n/a	rc, out2, err2 = assert_python_ok('-Sc', code)
1438	n/a	self.assertEqual(len(out2), 8)
1439	n/a	self.assertNotEqual(out2, out)
1440	n/a
1441	n/a
1442	n/a	@contextlib.contextmanager
1443	n/a	def _execvpe_mockup(defpath=None):
1444	n/a	"""
1445	n/a	Stubs out execv and execve functions when used as context manager.
1446	n/a	Records exec calls. The mock execv and execve functions always raise an
1447	n/a	exception as they would normally never return.
1448	n/a	"""
1449	n/a	# A list of tuples containing (function name, first arg, args)
1450	n/a	# of calls to execv or execve that have been made.
1451	n/a	calls = []
1452	n/a
1453	n/a	def mock_execv(name, *args):
1454	n/a	calls.append(('execv', name, args))
1455	n/a	raise RuntimeError("execv called")
1456	n/a
1457	n/a	def mock_execve(name, *args):
1458	n/a	calls.append(('execve', name, args))
1459	n/a	raise OSError(errno.ENOTDIR, "execve called")
1460	n/a
1461	n/a	try:
1462	n/a	orig_execv = os.execv
1463	n/a	orig_execve = os.execve
1464	n/a	orig_defpath = os.defpath
1465	n/a	os.execv = mock_execv
1466	n/a	os.execve = mock_execve
1467	n/a	if defpath is not None:
1468	n/a	os.defpath = defpath
1469	n/a	yield calls
1470	n/a	finally:
1471	n/a	os.execv = orig_execv
1472	n/a	os.execve = orig_execve
1473	n/a	os.defpath = orig_defpath
1474	n/a
1475	n/a
1476	n/a	class ExecTests(unittest.TestCase):
1477	n/a	@unittest.skipIf(USING_LINUXTHREADS,
1478	n/a	"avoid triggering a linuxthreads bug: see issue #4970")
1479	n/a	def test_execvpe_with_bad_program(self):
1480	n/a	self.assertRaises(OSError, os.execvpe, 'no such app-',
1481	n/a	['no such app-'], None)
1482	n/a
1483	n/a	def test_execv_with_bad_arglist(self):
1484	n/a	self.assertRaises(ValueError, os.execv, 'notepad', ())
1485	n/a	self.assertRaises(ValueError, os.execv, 'notepad', [])
1486	n/a	self.assertRaises(ValueError, os.execv, 'notepad', ('',))
1487	n/a	self.assertRaises(ValueError, os.execv, 'notepad', [''])
1488	n/a
1489	n/a	def test_execvpe_with_bad_arglist(self):
1490	n/a	self.assertRaises(ValueError, os.execvpe, 'notepad', [], None)
1491	n/a	self.assertRaises(ValueError, os.execvpe, 'notepad', [], {})
1492	n/a	self.assertRaises(ValueError, os.execvpe, 'notepad', [''], {})
1493	n/a
1494	n/a	@unittest.skipUnless(hasattr(os, '_execvpe'),
1495	n/a	"No internal os._execvpe function to test.")
1496	n/a	def _test_internal_execvpe(self, test_type):
1497	n/a	program_path = os.sep + 'absolutepath'
1498	n/a	if test_type is bytes:
1499	n/a	program = b'executable'
1500	n/a	fullpath = os.path.join(os.fsencode(program_path), program)
1501	n/a	native_fullpath = fullpath
1502	n/a	arguments = [b'progname', 'arg1', 'arg2']
1503	n/a	else:
1504	n/a	program = 'executable'
1505	n/a	arguments = ['progname', 'arg1', 'arg2']
1506	n/a	fullpath = os.path.join(program_path, program)
1507	n/a	if os.name != "nt":
1508	n/a	native_fullpath = os.fsencode(fullpath)
1509	n/a	else:
1510	n/a	native_fullpath = fullpath
1511	n/a	env = {'spam': 'beans'}
1512	n/a
1513	n/a	# test os._execvpe() with an absolute path
1514	n/a	with _execvpe_mockup() as calls:
1515	n/a	self.assertRaises(RuntimeError,
1516	n/a	os._execvpe, fullpath, arguments)
1517	n/a	self.assertEqual(len(calls), 1)
1518	n/a	self.assertEqual(calls[0], ('execv', fullpath, (arguments,)))
1519	n/a
1520	n/a	# test os._execvpe() with a relative path:
1521	n/a	# os.get_exec_path() returns defpath
1522	n/a	with _execvpe_mockup(defpath=program_path) as calls:
1523	n/a	self.assertRaises(OSError,
1524	n/a	os._execvpe, program, arguments, env=env)
1525	n/a	self.assertEqual(len(calls), 1)
1526	n/a	self.assertSequenceEqual(calls[0],
1527	n/a	('execve', native_fullpath, (arguments, env)))
1528	n/a
1529	n/a	# test os._execvpe() with a relative path:
1530	n/a	# os.get_exec_path() reads the 'PATH' variable
1531	n/a	with _execvpe_mockup() as calls:
1532	n/a	env_path = env.copy()
1533	n/a	if test_type is bytes:
1534	n/a	env_path[b'PATH'] = program_path
1535	n/a	else:
1536	n/a	env_path['PATH'] = program_path
1537	n/a	self.assertRaises(OSError,
1538	n/a	os._execvpe, program, arguments, env=env_path)
1539	n/a	self.assertEqual(len(calls), 1)
1540	n/a	self.assertSequenceEqual(calls[0],
1541	n/a	('execve', native_fullpath, (arguments, env_path)))
1542	n/a
1543	n/a	def test_internal_execvpe_str(self):
1544	n/a	self._test_internal_execvpe(str)
1545	n/a	if os.name != "nt":
1546	n/a	self._test_internal_execvpe(bytes)
1547	n/a
1548	n/a
1549	n/a	@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
1550	n/a	class Win32ErrorTests(unittest.TestCase):
1551	n/a	def setUp(self):
1552	n/a	try:
1553	n/a	os.stat(support.TESTFN)
1554	n/a	except FileNotFoundError:
1555	n/a	exists = False
1556	n/a	except OSError as exc:
1557	n/a	exists = True
1558	n/a	self.fail("file %s must not exist; os.stat failed with %s"
1559	n/a	% (support.TESTFN, exc))
1560	n/a	else:
1561	n/a	self.fail("file %s must not exist" % support.TESTFN)
1562	n/a
1563	n/a	def test_rename(self):
1564	n/a	self.assertRaises(OSError, os.rename, support.TESTFN, support.TESTFN+".bak")
1565	n/a
1566	n/a	def test_remove(self):
1567	n/a	self.assertRaises(OSError, os.remove, support.TESTFN)
1568	n/a
1569	n/a	def test_chdir(self):
1570	n/a	self.assertRaises(OSError, os.chdir, support.TESTFN)
1571	n/a
1572	n/a	def test_mkdir(self):
1573	n/a	self.addCleanup(support.unlink, support.TESTFN)
1574	n/a
1575	n/a	with open(support.TESTFN, "x") as f:
1576	n/a	self.assertRaises(OSError, os.mkdir, support.TESTFN)
1577	n/a
1578	n/a	def test_utime(self):
1579	n/a	self.assertRaises(OSError, os.utime, support.TESTFN, None)
1580	n/a
1581	n/a	def test_chmod(self):
1582	n/a	self.assertRaises(OSError, os.chmod, support.TESTFN, 0)
1583	n/a
1584	n/a
1585	n/a	class TestInvalidFD(unittest.TestCase):
1586	n/a	singles = ["fchdir", "dup", "fdopen", "fdatasync", "fstat",
1587	n/a	"fstatvfs", "fsync", "tcgetpgrp", "ttyname"]
1588	n/a	#singles.append("close")
1589	n/a	#We omit close because it doesn't raise an exception on some platforms
1590	n/a	def get_single(f):
1591	n/a	def helper(self):
1592	n/a	if hasattr(os, f):
1593	n/a	self.check(getattr(os, f))
1594	n/a	return helper
1595	n/a	for f in singles:
1596	n/a	locals()["test_"+f] = get_single(f)
1597	n/a
1598	n/a	def check(self, f, *args):
1599	n/a	try:
1600	n/a	f(support.make_bad_fd(), *args)
1601	n/a	except OSError as e:
1602	n/a	self.assertEqual(e.errno, errno.EBADF)
1603	n/a	else:
1604	n/a	self.fail("%r didn't raise an OSError with a bad file descriptor"
1605	n/a	% f)
1606	n/a
1607	n/a	@unittest.skipUnless(hasattr(os, 'isatty'), 'test needs os.isatty()')
1608	n/a	def test_isatty(self):
1609	n/a	self.assertEqual(os.isatty(support.make_bad_fd()), False)
1610	n/a
1611	n/a	@unittest.skipUnless(hasattr(os, 'closerange'), 'test needs os.closerange()')
1612	n/a	def test_closerange(self):
1613	n/a	fd = support.make_bad_fd()
1614	n/a	# Make sure none of the descriptors we are about to close are
1615	n/a	# currently valid (issue 6542).
1616	n/a	for i in range(10):
1617	n/a	try: os.fstat(fd+i)
1618	n/a	except OSError:
1619	n/a	pass
1620	n/a	else:
1621	n/a	break
1622	n/a	if i < 2:
1623	n/a	raise unittest.SkipTest(
1624	n/a	"Unable to acquire a range of invalid file descriptors")
1625	n/a	self.assertEqual(os.closerange(fd, fd + i-1), None)
1626	n/a
1627	n/a	@unittest.skipUnless(hasattr(os, 'dup2'), 'test needs os.dup2()')
1628	n/a	def test_dup2(self):
1629	n/a	self.check(os.dup2, 20)
1630	n/a
1631	n/a	@unittest.skipUnless(hasattr(os, 'fchmod'), 'test needs os.fchmod()')
1632	n/a	def test_fchmod(self):
1633	n/a	self.check(os.fchmod, 0)
1634	n/a
1635	n/a	@unittest.skipUnless(hasattr(os, 'fchown'), 'test needs os.fchown()')
1636	n/a	def test_fchown(self):
1637	n/a	self.check(os.fchown, -1, -1)
1638	n/a
1639	n/a	@unittest.skipUnless(hasattr(os, 'fpathconf'), 'test needs os.fpathconf()')
1640	n/a	def test_fpathconf(self):
1641	n/a	self.check(os.pathconf, "PC_NAME_MAX")
1642	n/a	self.check(os.fpathconf, "PC_NAME_MAX")
1643	n/a
1644	n/a	@unittest.skipUnless(hasattr(os, 'ftruncate'), 'test needs os.ftruncate()')
1645	n/a	def test_ftruncate(self):
1646	n/a	self.check(os.truncate, 0)
1647	n/a	self.check(os.ftruncate, 0)
1648	n/a
1649	n/a	@unittest.skipUnless(hasattr(os, 'lseek'), 'test needs os.lseek()')
1650	n/a	def test_lseek(self):
1651	n/a	self.check(os.lseek, 0, 0)
1652	n/a
1653	n/a	@unittest.skipUnless(hasattr(os, 'read'), 'test needs os.read()')
1654	n/a	def test_read(self):
1655	n/a	self.check(os.read, 1)
1656	n/a
1657	n/a	@unittest.skipUnless(hasattr(os, 'readv'), 'test needs os.readv()')
1658	n/a	def test_readv(self):
1659	n/a	buf = bytearray(10)
1660	n/a	self.check(os.readv, [buf])
1661	n/a
1662	n/a	@unittest.skipUnless(hasattr(os, 'tcsetpgrp'), 'test needs os.tcsetpgrp()')
1663	n/a	def test_tcsetpgrpt(self):
1664	n/a	self.check(os.tcsetpgrp, 0)
1665	n/a
1666	n/a	@unittest.skipUnless(hasattr(os, 'write'), 'test needs os.write()')
1667	n/a	def test_write(self):
1668	n/a	self.check(os.write, b" ")
1669	n/a
1670	n/a	@unittest.skipUnless(hasattr(os, 'writev'), 'test needs os.writev()')
1671	n/a	def test_writev(self):
1672	n/a	self.check(os.writev, [b'abc'])
1673	n/a
1674	n/a	def test_inheritable(self):
1675	n/a	self.check(os.get_inheritable)
1676	n/a	self.check(os.set_inheritable, True)
1677	n/a
1678	n/a	@unittest.skipUnless(hasattr(os, 'get_blocking'),
1679	n/a	'needs os.get_blocking() and os.set_blocking()')
1680	n/a	def test_blocking(self):
1681	n/a	self.check(os.get_blocking)
1682	n/a	self.check(os.set_blocking, True)
1683	n/a
1684	n/a
1685	n/a	class LinkTests(unittest.TestCase):
1686	n/a	def setUp(self):
1687	n/a	self.file1 = support.TESTFN
1688	n/a	self.file2 = os.path.join(support.TESTFN + "2")
1689	n/a
1690	n/a	def tearDown(self):
1691	n/a	for file in (self.file1, self.file2):
1692	n/a	if os.path.exists(file):
1693	n/a	os.unlink(file)
1694	n/a
1695	n/a	def _test_link(self, file1, file2):
1696	n/a	create_file(file1)
1697	n/a
1698	n/a	os.link(file1, file2)
1699	n/a	with open(file1, "r") as f1, open(file2, "r") as f2:
1700	n/a	self.assertTrue(os.path.sameopenfile(f1.fileno(), f2.fileno()))
1701	n/a
1702	n/a	def test_link(self):
1703	n/a	self._test_link(self.file1, self.file2)
1704	n/a
1705	n/a	def test_link_bytes(self):
1706	n/a	self._test_link(bytes(self.file1, sys.getfilesystemencoding()),
1707	n/a	bytes(self.file2, sys.getfilesystemencoding()))
1708	n/a
1709	n/a	def test_unicode_name(self):
1710	n/a	try:
1711	n/a	os.fsencode("\xf1")
1712	n/a	except UnicodeError:
1713	n/a	raise unittest.SkipTest("Unable to encode for this platform.")
1714	n/a
1715	n/a	self.file1 += "\xf1"
1716	n/a	self.file2 = self.file1 + "2"
1717	n/a	self._test_link(self.file1, self.file2)
1718	n/a
1719	n/a	@unittest.skipIf(sys.platform == "win32", "Posix specific tests")
1720	n/a	class PosixUidGidTests(unittest.TestCase):
1721	n/a	@unittest.skipUnless(hasattr(os, 'setuid'), 'test needs os.setuid()')
1722	n/a	def test_setuid(self):
1723	n/a	if os.getuid() != 0:
1724	n/a	self.assertRaises(OSError, os.setuid, 0)
1725	n/a	self.assertRaises(OverflowError, os.setuid, 1<<32)
1726	n/a
1727	n/a	@unittest.skipUnless(hasattr(os, 'setgid'), 'test needs os.setgid()')
1728	n/a	def test_setgid(self):
1729	n/a	if os.getuid() != 0 and not HAVE_WHEEL_GROUP:
1730	n/a	self.assertRaises(OSError, os.setgid, 0)
1731	n/a	self.assertRaises(OverflowError, os.setgid, 1<<32)
1732	n/a
1733	n/a	@unittest.skipUnless(hasattr(os, 'seteuid'), 'test needs os.seteuid()')
1734	n/a	def test_seteuid(self):
1735	n/a	if os.getuid() != 0:
1736	n/a	self.assertRaises(OSError, os.seteuid, 0)
1737	n/a	self.assertRaises(OverflowError, os.seteuid, 1<<32)
1738	n/a
1739	n/a	@unittest.skipUnless(hasattr(os, 'setegid'), 'test needs os.setegid()')
1740	n/a	def test_setegid(self):
1741	n/a	if os.getuid() != 0 and not HAVE_WHEEL_GROUP:
1742	n/a	self.assertRaises(OSError, os.setegid, 0)
1743	n/a	self.assertRaises(OverflowError, os.setegid, 1<<32)
1744	n/a
1745	n/a	@unittest.skipUnless(hasattr(os, 'setreuid'), 'test needs os.setreuid()')
1746	n/a	def test_setreuid(self):
1747	n/a	if os.getuid() != 0:
1748	n/a	self.assertRaises(OSError, os.setreuid, 0, 0)
1749	n/a	self.assertRaises(OverflowError, os.setreuid, 1<<32, 0)
1750	n/a	self.assertRaises(OverflowError, os.setreuid, 0, 1<<32)
1751	n/a
1752	n/a	@unittest.skipUnless(hasattr(os, 'setreuid'), 'test needs os.setreuid()')
1753	n/a	def test_setreuid_neg1(self):
1754	n/a	# Needs to accept -1. We run this in a subprocess to avoid
1755	n/a	# altering the test runner's process state (issue8045).
1756	n/a	subprocess.check_call([
1757	n/a	sys.executable, '-c',
1758	n/a	'import os,sys;os.setreuid(-1,-1);sys.exit(0)'])
1759	n/a
1760	n/a	@unittest.skipUnless(hasattr(os, 'setregid'), 'test needs os.setregid()')
1761	n/a	def test_setregid(self):
1762	n/a	if os.getuid() != 0 and not HAVE_WHEEL_GROUP:
1763	n/a	self.assertRaises(OSError, os.setregid, 0, 0)
1764	n/a	self.assertRaises(OverflowError, os.setregid, 1<<32, 0)
1765	n/a	self.assertRaises(OverflowError, os.setregid, 0, 1<<32)
1766	n/a
1767	n/a	@unittest.skipUnless(hasattr(os, 'setregid'), 'test needs os.setregid()')
1768	n/a	def test_setregid_neg1(self):
1769	n/a	# Needs to accept -1. We run this in a subprocess to avoid
1770	n/a	# altering the test runner's process state (issue8045).
1771	n/a	subprocess.check_call([
1772	n/a	sys.executable, '-c',
1773	n/a	'import os,sys;os.setregid(-1,-1);sys.exit(0)'])
1774	n/a
1775	n/a	@unittest.skipIf(sys.platform == "win32", "Posix specific tests")
1776	n/a	class Pep383Tests(unittest.TestCase):
1777	n/a	def setUp(self):
1778	n/a	if support.TESTFN_UNENCODABLE:
1779	n/a	self.dir = support.TESTFN_UNENCODABLE
1780	n/a	elif support.TESTFN_NONASCII:
1781	n/a	self.dir = support.TESTFN_NONASCII
1782	n/a	else:
1783	n/a	self.dir = support.TESTFN
1784	n/a	self.bdir = os.fsencode(self.dir)
1785	n/a
1786	n/a	bytesfn = []
1787	n/a	def add_filename(fn):
1788	n/a	try:
1789	n/a	fn = os.fsencode(fn)
1790	n/a	except UnicodeEncodeError:
1791	n/a	return
1792	n/a	bytesfn.append(fn)
1793	n/a	add_filename(support.TESTFN_UNICODE)
1794	n/a	if support.TESTFN_UNENCODABLE:
1795	n/a	add_filename(support.TESTFN_UNENCODABLE)
1796	n/a	if support.TESTFN_NONASCII:
1797	n/a	add_filename(support.TESTFN_NONASCII)
1798	n/a	if not bytesfn:
1799	n/a	self.skipTest("couldn't create any non-ascii filename")
1800	n/a
1801	n/a	self.unicodefn = set()
1802	n/a	os.mkdir(self.dir)
1803	n/a	try:
1804	n/a	for fn in bytesfn:
1805	n/a	support.create_empty_file(os.path.join(self.bdir, fn))
1806	n/a	fn = os.fsdecode(fn)
1807	n/a	if fn in self.unicodefn:
1808	n/a	raise ValueError("duplicate filename")
1809	n/a	self.unicodefn.add(fn)
1810	n/a	except:
1811	n/a	shutil.rmtree(self.dir)
1812	n/a	raise
1813	n/a
1814	n/a	def tearDown(self):
1815	n/a	shutil.rmtree(self.dir)
1816	n/a
1817	n/a	def test_listdir(self):
1818	n/a	expected = self.unicodefn
1819	n/a	found = set(os.listdir(self.dir))
1820	n/a	self.assertEqual(found, expected)
1821	n/a	# test listdir without arguments
1822	n/a	current_directory = os.getcwd()
1823	n/a	try:
1824	n/a	os.chdir(os.sep)
1825	n/a	self.assertEqual(set(os.listdir()), set(os.listdir(os.sep)))
1826	n/a	finally:
1827	n/a	os.chdir(current_directory)
1828	n/a
1829	n/a	def test_open(self):
1830	n/a	for fn in self.unicodefn:
1831	n/a	f = open(os.path.join(self.dir, fn), 'rb')
1832	n/a	f.close()
1833	n/a
1834	n/a	@unittest.skipUnless(hasattr(os, 'statvfs'),
1835	n/a	"need os.statvfs()")
1836	n/a	def test_statvfs(self):
1837	n/a	# issue #9645
1838	n/a	for fn in self.unicodefn:
1839	n/a	# should not fail with file not found error
1840	n/a	fullname = os.path.join(self.dir, fn)
1841	n/a	os.statvfs(fullname)
1842	n/a
1843	n/a	def test_stat(self):
1844	n/a	for fn in self.unicodefn:
1845	n/a	os.stat(os.path.join(self.dir, fn))
1846	n/a
1847	n/a	@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
1848	n/a	class Win32KillTests(unittest.TestCase):
1849	n/a	def _kill(self, sig):
1850	n/a	# Start sys.executable as a subprocess and communicate from the
1851	n/a	# subprocess to the parent that the interpreter is ready. When it
1852	n/a	# becomes ready, send sig via os.kill to the subprocess and check
1853	n/a	# that the return code is equal to sig.
1854	n/a	import ctypes
1855	n/a	from ctypes import wintypes
1856	n/a	import msvcrt
1857	n/a
1858	n/a	# Since we can't access the contents of the process' stdout until the
1859	n/a	# process has exited, use PeekNamedPipe to see what's inside stdout
1860	n/a	# without waiting. This is done so we can tell that the interpreter
1861	n/a	# is started and running at a point where it could handle a signal.
1862	n/a	PeekNamedPipe = ctypes.windll.kernel32.PeekNamedPipe
1863	n/a	PeekNamedPipe.restype = wintypes.BOOL
1864	n/a	PeekNamedPipe.argtypes = (wintypes.HANDLE, # Pipe handle
1865	n/a	ctypes.POINTER(ctypes.c_char), # stdout buf
1866	n/a	wintypes.DWORD, # Buffer size
1867	n/a	ctypes.POINTER(wintypes.DWORD), # bytes read
1868	n/a	ctypes.POINTER(wintypes.DWORD), # bytes avail
1869	n/a	ctypes.POINTER(wintypes.DWORD)) # bytes left
1870	n/a	msg = "running"
1871	n/a	proc = subprocess.Popen([sys.executable, "-c",
1872	n/a	"import sys;"
1873	n/a	"sys.stdout.write('{}');"
1874	n/a	"sys.stdout.flush();"
1875	n/a	"input()".format(msg)],
1876	n/a	stdout=subprocess.PIPE,
1877	n/a	stderr=subprocess.PIPE,
1878	n/a	stdin=subprocess.PIPE)
1879	n/a	self.addCleanup(proc.stdout.close)
1880	n/a	self.addCleanup(proc.stderr.close)
1881	n/a	self.addCleanup(proc.stdin.close)
1882	n/a
1883	n/a	count, max = 0, 100
1884	n/a	while count < max and proc.poll() is None:
1885	n/a	# Create a string buffer to store the result of stdout from the pipe
1886	n/a	buf = ctypes.create_string_buffer(len(msg))
1887	n/a	# Obtain the text currently in proc.stdout
1888	n/a	# Bytes read/avail/left are left as NULL and unused
1889	n/a	rslt = PeekNamedPipe(msvcrt.get_osfhandle(proc.stdout.fileno()),
1890	n/a	buf, ctypes.sizeof(buf), None, None, None)
1891	n/a	self.assertNotEqual(rslt, 0, "PeekNamedPipe failed")
1892	n/a	if buf.value:
1893	n/a	self.assertEqual(msg, buf.value.decode())
1894	n/a	break
1895	n/a	time.sleep(0.1)
1896	n/a	count += 1
1897	n/a	else:
1898	n/a	self.fail("Did not receive communication from the subprocess")
1899	n/a
1900	n/a	os.kill(proc.pid, sig)
1901	n/a	self.assertEqual(proc.wait(), sig)
1902	n/a
1903	n/a	def test_kill_sigterm(self):
1904	n/a	# SIGTERM doesn't mean anything special, but make sure it works
1905	n/a	self._kill(signal.SIGTERM)
1906	n/a
1907	n/a	def test_kill_int(self):
1908	n/a	# os.kill on Windows can take an int which gets set as the exit code
1909	n/a	self._kill(100)
1910	n/a
1911	n/a	def _kill_with_event(self, event, name):
1912	n/a	tagname = "test_os_%s" % uuid.uuid1()
1913	n/a	m = mmap.mmap(-1, 1, tagname)
1914	n/a	m[0] = 0
1915	n/a	# Run a script which has console control handling enabled.
1916	n/a	proc = subprocess.Popen([sys.executable,
1917	n/a	os.path.join(os.path.dirname(__file__),
1918	n/a	"win_console_handler.py"), tagname],
1919	n/a	creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
1920	n/a	# Let the interpreter startup before we send signals. See #3137.
1921	n/a	count, max = 0, 100
1922	n/a	while count < max and proc.poll() is None:
1923	n/a	if m[0] == 1:
1924	n/a	break
1925	n/a	time.sleep(0.1)
1926	n/a	count += 1
1927	n/a	else:
1928	n/a	# Forcefully kill the process if we weren't able to signal it.
1929	n/a	os.kill(proc.pid, signal.SIGINT)
1930	n/a	self.fail("Subprocess didn't finish initialization")
1931	n/a	os.kill(proc.pid, event)
1932	n/a	# proc.send_signal(event) could also be done here.
1933	n/a	# Allow time for the signal to be passed and the process to exit.
1934	n/a	time.sleep(0.5)
1935	n/a	if not proc.poll():
1936	n/a	# Forcefully kill the process if we weren't able to signal it.
1937	n/a	os.kill(proc.pid, signal.SIGINT)
1938	n/a	self.fail("subprocess did not stop on {}".format(name))
1939	n/a
1940	n/a	@unittest.skip("subprocesses aren't inheriting Ctrl+C property")
1941	n/a	def test_CTRL_C_EVENT(self):
1942	n/a	from ctypes import wintypes
1943	n/a	import ctypes
1944	n/a
1945	n/a	# Make a NULL value by creating a pointer with no argument.
1946	n/a	NULL = ctypes.POINTER(ctypes.c_int)()
1947	n/a	SetConsoleCtrlHandler = ctypes.windll.kernel32.SetConsoleCtrlHandler
1948	n/a	SetConsoleCtrlHandler.argtypes = (ctypes.POINTER(ctypes.c_int),
1949	n/a	wintypes.BOOL)
1950	n/a	SetConsoleCtrlHandler.restype = wintypes.BOOL
1951	n/a
1952	n/a	# Calling this with NULL and FALSE causes the calling process to
1953	n/a	# handle Ctrl+C, rather than ignore it. This property is inherited
1954	n/a	# by subprocesses.
1955	n/a	SetConsoleCtrlHandler(NULL, 0)
1956	n/a
1957	n/a	self._kill_with_event(signal.CTRL_C_EVENT, "CTRL_C_EVENT")
1958	n/a
1959	n/a	def test_CTRL_BREAK_EVENT(self):
1960	n/a	self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT")
1961	n/a
1962	n/a
1963	n/a	@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
1964	n/a	class Win32ListdirTests(unittest.TestCase):
1965	n/a	"""Test listdir on Windows."""
1966	n/a
1967	n/a	def setUp(self):
1968	n/a	self.created_paths = []
1969	n/a	for i in range(2):
1970	n/a	dir_name = 'SUB%d' % i
1971	n/a	dir_path = os.path.join(support.TESTFN, dir_name)
1972	n/a	file_name = 'FILE%d' % i
1973	n/a	file_path = os.path.join(support.TESTFN, file_name)
1974	n/a	os.makedirs(dir_path)
1975	n/a	with open(file_path, 'w') as f:
1976	n/a	f.write("I'm %s and proud of it. Blame test_os.\n" % file_path)
1977	n/a	self.created_paths.extend([dir_name, file_name])
1978	n/a	self.created_paths.sort()
1979	n/a
1980	n/a	def tearDown(self):
1981	n/a	shutil.rmtree(support.TESTFN)
1982	n/a
1983	n/a	def test_listdir_no_extended_path(self):
1984	n/a	"""Test when the path is not an "extended" path."""
1985	n/a	# unicode
1986	n/a	self.assertEqual(
1987	n/a	sorted(os.listdir(support.TESTFN)),
1988	n/a	self.created_paths)
1989	n/a
1990	n/a	# bytes
1991	n/a	self.assertEqual(
1992	n/a	sorted(os.listdir(os.fsencode(support.TESTFN))),
1993	n/a	[os.fsencode(path) for path in self.created_paths])
1994	n/a
1995	n/a	def test_listdir_extended_path(self):
1996	n/a	"""Test when the path starts with '\\\\?\\'."""
1997	n/a	# See: http://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx#maxpath
1998	n/a	# unicode
1999	n/a	path = '\\\\?\\' + os.path.abspath(support.TESTFN)
2000	n/a	self.assertEqual(
2001	n/a	sorted(os.listdir(path)),
2002	n/a	self.created_paths)
2003	n/a
2004	n/a	# bytes
2005	n/a	path = b'\\\\?\\' + os.fsencode(os.path.abspath(support.TESTFN))
2006	n/a	self.assertEqual(
2007	n/a	sorted(os.listdir(path)),
2008	n/a	[os.fsencode(path) for path in self.created_paths])
2009	n/a
2010	n/a
2011	n/a	@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
2012	n/a	@support.skip_unless_symlink
2013	n/a	class Win32SymlinkTests(unittest.TestCase):
2014	n/a	filelink = 'filelinktest'
2015	n/a	filelink_target = os.path.abspath(__file__)
2016	n/a	dirlink = 'dirlinktest'
2017	n/a	dirlink_target = os.path.dirname(filelink_target)
2018	n/a	missing_link = 'missing link'
2019	n/a
2020	n/a	def setUp(self):
2021	n/a	assert os.path.exists(self.dirlink_target)
2022	n/a	assert os.path.exists(self.filelink_target)
2023	n/a	assert not os.path.exists(self.dirlink)
2024	n/a	assert not os.path.exists(self.filelink)
2025	n/a	assert not os.path.exists(self.missing_link)
2026	n/a
2027	n/a	def tearDown(self):
2028	n/a	if os.path.exists(self.filelink):
2029	n/a	os.remove(self.filelink)
2030	n/a	if os.path.exists(self.dirlink):
2031	n/a	os.rmdir(self.dirlink)
2032	n/a	if os.path.lexists(self.missing_link):
2033	n/a	os.remove(self.missing_link)
2034	n/a
2035	n/a	def test_directory_link(self):
2036	n/a	os.symlink(self.dirlink_target, self.dirlink)
2037	n/a	self.assertTrue(os.path.exists(self.dirlink))
2038	n/a	self.assertTrue(os.path.isdir(self.dirlink))
2039	n/a	self.assertTrue(os.path.islink(self.dirlink))
2040	n/a	self.check_stat(self.dirlink, self.dirlink_target)
2041	n/a
2042	n/a	def test_file_link(self):
2043	n/a	os.symlink(self.filelink_target, self.filelink)
2044	n/a	self.assertTrue(os.path.exists(self.filelink))
2045	n/a	self.assertTrue(os.path.isfile(self.filelink))
2046	n/a	self.assertTrue(os.path.islink(self.filelink))
2047	n/a	self.check_stat(self.filelink, self.filelink_target)
2048	n/a
2049	n/a	def _create_missing_dir_link(self):
2050	n/a	'Create a "directory" link to a non-existent target'
2051	n/a	linkname = self.missing_link
2052	n/a	if os.path.lexists(linkname):
2053	n/a	os.remove(linkname)
2054	n/a	target = r'c:\\target does not exist.29r3c740'
2055	n/a	assert not os.path.exists(target)
2056	n/a	target_is_dir = True
2057	n/a	os.symlink(target, linkname, target_is_dir)
2058	n/a
2059	n/a	def test_remove_directory_link_to_missing_target(self):
2060	n/a	self._create_missing_dir_link()
2061	n/a	# For compatibility with Unix, os.remove will check the
2062	n/a	# directory status and call RemoveDirectory if the symlink
2063	n/a	# was created with target_is_dir==True.
2064	n/a	os.remove(self.missing_link)
2065	n/a
2066	n/a	@unittest.skip("currently fails; consider for improvement")
2067	n/a	def test_isdir_on_directory_link_to_missing_target(self):
2068	n/a	self._create_missing_dir_link()
2069	n/a	# consider having isdir return true for directory links
2070	n/a	self.assertTrue(os.path.isdir(self.missing_link))
2071	n/a
2072	n/a	@unittest.skip("currently fails; consider for improvement")
2073	n/a	def test_rmdir_on_directory_link_to_missing_target(self):
2074	n/a	self._create_missing_dir_link()
2075	n/a	# consider allowing rmdir to remove directory links
2076	n/a	os.rmdir(self.missing_link)
2077	n/a
2078	n/a	def check_stat(self, link, target):
2079	n/a	self.assertEqual(os.stat(link), os.stat(target))
2080	n/a	self.assertNotEqual(os.lstat(link), os.stat(link))
2081	n/a
2082	n/a	bytes_link = os.fsencode(link)
2083	n/a	self.assertEqual(os.stat(bytes_link), os.stat(target))
2084	n/a	self.assertNotEqual(os.lstat(bytes_link), os.stat(bytes_link))
2085	n/a
2086	n/a	def test_12084(self):
2087	n/a	level1 = os.path.abspath(support.TESTFN)
2088	n/a	level2 = os.path.join(level1, "level2")
2089	n/a	level3 = os.path.join(level2, "level3")
2090	n/a	self.addCleanup(support.rmtree, level1)
2091	n/a
2092	n/a	os.mkdir(level1)
2093	n/a	os.mkdir(level2)
2094	n/a	os.mkdir(level3)
2095	n/a
2096	n/a	file1 = os.path.abspath(os.path.join(level1, "file1"))
2097	n/a	create_file(file1)
2098	n/a
2099	n/a	orig_dir = os.getcwd()
2100	n/a	try:
2101	n/a	os.chdir(level2)
2102	n/a	link = os.path.join(level2, "link")
2103	n/a	os.symlink(os.path.relpath(file1), "link")
2104	n/a	self.assertIn("link", os.listdir(os.getcwd()))
2105	n/a
2106	n/a	# Check os.stat calls from the same dir as the link
2107	n/a	self.assertEqual(os.stat(file1), os.stat("link"))
2108	n/a
2109	n/a	# Check os.stat calls from a dir below the link
2110	n/a	os.chdir(level1)
2111	n/a	self.assertEqual(os.stat(file1),
2112	n/a	os.stat(os.path.relpath(link)))
2113	n/a
2114	n/a	# Check os.stat calls from a dir above the link
2115	n/a	os.chdir(level3)
2116	n/a	self.assertEqual(os.stat(file1),
2117	n/a	os.stat(os.path.relpath(link)))
2118	n/a	finally:
2119	n/a	os.chdir(orig_dir)
2120	n/a
2121	n/a
2122	n/a	@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
2123	n/a	class Win32JunctionTests(unittest.TestCase):
2124	n/a	junction = 'junctiontest'
2125	n/a	junction_target = os.path.dirname(os.path.abspath(__file__))
2126	n/a
2127	n/a	def setUp(self):
2128	n/a	assert os.path.exists(self.junction_target)
2129	n/a	assert not os.path.exists(self.junction)
2130	n/a
2131	n/a	def tearDown(self):
2132	n/a	if os.path.exists(self.junction):
2133	n/a	# os.rmdir delegates to Windows' RemoveDirectoryW,
2134	n/a	# which removes junction points safely.
2135	n/a	os.rmdir(self.junction)
2136	n/a
2137	n/a	def test_create_junction(self):
2138	n/a	_winapi.CreateJunction(self.junction_target, self.junction)
2139	n/a	self.assertTrue(os.path.exists(self.junction))
2140	n/a	self.assertTrue(os.path.isdir(self.junction))
2141	n/a
2142	n/a	# Junctions are not recognized as links.
2143	n/a	self.assertFalse(os.path.islink(self.junction))
2144	n/a
2145	n/a	def test_unlink_removes_junction(self):
2146	n/a	_winapi.CreateJunction(self.junction_target, self.junction)
2147	n/a	self.assertTrue(os.path.exists(self.junction))
2148	n/a
2149	n/a	os.unlink(self.junction)
2150	n/a	self.assertFalse(os.path.exists(self.junction))
2151	n/a
2152	n/a
2153	n/a	@support.skip_unless_symlink
2154	n/a	class NonLocalSymlinkTests(unittest.TestCase):
2155	n/a
2156	n/a	def setUp(self):
2157	n/a	r"""
2158	n/a	Create this structure:
2159	n/a
2160	n/a	base
2161	n/a	\___ some_dir
2162	n/a	"""
2163	n/a	os.makedirs('base/some_dir')
2164	n/a
2165	n/a	def tearDown(self):
2166	n/a	shutil.rmtree('base')
2167	n/a
2168	n/a	def test_directory_link_nonlocal(self):
2169	n/a	"""
2170	n/a	The symlink target should resolve relative to the link, not relative
2171	n/a	to the current directory.
2172	n/a
2173	n/a	Then, link base/some_link -> base/some_dir and ensure that some_link
2174	n/a	is resolved as a directory.
2175	n/a
2176	n/a	In issue13772, it was discovered that directory detection failed if
2177	n/a	the symlink target was not specified relative to the current
2178	n/a	directory, which was a defect in the implementation.
2179	n/a	"""
2180	n/a	src = os.path.join('base', 'some_link')
2181	n/a	os.symlink('some_dir', src)
2182	n/a	assert os.path.isdir(src)
2183	n/a
2184	n/a
2185	n/a	class FSEncodingTests(unittest.TestCase):
2186	n/a	def test_nop(self):
2187	n/a	self.assertEqual(os.fsencode(b'abc\xff'), b'abc\xff')
2188	n/a	self.assertEqual(os.fsdecode('abc\u0141'), 'abc\u0141')
2189	n/a
2190	n/a	def test_identity(self):
2191	n/a	# assert fsdecode(fsencode(x)) == x
2192	n/a	for fn in ('unicode\u0141', 'latin\xe9', 'ascii'):
2193	n/a	try:
2194	n/a	bytesfn = os.fsencode(fn)
2195	n/a	except UnicodeEncodeError:
2196	n/a	continue
2197	n/a	self.assertEqual(os.fsdecode(bytesfn), fn)
2198	n/a
2199	n/a
2200	n/a
2201	n/a	class DeviceEncodingTests(unittest.TestCase):
2202	n/a
2203	n/a	def test_bad_fd(self):
2204	n/a	# Return None when an fd doesn't actually exist.
2205	n/a	self.assertIsNone(os.device_encoding(123456))
2206	n/a
2207	n/a	@unittest.skipUnless(os.isatty(0) and (sys.platform.startswith('win') or
2208	n/a	(hasattr(locale, 'nl_langinfo') and hasattr(locale, 'CODESET'))),
2209	n/a	'test requires a tty and either Windows or nl_langinfo(CODESET)')
2210	n/a	def test_device_encoding(self):
2211	n/a	encoding = os.device_encoding(0)
2212	n/a	self.assertIsNotNone(encoding)
2213	n/a	self.assertTrue(codecs.lookup(encoding))
2214	n/a
2215	n/a
2216	n/a	class PidTests(unittest.TestCase):
2217	n/a	@unittest.skipUnless(hasattr(os, 'getppid'), "test needs os.getppid")
2218	n/a	def test_getppid(self):
2219	n/a	p = subprocess.Popen([sys.executable, '-c',
2220	n/a	'import os; print(os.getppid())'],
2221	n/a	stdout=subprocess.PIPE)
2222	n/a	stdout, _ = p.communicate()
2223	n/a	# We are the parent of our subprocess
2224	n/a	self.assertEqual(int(stdout), os.getpid())
2225	n/a
2226	n/a	def test_waitpid(self):
2227	n/a	args = [sys.executable, '-c', 'pass']
2228	n/a	# Add an implicit test for PyUnicode_FSConverter().
2229	n/a	pid = os.spawnv(os.P_NOWAIT, _PathLike(args[0]), args)
2230	n/a	status = os.waitpid(pid, 0)
2231	n/a	self.assertEqual(status, (pid, 0))
2232	n/a
2233	n/a
2234	n/a	class SpawnTests(unittest.TestCase):
2235	n/a	def create_args(self, *, with_env=False, use_bytes=False):
2236	n/a	self.exitcode = 17
2237	n/a
2238	n/a	filename = support.TESTFN
2239	n/a	self.addCleanup(support.unlink, filename)
2240	n/a
2241	n/a	if not with_env:
2242	n/a	code = 'import sys; sys.exit(%s)' % self.exitcode
2243	n/a	else:
2244	n/a	self.env = dict(os.environ)
2245	n/a	# create an unique key
2246	n/a	self.key = str(uuid.uuid4())
2247	n/a	self.env[self.key] = self.key
2248	n/a	# read the variable from os.environ to check that it exists
2249	n/a	code = ('import sys, os; magic = os.environ[%r]; sys.exit(%s)'
2250	n/a	% (self.key, self.exitcode))
2251	n/a
2252	n/a	with open(filename, "w") as fp:
2253	n/a	fp.write(code)
2254	n/a
2255	n/a	args = [sys.executable, filename]
2256	n/a	if use_bytes:
2257	n/a	args = [os.fsencode(a) for a in args]
2258	n/a	self.env = {os.fsencode(k): os.fsencode(v)
2259	n/a	for k, v in self.env.items()}
2260	n/a
2261	n/a	return args
2262	n/a
2263	n/a	@requires_os_func('spawnl')
2264	n/a	def test_spawnl(self):
2265	n/a	args = self.create_args()
2266	n/a	exitcode = os.spawnl(os.P_WAIT, args[0], *args)
2267	n/a	self.assertEqual(exitcode, self.exitcode)
2268	n/a
2269	n/a	@requires_os_func('spawnle')
2270	n/a	def test_spawnle(self):
2271	n/a	args = self.create_args(with_env=True)
2272	n/a	exitcode = os.spawnle(os.P_WAIT, args[0], *args, self.env)
2273	n/a	self.assertEqual(exitcode, self.exitcode)
2274	n/a
2275	n/a	@requires_os_func('spawnlp')
2276	n/a	def test_spawnlp(self):
2277	n/a	args = self.create_args()
2278	n/a	exitcode = os.spawnlp(os.P_WAIT, args[0], *args)
2279	n/a	self.assertEqual(exitcode, self.exitcode)
2280	n/a
2281	n/a	@requires_os_func('spawnlpe')
2282	n/a	def test_spawnlpe(self):
2283	n/a	args = self.create_args(with_env=True)
2284	n/a	exitcode = os.spawnlpe(os.P_WAIT, args[0], *args, self.env)
2285	n/a	self.assertEqual(exitcode, self.exitcode)
2286	n/a
2287	n/a	@requires_os_func('spawnv')
2288	n/a	def test_spawnv(self):
2289	n/a	args = self.create_args()
2290	n/a	exitcode = os.spawnv(os.P_WAIT, args[0], args)
2291	n/a	self.assertEqual(exitcode, self.exitcode)
2292	n/a
2293	n/a	@requires_os_func('spawnve')
2294	n/a	def test_spawnve(self):
2295	n/a	args = self.create_args(with_env=True)
2296	n/a	exitcode = os.spawnve(os.P_WAIT, args[0], args, self.env)
2297	n/a	self.assertEqual(exitcode, self.exitcode)
2298	n/a
2299	n/a	@requires_os_func('spawnvp')
2300	n/a	def test_spawnvp(self):
2301	n/a	args = self.create_args()
2302	n/a	exitcode = os.spawnvp(os.P_WAIT, args[0], args)
2303	n/a	self.assertEqual(exitcode, self.exitcode)
2304	n/a
2305	n/a	@requires_os_func('spawnvpe')
2306	n/a	def test_spawnvpe(self):
2307	n/a	args = self.create_args(with_env=True)
2308	n/a	exitcode = os.spawnvpe(os.P_WAIT, args[0], args, self.env)
2309	n/a	self.assertEqual(exitcode, self.exitcode)
2310	n/a
2311	n/a	@requires_os_func('spawnv')
2312	n/a	def test_nowait(self):
2313	n/a	args = self.create_args()
2314	n/a	pid = os.spawnv(os.P_NOWAIT, args[0], args)
2315	n/a	result = os.waitpid(pid, 0)
2316	n/a	self.assertEqual(result[0], pid)
2317	n/a	status = result[1]
2318	n/a	if hasattr(os, 'WIFEXITED'):
2319	n/a	self.assertTrue(os.WIFEXITED(status))
2320	n/a	self.assertEqual(os.WEXITSTATUS(status), self.exitcode)
2321	n/a	else:
2322	n/a	self.assertEqual(status, self.exitcode << 8)
2323	n/a
2324	n/a	@requires_os_func('spawnve')
2325	n/a	def test_spawnve_bytes(self):
2326	n/a	# Test bytes handling in parse_arglist and parse_envlist (#28114)
2327	n/a	args = self.create_args(with_env=True, use_bytes=True)
2328	n/a	exitcode = os.spawnve(os.P_WAIT, args[0], args, self.env)
2329	n/a	self.assertEqual(exitcode, self.exitcode)
2330	n/a
2331	n/a	@requires_os_func('spawnl')
2332	n/a	def test_spawnl_noargs(self):
2333	n/a	args = self.create_args()
2334	n/a	self.assertRaises(ValueError, os.spawnl, os.P_NOWAIT, args[0])
2335	n/a	self.assertRaises(ValueError, os.spawnl, os.P_NOWAIT, args[0], '')
2336	n/a
2337	n/a	@requires_os_func('spawnle')
2338	n/a	def test_spawnle_noargs(self):
2339	n/a	args = self.create_args()
2340	n/a	self.assertRaises(ValueError, os.spawnle, os.P_NOWAIT, args[0], {})
2341	n/a	self.assertRaises(ValueError, os.spawnle, os.P_NOWAIT, args[0], '', {})
2342	n/a
2343	n/a	@requires_os_func('spawnv')
2344	n/a	def test_spawnv_noargs(self):
2345	n/a	args = self.create_args()
2346	n/a	self.assertRaises(ValueError, os.spawnv, os.P_NOWAIT, args[0], ())
2347	n/a	self.assertRaises(ValueError, os.spawnv, os.P_NOWAIT, args[0], [])
2348	n/a	self.assertRaises(ValueError, os.spawnv, os.P_NOWAIT, args[0], ('',))
2349	n/a	self.assertRaises(ValueError, os.spawnv, os.P_NOWAIT, args[0], [''])
2350	n/a
2351	n/a	@requires_os_func('spawnve')
2352	n/a	def test_spawnve_noargs(self):
2353	n/a	args = self.create_args()
2354	n/a	self.assertRaises(ValueError, os.spawnve, os.P_NOWAIT, args[0], (), {})
2355	n/a	self.assertRaises(ValueError, os.spawnve, os.P_NOWAIT, args[0], [], {})
2356	n/a	self.assertRaises(ValueError, os.spawnve, os.P_NOWAIT, args[0], ('',), {})
2357	n/a	self.assertRaises(ValueError, os.spawnve, os.P_NOWAIT, args[0], [''], {})
2358	n/a
2359	n/a	# The introduction of this TestCase caused at least two different errors on
2360	n/a	# *nix buildbots. Temporarily skip this to let the buildbots move along.
2361	n/a	@unittest.skip("Skip due to platform/environment differences on *NIX buildbots")
2362	n/a	@unittest.skipUnless(hasattr(os, 'getlogin'), "test needs os.getlogin")
2363	n/a	class LoginTests(unittest.TestCase):
2364	n/a	def test_getlogin(self):
2365	n/a	user_name = os.getlogin()
2366	n/a	self.assertNotEqual(len(user_name), 0)
2367	n/a
2368	n/a
2369	n/a	@unittest.skipUnless(hasattr(os, 'getpriority') and hasattr(os, 'setpriority'),
2370	n/a	"needs os.getpriority and os.setpriority")
2371	n/a	class ProgramPriorityTests(unittest.TestCase):
2372	n/a	"""Tests for os.getpriority() and os.setpriority()."""
2373	n/a
2374	n/a	def test_set_get_priority(self):
2375	n/a
2376	n/a	base = os.getpriority(os.PRIO_PROCESS, os.getpid())
2377	n/a	os.setpriority(os.PRIO_PROCESS, os.getpid(), base + 1)
2378	n/a	try:
2379	n/a	new_prio = os.getpriority(os.PRIO_PROCESS, os.getpid())
2380	n/a	if base >= 19 and new_prio <= 19:
2381	n/a	raise unittest.SkipTest("unable to reliably test setpriority "
2382	n/a	"at current nice level of %s" % base)
2383	n/a	else:
2384	n/a	self.assertEqual(new_prio, base + 1)
2385	n/a	finally:
2386	n/a	try:
2387	n/a	os.setpriority(os.PRIO_PROCESS, os.getpid(), base)
2388	n/a	except OSError as err:
2389	n/a	if err.errno != errno.EACCES:
2390	n/a	raise
2391	n/a
2392	n/a
2393	n/a	if threading is not None:
2394	n/a	class SendfileTestServer(asyncore.dispatcher, threading.Thread):
2395	n/a
2396	n/a	class Handler(asynchat.async_chat):
2397	n/a
2398	n/a	def __init__(self, conn):
2399	n/a	asynchat.async_chat.__init__(self, conn)
2400	n/a	self.in_buffer = []
2401	n/a	self.closed = False
2402	n/a	self.push(b"220 ready\r\n")
2403	n/a
2404	n/a	def handle_read(self):
2405	n/a	data = self.recv(4096)
2406	n/a	self.in_buffer.append(data)
2407	n/a
2408	n/a	def get_data(self):
2409	n/a	return b''.join(self.in_buffer)
2410	n/a
2411	n/a	def handle_close(self):
2412	n/a	self.close()
2413	n/a	self.closed = True
2414	n/a
2415	n/a	def handle_error(self):
2416	n/a	raise
2417	n/a
2418	n/a	def __init__(self, address):
2419	n/a	threading.Thread.__init__(self)
2420	n/a	asyncore.dispatcher.__init__(self)
2421	n/a	self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
2422	n/a	self.bind(address)
2423	n/a	self.listen(5)
2424	n/a	self.host, self.port = self.socket.getsockname()[:2]
2425	n/a	self.handler_instance = None
2426	n/a	self._active = False
2427	n/a	self._active_lock = threading.Lock()
2428	n/a
2429	n/a	# --- public API
2430	n/a
2431	n/a	@property
2432	n/a	def running(self):
2433	n/a	return self._active
2434	n/a
2435	n/a	def start(self):
2436	n/a	assert not self.running
2437	n/a	self.__flag = threading.Event()
2438	n/a	threading.Thread.start(self)
2439	n/a	self.__flag.wait()
2440	n/a
2441	n/a	def stop(self):
2442	n/a	assert self.running
2443	n/a	self._active = False
2444	n/a	self.join()
2445	n/a
2446	n/a	def wait(self):
2447	n/a	# wait for handler connection to be closed, then stop the server
2448	n/a	while not getattr(self.handler_instance, "closed", False):
2449	n/a	time.sleep(0.001)
2450	n/a	self.stop()
2451	n/a
2452	n/a	# --- internals
2453	n/a
2454	n/a	def run(self):
2455	n/a	self._active = True
2456	n/a	self.__flag.set()
2457	n/a	while self._active and asyncore.socket_map:
2458	n/a	self._active_lock.acquire()
2459	n/a	asyncore.loop(timeout=0.001, count=1)
2460	n/a	self._active_lock.release()
2461	n/a	asyncore.close_all()
2462	n/a
2463	n/a	def handle_accept(self):
2464	n/a	conn, addr = self.accept()
2465	n/a	self.handler_instance = self.Handler(conn)
2466	n/a
2467	n/a	def handle_connect(self):
2468	n/a	self.close()
2469	n/a	handle_read = handle_connect
2470	n/a
2471	n/a	def writable(self):
2472	n/a	return 0
2473	n/a
2474	n/a	def handle_error(self):
2475	n/a	raise
2476	n/a
2477	n/a
2478	n/a	@unittest.skipUnless(threading is not None, "test needs threading module")
2479	n/a	@unittest.skipUnless(hasattr(os, 'sendfile'), "test needs os.sendfile()")
2480	n/a	class TestSendfile(unittest.TestCase):
2481	n/a
2482	n/a	DATA = b"12345abcde" * 16 * 1024 # 160 KB
2483	n/a	SUPPORT_HEADERS_TRAILERS = not sys.platform.startswith("linux") and \
2484	n/a	not sys.platform.startswith("solaris") and \
2485	n/a	not sys.platform.startswith("sunos")
2486	n/a	requires_headers_trailers = unittest.skipUnless(SUPPORT_HEADERS_TRAILERS,
2487	n/a	'requires headers and trailers support')
2488	n/a
2489	n/a	@classmethod
2490	n/a	def setUpClass(cls):
2491	n/a	cls.key = support.threading_setup()
2492	n/a	create_file(support.TESTFN, cls.DATA)
2493	n/a
2494	n/a	@classmethod
2495	n/a	def tearDownClass(cls):
2496	n/a	support.threading_cleanup(*cls.key)
2497	n/a	support.unlink(support.TESTFN)
2498	n/a
2499	n/a	def setUp(self):
2500	n/a	self.server = SendfileTestServer((support.HOST, 0))
2501	n/a	self.server.start()
2502	n/a	self.client = socket.socket()
2503	n/a	self.client.connect((self.server.host, self.server.port))
2504	n/a	self.client.settimeout(1)
2505	n/a	# synchronize by waiting for "220 ready" response
2506	n/a	self.client.recv(1024)
2507	n/a	self.sockno = self.client.fileno()
2508	n/a	self.file = open(support.TESTFN, 'rb')
2509	n/a	self.fileno = self.file.fileno()
2510	n/a
2511	n/a	def tearDown(self):
2512	n/a	self.file.close()
2513	n/a	self.client.close()
2514	n/a	if self.server.running:
2515	n/a	self.server.stop()
2516	n/a
2517	n/a	def sendfile_wrapper(self, sock, file, offset, nbytes, headers=[], trailers=[]):
2518	n/a	"""A higher level wrapper representing how an application is
2519	n/a	supposed to use sendfile().
2520	n/a	"""
2521	n/a	while 1:
2522	n/a	try:
2523	n/a	if self.SUPPORT_HEADERS_TRAILERS:
2524	n/a	return os.sendfile(sock, file, offset, nbytes, headers,
2525	n/a	trailers)
2526	n/a	else:
2527	n/a	return os.sendfile(sock, file, offset, nbytes)
2528	n/a	except OSError as err:
2529	n/a	if err.errno == errno.ECONNRESET:
2530	n/a	# disconnected
2531	n/a	raise
2532	n/a	elif err.errno in (errno.EAGAIN, errno.EBUSY):
2533	n/a	# we have to retry send data
2534	n/a	continue
2535	n/a	else:
2536	n/a	raise
2537	n/a
2538	n/a	def test_send_whole_file(self):
2539	n/a	# normal send
2540	n/a	total_sent = 0
2541	n/a	offset = 0
2542	n/a	nbytes = 4096
2543	n/a	while total_sent < len(self.DATA):
2544	n/a	sent = self.sendfile_wrapper(self.sockno, self.fileno, offset, nbytes)
2545	n/a	if sent == 0:
2546	n/a	break
2547	n/a	offset += sent
2548	n/a	total_sent += sent
2549	n/a	self.assertTrue(sent <= nbytes)
2550	n/a	self.assertEqual(offset, total_sent)
2551	n/a
2552	n/a	self.assertEqual(total_sent, len(self.DATA))
2553	n/a	self.client.shutdown(socket.SHUT_RDWR)
2554	n/a	self.client.close()
2555	n/a	self.server.wait()
2556	n/a	data = self.server.handler_instance.get_data()
2557	n/a	self.assertEqual(len(data), len(self.DATA))
2558	n/a	self.assertEqual(data, self.DATA)
2559	n/a
2560	n/a	def test_send_at_certain_offset(self):
2561	n/a	# start sending a file at a certain offset
2562	n/a	total_sent = 0
2563	n/a	offset = len(self.DATA) // 2
2564	n/a	must_send = len(self.DATA) - offset
2565	n/a	nbytes = 4096
2566	n/a	while total_sent < must_send:
2567	n/a	sent = self.sendfile_wrapper(self.sockno, self.fileno, offset, nbytes)
2568	n/a	if sent == 0:
2569	n/a	break
2570	n/a	offset += sent
2571	n/a	total_sent += sent
2572	n/a	self.assertTrue(sent <= nbytes)
2573	n/a
2574	n/a	self.client.shutdown(socket.SHUT_RDWR)
2575	n/a	self.client.close()
2576	n/a	self.server.wait()
2577	n/a	data = self.server.handler_instance.get_data()
2578	n/a	expected = self.DATA[len(self.DATA) // 2:]
2579	n/a	self.assertEqual(total_sent, len(expected))
2580	n/a	self.assertEqual(len(data), len(expected))
2581	n/a	self.assertEqual(data, expected)
2582	n/a
2583	n/a	def test_offset_overflow(self):
2584	n/a	# specify an offset > file size
2585	n/a	offset = len(self.DATA) + 4096
2586	n/a	try:
2587	n/a	sent = os.sendfile(self.sockno, self.fileno, offset, 4096)
2588	n/a	except OSError as e:
2589	n/a	# Solaris can raise EINVAL if offset >= file length, ignore.
2590	n/a	if e.errno != errno.EINVAL:
2591	n/a	raise
2592	n/a	else:
2593	n/a	self.assertEqual(sent, 0)
2594	n/a	self.client.shutdown(socket.SHUT_RDWR)
2595	n/a	self.client.close()
2596	n/a	self.server.wait()
2597	n/a	data = self.server.handler_instance.get_data()
2598	n/a	self.assertEqual(data, b'')
2599	n/a
2600	n/a	def test_invalid_offset(self):
2601	n/a	with self.assertRaises(OSError) as cm:
2602	n/a	os.sendfile(self.sockno, self.fileno, -1, 4096)
2603	n/a	self.assertEqual(cm.exception.errno, errno.EINVAL)
2604	n/a
2605	n/a	def test_keywords(self):
2606	n/a	# Keyword arguments should be supported
2607	n/a	os.sendfile(out=self.sockno, offset=0, count=4096,
2608	n/a	**{'in': self.fileno})
2609	n/a	if self.SUPPORT_HEADERS_TRAILERS:
2610	n/a	os.sendfile(self.sockno, self.fileno, offset=0, count=4096,
2611	n/a	headers=(), trailers=(), flags=0)
2612	n/a
2613	n/a	# --- headers / trailers tests
2614	n/a
2615	n/a	@requires_headers_trailers
2616	n/a	def test_headers(self):
2617	n/a	total_sent = 0
2618	n/a	sent = os.sendfile(self.sockno, self.fileno, 0, 4096,
2619	n/a	headers=[b"x" * 512])
2620	n/a	total_sent += sent
2621	n/a	offset = 4096
2622	n/a	nbytes = 4096
2623	n/a	while 1:
2624	n/a	sent = self.sendfile_wrapper(self.sockno, self.fileno,
2625	n/a	offset, nbytes)
2626	n/a	if sent == 0:
2627	n/a	break
2628	n/a	total_sent += sent
2629	n/a	offset += sent
2630	n/a
2631	n/a	expected_data = b"x" * 512 + self.DATA
2632	n/a	self.assertEqual(total_sent, len(expected_data))
2633	n/a	self.client.close()
2634	n/a	self.server.wait()
2635	n/a	data = self.server.handler_instance.get_data()
2636	n/a	self.assertEqual(hash(data), hash(expected_data))
2637	n/a
2638	n/a	@requires_headers_trailers
2639	n/a	def test_trailers(self):
2640	n/a	TESTFN2 = support.TESTFN + "2"
2641	n/a	file_data = b"abcdef"
2642	n/a
2643	n/a	self.addCleanup(support.unlink, TESTFN2)
2644	n/a	create_file(TESTFN2, file_data)
2645	n/a
2646	n/a	with open(TESTFN2, 'rb') as f:
2647	n/a	os.sendfile(self.sockno, f.fileno(), 0, len(file_data),
2648	n/a	trailers=[b"1234"])
2649	n/a	self.client.close()
2650	n/a	self.server.wait()
2651	n/a	data = self.server.handler_instance.get_data()
2652	n/a	self.assertEqual(data, b"abcdef1234")
2653	n/a
2654	n/a	@requires_headers_trailers
2655	n/a	@unittest.skipUnless(hasattr(os, 'SF_NODISKIO'),
2656	n/a	'test needs os.SF_NODISKIO')
2657	n/a	def test_flags(self):
2658	n/a	try:
2659	n/a	os.sendfile(self.sockno, self.fileno, 0, 4096,
2660	n/a	flags=os.SF_NODISKIO)
2661	n/a	except OSError as err:
2662	n/a	if err.errno not in (errno.EBUSY, errno.EAGAIN):
2663	n/a	raise
2664	n/a
2665	n/a
2666	n/a	def supports_extended_attributes():
2667	n/a	if not hasattr(os, "setxattr"):
2668	n/a	return False
2669	n/a
2670	n/a	try:
2671	n/a	with open(support.TESTFN, "xb", 0) as fp:
2672	n/a	try:
2673	n/a	os.setxattr(fp.fileno(), b"user.test", b"")
2674	n/a	except OSError:
2675	n/a	return False
2676	n/a	finally:
2677	n/a	support.unlink(support.TESTFN)
2678	n/a
2679	n/a	return True
2680	n/a
2681	n/a
2682	n/a	@unittest.skipUnless(supports_extended_attributes(),
2683	n/a	"no non-broken extended attribute support")
2684	n/a	# Kernels < 2.6.39 don't respect setxattr flags.
2685	n/a	@support.requires_linux_version(2, 6, 39)
2686	n/a	class ExtendedAttributeTests(unittest.TestCase):
2687	n/a
2688	n/a	def _check_xattrs_str(self, s, getxattr, setxattr, removexattr, listxattr, **kwargs):
2689	n/a	fn = support.TESTFN
2690	n/a	self.addCleanup(support.unlink, fn)
2691	n/a	create_file(fn)
2692	n/a
2693	n/a	with self.assertRaises(OSError) as cm:
2694	n/a	getxattr(fn, s("user.test"), **kwargs)
2695	n/a	self.assertEqual(cm.exception.errno, errno.ENODATA)
2696	n/a
2697	n/a	init_xattr = listxattr(fn)
2698	n/a	self.assertIsInstance(init_xattr, list)
2699	n/a
2700	n/a	setxattr(fn, s("user.test"), b"", **kwargs)
2701	n/a	xattr = set(init_xattr)
2702	n/a	xattr.add("user.test")
2703	n/a	self.assertEqual(set(listxattr(fn)), xattr)
2704	n/a	self.assertEqual(getxattr(fn, b"user.test", **kwargs), b"")
2705	n/a	setxattr(fn, s("user.test"), b"hello", os.XATTR_REPLACE, **kwargs)
2706	n/a	self.assertEqual(getxattr(fn, b"user.test", **kwargs), b"hello")
2707	n/a
2708	n/a	with self.assertRaises(OSError) as cm:
2709	n/a	setxattr(fn, s("user.test"), b"bye", os.XATTR_CREATE, **kwargs)
2710	n/a	self.assertEqual(cm.exception.errno, errno.EEXIST)
2711	n/a
2712	n/a	with self.assertRaises(OSError) as cm:
2713	n/a	setxattr(fn, s("user.test2"), b"bye", os.XATTR_REPLACE, **kwargs)
2714	n/a	self.assertEqual(cm.exception.errno, errno.ENODATA)
2715	n/a
2716	n/a	setxattr(fn, s("user.test2"), b"foo", os.XATTR_CREATE, **kwargs)
2717	n/a	xattr.add("user.test2")
2718	n/a	self.assertEqual(set(listxattr(fn)), xattr)
2719	n/a	removexattr(fn, s("user.test"), **kwargs)
2720	n/a
2721	n/a	with self.assertRaises(OSError) as cm:
2722	n/a	getxattr(fn, s("user.test"), **kwargs)
2723	n/a	self.assertEqual(cm.exception.errno, errno.ENODATA)
2724	n/a
2725	n/a	xattr.remove("user.test")
2726	n/a	self.assertEqual(set(listxattr(fn)), xattr)
2727	n/a	self.assertEqual(getxattr(fn, s("user.test2"), **kwargs), b"foo")
2728	n/a	setxattr(fn, s("user.test"), b"a"1024, *kwargs)
2729	n/a	self.assertEqual(getxattr(fn, s("user.test"), *kwargs), b"a"1024)
2730	n/a	removexattr(fn, s("user.test"), **kwargs)
2731	n/a	many = sorted("user.test{}".format(i) for i in range(100))
2732	n/a	for thing in many:
2733	n/a	setxattr(fn, thing, b"x", **kwargs)
2734	n/a	self.assertEqual(set(listxattr(fn)), set(init_xattr) \| set(many))
2735	n/a
2736	n/a	def _check_xattrs(self, args, *kwargs):
2737	n/a	self._check_xattrs_str(str, args, *kwargs)
2738	n/a	support.unlink(support.TESTFN)
2739	n/a
2740	n/a	self._check_xattrs_str(os.fsencode, args, *kwargs)
2741	n/a	support.unlink(support.TESTFN)
2742	n/a
2743	n/a	def test_simple(self):
2744	n/a	self._check_xattrs(os.getxattr, os.setxattr, os.removexattr,
2745	n/a	os.listxattr)
2746	n/a
2747	n/a	def test_lpath(self):
2748	n/a	self._check_xattrs(os.getxattr, os.setxattr, os.removexattr,
2749	n/a	os.listxattr, follow_symlinks=False)
2750	n/a
2751	n/a	def test_fds(self):
2752	n/a	def getxattr(path, *args):
2753	n/a	with open(path, "rb") as fp:
2754	n/a	return os.getxattr(fp.fileno(), *args)
2755	n/a	def setxattr(path, *args):
2756	n/a	with open(path, "wb", 0) as fp:
2757	n/a	os.setxattr(fp.fileno(), *args)
2758	n/a	def removexattr(path, *args):
2759	n/a	with open(path, "wb", 0) as fp:
2760	n/a	os.removexattr(fp.fileno(), *args)
2761	n/a	def listxattr(path, *args):
2762	n/a	with open(path, "rb") as fp:
2763	n/a	return os.listxattr(fp.fileno(), *args)
2764	n/a	self._check_xattrs(getxattr, setxattr, removexattr, listxattr)
2765	n/a
2766	n/a
2767	n/a	@unittest.skipUnless(hasattr(os, 'get_terminal_size'), "requires os.get_terminal_size")
2768	n/a	class TermsizeTests(unittest.TestCase):
2769	n/a	def test_does_not_crash(self):
2770	n/a	"""Check if get_terminal_size() returns a meaningful value.
2771	n/a
2772	n/a	There's no easy portable way to actually check the size of the
2773	n/a	terminal, so let's check if it returns something sensible instead.
2774	n/a	"""
2775	n/a	try:
2776	n/a	size = os.get_terminal_size()
2777	n/a	except OSError as e:
2778	n/a	if sys.platform == "win32" or e.errno in (errno.EINVAL, errno.ENOTTY):
2779	n/a	# Under win32 a generic OSError can be thrown if the
2780	n/a	# handle cannot be retrieved
2781	n/a	self.skipTest("failed to query terminal size")
2782	n/a	raise
2783	n/a
2784	n/a	self.assertGreaterEqual(size.columns, 0)
2785	n/a	self.assertGreaterEqual(size.lines, 0)
2786	n/a
2787	n/a	def test_stty_match(self):
2788	n/a	"""Check if stty returns the same results
2789	n/a
2790	n/a	stty actually tests stdin, so get_terminal_size is invoked on
2791	n/a	stdin explicitly. If stty succeeded, then get_terminal_size()
2792	n/a	should work too.
2793	n/a	"""
2794	n/a	try:
2795	n/a	size = subprocess.check_output(['stty', 'size']).decode().split()
2796	n/a	except (FileNotFoundError, subprocess.CalledProcessError):
2797	n/a	self.skipTest("stty invocation failed")
2798	n/a	expected = (int(size[1]), int(size[0])) # reversed order
2799	n/a
2800	n/a	try:
2801	n/a	actual = os.get_terminal_size(sys.__stdin__.fileno())
2802	n/a	except OSError as e:
2803	n/a	if sys.platform == "win32" or e.errno in (errno.EINVAL, errno.ENOTTY):
2804	n/a	# Under win32 a generic OSError can be thrown if the
2805	n/a	# handle cannot be retrieved
2806	n/a	self.skipTest("failed to query terminal size")
2807	n/a	raise
2808	n/a	self.assertEqual(expected, actual)
2809	n/a
2810	n/a
2811	n/a	class OSErrorTests(unittest.TestCase):
2812	n/a	def setUp(self):
2813	n/a	class Str(str):
2814	n/a	pass
2815	n/a
2816	n/a	self.bytes_filenames = []
2817	n/a	self.unicode_filenames = []
2818	n/a	if support.TESTFN_UNENCODABLE is not None:
2819	n/a	decoded = support.TESTFN_UNENCODABLE
2820	n/a	else:
2821	n/a	decoded = support.TESTFN
2822	n/a	self.unicode_filenames.append(decoded)
2823	n/a	self.unicode_filenames.append(Str(decoded))
2824	n/a	if support.TESTFN_UNDECODABLE is not None:
2825	n/a	encoded = support.TESTFN_UNDECODABLE
2826	n/a	else:
2827	n/a	encoded = os.fsencode(support.TESTFN)
2828	n/a	self.bytes_filenames.append(encoded)
2829	n/a	self.bytes_filenames.append(bytearray(encoded))
2830	n/a	self.bytes_filenames.append(memoryview(encoded))
2831	n/a
2832	n/a	self.filenames = self.bytes_filenames + self.unicode_filenames
2833	n/a
2834	n/a	def test_oserror_filename(self):
2835	n/a	funcs = [
2836	n/a	(self.filenames, os.chdir,),
2837	n/a	(self.filenames, os.chmod, 0o777),
2838	n/a	(self.filenames, os.lstat,),
2839	n/a	(self.filenames, os.open, os.O_RDONLY),
2840	n/a	(self.filenames, os.rmdir,),
2841	n/a	(self.filenames, os.stat,),
2842	n/a	(self.filenames, os.unlink,),
2843	n/a	]
2844	n/a	if sys.platform == "win32":
2845	n/a	funcs.extend((
2846	n/a	(self.bytes_filenames, os.rename, b"dst"),
2847	n/a	(self.bytes_filenames, os.replace, b"dst"),
2848	n/a	(self.unicode_filenames, os.rename, "dst"),
2849	n/a	(self.unicode_filenames, os.replace, "dst"),
2850	n/a	(self.unicode_filenames, os.listdir, ),
2851	n/a	))
2852	n/a	else:
2853	n/a	funcs.extend((
2854	n/a	(self.filenames, os.listdir,),
2855	n/a	(self.filenames, os.rename, "dst"),
2856	n/a	(self.filenames, os.replace, "dst"),
2857	n/a	))
2858	n/a	if hasattr(os, "chown"):
2859	n/a	funcs.append((self.filenames, os.chown, 0, 0))
2860	n/a	if hasattr(os, "lchown"):
2861	n/a	funcs.append((self.filenames, os.lchown, 0, 0))
2862	n/a	if hasattr(os, "truncate"):
2863	n/a	funcs.append((self.filenames, os.truncate, 0))
2864	n/a	if hasattr(os, "chflags"):
2865	n/a	funcs.append((self.filenames, os.chflags, 0))
2866	n/a	if hasattr(os, "lchflags"):
2867	n/a	funcs.append((self.filenames, os.lchflags, 0))
2868	n/a	if hasattr(os, "chroot"):
2869	n/a	funcs.append((self.filenames, os.chroot,))
2870	n/a	if hasattr(os, "link"):
2871	n/a	if sys.platform == "win32":
2872	n/a	funcs.append((self.bytes_filenames, os.link, b"dst"))
2873	n/a	funcs.append((self.unicode_filenames, os.link, "dst"))
2874	n/a	else:
2875	n/a	funcs.append((self.filenames, os.link, "dst"))
2876	n/a	if hasattr(os, "listxattr"):
2877	n/a	funcs.extend((
2878	n/a	(self.filenames, os.listxattr,),
2879	n/a	(self.filenames, os.getxattr, "user.test"),
2880	n/a	(self.filenames, os.setxattr, "user.test", b'user'),
2881	n/a	(self.filenames, os.removexattr, "user.test"),
2882	n/a	))
2883	n/a	if hasattr(os, "lchmod"):
2884	n/a	funcs.append((self.filenames, os.lchmod, 0o777))
2885	n/a	if hasattr(os, "readlink"):
2886	n/a	if sys.platform == "win32":
2887	n/a	funcs.append((self.unicode_filenames, os.readlink,))
2888	n/a	else:
2889	n/a	funcs.append((self.filenames, os.readlink,))
2890	n/a
2891	n/a
2892	n/a	for filenames, func, *func_args in funcs:
2893	n/a	for name in filenames:
2894	n/a	try:
2895	n/a	if isinstance(name, (str, bytes)):
2896	n/a	func(name, *func_args)
2897	n/a	else:
2898	n/a	with self.assertWarnsRegex(DeprecationWarning, 'should be'):
2899	n/a	func(name, *func_args)
2900	n/a	except OSError as err:
2901	n/a	self.assertIs(err.filename, name, str(func))
2902	n/a	except UnicodeDecodeError:
2903	n/a	pass
2904	n/a	else:
2905	n/a	self.fail("No exception thrown by {}".format(func))
2906	n/a
2907	n/a	class CPUCountTests(unittest.TestCase):
2908	n/a	def test_cpu_count(self):
2909	n/a	cpus = os.cpu_count()
2910	n/a	if cpus is not None:
2911	n/a	self.assertIsInstance(cpus, int)
2912	n/a	self.assertGreater(cpus, 0)
2913	n/a	else:
2914	n/a	self.skipTest("Could not determine the number of CPUs")
2915	n/a
2916	n/a
2917	n/a	class FDInheritanceTests(unittest.TestCase):
2918	n/a	def test_get_set_inheritable(self):
2919	n/a	fd = os.open(__file__, os.O_RDONLY)
2920	n/a	self.addCleanup(os.close, fd)
2921	n/a	self.assertEqual(os.get_inheritable(fd), False)
2922	n/a
2923	n/a	os.set_inheritable(fd, True)
2924	n/a	self.assertEqual(os.get_inheritable(fd), True)
2925	n/a
2926	n/a	@unittest.skipIf(fcntl is None, "need fcntl")
2927	n/a	def test_get_inheritable_cloexec(self):
2928	n/a	fd = os.open(__file__, os.O_RDONLY)
2929	n/a	self.addCleanup(os.close, fd)
2930	n/a	self.assertEqual(os.get_inheritable(fd), False)
2931	n/a
2932	n/a	# clear FD_CLOEXEC flag
2933	n/a	flags = fcntl.fcntl(fd, fcntl.F_GETFD)
2934	n/a	flags &= ~fcntl.FD_CLOEXEC
2935	n/a	fcntl.fcntl(fd, fcntl.F_SETFD, flags)
2936	n/a
2937	n/a	self.assertEqual(os.get_inheritable(fd), True)
2938	n/a
2939	n/a	@unittest.skipIf(fcntl is None, "need fcntl")
2940	n/a	def test_set_inheritable_cloexec(self):
2941	n/a	fd = os.open(__file__, os.O_RDONLY)
2942	n/a	self.addCleanup(os.close, fd)
2943	n/a	self.assertEqual(fcntl.fcntl(fd, fcntl.F_GETFD) & fcntl.FD_CLOEXEC,
2944	n/a	fcntl.FD_CLOEXEC)
2945	n/a
2946	n/a	os.set_inheritable(fd, True)
2947	n/a	self.assertEqual(fcntl.fcntl(fd, fcntl.F_GETFD) & fcntl.FD_CLOEXEC,
2948	n/a	0)
2949	n/a
2950	n/a	def test_open(self):
2951	n/a	fd = os.open(__file__, os.O_RDONLY)
2952	n/a	self.addCleanup(os.close, fd)
2953	n/a	self.assertEqual(os.get_inheritable(fd), False)
2954	n/a
2955	n/a	@unittest.skipUnless(hasattr(os, 'pipe'), "need os.pipe()")
2956	n/a	def test_pipe(self):
2957	n/a	rfd, wfd = os.pipe()
2958	n/a	self.addCleanup(os.close, rfd)
2959	n/a	self.addCleanup(os.close, wfd)
2960	n/a	self.assertEqual(os.get_inheritable(rfd), False)
2961	n/a	self.assertEqual(os.get_inheritable(wfd), False)
2962	n/a
2963	n/a	def test_dup(self):
2964	n/a	fd1 = os.open(__file__, os.O_RDONLY)
2965	n/a	self.addCleanup(os.close, fd1)
2966	n/a
2967	n/a	fd2 = os.dup(fd1)
2968	n/a	self.addCleanup(os.close, fd2)
2969	n/a	self.assertEqual(os.get_inheritable(fd2), False)
2970	n/a
2971	n/a	@unittest.skipUnless(hasattr(os, 'dup2'), "need os.dup2()")
2972	n/a	def test_dup2(self):
2973	n/a	fd = os.open(__file__, os.O_RDONLY)
2974	n/a	self.addCleanup(os.close, fd)
2975	n/a
2976	n/a	# inheritable by default
2977	n/a	fd2 = os.open(__file__, os.O_RDONLY)
2978	n/a	try:
2979	n/a	os.dup2(fd, fd2)
2980	n/a	self.assertEqual(os.get_inheritable(fd2), True)
2981	n/a	finally:
2982	n/a	os.close(fd2)
2983	n/a
2984	n/a	# force non-inheritable
2985	n/a	fd3 = os.open(__file__, os.O_RDONLY)
2986	n/a	try:
2987	n/a	os.dup2(fd, fd3, inheritable=False)
2988	n/a	self.assertEqual(os.get_inheritable(fd3), False)
2989	n/a	finally:
2990	n/a	os.close(fd3)
2991	n/a
2992	n/a	@unittest.skipUnless(hasattr(os, 'openpty'), "need os.openpty()")
2993	n/a	def test_openpty(self):
2994	n/a	master_fd, slave_fd = os.openpty()
2995	n/a	self.addCleanup(os.close, master_fd)
2996	n/a	self.addCleanup(os.close, slave_fd)
2997	n/a	self.assertEqual(os.get_inheritable(master_fd), False)
2998	n/a	self.assertEqual(os.get_inheritable(slave_fd), False)
2999	n/a
3000	n/a
3001	n/a	class PathTConverterTests(unittest.TestCase):
3002	n/a	# tuples of (function name, allows fd arguments, additional arguments to
3003	n/a	# function, cleanup function)
3004	n/a	functions = [
3005	n/a	('stat', True, (), None),
3006	n/a	('lstat', False, (), None),
3007	n/a	('access', False, (os.F_OK,), None),
3008	n/a	('chflags', False, (0,), None),
3009	n/a	('lchflags', False, (0,), None),
3010	n/a	('open', False, (0,), getattr(os, 'close', None)),
3011	n/a	]
3012	n/a
3013	n/a	def test_path_t_converter(self):
3014	n/a	str_filename = support.TESTFN
3015	n/a	if os.name == 'nt':
3016	n/a	bytes_fspath = bytes_filename = None
3017	n/a	else:
3018	n/a	bytes_filename = support.TESTFN.encode('ascii')
3019	n/a	bytes_fspath = _PathLike(bytes_filename)
3020	n/a	fd = os.open(_PathLike(str_filename), os.O_WRONLY\|os.O_CREAT)
3021	n/a	self.addCleanup(support.unlink, support.TESTFN)
3022	n/a	self.addCleanup(os.close, fd)
3023	n/a
3024	n/a	int_fspath = _PathLike(fd)
3025	n/a	str_fspath = _PathLike(str_filename)
3026	n/a
3027	n/a	for name, allow_fd, extra_args, cleanup_fn in self.functions:
3028	n/a	with self.subTest(name=name):
3029	n/a	try:
3030	n/a	fn = getattr(os, name)
3031	n/a	except AttributeError:
3032	n/a	continue
3033	n/a
3034	n/a	for path in (str_filename, bytes_filename, str_fspath,
3035	n/a	bytes_fspath):
3036	n/a	if path is None:
3037	n/a	continue
3038	n/a	with self.subTest(name=name, path=path):
3039	n/a	result = fn(path, *extra_args)
3040	n/a	if cleanup_fn is not None:
3041	n/a	cleanup_fn(result)
3042	n/a
3043	n/a	with self.assertRaisesRegex(
3044	n/a	TypeError, 'should be string, bytes'):
3045	n/a	fn(int_fspath, *extra_args)
3046	n/a
3047	n/a	if allow_fd:
3048	n/a	result = fn(fd, *extra_args) # should not fail
3049	n/a	if cleanup_fn is not None:
3050	n/a	cleanup_fn(result)
3051	n/a	else:
3052	n/a	with self.assertRaisesRegex(
3053	n/a	TypeError,
3054	n/a	'os.PathLike'):
3055	n/a	fn(fd, *extra_args)
3056	n/a
3057	n/a
3058	n/a	@unittest.skipUnless(hasattr(os, 'get_blocking'),
3059	n/a	'needs os.get_blocking() and os.set_blocking()')
3060	n/a	class BlockingTests(unittest.TestCase):
3061	n/a	def test_blocking(self):
3062	n/a	fd = os.open(__file__, os.O_RDONLY)
3063	n/a	self.addCleanup(os.close, fd)
3064	n/a	self.assertEqual(os.get_blocking(fd), True)
3065	n/a
3066	n/a	os.set_blocking(fd, False)
3067	n/a	self.assertEqual(os.get_blocking(fd), False)
3068	n/a
3069	n/a	os.set_blocking(fd, True)
3070	n/a	self.assertEqual(os.get_blocking(fd), True)
3071	n/a
3072	n/a
3073	n/a
3074	n/a	class ExportsTests(unittest.TestCase):
3075	n/a	def test_os_all(self):
3076	n/a	self.assertIn('open', os.__all__)
3077	n/a	self.assertIn('walk', os.__all__)
3078	n/a
3079	n/a
3080	n/a	class TestScandir(unittest.TestCase):
3081	n/a	check_no_resource_warning = support.check_no_resource_warning
3082	n/a
3083	n/a	def setUp(self):
3084	n/a	self.path = os.path.realpath(support.TESTFN)
3085	n/a	self.bytes_path = os.fsencode(self.path)
3086	n/a	self.addCleanup(support.rmtree, self.path)
3087	n/a	os.mkdir(self.path)
3088	n/a
3089	n/a	def create_file(self, name="file.txt"):
3090	n/a	path = self.bytes_path if isinstance(name, bytes) else self.path
3091	n/a	filename = os.path.join(path, name)
3092	n/a	create_file(filename, b'python')
3093	n/a	return filename
3094	n/a
3095	n/a	def get_entries(self, names):
3096	n/a	entries = dict((entry.name, entry)
3097	n/a	for entry in os.scandir(self.path))
3098	n/a	self.assertEqual(sorted(entries.keys()), names)
3099	n/a	return entries
3100	n/a
3101	n/a	def assert_stat_equal(self, stat1, stat2, skip_fields):
3102	n/a	if skip_fields:
3103	n/a	for attr in dir(stat1):
3104	n/a	if not attr.startswith("st_"):
3105	n/a	continue
3106	n/a	if attr in ("st_dev", "st_ino", "st_nlink"):
3107	n/a	continue
3108	n/a	self.assertEqual(getattr(stat1, attr),
3109	n/a	getattr(stat2, attr),
3110	n/a	(stat1, stat2, attr))
3111	n/a	else:
3112	n/a	self.assertEqual(stat1, stat2)
3113	n/a
3114	n/a	def check_entry(self, entry, name, is_dir, is_file, is_symlink):
3115	n/a	self.assertIsInstance(entry, os.DirEntry)
3116	n/a	self.assertEqual(entry.name, name)
3117	n/a	self.assertEqual(entry.path, os.path.join(self.path, name))
3118	n/a	self.assertEqual(entry.inode(),
3119	n/a	os.stat(entry.path, follow_symlinks=False).st_ino)
3120	n/a
3121	n/a	entry_stat = os.stat(entry.path)
3122	n/a	self.assertEqual(entry.is_dir(),
3123	n/a	stat.S_ISDIR(entry_stat.st_mode))
3124	n/a	self.assertEqual(entry.is_file(),
3125	n/a	stat.S_ISREG(entry_stat.st_mode))
3126	n/a	self.assertEqual(entry.is_symlink(),
3127	n/a	os.path.islink(entry.path))
3128	n/a
3129	n/a	entry_lstat = os.stat(entry.path, follow_symlinks=False)
3130	n/a	self.assertEqual(entry.is_dir(follow_symlinks=False),
3131	n/a	stat.S_ISDIR(entry_lstat.st_mode))
3132	n/a	self.assertEqual(entry.is_file(follow_symlinks=False),
3133	n/a	stat.S_ISREG(entry_lstat.st_mode))
3134	n/a
3135	n/a	self.assert_stat_equal(entry.stat(),
3136	n/a	entry_stat,
3137	n/a	os.name == 'nt' and not is_symlink)
3138	n/a	self.assert_stat_equal(entry.stat(follow_symlinks=False),
3139	n/a	entry_lstat,
3140	n/a	os.name == 'nt')
3141	n/a
3142	n/a	def test_attributes(self):
3143	n/a	link = hasattr(os, 'link')
3144	n/a	symlink = support.can_symlink()
3145	n/a
3146	n/a	dirname = os.path.join(self.path, "dir")
3147	n/a	os.mkdir(dirname)
3148	n/a	filename = self.create_file("file.txt")
3149	n/a	if link:
3150	n/a	os.link(filename, os.path.join(self.path, "link_file.txt"))
3151	n/a	if symlink:
3152	n/a	os.symlink(dirname, os.path.join(self.path, "symlink_dir"),
3153	n/a	target_is_directory=True)
3154	n/a	os.symlink(filename, os.path.join(self.path, "symlink_file.txt"))
3155	n/a
3156	n/a	names = ['dir', 'file.txt']
3157	n/a	if link:
3158	n/a	names.append('link_file.txt')
3159	n/a	if symlink:
3160	n/a	names.extend(('symlink_dir', 'symlink_file.txt'))
3161	n/a	entries = self.get_entries(names)
3162	n/a
3163	n/a	entry = entries['dir']
3164	n/a	self.check_entry(entry, 'dir', True, False, False)
3165	n/a
3166	n/a	entry = entries['file.txt']
3167	n/a	self.check_entry(entry, 'file.txt', False, True, False)
3168	n/a
3169	n/a	if link:
3170	n/a	entry = entries['link_file.txt']
3171	n/a	self.check_entry(entry, 'link_file.txt', False, True, False)
3172	n/a
3173	n/a	if symlink:
3174	n/a	entry = entries['symlink_dir']
3175	n/a	self.check_entry(entry, 'symlink_dir', True, False, True)
3176	n/a
3177	n/a	entry = entries['symlink_file.txt']
3178	n/a	self.check_entry(entry, 'symlink_file.txt', False, True, True)
3179	n/a
3180	n/a	def get_entry(self, name):
3181	n/a	path = self.bytes_path if isinstance(name, bytes) else self.path
3182	n/a	entries = list(os.scandir(path))
3183	n/a	self.assertEqual(len(entries), 1)
3184	n/a
3185	n/a	entry = entries[0]
3186	n/a	self.assertEqual(entry.name, name)
3187	n/a	return entry
3188	n/a
3189	n/a	def create_file_entry(self, name='file.txt'):
3190	n/a	filename = self.create_file(name=name)
3191	n/a	return self.get_entry(os.path.basename(filename))
3192	n/a
3193	n/a	def test_current_directory(self):
3194	n/a	filename = self.create_file()
3195	n/a	old_dir = os.getcwd()
3196	n/a	try:
3197	n/a	os.chdir(self.path)
3198	n/a
3199	n/a	# call scandir() without parameter: it must list the content
3200	n/a	# of the current directory
3201	n/a	entries = dict((entry.name, entry) for entry in os.scandir())
3202	n/a	self.assertEqual(sorted(entries.keys()),
3203	n/a	[os.path.basename(filename)])
3204	n/a	finally:
3205	n/a	os.chdir(old_dir)
3206	n/a
3207	n/a	def test_repr(self):
3208	n/a	entry = self.create_file_entry()
3209	n/a	self.assertEqual(repr(entry), "<DirEntry 'file.txt'>")
3210	n/a
3211	n/a	def test_fspath_protocol(self):
3212	n/a	entry = self.create_file_entry()
3213	n/a	self.assertEqual(os.fspath(entry), os.path.join(self.path, 'file.txt'))
3214	n/a
3215	n/a	def test_fspath_protocol_bytes(self):
3216	n/a	bytes_filename = os.fsencode('bytesfile.txt')
3217	n/a	bytes_entry = self.create_file_entry(name=bytes_filename)
3218	n/a	fspath = os.fspath(bytes_entry)
3219	n/a	self.assertIsInstance(fspath, bytes)
3220	n/a	self.assertEqual(fspath,
3221	n/a	os.path.join(os.fsencode(self.path),bytes_filename))
3222	n/a
3223	n/a	def test_removed_dir(self):
3224	n/a	path = os.path.join(self.path, 'dir')
3225	n/a
3226	n/a	os.mkdir(path)
3227	n/a	entry = self.get_entry('dir')
3228	n/a	os.rmdir(path)
3229	n/a
3230	n/a	# On POSIX, is_dir() result depends if scandir() filled d_type or not
3231	n/a	if os.name == 'nt':
3232	n/a	self.assertTrue(entry.is_dir())
3233	n/a	self.assertFalse(entry.is_file())
3234	n/a	self.assertFalse(entry.is_symlink())
3235	n/a	if os.name == 'nt':
3236	n/a	self.assertRaises(FileNotFoundError, entry.inode)
3237	n/a	# don't fail
3238	n/a	entry.stat()
3239	n/a	entry.stat(follow_symlinks=False)
3240	n/a	else:
3241	n/a	self.assertGreater(entry.inode(), 0)
3242	n/a	self.assertRaises(FileNotFoundError, entry.stat)
3243	n/a	self.assertRaises(FileNotFoundError, entry.stat, follow_symlinks=False)
3244	n/a
3245	n/a	def test_removed_file(self):
3246	n/a	entry = self.create_file_entry()
3247	n/a	os.unlink(entry.path)
3248	n/a
3249	n/a	self.assertFalse(entry.is_dir())
3250	n/a	# On POSIX, is_dir() result depends if scandir() filled d_type or not
3251	n/a	if os.name == 'nt':
3252	n/a	self.assertTrue(entry.is_file())
3253	n/a	self.assertFalse(entry.is_symlink())
3254	n/a	if os.name == 'nt':
3255	n/a	self.assertRaises(FileNotFoundError, entry.inode)
3256	n/a	# don't fail
3257	n/a	entry.stat()
3258	n/a	entry.stat(follow_symlinks=False)
3259	n/a	else:
3260	n/a	self.assertGreater(entry.inode(), 0)
3261	n/a	self.assertRaises(FileNotFoundError, entry.stat)
3262	n/a	self.assertRaises(FileNotFoundError, entry.stat, follow_symlinks=False)
3263	n/a
3264	n/a	def test_broken_symlink(self):
3265	n/a	if not support.can_symlink():
3266	n/a	return self.skipTest('cannot create symbolic link')
3267	n/a
3268	n/a	filename = self.create_file("file.txt")
3269	n/a	os.symlink(filename,
3270	n/a	os.path.join(self.path, "symlink.txt"))
3271	n/a	entries = self.get_entries(['file.txt', 'symlink.txt'])
3272	n/a	entry = entries['symlink.txt']
3273	n/a	os.unlink(filename)
3274	n/a
3275	n/a	self.assertGreater(entry.inode(), 0)
3276	n/a	self.assertFalse(entry.is_dir())
3277	n/a	self.assertFalse(entry.is_file()) # broken symlink returns False
3278	n/a	self.assertFalse(entry.is_dir(follow_symlinks=False))
3279	n/a	self.assertFalse(entry.is_file(follow_symlinks=False))
3280	n/a	self.assertTrue(entry.is_symlink())
3281	n/a	self.assertRaises(FileNotFoundError, entry.stat)
3282	n/a	# don't fail
3283	n/a	entry.stat(follow_symlinks=False)
3284	n/a
3285	n/a	def test_bytes(self):
3286	n/a	self.create_file("file.txt")
3287	n/a
3288	n/a	path_bytes = os.fsencode(self.path)
3289	n/a	entries = list(os.scandir(path_bytes))
3290	n/a	self.assertEqual(len(entries), 1, entries)
3291	n/a	entry = entries[0]
3292	n/a
3293	n/a	self.assertEqual(entry.name, b'file.txt')
3294	n/a	self.assertEqual(entry.path,
3295	n/a	os.fsencode(os.path.join(self.path, 'file.txt')))
3296	n/a
3297	n/a	def test_empty_path(self):
3298	n/a	self.assertRaises(FileNotFoundError, os.scandir, '')
3299	n/a
3300	n/a	def test_consume_iterator_twice(self):
3301	n/a	self.create_file("file.txt")
3302	n/a	iterator = os.scandir(self.path)
3303	n/a
3304	n/a	entries = list(iterator)
3305	n/a	self.assertEqual(len(entries), 1, entries)
3306	n/a
3307	n/a	# check than consuming the iterator twice doesn't raise exception
3308	n/a	entries2 = list(iterator)
3309	n/a	self.assertEqual(len(entries2), 0, entries2)
3310	n/a
3311	n/a	def test_bad_path_type(self):
3312	n/a	for obj in [1234, 1.234, {}, []]:
3313	n/a	self.assertRaises(TypeError, os.scandir, obj)
3314	n/a
3315	n/a	def test_close(self):
3316	n/a	self.create_file("file.txt")
3317	n/a	self.create_file("file2.txt")
3318	n/a	iterator = os.scandir(self.path)
3319	n/a	next(iterator)
3320	n/a	iterator.close()
3321	n/a	# multiple closes
3322	n/a	iterator.close()
3323	n/a	with self.check_no_resource_warning():
3324	n/a	del iterator
3325	n/a
3326	n/a	def test_context_manager(self):
3327	n/a	self.create_file("file.txt")
3328	n/a	self.create_file("file2.txt")
3329	n/a	with os.scandir(self.path) as iterator:
3330	n/a	next(iterator)
3331	n/a	with self.check_no_resource_warning():
3332	n/a	del iterator
3333	n/a
3334	n/a	def test_context_manager_close(self):
3335	n/a	self.create_file("file.txt")
3336	n/a	self.create_file("file2.txt")
3337	n/a	with os.scandir(self.path) as iterator:
3338	n/a	next(iterator)
3339	n/a	iterator.close()
3340	n/a
3341	n/a	def test_context_manager_exception(self):
3342	n/a	self.create_file("file.txt")
3343	n/a	self.create_file("file2.txt")
3344	n/a	with self.assertRaises(ZeroDivisionError):
3345	n/a	with os.scandir(self.path) as iterator:
3346	n/a	next(iterator)
3347	n/a	1/0
3348	n/a	with self.check_no_resource_warning():
3349	n/a	del iterator
3350	n/a
3351	n/a	def test_resource_warning(self):
3352	n/a	self.create_file("file.txt")
3353	n/a	self.create_file("file2.txt")
3354	n/a	iterator = os.scandir(self.path)
3355	n/a	next(iterator)
3356	n/a	with self.assertWarns(ResourceWarning):
3357	n/a	del iterator
3358	n/a	support.gc_collect()
3359	n/a	# exhausted iterator
3360	n/a	iterator = os.scandir(self.path)
3361	n/a	list(iterator)
3362	n/a	with self.check_no_resource_warning():
3363	n/a	del iterator
3364	n/a
3365	n/a
3366	n/a	class TestPEP519(unittest.TestCase):
3367	n/a
3368	n/a	# Abstracted so it can be overridden to test pure Python implementation
3369	n/a	# if a C version is provided.
3370	n/a	fspath = staticmethod(os.fspath)
3371	n/a
3372	n/a	def test_return_bytes(self):
3373	n/a	for b in b'hello', b'goodbye', b'some/path/and/file':
3374	n/a	self.assertEqual(b, self.fspath(b))
3375	n/a
3376	n/a	def test_return_string(self):
3377	n/a	for s in 'hello', 'goodbye', 'some/path/and/file':
3378	n/a	self.assertEqual(s, self.fspath(s))
3379	n/a
3380	n/a	def test_fsencode_fsdecode(self):
3381	n/a	for p in "path/like/object", b"path/like/object":
3382	n/a	pathlike = _PathLike(p)
3383	n/a
3384	n/a	self.assertEqual(p, self.fspath(pathlike))
3385	n/a	self.assertEqual(b"path/like/object", os.fsencode(pathlike))
3386	n/a	self.assertEqual("path/like/object", os.fsdecode(pathlike))
3387	n/a
3388	n/a	def test_pathlike(self):
3389	n/a	self.assertEqual('#feelthegil', self.fspath(_PathLike('#feelthegil')))
3390	n/a	self.assertTrue(issubclass(_PathLike, os.PathLike))
3391	n/a	self.assertTrue(isinstance(_PathLike(), os.PathLike))
3392	n/a
3393	n/a	def test_garbage_in_exception_out(self):
3394	n/a	vapor = type('blah', (), {})
3395	n/a	for o in int, type, os, vapor():
3396	n/a	self.assertRaises(TypeError, self.fspath, o)
3397	n/a
3398	n/a	def test_argument_required(self):
3399	n/a	self.assertRaises(TypeError, self.fspath)
3400	n/a
3401	n/a	def test_bad_pathlike(self):
3402	n/a	# __fspath__ returns a value other than str or bytes.
3403	n/a	self.assertRaises(TypeError, self.fspath, _PathLike(42))
3404	n/a	# __fspath__ attribute that is not callable.
3405	n/a	c = type('foo', (), {})
3406	n/a	c.__fspath__ = 1
3407	n/a	self.assertRaises(TypeError, self.fspath, c())
3408	n/a	# __fspath__ raises an exception.
3409	n/a	self.assertRaises(ZeroDivisionError, self.fspath,
3410	n/a	_PathLike(ZeroDivisionError()))
3411	n/a
3412	n/a	# Only test if the C version is provided, otherwise TestPEP519 already tested
3413	n/a	# the pure Python implementation.
3414	n/a	if hasattr(os, "_fspath"):
3415	n/a	class TestPEP519PurePython(TestPEP519):
3416	n/a
3417	n/a	"""Explicitly test the pure Python implementation of os.fspath()."""
3418	n/a
3419	n/a	fspath = staticmethod(os._fspath)
3420	n/a
3421	n/a
3422	n/a	if __name__ == "__main__":
3423	n/a	unittest.main()