diff options
Diffstat (limited to 'Lib/test/test_os.py')
| -rw-r--r-- | Lib/test/test_os.py | 848 |
1 files changed, 811 insertions, 37 deletions
diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py index cbf0c16..35aa7fa 100644 --- a/Lib/test/test_os.py +++ b/Lib/test/test_os.py @@ -7,8 +7,21 @@ import errno import unittest import warnings import sys +import signal +import subprocess +import time import shutil from test import support +import contextlib +import mmap +import uuid +import asyncore +import asynchat +import socket +try: + import threading +except ImportError: + threading = None # Detect whether we're on a Linux system that uses the (now outdated # and unmaintained) linuxthreads threading library. There's an issue @@ -54,6 +67,7 @@ class FileTests(unittest.TestCase): os.closerange(first, first + 2) self.assertRaises(OSError, os.write, first, b"a") + @support.cpython_only def test_rename(self): path = support.TESTFN old = sys.getrefcount(path) @@ -83,6 +97,25 @@ class FileTests(unittest.TestCase): self.assertEqual(fobj.read().splitlines(), [b"bacon", b"eggs", b"spam"]) + def write_windows_console(self, *args): + retcode = subprocess.call(args, + # use a new console to not flood the test output + creationflags=subprocess.CREATE_NEW_CONSOLE, + # use a shell to hide the console window (SW_HIDE) + shell=True) + self.assertEqual(retcode, 0) + + @unittest.skipUnless(sys.platform == 'win32', + 'test specific to the Windows console') + def test_write_windows_console(self): + # Issue #11395: the Windows console returns an error (12: not enough + # space error) on writing into stdout if stdout mode is binary and the + # length is greater than 66,000 bytes (or less, depending on heap + # usage). + code = "print('x' * 100000)" + self.write_windows_console(sys.executable, "-c", code) + self.write_windows_console(sys.executable, "-u", "-c", code) + class TemporaryFileTests(unittest.TestCase): def setUp(self): @@ -165,7 +198,6 @@ class TemporaryFileTests(unittest.TestCase): self.assertTrue(s == "foobar") def test_tmpnam(self): - import sys if not hasattr(os, "tmpnam"): return warnings.filterwarnings("ignore", "tmpnam", RuntimeWarning, @@ -215,19 +247,17 @@ class StatAttributeTests(unittest.TestCase): os.unlink(self.fname) os.rmdir(support.TESTFN) - def test_stat_attributes(self): + def check_stat_attributes(self, fname): if not hasattr(os, "stat"): return import stat - result = os.stat(self.fname) + result = os.stat(fname) # Make sure direct access works self.assertEqual(result[stat.ST_SIZE], 3) self.assertEqual(result.st_size, 3) - import sys - # Make sure all the attributes are there members = dir(result) for name in dir(stat): @@ -238,8 +268,8 @@ class StatAttributeTests(unittest.TestCase): else: def trunc(x): return x self.assertEqual(trunc(getattr(result, attr)), - result[getattr(stat, name)]) - self.assertTrue(attr in members) + result[getattr(stat, name)]) + self.assertIn(attr, members) try: result[200] @@ -279,6 +309,15 @@ class StatAttributeTests(unittest.TestCase): except TypeError: pass + def test_stat_attributes(self): + self.check_stat_attributes(self.fname) + + def test_stat_attributes_bytes(self): + try: + fname = self.fname.encode(sys.getfilesystemencoding()) + except UnicodeEncodeError: + self.skipTest("cannot encode %a for the filesystem" % self.fname) + self.check_stat_attributes(fname) def test_statvfs_attributes(self): if not hasattr(os, "statvfs"): @@ -374,12 +413,17 @@ class EnvironTests(mapping_tests.BasicTestMappingProtocol): def setUp(self): self.__save = dict(os.environ) + if os.supports_bytes_environ: + self.__saveb = dict(os.environb) for key, value in self._reference().items(): os.environ[key] = value def tearDown(self): os.environ.clear() os.environ.update(self.__save) + if os.supports_bytes_environ: + os.environb.clear() + os.environb.update(self.__saveb) def _reference(self): return {"KEY1":"VALUE1", "KEY2":"VALUE2", "KEY3":"VALUE3"} @@ -422,9 +466,73 @@ class EnvironTests(mapping_tests.BasicTestMappingProtocol): def test___repr__(self): """Check that the repr() of os.environ looks like environ({...}).""" env = os.environ - self.assertTrue(isinstance(env.data, dict)) - self.assertEqual(repr(env), 'environ({!r})'.format(env.data)) + self.assertEqual(repr(env), 'environ({{{}}})'.format(', '.join( + '{!r}: {!r}'.format(key, value) + for key, value in env.items()))) + + def test_get_exec_path(self): + defpath_list = os.defpath.split(os.pathsep) + test_path = ['/monty', '/python', '', '/flying/circus'] + test_env = {'PATH': os.pathsep.join(test_path)} + + saved_environ = os.environ + try: + os.environ = dict(test_env) + # Test that defaulting to os.environ works. + self.assertSequenceEqual(test_path, os.get_exec_path()) + self.assertSequenceEqual(test_path, os.get_exec_path(env=None)) + finally: + os.environ = saved_environ + + # No PATH environment variable + self.assertSequenceEqual(defpath_list, os.get_exec_path({})) + # Empty PATH environment variable + self.assertSequenceEqual(('',), os.get_exec_path({'PATH':''})) + # Supplied PATH environment variable + self.assertSequenceEqual(test_path, os.get_exec_path(test_env)) + if os.supports_bytes_environ: + # env cannot contain 'PATH' and b'PATH' keys + try: + # ignore BytesWarning warning + with warnings.catch_warnings(record=True): + mixed_env = {'PATH': '1', b'PATH': b'2'} + except BytesWarning: + # mixed_env cannot be created with python -bb + pass + else: + self.assertRaises(ValueError, os.get_exec_path, mixed_env) + + # bytes key and/or value + self.assertSequenceEqual(os.get_exec_path({b'PATH': b'abc'}), + ['abc']) + self.assertSequenceEqual(os.get_exec_path({b'PATH': 'abc'}), + ['abc']) + self.assertSequenceEqual(os.get_exec_path({'PATH': b'abc'}), + ['abc']) + + @unittest.skipUnless(os.supports_bytes_environ, + "os.environb required for this test.") + def test_environb(self): + # os.environ -> os.environb + value = 'euro\u20ac' + try: + value_bytes = value.encode(sys.getfilesystemencoding(), + 'surrogateescape') + except UnicodeEncodeError: + msg = "U+20AC character is not encodable to %s" % ( + sys.getfilesystemencoding(),) + self.skipTest(msg) + os.environ['unicode'] = value + self.assertEqual(os.environ['unicode'], value) + self.assertEqual(os.environb[b'unicode'], value_bytes) + + # os.environb -> os.environ + value = b'\xff' + os.environb[b'bytes'] = value + self.assertEqual(os.environb[b'bytes'], value) + value_str = value.decode(sys.getfilesystemencoding(), 'surrogateescape') + self.assertEqual(os.environ['bytes'], value_str) class WalkTests(unittest.TestCase): """Tests for os.walk().""" @@ -464,7 +572,7 @@ class WalkTests(unittest.TestCase): f = open(path, "w") f.write("I'm " + path + " and proud of it. Blame test_os.\n") f.close() - if hasattr(os, "symlink"): + if support.can_symlink(): os.symlink(os.path.abspath(t2_path), link_path) sub2_tree = (sub2_path, ["link"], ["tmp3"]) else: @@ -508,7 +616,7 @@ class WalkTests(unittest.TestCase): self.assertEqual(all[flipped + 1], (sub1_path, ["SUB11"], ["tmp2"])) self.assertEqual(all[2 - 2 * flipped], sub2_tree) - if hasattr(os, "symlink"): + if support.can_symlink(): # Walk, following symlinks. for root, dirs, files in os.walk(walk_path, followlinks=True): if root == link_path: @@ -553,6 +661,28 @@ class MakedirTests(unittest.TestCase): 'dir5', 'dir6') os.makedirs(path) + def test_exist_ok_existing_directory(self): + path = os.path.join(support.TESTFN, 'dir1') + mode = 0o777 + old_mask = os.umask(0o022) + os.makedirs(path, mode) + self.assertRaises(OSError, os.makedirs, path, mode) + self.assertRaises(OSError, os.makedirs, path, mode, exist_ok=False) + self.assertRaises(OSError, os.makedirs, path, 0o776, exist_ok=True) + os.makedirs(path, mode=mode, exist_ok=True) + os.umask(old_mask) + + def test_exist_ok_existing_regular_file(self): + base = support.TESTFN + path = os.path.join(support.TESTFN, 'dir1') + f = open(path, 'w') + f.write('abc') + f.close() + self.assertRaises(OSError, os.makedirs, path) + self.assertRaises(OSError, os.makedirs, path, exist_ok=False) + self.assertRaises(OSError, os.makedirs, path, exist_ok=True) + os.remove(path) + def tearDown(self): path = os.path.join(support.TESTFN, 'dir1', 'dir2', 'dir3', 'dir4', 'dir5', 'dir6') @@ -583,6 +713,39 @@ class URandomTests(unittest.TestCase): except NotImplementedError: pass +@contextlib.contextmanager +def _execvpe_mockup(defpath=None): + """ + Stubs out execv and execve functions when used as context manager. + Records exec calls. The mock execv and execve functions always raise an + exception as they would normally never return. + """ + # A list of tuples containing (function name, first arg, args) + # of calls to execv or execve that have been made. + calls = [] + + def mock_execv(name, *args): + calls.append(('execv', name, args)) + raise RuntimeError("execv called") + + def mock_execve(name, *args): + calls.append(('execve', name, args)) + raise OSError(errno.ENOTDIR, "execve called") + + try: + orig_execv = os.execv + orig_execve = os.execve + orig_defpath = os.defpath + os.execv = mock_execv + os.execve = mock_execve + if defpath is not None: + os.defpath = defpath + yield calls + finally: + os.execv = orig_execv + os.execve = orig_execve + os.defpath = orig_defpath + class ExecTests(unittest.TestCase): @unittest.skipIf(USING_LINUXTHREADS, "avoid triggering a linuxthreads bug: see issue #4970") @@ -593,13 +756,60 @@ class ExecTests(unittest.TestCase): def test_execvpe_with_bad_arglist(self): self.assertRaises(ValueError, os.execvpe, 'notepad', [], None) -class ArgTests(unittest.TestCase): - def test_bytearray(self): - # Issue #7561: posix module didn't release bytearray exports properly. - b = bytearray(os.sep.encode('ascii')) - self.assertRaises(OSError, os.mkdir, b) - # Check object is still resizable. - b[:] = b'' + @unittest.skipUnless(hasattr(os, '_execvpe'), + "No internal os._execvpe function to test.") + def _test_internal_execvpe(self, test_type): + program_path = os.sep + 'absolutepath' + if test_type is bytes: + program = b'executable' + fullpath = os.path.join(os.fsencode(program_path), program) + native_fullpath = fullpath + arguments = [b'progname', 'arg1', 'arg2'] + else: + program = 'executable' + arguments = ['progname', 'arg1', 'arg2'] + fullpath = os.path.join(program_path, program) + if os.name != "nt": + native_fullpath = os.fsencode(fullpath) + else: + native_fullpath = fullpath + env = {'spam': 'beans'} + + # test os._execvpe() with an absolute path + with _execvpe_mockup() as calls: + self.assertRaises(RuntimeError, + os._execvpe, fullpath, arguments) + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0], ('execv', fullpath, (arguments,))) + + # test os._execvpe() with a relative path: + # os.get_exec_path() returns defpath + with _execvpe_mockup(defpath=program_path) as calls: + self.assertRaises(OSError, + os._execvpe, program, arguments, env=env) + self.assertEqual(len(calls), 1) + self.assertSequenceEqual(calls[0], + ('execve', native_fullpath, (arguments, env))) + + # test os._execvpe() with a relative path: + # os.get_exec_path() reads the 'PATH' variable + with _execvpe_mockup() as calls: + env_path = env.copy() + if test_type is bytes: + env_path[b'PATH'] = program_path + else: + env_path['PATH'] = program_path + self.assertRaises(OSError, + os._execvpe, program, arguments, env=env_path) + self.assertEqual(len(calls), 1) + self.assertSequenceEqual(calls[0], + ('execve', native_fullpath, (arguments, env_path))) + + def test_internal_execvpe_str(self): + self._test_internal_execvpe(str) + if os.name != "nt": + self._test_internal_execvpe(bytes) + class Win32ErrorTests(unittest.TestCase): def test_rename(self): @@ -703,6 +913,42 @@ class TestInvalidFD(unittest.TestCase): if hasattr(os, "write"): self.check(os.write, b" ") + +class LinkTests(unittest.TestCase): + def setUp(self): + self.file1 = support.TESTFN + self.file2 = os.path.join(support.TESTFN + "2") + + def tearDown(self): + for file in (self.file1, self.file2): + if os.path.exists(file): + os.unlink(file) + + def _test_link(self, file1, file2): + with open(file1, "w") as f1: + f1.write("test") + + os.link(file1, file2) + with open(file1, "r") as f1, open(file2, "r") as f2: + self.assertTrue(os.path.sameopenfile(f1.fileno(), f2.fileno())) + + def test_link(self): + self._test_link(self.file1, self.file2) + + def test_link_bytes(self): + self._test_link(bytes(self.file1, sys.getfilesystemencoding()), + bytes(self.file2, sys.getfilesystemencoding())) + + def test_unicode_name(self): + try: + os.fsencode("\xf1") + except UnicodeError: + raise unittest.SkipTest("Unable to encode for this platform.") + + self.file1 += "\xf1" + self.file2 = self.file1 + "2" + self._test_link(self.file1, self.file2) + if sys.platform != 'win32': class Win32ErrorTests(unittest.TestCase): pass @@ -742,7 +988,6 @@ if sys.platform != 'win32': def test_setreuid_neg1(self): # Needs to accept -1. We run this in a subprocess to avoid # altering the test runner's process state (issue8045). - import subprocess subprocess.check_call([ sys.executable, '-c', 'import os,sys;os.setreuid(-1,-1);sys.exit(0)']) @@ -757,34 +1002,51 @@ if sys.platform != 'win32': def test_setregid_neg1(self): # Needs to accept -1. We run this in a subprocess to avoid # altering the test runner's process state (issue8045). - import subprocess subprocess.check_call([ sys.executable, '-c', 'import os,sys;os.setregid(-1,-1);sys.exit(0)']) - @unittest.skipIf(sys.platform == 'darwin', "tests don't apply to OS X") class Pep383Tests(unittest.TestCase): - filenames = [b'foo\xf6bar', 'foo\xf6bar'.encode("utf-8")] - def setUp(self): - self.fsencoding = sys.getfilesystemencoding() - sys.setfilesystemencoding("utf-8") - self.dir = support.TESTFN - self.bdir = self.dir.encode("utf-8", "surrogateescape") + if support.TESTFN_UNENCODABLE: + self.dir = support.TESTFN_UNENCODABLE + else: + self.dir = support.TESTFN + self.bdir = os.fsencode(self.dir) + + bytesfn = [] + def add_filename(fn): + try: + fn = os.fsencode(fn) + except UnicodeEncodeError: + return + bytesfn.append(fn) + add_filename(support.TESTFN_UNICODE) + if support.TESTFN_UNENCODABLE: + add_filename(support.TESTFN_UNENCODABLE) + if not bytesfn: + self.skipTest("couldn't create any non-ascii filename") + + self.unicodefn = set() os.mkdir(self.dir) - self.unicodefn = [] - for fn in self.filenames: - f = open(os.path.join(self.bdir, fn), "w") - f.close() - self.unicodefn.append(fn.decode("utf-8", "surrogateescape")) + try: + for fn in bytesfn: + f = open(os.path.join(self.bdir, fn), "w") + f.close() + fn = os.fsdecode(fn) + if fn in self.unicodefn: + raise ValueError("duplicate filename") + self.unicodefn.add(fn) + except: + shutil.rmtree(self.dir) + raise def tearDown(self): shutil.rmtree(self.dir) - sys.setfilesystemencoding(self.fsencoding) def test_listdir(self): - expected = set(self.unicodefn) - found = set(os.listdir(support.TESTFN)) + expected = self.unicodefn + found = set(os.listdir(self.dir)) self.assertEqual(found, expected) def test_open(self): @@ -801,9 +1063,513 @@ else: class Pep383Tests(unittest.TestCase): pass +@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests") +class Win32KillTests(unittest.TestCase): + def _kill(self, sig): + # Start sys.executable as a subprocess and communicate from the + # subprocess to the parent that the interpreter is ready. When it + # becomes ready, send *sig* via os.kill to the subprocess and check + # that the return code is equal to *sig*. + import ctypes + from ctypes import wintypes + import msvcrt + + # Since we can't access the contents of the process' stdout until the + # process has exited, use PeekNamedPipe to see what's inside stdout + # without waiting. This is done so we can tell that the interpreter + # is started and running at a point where it could handle a signal. + PeekNamedPipe = ctypes.windll.kernel32.PeekNamedPipe + PeekNamedPipe.restype = wintypes.BOOL + PeekNamedPipe.argtypes = (wintypes.HANDLE, # Pipe handle + ctypes.POINTER(ctypes.c_char), # stdout buf + wintypes.DWORD, # Buffer size + ctypes.POINTER(wintypes.DWORD), # bytes read + ctypes.POINTER(wintypes.DWORD), # bytes avail + ctypes.POINTER(wintypes.DWORD)) # bytes left + msg = "running" + proc = subprocess.Popen([sys.executable, "-c", + "import sys;" + "sys.stdout.write('{}');" + "sys.stdout.flush();" + "input()".format(msg)], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + stdin=subprocess.PIPE) + self.addCleanup(proc.stdout.close) + self.addCleanup(proc.stderr.close) + self.addCleanup(proc.stdin.close) + + count, max = 0, 100 + while count < max and proc.poll() is None: + # Create a string buffer to store the result of stdout from the pipe + buf = ctypes.create_string_buffer(len(msg)) + # Obtain the text currently in proc.stdout + # Bytes read/avail/left are left as NULL and unused + rslt = PeekNamedPipe(msvcrt.get_osfhandle(proc.stdout.fileno()), + buf, ctypes.sizeof(buf), None, None, None) + self.assertNotEqual(rslt, 0, "PeekNamedPipe failed") + if buf.value: + self.assertEqual(msg, buf.value.decode()) + break + time.sleep(0.1) + count += 1 + else: + self.fail("Did not receive communication from the subprocess") + + os.kill(proc.pid, sig) + self.assertEqual(proc.wait(), sig) + + def test_kill_sigterm(self): + # SIGTERM doesn't mean anything special, but make sure it works + self._kill(signal.SIGTERM) + + def test_kill_int(self): + # os.kill on Windows can take an int which gets set as the exit code + self._kill(100) + + def _kill_with_event(self, event, name): + tagname = "test_os_%s" % uuid.uuid1() + m = mmap.mmap(-1, 1, tagname) + m[0] = 0 + # Run a script which has console control handling enabled. + proc = subprocess.Popen([sys.executable, + os.path.join(os.path.dirname(__file__), + "win_console_handler.py"), tagname], + creationflags=subprocess.CREATE_NEW_PROCESS_GROUP) + # Let the interpreter startup before we send signals. See #3137. + count, max = 0, 100 + while count < max and proc.poll() is None: + if m[0] == 1: + break + time.sleep(0.1) + count += 1 + else: + # Forcefully kill the process if we weren't able to signal it. + os.kill(proc.pid, signal.SIGINT) + self.fail("Subprocess didn't finish initialization") + os.kill(proc.pid, event) + # proc.send_signal(event) could also be done here. + # Allow time for the signal to be passed and the process to exit. + time.sleep(0.5) + if not proc.poll(): + # Forcefully kill the process if we weren't able to signal it. + os.kill(proc.pid, signal.SIGINT) + self.fail("subprocess did not stop on {}".format(name)) + + @unittest.skip("subprocesses aren't inheriting CTRL+C property") + def test_CTRL_C_EVENT(self): + from ctypes import wintypes + import ctypes + + # Make a NULL value by creating a pointer with no argument. + NULL = ctypes.POINTER(ctypes.c_int)() + SetConsoleCtrlHandler = ctypes.windll.kernel32.SetConsoleCtrlHandler + SetConsoleCtrlHandler.argtypes = (ctypes.POINTER(ctypes.c_int), + wintypes.BOOL) + SetConsoleCtrlHandler.restype = wintypes.BOOL + + # Calling this with NULL and FALSE causes the calling process to + # handle CTRL+C, rather than ignore it. This property is inherited + # by subprocesses. + SetConsoleCtrlHandler(NULL, 0) + + self._kill_with_event(signal.CTRL_C_EVENT, "CTRL_C_EVENT") + + def test_CTRL_BREAK_EVENT(self): + self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT") + + +@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests") +@support.skip_unless_symlink +class Win32SymlinkTests(unittest.TestCase): + filelink = 'filelinktest' + filelink_target = os.path.abspath(__file__) + dirlink = 'dirlinktest' + dirlink_target = os.path.dirname(filelink_target) + missing_link = 'missing link' + + def setUp(self): + assert os.path.exists(self.dirlink_target) + assert os.path.exists(self.filelink_target) + assert not os.path.exists(self.dirlink) + assert not os.path.exists(self.filelink) + assert not os.path.exists(self.missing_link) + + def tearDown(self): + if os.path.exists(self.filelink): + os.remove(self.filelink) + if os.path.exists(self.dirlink): + os.rmdir(self.dirlink) + if os.path.lexists(self.missing_link): + os.remove(self.missing_link) + + def test_directory_link(self): + os.symlink(self.dirlink_target, self.dirlink) + self.assertTrue(os.path.exists(self.dirlink)) + self.assertTrue(os.path.isdir(self.dirlink)) + self.assertTrue(os.path.islink(self.dirlink)) + self.check_stat(self.dirlink, self.dirlink_target) + + def test_file_link(self): + os.symlink(self.filelink_target, self.filelink) + self.assertTrue(os.path.exists(self.filelink)) + self.assertTrue(os.path.isfile(self.filelink)) + self.assertTrue(os.path.islink(self.filelink)) + self.check_stat(self.filelink, self.filelink_target) + + def _create_missing_dir_link(self): + 'Create a "directory" link to a non-existent target' + linkname = self.missing_link + if os.path.lexists(linkname): + os.remove(linkname) + target = r'c:\\target does not exist.29r3c740' + assert not os.path.exists(target) + target_is_dir = True + os.symlink(target, linkname, target_is_dir) + + def test_remove_directory_link_to_missing_target(self): + self._create_missing_dir_link() + # For compatibility with Unix, os.remove will check the + # directory status and call RemoveDirectory if the symlink + # was created with target_is_dir==True. + os.remove(self.missing_link) + + @unittest.skip("currently fails; consider for improvement") + def test_isdir_on_directory_link_to_missing_target(self): + self._create_missing_dir_link() + # consider having isdir return true for directory links + self.assertTrue(os.path.isdir(self.missing_link)) + + @unittest.skip("currently fails; consider for improvement") + def test_rmdir_on_directory_link_to_missing_target(self): + self._create_missing_dir_link() + # consider allowing rmdir to remove directory links + os.rmdir(self.missing_link) + + def check_stat(self, link, target): + self.assertEqual(os.stat(link), os.stat(target)) + self.assertNotEqual(os.lstat(link), os.stat(link)) + + +class FSEncodingTests(unittest.TestCase): + def test_nop(self): + self.assertEqual(os.fsencode(b'abc\xff'), b'abc\xff') + self.assertEqual(os.fsdecode('abc\u0141'), 'abc\u0141') + + def test_identity(self): + # assert fsdecode(fsencode(x)) == x + for fn in ('unicode\u0141', 'latin\xe9', 'ascii'): + try: + bytesfn = os.fsencode(fn) + except UnicodeEncodeError: + continue + self.assertEqual(os.fsdecode(bytesfn), fn) + + +class PidTests(unittest.TestCase): + @unittest.skipUnless(hasattr(os, 'getppid'), "test needs os.getppid") + def test_getppid(self): + p = subprocess.Popen([sys.executable, '-c', + 'import os; print(os.getppid())'], + stdout=subprocess.PIPE) + stdout, _ = p.communicate() + # We are the parent of our subprocess + self.assertEqual(int(stdout), os.getpid()) + + +# The introduction of this TestCase caused at least two different errors on +# *nix buildbots. Temporarily skip this to let the buildbots move along. +@unittest.skip("Skip due to platform/environment differences on *NIX buildbots") +@unittest.skipUnless(hasattr(os, 'getlogin'), "test needs os.getlogin") +class LoginTests(unittest.TestCase): + def test_getlogin(self): + user_name = os.getlogin() + self.assertNotEqual(len(user_name), 0) + + +@unittest.skipUnless(hasattr(os, 'getpriority') and hasattr(os, 'setpriority'), + "needs os.getpriority and os.setpriority") +class ProgramPriorityTests(unittest.TestCase): + """Tests for os.getpriority() and os.setpriority().""" + + def test_set_get_priority(self): + + base = os.getpriority(os.PRIO_PROCESS, os.getpid()) + os.setpriority(os.PRIO_PROCESS, os.getpid(), base + 1) + try: + new_prio = os.getpriority(os.PRIO_PROCESS, os.getpid()) + if base >= 19 and new_prio <= 19: + raise unittest.SkipTest( + "unable to reliably test setpriority at current nice level of %s" % base) + else: + self.assertEqual(new_prio, base + 1) + finally: + try: + os.setpriority(os.PRIO_PROCESS, os.getpid(), base) + except OSError as err: + if err.errno != errno.EACCES: + raise + + +class SendfileTestServer(asyncore.dispatcher, threading.Thread): + + class Handler(asynchat.async_chat): + + def __init__(self, conn): + asynchat.async_chat.__init__(self, conn) + self.in_buffer = [] + self.closed = False + self.push(b"220 ready\r\n") + + def handle_read(self): + data = self.recv(4096) + self.in_buffer.append(data) + + def get_data(self): + return b''.join(self.in_buffer) + + def handle_close(self): + self.close() + self.closed = True + + def handle_error(self): + raise + + def __init__(self, address): + threading.Thread.__init__(self) + asyncore.dispatcher.__init__(self) + self.create_socket(socket.AF_INET, socket.SOCK_STREAM) + self.bind(address) + self.listen(5) + self.host, self.port = self.socket.getsockname()[:2] + self.handler_instance = None + self._active = False + self._active_lock = threading.Lock() + + # --- public API + + @property + def running(self): + return self._active + + def start(self): + assert not self.running + self.__flag = threading.Event() + threading.Thread.start(self) + self.__flag.wait() + + def stop(self): + assert self.running + self._active = False + self.join() + + def wait(self): + # wait for handler connection to be closed, then stop the server + while not getattr(self.handler_instance, "closed", False): + time.sleep(0.001) + self.stop() + + # --- internals + + def run(self): + self._active = True + self.__flag.set() + while self._active and asyncore.socket_map: + self._active_lock.acquire() + asyncore.loop(timeout=0.001, count=1) + self._active_lock.release() + asyncore.close_all() + + def handle_accept(self): + conn, addr = self.accept() + self.handler_instance = self.Handler(conn) + + def handle_connect(self): + self.close() + handle_read = handle_connect + + def writable(self): + return 0 + + def handle_error(self): + raise + + +@unittest.skipUnless(threading is not None, "test needs threading module") +@unittest.skipUnless(hasattr(os, 'sendfile'), "test needs os.sendfile()") +class TestSendfile(unittest.TestCase): + + DATA = b"12345abcde" * 16 * 1024 # 160 KB + SUPPORT_HEADERS_TRAILERS = not sys.platform.startswith("linux") and \ + not sys.platform.startswith("solaris") and \ + not sys.platform.startswith("sunos") + + @classmethod + def setUpClass(cls): + with open(support.TESTFN, "wb") as f: + f.write(cls.DATA) + + @classmethod + def tearDownClass(cls): + support.unlink(support.TESTFN) + + def setUp(self): + self.server = SendfileTestServer((support.HOST, 0)) + self.server.start() + self.client = socket.socket() + self.client.connect((self.server.host, self.server.port)) + self.client.settimeout(1) + # synchronize by waiting for "220 ready" response + self.client.recv(1024) + self.sockno = self.client.fileno() + self.file = open(support.TESTFN, 'rb') + self.fileno = self.file.fileno() + + def tearDown(self): + self.file.close() + self.client.close() + if self.server.running: + self.server.stop() + + def sendfile_wrapper(self, sock, file, offset, nbytes, headers=[], trailers=[]): + """A higher level wrapper representing how an application is + supposed to use sendfile(). + """ + while 1: + try: + if self.SUPPORT_HEADERS_TRAILERS: + return os.sendfile(sock, file, offset, nbytes, headers, + trailers) + else: + return os.sendfile(sock, file, offset, nbytes) + except OSError as err: + if err.errno == errno.ECONNRESET: + # disconnected + raise + elif err.errno in (errno.EAGAIN, errno.EBUSY): + # we have to retry send data + continue + else: + raise + + def test_send_whole_file(self): + # normal send + total_sent = 0 + offset = 0 + nbytes = 4096 + while total_sent < len(self.DATA): + sent = self.sendfile_wrapper(self.sockno, self.fileno, offset, nbytes) + if sent == 0: + break + offset += sent + total_sent += sent + self.assertTrue(sent <= nbytes) + self.assertEqual(offset, total_sent) + + self.assertEqual(total_sent, len(self.DATA)) + self.client.shutdown(socket.SHUT_RDWR) + self.client.close() + self.server.wait() + data = self.server.handler_instance.get_data() + self.assertEqual(len(data), len(self.DATA)) + self.assertEqual(data, self.DATA) + + def test_send_at_certain_offset(self): + # start sending a file at a certain offset + total_sent = 0 + offset = len(self.DATA) // 2 + must_send = len(self.DATA) - offset + nbytes = 4096 + while total_sent < must_send: + sent = self.sendfile_wrapper(self.sockno, self.fileno, offset, nbytes) + if sent == 0: + break + offset += sent + total_sent += sent + self.assertTrue(sent <= nbytes) + + self.client.shutdown(socket.SHUT_RDWR) + self.client.close() + self.server.wait() + data = self.server.handler_instance.get_data() + expected = self.DATA[len(self.DATA) // 2:] + self.assertEqual(total_sent, len(expected)) + self.assertEqual(len(data), len(expected)) + self.assertEqual(data, expected) + + def test_offset_overflow(self): + # specify an offset > file size + offset = len(self.DATA) + 4096 + try: + sent = os.sendfile(self.sockno, self.fileno, offset, 4096) + except OSError as e: + # Solaris can raise EINVAL if offset >= file length, ignore. + if e.errno != errno.EINVAL: + raise + else: + self.assertEqual(sent, 0) + self.client.shutdown(socket.SHUT_RDWR) + self.client.close() + self.server.wait() + data = self.server.handler_instance.get_data() + self.assertEqual(data, b'') + + def test_invalid_offset(self): + with self.assertRaises(OSError) as cm: + os.sendfile(self.sockno, self.fileno, -1, 4096) + self.assertEqual(cm.exception.errno, errno.EINVAL) + + # --- headers / trailers tests + + if SUPPORT_HEADERS_TRAILERS: + + def test_headers(self): + total_sent = 0 + sent = os.sendfile(self.sockno, self.fileno, 0, 4096, + headers=[b"x" * 512]) + total_sent += sent + offset = 4096 + nbytes = 4096 + while 1: + sent = self.sendfile_wrapper(self.sockno, self.fileno, + offset, nbytes) + if sent == 0: + break + total_sent += sent + offset += sent + + expected_data = b"x" * 512 + self.DATA + self.assertEqual(total_sent, len(expected_data)) + self.client.close() + self.server.wait() + data = self.server.handler_instance.get_data() + self.assertEqual(hash(data), hash(expected_data)) + + def test_trailers(self): + TESTFN2 = support.TESTFN + "2" + with open(TESTFN2, 'wb') as f: + f.write(b"abcde") + with open(TESTFN2, 'rb')as f: + self.addCleanup(os.remove, TESTFN2) + os.sendfile(self.sockno, f.fileno(), 0, 4096, + trailers=[b"12345"]) + self.client.close() + self.server.wait() + data = self.server.handler_instance.get_data() + self.assertEqual(data, b"abcde12345") + + if hasattr(os, "SF_NODISKIO"): + def test_flags(self): + try: + os.sendfile(self.sockno, self.fileno, 0, 4096, + flags=os.SF_NODISKIO) + except OSError as err: + if err.errno not in (errno.EBUSY, errno.EAGAIN): + raise + + def test_main(): support.run_unittest( - ArgTests, FileTests, StatAttributeTests, EnvironTests, @@ -815,7 +1581,15 @@ def test_main(): Win32ErrorTests, TestInvalidFD, PosixUidGidTests, - Pep383Tests + Pep383Tests, + Win32KillTests, + Win32SymlinkTests, + FSEncodingTests, + PidTests, + LoginTests, + LinkTests, + TestSendfile, + ProgramPriorityTests, ) if __name__ == "__main__": |