From dd6e947122249de4e4b15b37a3df453d9a67474e Mon Sep 17 00:00:00 2001 From: CPython Developers <> Date: Mon, 2 Feb 2026 12:58:01 +0900 Subject: [PATCH] Update os from v3.14.2 --- Lib/os.py | 115 +++++++-------- Lib/test/test_os.py | 308 ++++++++++++++++++++++++++++++++--------- Lib/test/test_popen.py | 11 +- 3 files changed, 309 insertions(+), 125 deletions(-) diff --git a/Lib/os.py b/Lib/os.py index b4c9f84c3..ac03b4163 100644 --- a/Lib/os.py +++ b/Lib/os.py @@ -10,7 +10,7 @@ This exports: - os.extsep is the extension separator (always '.') - os.altsep is the alternate pathname separator (None or '/') - os.pathsep is the component separator used in $PATH etc - - os.linesep is the line separator in text files ('\r' or '\n' or '\r\n') + - os.linesep is the line separator in text files ('\n' or '\r\n') - os.defpath is the default search path for executables - os.devnull is the file path of the null device ('/dev/null', etc.) @@ -64,6 +64,10 @@ if 'posix' in _names: from posix import _have_functions except ImportError: pass + try: + from posix import _create_environ + except ImportError: + pass import posix __all__.extend(_get_exports_list(posix)) @@ -88,6 +92,10 @@ elif 'nt' in _names: from nt import _have_functions except ImportError: pass + try: + from nt import _create_environ + except ImportError: + pass else: raise ImportError('no os specific module found') @@ -366,61 +374,45 @@ def walk(top, topdown=True, onerror=None, followlinks=False): # minor reason when (say) a thousand readable directories are still # left to visit. try: - scandir_it = scandir(top) + with scandir(top) as entries: + for entry in entries: + try: + if followlinks is _walk_symlinks_as_files: + is_dir = entry.is_dir(follow_symlinks=False) and not entry.is_junction() + else: + is_dir = entry.is_dir() + except OSError: + # If is_dir() raises an OSError, consider the entry not to + # be a directory, same behaviour as os.path.isdir(). + is_dir = False + + if is_dir: + dirs.append(entry.name) + else: + nondirs.append(entry.name) + + if not topdown and is_dir: + # Bottom-up: traverse into sub-directory, but exclude + # symlinks to directories if followlinks is False + if followlinks: + walk_into = True + else: + try: + is_symlink = entry.is_symlink() + except OSError: + # If is_symlink() raises an OSError, consider the + # entry not to be a symbolic link, same behaviour + # as os.path.islink(). + is_symlink = False + walk_into = not is_symlink + + if walk_into: + walk_dirs.append(entry.path) except OSError as error: if onerror is not None: onerror(error) continue - cont = False - with scandir_it: - while True: - try: - try: - entry = next(scandir_it) - except StopIteration: - break - except OSError as error: - if onerror is not None: - onerror(error) - cont = True - break - - try: - if followlinks is _walk_symlinks_as_files: - is_dir = entry.is_dir(follow_symlinks=False) and not entry.is_junction() - else: - is_dir = entry.is_dir() - except OSError: - # If is_dir() raises an OSError, consider the entry not to - # be a directory, same behaviour as os.path.isdir(). - is_dir = False - - if is_dir: - dirs.append(entry.name) - else: - nondirs.append(entry.name) - - if not topdown and is_dir: - # Bottom-up: traverse into sub-directory, but exclude - # symlinks to directories if followlinks is False - if followlinks: - walk_into = True - else: - try: - is_symlink = entry.is_symlink() - except OSError: - # If is_symlink() raises an OSError, consider the - # entry not to be a symbolic link, same behaviour - # as os.path.islink(). - is_symlink = False - walk_into = not is_symlink - - if walk_into: - walk_dirs.append(entry.path) - if cont: - continue - if topdown: # Yield before sub-directory traversal if going top down yield top, dirs, nondirs @@ -774,7 +766,7 @@ class _Environ(MutableMapping): new.update(self) return new -def _createenviron(): +def _create_environ_mapping(): if name == 'nt': # Where Env Var Names Must Be UPPERCASE def check_str(value): @@ -804,10 +796,25 @@ def _createenviron(): encode, decode) # unicode environ -environ = _createenviron() -del _createenviron +environ = _create_environ_mapping() +del _create_environ_mapping +if _exists("_create_environ"): + def reload_environ(): + data = _create_environ() + if name == 'nt': + encodekey = environ.encodekey + data = {encodekey(key): value + for key, value in data.items()} + + # modify in-place to keep os.environb in sync + env_data = environ._data + env_data.clear() + env_data.update(data) + + __all__.append("reload_environ") + def getenv(key, default=None): """Get an environment variable, return None if it doesn't exist. The optional second argument can specify an alternate default. diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py index 653a05dd0..2d08d82f2 100644 --- a/Lib/test/test_os.py +++ b/Lib/test/test_os.py @@ -104,7 +104,7 @@ requires_splice_pipe = unittest.skipIf(sys.platform.startswith("aix"), def tearDownModule(): - asyncio.set_event_loop_policy(None) + asyncio.events._set_event_loop_policy(None) class MiscTests(unittest.TestCase): @@ -188,9 +188,6 @@ class FileTests(unittest.TestCase): self.assertTrue(os.access(os_helper.TESTFN, os.W_OK)) @unittest.skipIf(sys.platform == 'win32', "TODO: RUSTPYTHON; BrokenPipeError: (32, 'The process cannot access the file because it is being used by another process. (os error 32)')") - @unittest.skipIf( - support.is_emscripten, "Test is unstable under Emscripten." - ) @unittest.skipIf( support.is_wasi, "WASI does not support dup." ) @@ -233,6 +230,97 @@ class FileTests(unittest.TestCase): self.assertEqual(type(s), bytes) self.assertEqual(s, b"spam") + @unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: module 'os' has no attribute 'readinto'. Did you mean: 'readlink'? + def test_readinto(self): + with open(os_helper.TESTFN, "w+b") as fobj: + fobj.write(b"spam") + fobj.flush() + fd = fobj.fileno() + os.lseek(fd, 0, 0) + # Oversized so readinto without hitting end. + buffer = bytearray(7) + s = os.readinto(fd, buffer) + self.assertEqual(type(s), int) + self.assertEqual(s, 4) + # Should overwrite the first 4 bytes of the buffer. + self.assertEqual(buffer[:4], b"spam") + + # Readinto at EOF should return 0 and not touch buffer. + buffer[:] = b"notspam" + s = os.readinto(fd, buffer) + self.assertEqual(type(s), int) + self.assertEqual(s, 0) + self.assertEqual(bytes(buffer), b"notspam") + s = os.readinto(fd, buffer) + self.assertEqual(s, 0) + self.assertEqual(bytes(buffer), b"notspam") + + # Readinto a 0 length bytearray when at EOF should return 0 + self.assertEqual(os.readinto(fd, bytearray()), 0) + + # Readinto a 0 length bytearray with data available should return 0. + os.lseek(fd, 0, 0) + self.assertEqual(os.readinto(fd, bytearray()), 0) + + @unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: module 'os' has no attribute 'readinto'. Did you mean: 'readlink'? + @unittest.skipUnless(hasattr(os, 'get_blocking'), + 'needs os.get_blocking() and os.set_blocking()') + @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()") + @unittest.skipIf(support.is_emscripten, "set_blocking does not work correctly") + def test_readinto_non_blocking(self): + # Verify behavior of a readinto which would block on a non-blocking fd. + r, w = os.pipe() + try: + os.set_blocking(r, False) + with self.assertRaises(BlockingIOError): + os.readinto(r, bytearray(5)) + + # Pass some data through + os.write(w, b"spam") + self.assertEqual(os.readinto(r, bytearray(4)), 4) + + # Still don't block or return 0. + with self.assertRaises(BlockingIOError): + os.readinto(r, bytearray(5)) + + # At EOF should return size 0 + os.close(w) + w = None + self.assertEqual(os.readinto(r, bytearray(5)), 0) + self.assertEqual(os.readinto(r, bytearray(5)), 0) # Still EOF + + finally: + os.close(r) + if w is not None: + os.close(w) + + @unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: module 'os' has no attribute 'readinto'. Did you mean: 'readlink'? + def test_readinto_badarg(self): + with open(os_helper.TESTFN, "w+b") as fobj: + fobj.write(b"spam") + fobj.flush() + fd = fobj.fileno() + os.lseek(fd, 0, 0) + + for bad_arg in ("test", bytes(), 14): + with self.subTest(f"bad buffer {type(bad_arg)}"): + with self.assertRaises(TypeError): + os.readinto(fd, bad_arg) + + with self.subTest("doesn't work on file objects"): + with self.assertRaises(TypeError): + os.readinto(fobj, bytearray(5)) + + # takes two args + with self.assertRaises(TypeError): + os.readinto(fd) + + # No data should have been read with the bad arguments. + buffer = bytearray(4) + s = os.readinto(fd, buffer) + self.assertEqual(s, 4) + self.assertEqual(buffer, b"spam") + @support.cpython_only # Skip the test on 32-bit platforms: the number of bytes must fit in a # Py_ssize_t type @@ -252,6 +340,29 @@ class FileTests(unittest.TestCase): # operating system is free to return less bytes than requested. self.assertEqual(data, b'test') + + @support.cpython_only + # Skip the test on 32-bit platforms: the number of bytes must fit in a + # Py_ssize_t type + @unittest.skipUnless(INT_MAX < PY_SSIZE_T_MAX, + "needs INT_MAX < PY_SSIZE_T_MAX") + @support.bigmemtest(size=INT_MAX + 10, memuse=1, dry_run=False) + def test_large_readinto(self, size): + self.addCleanup(os_helper.unlink, os_helper.TESTFN) + create_file(os_helper.TESTFN, b'test') + + # Issue #21932: For readinto the buffer contains the length rather than + # a length being passed explicitly to read, should still get capped to a + # valid size / not raise an OverflowError for sizes larger than INT_MAX. + buffer = bytearray(INT_MAX + 10) + with open(os_helper.TESTFN, "rb") as fp: + length = os.readinto(fp.fileno(), buffer) + + # The test does not try to read more than 2 GiB at once because the + # operating system is free to return less bytes than requested. + self.assertEqual(length, 4) + self.assertEqual(buffer[:4], b'test') + def test_write(self): # os.write() accepts bytes- and buffer-like objects but not strings fd = os.open(os_helper.TESTFN, os.O_CREAT | os.O_WRONLY) @@ -710,7 +821,7 @@ class StatAttributeTests(unittest.TestCase): self.assertEqual(ctx.exception.errno, errno.EBADF) def check_file_attributes(self, result): - self.assertTrue(hasattr(result, 'st_file_attributes')) + self.assertHasAttr(result, 'st_file_attributes') self.assertTrue(isinstance(result.st_file_attributes, int)) self.assertTrue(0 <= result.st_file_attributes <= 0xFFFFFFFF) @@ -805,14 +916,28 @@ class UtimeTests(unittest.TestCase): set_time(filename, (atime_ns, mtime_ns)) st = os.stat(filename) - if support_subsecond: - self.assertAlmostEqual(st.st_atime, atime_ns * 1e-9, delta=1e-6) - self.assertAlmostEqual(st.st_mtime, mtime_ns * 1e-9, delta=1e-6) + if support.is_emscripten: + # Emscripten timestamps are roundtripped through a 53 bit integer of + # nanoseconds. If we want to represent ~50 years which is an 11 + # digits number of seconds: + # 2*log10(60) + log10(24) + log10(365) + log10(60) + log10(50) + # is about 11. Because 53 * log10(2) is about 16, we only have 5 + # digits worth of sub-second precision. + # Some day it would be good to fix this upstream. + delta=1e-5 + self.assertAlmostEqual(st.st_atime, atime_ns * 1e-9, delta=1e-5) + self.assertAlmostEqual(st.st_mtime, mtime_ns * 1e-9, delta=1e-5) + self.assertAlmostEqual(st.st_atime_ns, atime_ns, delta=1e9 * 1e-5) + self.assertAlmostEqual(st.st_mtime_ns, mtime_ns, delta=1e9 * 1e-5) else: - self.assertEqual(st.st_atime, atime_ns * 1e-9) - self.assertEqual(st.st_mtime, mtime_ns * 1e-9) - self.assertEqual(st.st_atime_ns, atime_ns) - self.assertEqual(st.st_mtime_ns, mtime_ns) + if support_subsecond: + self.assertAlmostEqual(st.st_atime, atime_ns * 1e-9, delta=1e-6) + self.assertAlmostEqual(st.st_mtime, mtime_ns * 1e-9, delta=1e-6) + else: + self.assertEqual(st.st_atime, atime_ns * 1e-9) + self.assertEqual(st.st_mtime, mtime_ns * 1e-9) + self.assertEqual(st.st_atime_ns, atime_ns) + self.assertEqual(st.st_mtime_ns, mtime_ns) def test_utime(self): def set_time(filename, ns): @@ -825,9 +950,7 @@ class UtimeTests(unittest.TestCase): # Convert a number of nanosecond (int) to a number of seconds (float). # Round towards infinity by adding 0.5 nanosecond to avoid rounding # issue, os.utime() rounds towards minus infinity. - # XXX: RUSTPYTHON os.utime() use `[Duration::from_secs_f64](https://doc.rust-lang.org/std/time/struct.Duration.html#method.try_from_secs_f64)` - # return (ns * 1e-9) + 0.5e-9 - return (ns * 1e-9) + return (ns * 1e-9) + 0.5e-9 def test_utime_by_indexed(self): # pass times as floating-point seconds as the second indexed parameter @@ -1300,6 +1423,53 @@ class EnvironTests(mapping_tests.BasicTestMappingProtocol): self._test_underlying_process_env('_A_', '') self._test_underlying_process_env(overridden_key, original_value) + @unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: module 'os' has no attribute 'reload_environ' + def test_reload_environ(self): + # Test os.reload_environ() + has_environb = hasattr(os, 'environb') + + # Test with putenv() which doesn't update os.environ + os.environ['test_env'] = 'python_value' + os.putenv("test_env", "new_value") + self.assertEqual(os.environ['test_env'], 'python_value') + if has_environb: + self.assertEqual(os.environb[b'test_env'], b'python_value') + + os.reload_environ() + self.assertEqual(os.environ['test_env'], 'new_value') + if has_environb: + self.assertEqual(os.environb[b'test_env'], b'new_value') + + # Test with unsetenv() which doesn't update os.environ + os.unsetenv('test_env') + self.assertEqual(os.environ['test_env'], 'new_value') + if has_environb: + self.assertEqual(os.environb[b'test_env'], b'new_value') + + os.reload_environ() + self.assertNotIn('test_env', os.environ) + if has_environb: + self.assertNotIn(b'test_env', os.environb) + + if has_environb: + # test reload_environ() on os.environb with putenv() + os.environb[b'test_env'] = b'python_value2' + os.putenv("test_env", "new_value2") + self.assertEqual(os.environb[b'test_env'], b'python_value2') + self.assertEqual(os.environ['test_env'], 'python_value2') + + os.reload_environ() + self.assertEqual(os.environb[b'test_env'], b'new_value2') + self.assertEqual(os.environ['test_env'], 'new_value2') + + # test reload_environ() on os.environb with unsetenv() + os.unsetenv('test_env') + self.assertEqual(os.environb[b'test_env'], b'new_value2') + self.assertEqual(os.environ['test_env'], 'new_value2') + + os.reload_environ() + self.assertNotIn(b'test_env', os.environb) + self.assertNotIn('test_env', os.environ) class WalkTests(unittest.TestCase): """Tests for os.walk().""" @@ -1370,9 +1540,7 @@ class WalkTests(unittest.TestCase): else: self.sub2_tree = (sub2_path, ["SUB21"], ["tmp3"]) - if not support.is_emscripten: - # Emscripten fails with inaccessible directory - os.chmod(sub21_path, 0) + os.chmod(sub21_path, 0) try: os.listdir(sub21_path) except PermissionError: @@ -1668,9 +1836,6 @@ class FwalkTests(WalkTests): # check that listdir() returns consistent information self.assertEqual(set(os.listdir(rootfd)), set(dirs) | set(files)) - @unittest.skipIf( - support.is_emscripten, "Cannot dup stdout on Emscripten" - ) @unittest.skipIf( support.is_android, "dup return value is unpredictable on Android" ) @@ -1687,9 +1852,6 @@ class FwalkTests(WalkTests): self.addCleanup(os.close, newfd) self.assertEqual(newfd, minfd) - @unittest.skipIf( - support.is_emscripten, "Cannot dup stdout on Emscripten" - ) @unittest.skipIf( support.is_android, "dup return value is unpredictable on Android" ) @@ -1725,15 +1887,15 @@ class BytesWalkTests(WalkTests): bdirs[:] = list(map(os.fsencode, dirs)) bfiles[:] = list(map(os.fsencode, files)) - @unittest.expectedFailure # TODO: RUSTPYTHON; WalkTests doesn't have these methods + @unittest.expectedFailure # TODO: RUSTPYTHON; WalkTests doesn't have these methods def test_compare_to_walk(self): return super().test_compare_to_walk() - @unittest.expectedFailure # TODO: RUSTPYTHON; WalkTests doesn't have these methods + @unittest.expectedFailure # TODO: RUSTPYTHON; WalkTests doesn't have these methods def test_dir_fd(self): return super().test_dir_fd() - @unittest.expectedFailure # TODO: RUSTPYTHON; WalkTests doesn't have these methods + @unittest.expectedFailure # TODO: RUSTPYTHON; WalkTests doesn't have these methods def test_yields_correct_dir_fd(self): return super().test_yields_correct_dir_fd() @@ -1770,10 +1932,12 @@ class MakedirTests(unittest.TestCase): os.makedirs(path) @unittest.skipIf( - support.is_emscripten or support.is_wasi, - "Emscripten's/WASI's umask is a stub." + support.is_wasi, + "WASI's umask is a stub." ) def test_mode(self): + # Note: in some cases, the umask might already be 2 in which case this + # will pass even if os.umask is actually broken. with os_helper.temp_umask(0o002): base = os_helper.TESTFN parent = os.path.join(base, 'dir1') @@ -1786,8 +1950,8 @@ class MakedirTests(unittest.TestCase): self.assertEqual(os.stat(parent).st_mode & 0o777, 0o775) @unittest.skipIf( - support.is_emscripten or support.is_wasi, - "Emscripten's/WASI's umask is a stub." + support.is_wasi, + "WASI's umask is a stub." ) def test_exist_ok_existing_directory(self): path = os.path.join(os_helper.TESTFN, 'dir1') @@ -1804,8 +1968,8 @@ class MakedirTests(unittest.TestCase): os.makedirs(os.path.abspath('/'), exist_ok=True) @unittest.skipIf( - support.is_emscripten or support.is_wasi, - "Emscripten's/WASI's umask is a stub." + support.is_wasi, + "WASI's umask is a stub." ) def test_exist_ok_s_isgid_directory(self): path = os.path.join(os_helper.TESTFN, 'dir1') @@ -2035,7 +2199,7 @@ class GetRandomTests(unittest.TestCase): self.assertEqual(empty, b'') def test_getrandom_random(self): - self.assertTrue(hasattr(os, 'GRND_RANDOM')) + self.assertHasAttr(os, 'GRND_RANDOM') # Don't test os.getrandom(1, os.GRND_RANDOM) to not consume the rare # resource /dev/random @@ -2319,9 +2483,13 @@ class Win32ErrorTests(unittest.TestCase): @unittest.skipIf(support.is_wasi, "Cannot create invalid FD on WASI.") class TestInvalidFD(unittest.TestCase): - singles = ["fchdir", "dup", "fdatasync", "fstat", - "fstatvfs", "fsync", "tcgetpgrp", "ttyname"] - singles_fildes = {"fchdir", "fdatasync", "fsync"} + singles = ["fchdir", "dup", "fstat", "fstatvfs", "tcgetpgrp", "ttyname"] + singles_fildes = {"fchdir"} + # systemd-nspawn --suppress-sync=true does not verify fd passed + # fdatasync() and fsync(), and always returns success + if not support.in_systemd_nspawn_sync_suppressed(): + singles += ["fdatasync", "fsync"] + singles_fildes |= {"fdatasync", "fsync"} #singles.append("close") #We omit close because it doesn't raise an exception on some platforms def get_single(f): @@ -2379,10 +2547,6 @@ class TestInvalidFD(unittest.TestCase): self.check(os.dup2, 20) @unittest.skipUnless(hasattr(os, 'dup2'), 'test needs os.dup2()') - @unittest.skipIf( - support.is_emscripten, - "dup2() with negative fds is broken on Emscripten (see gh-102179)" - ) def test_dup2_negative_fd(self): valid_fd = os.open(__file__, os.O_RDONLY) self.addCleanup(os.close, valid_fd) @@ -2406,20 +2570,23 @@ class TestInvalidFD(unittest.TestCase): def test_fchown(self): self.check(os.fchown, -1, -1) - @unittest.expectedFailure # TODO: RUSTPYTHON + @unittest.expectedFailure # TODO: RUSTPYTHON; OSError: [Errno 22] Invalid argument: 0 @unittest.skipUnless(hasattr(os, 'fpathconf'), 'test needs os.fpathconf()') - @unittest.skipIf( - support.is_emscripten or support.is_wasi, - "musl libc issue on Emscripten/WASI, bpo-46390" - ) def test_fpathconf(self): self.assertIn("PC_NAME_MAX", os.pathconf_names) - self.check(os.pathconf, "PC_NAME_MAX") - self.check(os.fpathconf, "PC_NAME_MAX") self.check_bool(os.pathconf, "PC_NAME_MAX") self.check_bool(os.fpathconf, "PC_NAME_MAX") - @unittest.expectedFailure # TODO: RUSTPYTHON + @unittest.skipUnless(hasattr(os, 'fpathconf'), 'test needs os.fpathconf()') + @unittest.skipIf( + support.linked_to_musl(), + 'musl pathconf ignores the file descriptor and returns a constant', + ) + def test_fpathconf_bad_fd(self): + self.check(os.pathconf, "PC_NAME_MAX") + self.check(os.fpathconf, "PC_NAME_MAX") + + @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: RuntimeWarning not raised @unittest.skipUnless(hasattr(os, 'ftruncate'), 'test needs os.ftruncate()') def test_ftruncate(self): self.check(os.truncate, 0) @@ -2434,6 +2601,10 @@ class TestInvalidFD(unittest.TestCase): def test_read(self): self.check(os.read, 1) + @unittest.skipUnless(hasattr(os, 'readinto'), 'test needs os.readinto()') + def test_readinto(self): + self.check(os.readinto, bytearray(5)) + @unittest.skipUnless(hasattr(os, 'readv'), 'test needs os.readv()') def test_readv(self): buf = bytearray(10) @@ -2462,13 +2633,15 @@ class TestInvalidFD(unittest.TestCase): self.check(os.get_blocking) self.check(os.set_blocking, True) - @unittest.expectedFailure # TODO: RUSTPYTHON + @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: RuntimeWarning not raised + def test_fsync(self): + return super().test_fsync() + + @unittest.expectedFailure # TODO: RUSTPYTHON; NotADirectoryError: [Errno 20] Not a directory def test_fchdir(self): return super().test_fchdir() - @unittest.expectedFailure # TODO: RUSTPYTHON - def test_fsync(self): - return super().test_fsync() + @unittest.skipUnless(hasattr(os, 'link'), 'requires os.link') @@ -3355,9 +3528,6 @@ class DeviceEncodingTests(unittest.TestCase): @unittest.skipUnless(os.isatty(0) and not win32_is_iot() and (sys.platform.startswith('win') or (hasattr(locale, 'nl_langinfo') and hasattr(locale, 'CODESET'))), 'test requires a tty and either Windows or nl_langinfo(CODESET)') - @unittest.skipIf( - support.is_emscripten, "Cannot get encoding of stdin on Emscripten" - ) def test_device_encoding(self): encoding = os.device_encoding(0) self.assertIsNotNone(encoding) @@ -3485,8 +3655,8 @@ class SpawnTests(unittest.TestCase): exitcode = os.spawnl(os.P_WAIT, program, *args) self.assertEqual(exitcode, self.exitcode) + @unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; fix spawnve on Windows") @requires_os_func('spawnle') - @unittest.skipIf(sys.platform == 'win32', "TODO: RUSTPYTHON; fix spawnve on Windows") def test_spawnle(self): program, args = self.create_args(with_env=True) exitcode = os.spawnle(os.P_WAIT, program, *args, self.env) @@ -3514,8 +3684,8 @@ class SpawnTests(unittest.TestCase): exitcode = os.spawnv(os.P_WAIT, FakePath(program), args) self.assertEqual(exitcode, self.exitcode) + @unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; fix spawnve on Windows") @requires_os_func('spawnve') - @unittest.skipIf(sys.platform == 'win32', "TODO: RUSTPYTHON; fix spawnve on Windows") def test_spawnve(self): program, args = self.create_args(with_env=True) exitcode = os.spawnve(os.P_WAIT, program, args, self.env) @@ -3539,7 +3709,7 @@ class SpawnTests(unittest.TestCase): pid = os.spawnv(os.P_NOWAIT, program, args) support.wait_process(pid, exitcode=self.exitcode) - @unittest.expectedFailure # TODO: RUSTPYTHON; fix spawnv bytes + @unittest.expectedFailure # TODO: RUSTPYTHON; fix spawnv bytes @requires_os_func('spawnve') def test_spawnve_bytes(self): # Test bytes handling in parse_arglist and parse_envlist (#28114) @@ -3623,8 +3793,8 @@ class SpawnTests(unittest.TestCase): exitcode = spawn(os.P_WAIT, program, args, newenv) self.assertEqual(exitcode, 0) + @unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; fix spawnve on Windows") @requires_os_func('spawnve') - @unittest.skipIf(sys.platform == 'win32', "TODO: RUSTPYTHON; fix spawnve on Windows") def test_spawnve_invalid_env(self): self._test_invalid_env(os.spawnve) @@ -4989,7 +5159,7 @@ class TestScandir(unittest.TestCase): entry_lstat, os.name == 'nt') - @unittest.skipIf(sys.platform == 'linux', 'TODO: RUSTPYTHON; flaky test') + @unittest.skipIf(sys.platform == "linux", "TODO: RUSTPYTHON; flaky test") def test_attributes(self): link = os_helper.can_hardlink() symlink = os_helper.can_symlink() @@ -5171,7 +5341,7 @@ class TestScandir(unittest.TestCase): with self.assertRaises(TypeError): os.scandir(path_bytes) - @unittest.expectedFailure # TODO: RUSTPYTHON + @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: not found in {, , , , , , , , , , , } @unittest.skipUnless(os.listdir in os.supports_fd, 'fd support for listdir required for this test.') def test_fd(self): @@ -5253,7 +5423,7 @@ class TestScandir(unittest.TestCase): with self.check_no_resource_warning(): del iterator - @unittest.expectedFailure # TODO: RUSTPYTHON + @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: ResourceWarning not triggered def test_resource_warning(self): self.create_file("file.txt") self.create_file("file2.txt") @@ -5293,8 +5463,8 @@ class TestPEP519(unittest.TestCase): def test_pathlike(self): self.assertEqual('#feelthegil', self.fspath(FakePath('#feelthegil'))) - self.assertTrue(issubclass(FakePath, os.PathLike)) - self.assertTrue(isinstance(FakePath('x'), os.PathLike)) + self.assertIsSubclass(FakePath, os.PathLike) + self.assertIsInstance(FakePath('x'), os.PathLike) def test_garbage_in_exception_out(self): vapor = type('blah', (), {}) @@ -5320,8 +5490,8 @@ class TestPEP519(unittest.TestCase): # true on abstract implementation. class A(os.PathLike): pass - self.assertFalse(issubclass(FakePath, A)) - self.assertTrue(issubclass(FakePath, os.PathLike)) + self.assertNotIsSubclass(FakePath, A) + self.assertIsSubclass(FakePath, os.PathLike) def test_pathlike_class_getitem(self): self.assertIsInstance(os.PathLike[bytes], types.GenericAlias) @@ -5331,7 +5501,7 @@ class TestPEP519(unittest.TestCase): __slots__ = () def __fspath__(self): return '' - self.assertFalse(hasattr(A(), '__dict__')) + self.assertNotHasAttr(A(), '__dict__') def test_fspath_set_to_None(self): class Foo: @@ -5435,7 +5605,7 @@ class ForkTests(unittest.TestCase): self.assertEqual(err.decode("utf-8"), "") self.assertEqual(out.decode("utf-8"), "") - @unittest.expectedFailure # TODO: RUSTPYTHON + @unittest.expectedFailure # TODO: RUSTPYTHON; AssertionError: b"can't fork at interpreter shutdown" not found in b"Exception ignored in: \nAttributeError: 'NoneType' object has no attribute 'fork'\n" def test_fork_at_finalization(self): code = """if 1: import atexit diff --git a/Lib/test/test_popen.py b/Lib/test/test_popen.py index e6bfc480c..34cda35b1 100644 --- a/Lib/test/test_popen.py +++ b/Lib/test/test_popen.py @@ -57,14 +57,21 @@ class PopenTest(unittest.TestCase): def test_contextmanager(self): with os.popen("echo hello") as f: self.assertEqual(f.read(), "hello\n") + self.assertFalse(f.closed) + self.assertTrue(f.closed) def test_iterating(self): with os.popen("echo hello") as f: self.assertEqual(list(f), ["hello\n"]) + self.assertFalse(f.closed) + self.assertTrue(f.closed) def test_keywords(self): - with os.popen(cmd="exit 0", mode="w", buffering=-1): - pass + with os.popen(cmd="echo hello", mode="r", buffering=-1) as f: + self.assertEqual(f.read(), "hello\n") + self.assertFalse(f.closed) + self.assertTrue(f.closed) + if __name__ == "__main__": unittest.main()