mirrors/RustPython
0
0 Fork 0
mirror of https://github.com/RustPython/RustPython.git synced 2026-10-29 03:04:38 +08:00
Code Activity
Files
copilot/fix-isalnum-function-behavior
RustPython/extra_tests/snippets/stdlib_os.py
T
Jeong, YunWon d5d19e0611 Fix 33 reproduced fuzzing and static-review defects (#8514) 
* mmap: guard move() against dest/src past the mapping size

An offset larger than the mapping made `size - dest` underflow, producing
an out-of-range slice index panic. Check the high side first, as write() does.

Assisted-by: Claude

* exceptions: keep ImportError name/path/name_from out of an empty instance dict

ImportError.__reduce__ unwrapped get_arg(0), which is None when the exception
was constructed with no positional argument, aborting on pickle.dumps(ImportError()).
Fall back to the full args tuple there, and expose name/path/name_from as class
attributes defaulting to None so a bare ImportError() reduces to (cls, ()).

Assisted-by: Claude

* asyncio: stop unwrapping the _current_tasks downcast in current_task()

_asyncio._current_tasks is a reassignable module attribute; the three sibling
functions already degrade gracefully when it is not a dict, current_task() did not.

Assisted-by: Claude

* asyncio: raise TypeError when FutureIter.throw()'s exception class returns a non-exception

exc_type.call() goes through a Python-controlled __new__, so its result can be
any object; the downcast was unwrapped. Report it as a TypeError instead.

Assisted-by: Claude

* sequence: use a fallible reservation for sequence repetition

The guard only rejects a repeat whose per-element size crosses MAX_MEMORY_SIZE,
so (1,) * (10**12) still reached Vec::with_capacity and aborted in the allocator.
Reserve fallibly and surface a MemoryError.

Assisted-by: Claude

* collections: reject an oversized deque repetition with MemoryError

deque * sys.maxsize reached the allocator and aborted with a capacity overflow.
Apply the MAX_MEMORY_SIZE guard the other sequences already carry.

Assisted-by: Claude

* itertools: raise OverflowError for an out-of-range r argument

combinations/combinations_with_replacement/permutations narrowed r with
to_usize().unwrap(), so r=2**64 panicked instead of raising.

Assisted-by: Claude

* math: stream the generic sumprod path instead of collecting both iterables

The big-int path collected each argument into a Vec before multiplying, so an
unbounded iterable exhausted memory. Advance both iterators in lockstep and
accumulate, reporting a length mismatch when only one is exhausted.

Assisted-by: Claude

* _imp: raise TypeError for a second positional argument to find_frozen

withdata is keyword-only and unimplemented; passing it positionally hit an
unimplemented!() and aborted.

Assisted-by: Claude

* _typing: check _idfunc arity before indexing args

_typing._idfunc() with no argument indexed args[0] out of bounds.

Assisted-by: Claude

* exceptions: require a sequence for the ExceptionGroup excs argument

The argument was collected before being validated, so an unbounded iterable
such as itertools.count() exhausted memory. Reject a non-sequence up front.

Assisted-by: Claude

* mmap: treat an inverted find/rfind range as empty

find(b"x", 5, 2) built a slice whose start exceeded its end and panicked.

Assisted-by: Claude

* collections: give deque and defaultdict a GC traverse

Neither type opted into traversal, so a reference cycle through a deque or a
defaultdict was never collected.

Assisted-by: Claude

* itertools: opt the iterator types into GC traverse

cycle and its siblings hold Python references but declared no traverse, so a
cycle built through one of them leaked.

Assisted-by: Claude

* _ctypes: mask an out-of-range int instead of panicking

c_char_p(2**64) and pointer item assignment called .expect() on the narrowing
conversion. Wrap the value to the target width, which is what the C
implementation stores.

Assisted-by: Claude

* hashlib: import _hashlib when a hash module is loaded

The hash object type is a static type owned by _hashlib; calling _md5.md5()
without _hashlib imported hit an uninitialized static type and panicked.

Assisted-by: Claude

* _csv: fall back to the built-in dialect defaults when none is registered

The dialect table is empty until csv.py registers 'excel', so _csv.reader([])
unwrapped a missing entry and panicked.

Assisted-by: Claude

* builtins: reject surrogates in compile()/eval() source instead of panicking

expect_str() panics on a str containing surrogates; convert with try_as_utf8
so eval(chr(0xd800)) raises.

Assisted-by: Claude

* _suggestions: require a list for _generate_suggestions candidates

The argument was collected before validation, so an unbounded iterable
exhausted memory.

Assisted-by: Claude

* lzma: size the filter chain before consuming it

filters= was collected into a Vec before the length check, so an unbounded
iterable exhausted memory. Take the length through the sequence protocol first.

Assisted-by: Claude

* classmethod: opt into GC traverse

staticmethod already declares traverse; classmethod did not, so a cycle through
the wrapped callable leaked.

Assisted-by: Claude

* posix: reject unbounded iterables in posix_spawn and setgroups

argv, setsigdef, setsigmask and setgroups bound an ArgIterable and collected it
before validating, so an infinite generator exhausted memory. Require a
list/tuple for argv, validate signals while streaming, and take setgroups
through the sequence protocol.

Assisted-by: Claude

* _ctypes: size Array slice assignment and _argtypes_ before collecting

Both eagerly collected their argument, so an unbounded iterable exhausted
memory before the length check ran.

Assisted-by: Claude

* sys: propagate the breakpointhook warning failure

warn() was unwrapped, so an unimportable $PYTHONBREAKPOINT under
-W error panicked instead of raising.

Assisted-by: Claude

* structseq: require the sequence argument when constructing a struct sequence

A no-argument construction produced an empty backing tuple, and reading any
named field then indexed out of bounds.

Assisted-by: Claude

* Guard the hash slot dispatch against unbounded recursion

`PyObject::hash` invoked the type's hash slot directly, so an element-wise
`__hash__` following a deeply nested object graph recursed one native frame
per level and overflowed the stack. Wrap the dispatch in `with_recursion`,
matching the repr and rich-compare dispatches in the same file, so
`hash(x)` on a nested tuple/GenericAlias/slice raises `RecursionError`.

Assisted-by: Claude

* genericalias: guard the __parameters__ walk against unbounded recursion

`make_parameters_from_slice` recursed into every list/tuple argument with
nothing counting the frames, so `list[L]` for a self-referential or deeply
nested `L` overflowed the native stack at subscript time. Wrap the descent
in `with_recursion`, which makes the walk fallible; `PyGenericAlias::new`,
`from_args` and `make_parameters` now return `PyResult` and every caller
propagates.

Assisted-by: Claude

* Fix the type confusion in PyAtomicRef's Debug impl

`PyAtomicRef<T>` stores a pointer to a `Py<T>`, as `Deref`, `load_raw`,
`swap` and `Drop` all read it, but `Debug` cast it to a bare `T` and
formatted the object header as payload bytes. For `PyFunction`, whose
`code: PyAtomicRef<PyCode>` has a pointer-chasing `Debug`, that
dereferenced header words and segfaulted.

Cast to `PyObject` instead, which is what `Drop` already does and which
also covers the `PyAtomicRef<PyObject>` and `PyAtomicRef<Option<T>>`
instantiations that have no `Py<T>`.

`_asyncio._enter_task` reached this through `{:?}` in its "Cannot enter
into task" message; format the two tasks with their Python repr, which is
what the message is meant to show.

Assisted-by: Claude

* _sre: disallow instantiating Match

`Match` inherited `object.__new__`, so `M.__new__(M)` produced an instance
whose `regs`/`string`/`pattern` were never filled in by a match run; the
mapping subscript path read them and dereferenced garbage. The type has no
public constructor, so mark it `DISALLOW_INSTANTIATION`, which makes
`M.__new__(M)` raise `TypeError: cannot create 're.Match' instances`.

Assisted-by: Claude

* utils: return the empty repr instead of asserting a non-empty collection

`collection_repr` took the first element with an `.expect()` justified by the
caller's preceding non-empty check. Another thread clearing the collection
between that check and the iteration made the iterator yield nothing and
panicked the worker. Fall back to the caller-supplied empty form, which is
the text those callers already produce for an empty collection.

Assisted-by: Claude

* itertools: advance cycle's index atomically

`cycle.__next__` did `fetch_add(1)` and then reset the index to 0 in a
separate store, so two threads replaying the saved items could both read an
index past the end of `saved` and panic on the slice access. Do the advance
and the wrap in one `fetch_update`.

Assisted-by: Claude

* _ctypes: reject a float argument to a foreign call without argtypes

`conv_param`, the conversion used when `argtypes` is not set, converted its
argument with `try_int`, which goes through `__int__` and so accepts a
float. `libc.strlen(1.5)` therefore passed 1 where the callee expects a
`char *` and the callee dereferenced it. Match `ConvParam`, which does a
`PyLong_Check` and converts nothing: take the branch only for an `int` (or
a subclass, so `True` still converts), and let a float fall through to
"Don't know how to convert parameter".

The branch below it converted a float to a C double, but `try_int` claimed
every float before it could run, so it was dead; `CArgValue::Double`
existed only for that branch and both go. Typed doubles are unaffected —
they travel as `CArgValue::Typed` with code 'd'.

Assisted-by: Claude

* Report the iterator itself from PyIter's traverse

`Traverse for PyIter<O>` delegated to the inherent `PyObject::traverse` of
the object it wraps, so it reported that iterator's referents instead of
the iterator. The iterator's own reference to those referents was then
never subtracted during the collector's reference-subtraction pass, the
referents kept a non-zero gc_refs, and every object reachable from them
was classified as a root.

Any cycle running through a type with a `PyIter` field therefore survived
collection: `map`, `filter`, `zip`, `enumerate`, `reversed` and the
`itertools` iterators all leaked, while the same cycle through a `list`,
`tuple` or `list_iterator` collected. Report the wrapped object, as the
`PyObjectRef`, `PyRef<T>` and `PyStackRef` impls do.

`itertools.tee` still leaks: its shared buffer is a `PyRc<PyItertoolsTeeData>`
rather than a Python object, so the collector cannot see through it.

Assisted-by: Claude

* Remove the obsolete expectedFailure on test_code_module.test_unicode_error

Compiling a source string containing a lone surrogate now raises
UnicodeEncodeError, so the test passes.

Assisted-by: Claude

* itertools: reserve the combination indices fallibly

`combinations` and `combinations_with_replacement` built their index vector
with an infallible allocation, so an `r` that passes the ssize_t check but
does not fit in memory aborted the process instead of raising MemoryError.

Assisted-by: Claude

* Apply the struct sequence constructor's dict argument

`structseq(sequence, dict)` discarded its second argument, so the hidden
fields past `n_sequence_fields` — `tm_zone`, `st_atime` and friends — were
always None when constructed directly or restored from a `(sequence, dict)`
pickle, and a non-dict second argument was accepted silently.

Take the dict, require it to be a dict, and fill the hidden slots the
sequence did not cover from it. A key that names a field the sequence
already supplied, or no field at all, is now a
"got duplicate or unexpected field name(s)" TypeError instead of being
dropped. Both arguments are bindable by name, as `sequence` and `dict`.

`os.stat_result` and `os.statvfs_result` did not accept a second argument
at all; they and `time.struct_time` now share the parsing.

Assisted-by: Claude

* _imp: report the argument count in find_frozen's arity error

Assisted-by: Claude

* Add regression tests for the reproduced crashers

One case per catalog entry, each asserting the behavior the fix produces:
recursion guards, the memory-unsafety sites, the overflow and unbounded
allocation guards, the eager-collection rejections, the unwrap sites, and
the cycles the collector now breaks. Every expected value was checked
against CPython 3.14.

Assisted-by: Claude

* Tolerate a changed-size RuntimeError in the set repr stress test

The mutator and reader threads race on purpose; a "changed size during
iteration" RuntimeError is a valid outcome of that race and should not fail
the test. The panic it guards against is not.

Assisted-by: Claude

* Move the crash regression tests into per-module snippets

crash_regressions.py collected every reproduced crasher in one file. Split it
into the snippet for the module each case exercises, and add stdlib_gc.py,
stdlib_asyncio.py, stdlib_lzma.py, stdlib_threading_set_repr.py and
stdlib_threading_itertools_cycle.py for the cases with no existing home.

The suite runs every snippet under the host CPython too, so the checks only
RustPython raises are guarded by sys.implementation.name: the hash and
__parameters__ recursion depth, and the deque repeat overflow. The float
ctypes argument accepts either TypeError or ctypes.ArgumentError.

Assisted-by: Claude
2026-08-14 00:57:10 +09:00

548 lines
17 KiB
Python
Raw Permalink Blame History

import itertools
import os
import stat
import sys
import time
from testutils import assert_raises
assert os.name == "posix" or os.name == "nt"
fd = os.open("README.md", os.O_RDONLY)
assert fd > 0
os.close(fd)
assert_raises(OSError, lambda: os.read(fd, 10))
assert_raises(FileNotFoundError, lambda: os.open("DOES_NOT_EXIST", os.O_RDONLY))
assert_raises(FileNotFoundError, lambda: os.open("DOES_NOT_EXIST", os.O_WRONLY))
assert_raises(
FileNotFoundError, lambda: os.rename("DOES_NOT_EXIST", "DOES_NOT_EXIST 2")
)
# sendfile only supports in_fd as non-socket on linux and solaris
if hasattr(os, "sendfile") and sys.platform.startswith("linux"):
src_fd = os.open("README.md", os.O_RDONLY)
dest_fd = os.open("destination.md", os.O_RDWR | os.O_CREAT)
src_len = os.stat("README.md").st_size
bytes_sent = os.sendfile(dest_fd, src_fd, 0, src_len)
assert src_len == bytes_sent
os.lseek(dest_fd, 0, 0)
assert os.read(src_fd, src_len) == os.read(dest_fd, bytes_sent)
os.close(src_fd)
os.close(dest_fd)
os.remove("destination.md")
try:
os.open("DOES_NOT_EXIST", 0)
except OSError as err:
assert err.errno == 2
assert os.O_RDONLY == 0
assert os.O_WRONLY == 1
assert os.O_RDWR == 2
ENV_KEY = "TEST_ENV_VAR"
ENV_VALUE = "value"
assert os.getenv(ENV_KEY) is None
assert ENV_KEY not in os.environ
assert os.getenv(ENV_KEY, 5) == 5
os.environ[ENV_KEY] = ENV_VALUE
assert ENV_KEY in os.environ
assert os.getenv(ENV_KEY) == ENV_VALUE
del os.environ[ENV_KEY]
assert ENV_KEY not in os.environ
assert os.getenv(ENV_KEY) is None
if os.name == "posix":
os.putenv(ENV_KEY, ENV_VALUE)
os.unsetenv(ENV_KEY)
assert os.getenv(ENV_KEY) is None
assert os.curdir == "."
assert os.pardir == ".."
assert os.extsep == "."
if os.name == "nt":
assert os.sep == "\\"
assert os.linesep == "\r\n"
assert os.altsep == "/"
assert os.pathsep == ";"
else:
assert os.sep == "/"
assert os.linesep == "\n"
assert os.altsep is None
assert os.pathsep == ":"
assert os.fspath("Testing") == "Testing"
assert os.fspath(b"Testing") == b"Testing"
assert_raises(TypeError, lambda: os.fspath([1, 2, 3]))
class TestWithTempDir:
def __enter__(self):
if os.name == "nt":
base_folder = os.environ["TEMP"]
else:
base_folder = "/tmp"
name = os.path.join(base_folder, "rustpython_test_os_" + str(int(time.time())))
while os.path.isdir(name):
name = name + "_"
os.mkdir(name)
self.name = name
return name
def __exit__(self, exc_type, exc_val, exc_tb):
pass
class TestWithTempCurrentDir:
def __enter__(self):
self.prev_cwd = os.getcwd()
def __exit__(self, exc_type, exc_val, exc_tb):
os.chdir(self.prev_cwd)
FILE_NAME = "test1"
FILE_NAME2 = "test2"
FILE_NAME3 = "test3"
SYMLINK_FILE = "symlink"
SYMLINK_FOLDER = "symlink1"
FOLDER = "dir1"
CONTENT = b"testing"
CONTENT2 = b"rustpython"
CONTENT3 = b"BOYA"
with TestWithTempDir() as tmpdir:
fname = os.path.join(tmpdir, FILE_NAME)
fd = os.open(fname, os.O_WRONLY | os.O_CREAT | os.O_EXCL)
assert os.write(fd, CONTENT2) == len(CONTENT2)
os.close(fd)
fd = os.open(fname, os.O_WRONLY | os.O_APPEND)
assert os.write(fd, CONTENT3) == len(CONTENT3)
os.close(fd)
assert_raises(
FileExistsError, lambda: os.open(fname, os.O_WRONLY | os.O_CREAT | os.O_EXCL)
)
fd = os.open(fname, os.O_RDONLY)
assert os.read(fd, len(CONTENT2)) == CONTENT2
assert os.read(fd, len(CONTENT3)) == CONTENT3
os.close(fd)
fname3 = os.path.join(tmpdir, FILE_NAME3)
os.rename(fname, fname3)
assert os.path.exists(fname) is False
assert os.path.exists(fname3) is True
fd = os.open(fname3, 0)
assert os.read(fd, len(CONTENT2) + len(CONTENT3)) == CONTENT2 + CONTENT3
os.close(fd)
assert not os.isatty(fd)
# TODO: get os.lseek working on windows
if os.name != "nt":
fd = os.open(fname3, 0)
assert os.read(fd, len(CONTENT2)) == CONTENT2
assert os.read(fd, len(CONTENT3)) == CONTENT3
os.lseek(fd, len(CONTENT2), os.SEEK_SET)
assert os.read(fd, len(CONTENT3)) == CONTENT3
os.close(fd)
os.rename(fname3, fname)
assert os.path.exists(fname3) is False
assert os.path.exists(fname) is True
# wait a little bit to ensure that the file times aren't the same
time.sleep(0.1)
fname2 = os.path.join(tmpdir, FILE_NAME2)
with open(fname2, "wb"):
pass
folder = os.path.join(tmpdir, FOLDER)
os.mkdir(folder)
symlink_file = os.path.join(tmpdir, SYMLINK_FILE)
os.symlink(fname, symlink_file)
symlink_folder = os.path.join(tmpdir, SYMLINK_FOLDER)
os.symlink(folder, symlink_folder)
names = set()
paths = set()
dirs = set()
dirs_no_symlink = set()
files = set()
files_no_symlink = set()
symlinks = set()
for dir_entry in os.scandir(tmpdir):
names.add(dir_entry.name)
paths.add(dir_entry.path)
if dir_entry.is_dir():
assert stat.S_ISDIR(dir_entry.stat().st_mode) is True
dirs.add(dir_entry.name)
if dir_entry.is_dir(follow_symlinks=False):
assert stat.S_ISDIR(dir_entry.stat().st_mode) is True
dirs_no_symlink.add(dir_entry.name)
if dir_entry.is_file():
files.add(dir_entry.name)
assert stat.S_ISREG(dir_entry.stat().st_mode) is True
if dir_entry.is_file(follow_symlinks=False):
files_no_symlink.add(dir_entry.name)
assert stat.S_ISREG(dir_entry.stat().st_mode) is True
if dir_entry.is_symlink():
symlinks.add(dir_entry.name)
assert names == set([FILE_NAME, FILE_NAME2, FOLDER, SYMLINK_FILE, SYMLINK_FOLDER])
assert paths == set([fname, fname2, folder, symlink_file, symlink_folder])
assert dirs == set([FOLDER, SYMLINK_FOLDER])
assert dirs_no_symlink == set([FOLDER])
assert files == set([FILE_NAME, FILE_NAME2, SYMLINK_FILE])
assert files_no_symlink == set([FILE_NAME, FILE_NAME2])
assert symlinks == set([SYMLINK_FILE, SYMLINK_FOLDER])
# Stat
stat_res = os.stat(fname)
print(stat_res.st_mode)
assert stat.S_ISREG(stat_res.st_mode) is True
print(stat_res.st_ino)
print(stat_res.st_dev)
print(stat_res.st_nlink)
print(stat_res.st_uid)
print(stat_res.st_gid)
print(stat_res.st_size)
assert stat_res.st_size == len(CONTENT2) + len(CONTENT3)
print(stat_res.st_atime)
print(stat_res.st_ctime)
print(stat_res.st_mtime)
# test that it all of these times are greater than the 10 May 2019,
# when this test was written
assert stat_res.st_atime > 1557500000
assert stat_res.st_ctime > 1557500000
assert stat_res.st_mtime > 1557500000
bytes_stats_res = os.stat(fname.encode())
stat_file2 = os.stat(fname2)
print(stat_file2.st_ctime)
assert stat_file2.st_ctime > stat_res.st_ctime
# wait a little bit to ensures that the access/modify time will change
time.sleep(0.1)
old_atime = stat_res.st_atime
old_mtime = stat_res.st_mtime
fd = os.open(fname, os.O_RDWR)
os.write(fd, CONTENT)
os.fsync(fd)
# wait a little bit to ensures that the access/modify time is different
time.sleep(0.1)
os.read(fd, 1)
os.fsync(fd)
os.close(fd)
# retrieve update file stats
stat_res = os.stat(fname)
print(stat_res.st_atime)
print(stat_res.st_ctime)
print(stat_res.st_mtime)
if os.name != "nt":
# access time on windows has a resolution ranging from 1 hour to 1 day
# https://docs.microsoft.com/en-gb/windows/desktop/api/minwinbase/ns-minwinbase-filetime
assert stat_res.st_atime > old_atime, "Access time should be update"
assert stat_res.st_atime > stat_res.st_mtime
assert stat_res.st_mtime > old_mtime, "Modified time should be update"
# stat default is follow_symlink=True
os.stat(fname).st_ino == os.stat(symlink_file).st_ino
os.stat(fname).st_mode == os.stat(symlink_file).st_mode
os.stat(fname, follow_symlinks=False).st_ino == os.stat(
symlink_file, follow_symlinks=False
).st_ino
os.stat(fname, follow_symlinks=False).st_mode == os.stat(
symlink_file, follow_symlinks=False
).st_mode
# os.chmod
if os.name != "nt":
os.chmod(fname, 0o666)
assert oct(os.stat(fname).st_mode) == "0o100666"
# os.chown
if os.name != "nt":
# setup
root_in_posix = False
if hasattr(os, "geteuid"):
root_in_posix = os.geteuid() == 0
try:
import pwd
all_users = [u.pw_uid for u in pwd.getpwall()]
except (ImportError, AttributeError):
all_users = []
fname1 = os.path.join(tmpdir, FILE_NAME)
fname2 = os.path.join(tmpdir, FILE_NAME2)
fd = os.open(fname2, os.O_RDONLY)
# test chown without root permissions
if not root_in_posix and len(all_users) > 1:
uid_1, uid_2 = all_users[:2]
gid = os.stat(fname1).st_gid
assert_raises(PermissionError, lambda: os.chown(fname1, uid_1, gid))
assert_raises(PermissionError, lambda: os.chown(fname1, uid_2, gid))
# test chown with root perm and file name
if root_in_posix and len(all_users) > 1:
uid_1, uid_2 = all_users[:2]
gid = os.stat(fname1).st_gid
os.chown(fname1, uid_1, gid)
uid = os.stat(fname1).st_uid
assert uid == uid_1
os.chown(fname1, uid_2, gid)
uid = os.stat(fname1).st_uid
assert uid == uid_2
# test chown with root perm and file descriptor
if root_in_posix and len(all_users) > 1:
uid_1, uid_2 = all_users[:2]
gid = os.stat(fd).st_gid
os.chown(fd, uid_1, gid)
uid = os.stat(fd).st_uid
assert uid == uid_1
os.chown(fd, uid_2, gid)
uid = os.stat(fd).st_uid
assert uid == uid_2
# test gid change
if hasattr(os, "getgroups"):
groups = os.getgroups()
if len(groups) > 1:
gid_1, gid_2 = groups[:2]
uid = os.stat(fname1).st_uid
os.chown(fname1, uid, gid_1)
gid = os.stat(fname1).st_gid
assert gid == gid_1
os.chown(fname1, uid, gid_2)
gid = os.stat(fname1).st_gid
assert gid == gid_2
# teardown
os.close(fd)
# os.path
assert os.path.exists(fname) is True
assert os.path.exists("NO_SUCH_FILE") is False
assert os.path.isfile(fname) is True
assert os.path.isdir(folder) is True
assert os.path.isfile(folder) is False
assert os.path.isdir(fname) is False
assert os.path.basename(fname) == FILE_NAME
assert os.path.dirname(fname) == tmpdir
with TestWithTempCurrentDir():
os.chdir(tmpdir)
assert os.path.realpath(os.getcwd()) == os.path.realpath(tmpdir)
assert os.path.exists(FILE_NAME)
# supports
assert isinstance(os.supports_fd, set)
assert isinstance(os.supports_dir_fd, set)
assert isinstance(os.supports_follow_symlinks, set)
# get pid
assert isinstance(os.getpid(), int)
# unix
if "win" not in sys.platform:
assert isinstance(os.getegid(), int)
assert isinstance(os.getgid(), int)
assert isinstance(os.getsid(os.getpid()), int)
assert isinstance(os.getuid(), int)
assert isinstance(os.geteuid(), int)
assert isinstance(os.getppid(), int)
assert isinstance(os.getpgid(os.getpid()), int)
if os.getuid() != 0:
assert_raises(PermissionError, lambda: os.setgid(42))
assert_raises(PermissionError, lambda: os.setegid(42))
assert_raises(PermissionError, lambda: os.setpgid(os.getpid(), 42))
assert_raises(PermissionError, lambda: os.setuid(42))
assert_raises(PermissionError, lambda: os.seteuid(42))
assert_raises(PermissionError, lambda: os.setreuid(42, 42))
assert_raises(PermissionError, lambda: os.setresuid(42, 42, 42))
# pty
a, b = os.openpty()
assert isinstance(a, int)
assert isinstance(b, int)
assert isinstance(os.ttyname(b), str)
assert_raises(OSError, lambda: os.ttyname(9999))
os.close(b)
os.close(a)
# os.get_blocking, os.set_blocking
# TODO: windows support should be added for below functions
# os.pipe,
# os.set_inheritable, os.get_inheritable,
rfd, wfd = os.pipe()
try:
os.write(wfd, CONTENT2)
assert os.read(rfd, len(CONTENT2)) == CONTENT2
assert not os.get_inheritable(rfd)
assert not os.get_inheritable(wfd)
os.set_inheritable(rfd, True)
os.set_inheritable(wfd, True)
assert os.get_inheritable(rfd)
assert os.get_inheritable(wfd)
os.set_inheritable(rfd, True)
os.set_inheritable(wfd, True)
os.set_inheritable(rfd, True)
os.set_inheritable(wfd, True)
assert os.get_inheritable(rfd)
assert os.get_inheritable(wfd)
assert os.get_blocking(rfd)
assert os.get_blocking(wfd)
os.set_blocking(rfd, False)
os.set_blocking(wfd, False)
assert not os.get_blocking(rfd)
assert not os.get_blocking(wfd)
os.set_blocking(rfd, True)
os.set_blocking(wfd, True)
os.set_blocking(rfd, True)
os.set_blocking(wfd, True)
assert os.get_blocking(rfd)
assert os.get_blocking(wfd)
finally:
os.close(rfd)
os.close(wfd)
# os.pipe2
if sys.platform.startswith("linux") or sys.platform.startswith("freebsd"):
rfd, wfd = os.pipe2(0)
try:
os.write(wfd, CONTENT2)
assert os.read(rfd, len(CONTENT2)) == CONTENT2
assert os.get_inheritable(rfd)
assert os.get_inheritable(wfd)
assert os.get_blocking(rfd)
assert os.get_blocking(wfd)
finally:
os.close(rfd)
os.close(wfd)
rfd, wfd = os.pipe2(os.O_CLOEXEC | os.O_NONBLOCK)
try:
os.write(wfd, CONTENT2)
assert os.read(rfd, len(CONTENT2)) == CONTENT2
assert not os.get_inheritable(rfd)
assert not os.get_inheritable(wfd)
assert not os.get_blocking(rfd)
assert not os.get_blocking(wfd)
finally:
os.close(rfd)
os.close(wfd)
with TestWithTempDir() as tmpdir:
for i in range(0, 4):
file_name = os.path.join(tmpdir, "file" + str(i))
with open(file_name, "w") as f:
f.write("test")
expected_files = ["file0", "file1", "file2", "file3"]
dir_iter = os.scandir(tmpdir)
collected_files = [dir_entry.name for dir_entry in dir_iter]
assert set(collected_files) == set(expected_files)
with assert_raises(StopIteration):
next(dir_iter)
dir_iter.close()
expected_files_bytes = [
(file.encode(), os.path.join(tmpdir, file).encode()) for file in expected_files
]
dir_iter_bytes = os.scandir(tmpdir.encode())
collected_files_bytes = [
(dir_entry.name, dir_entry.path) for dir_entry in dir_iter_bytes
]
assert set(collected_files_bytes) == set(expected_files_bytes)
dir_iter_bytes.close()
collected_files = os.listdir(tmpdir)
assert set(collected_files) == set(expected_files)
collected_files = os.listdir(tmpdir.encode())
assert set(collected_files) == set([file.encode() for file in expected_files])
with TestWithTempCurrentDir():
os.chdir(tmpdir)
with os.scandir() as dir_iter:
collected_files = [dir_entry.name for dir_entry in dir_iter]
assert set(collected_files) == set(expected_files)
# system()
if os.name in ("posix", "nt"):
assert os.system("echo test") == 0
assert os.system("&") != 0
for arg in [None, 1, 1.0, TabError]:
assert_raises(TypeError, os.system, arg)
# Testing for os.pathconf_names
if not sys.platform.startswith("win"):
assert len(os.pathconf_names) > 0
assert "PC_NAME_MAX" in os.pathconf_names
for option, index in os.pathconf_names.items():
if sys.platform == "darwin":
# TODO: check why it fails
if option in ["PC_MAX_CANON", "PC_MAX_INPUT", "PC_VDISABLE"]:
continue
assert os.pathconf("/", index) == os.pathconf("/", option)
# os.access - test with empty path and nonexistent files
assert os.access("", os.F_OK) is False
assert os.access("", os.R_OK) is False
assert os.access("", os.W_OK) is False
assert os.access("", os.X_OK) is False
assert os.access("nonexistent_file_12345", os.F_OK) is False
assert os.access("nonexistent_file_12345", os.W_OK) is False
assert os.access("README.md", os.F_OK) is True
assert os.access("README.md", os.R_OK) is True
# argv and the group list are sequences; an arbitrary iterable must be rejected
# rather than drained.
if hasattr(os, "posix_spawn"):
with assert_raises(TypeError):
os.posix_spawn("/bin/true", map(str, itertools.count()), os.environ)
if hasattr(os, "setgroups"):
with assert_raises(TypeError):
os.setgroups(itertools.count())
# The optional second argument fills the fields past the visible ones, and the
# getters must not index past what __new__ stored.
assert os.stat_result(tuple(range(10))).st_atime == 7
assert os.stat_result(tuple(range(10)), {"st_atime": 1.5}).st_atime == 1.5
with assert_raises(TypeError):
os.stat_result(tuple(range(10)), ["st_atime"])
View Git Blame Copy Permalink