mschoi/RustPython
1
0
Fork 0
forked from Rust-related/RustPython
Code Pull Requests Actions Activity

Revert "/pull/1616#pullrequestreview-324914220"

Browse Source 
This reverts commit 8ad4aa84d5.
This commit is contained in:
maybe-rachel
2019-11-30 21:34:47 -07:00
parent 8ad4aa84d5
commit 614d2ed5c5
2 changed files with 221 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

411
tests/snippets/stdlib_os.py
View File

@@ -15,9 +15,9 @@ 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'))
try:
os.open('DOES_NOT_EXIST', 0)
os.open('DOES_NOT_EXIST', 0)
except OSError as err:
assert err.errno == 2
assert err.errno == 2
@@ -39,24 +39,24 @@ assert ENV_KEY not in os.environ
assert os.getenv(ENV_KEY) == None
if os.name == "posix":
os.putenv(ENV_KEY, ENV_VALUE)
os.unsetenv(ENV_KEY)
assert os.getenv(ENV_KEY) == None
os.putenv(ENV_KEY, ENV_VALUE)
os.unsetenv(ENV_KEY)
assert os.getenv(ENV_KEY) == 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 == ";"
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 == None
assert os.pathsep == ":"
assert os.sep == "/"
assert os.linesep == "\n"
assert os.altsep == None
assert os.pathsep == ":"
assert os.fspath("Testing") == "Testing"
assert os.fspath(b"Testing") == b"Testing"
@@ -100,161 +100,224 @@ 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)
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)
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))
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)
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) == False
assert os.path.exists(fname3) == True
fname3 = os.path.join(tmpdir, FILE_NAME3)
os.rename(fname, fname3)
assert os.path.exists(fname) == False
assert os.path.exists(fname3) == True
fd = os.open(fname3, 0)
assert os.read(fd, len(CONTENT2) + len(CONTENT3)) == CONTENT2 + CONTENT3
os.close(fd)
fd = os.open(fname3, 0)
assert os.read(fd, len(CONTENT2) + len(CONTENT3)) == CONTENT2 + CONTENT3
os.close(fd)
os.rename(fname3, fname)
assert os.path.exists(fname3) == False
assert os.path.exists(fname) == True
os.rename(fname3, fname)
assert os.path.exists(fname3) == False
assert os.path.exists(fname) == True
# wait a little bit to ensure that the file times aren't the same
time.sleep(0.1)
# 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)
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)
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) == True
dirs.add(dir_entry.name)
if dir_entry.is_dir(follow_symlinks=False):
assert stat.S_ISDIR(dir_entry.stat().st_mode) == 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) == 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) == True
if dir_entry.is_symlink():
symlinks.add(dir_entry.name)
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) == True
dirs.add(dir_entry.name)
if dir_entry.is_dir(follow_symlinks=False):
assert stat.S_ISDIR(dir_entry.stat().st_mode) == 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) == 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) == 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])
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) == 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
# Stat
stat_res = os.stat(fname)
print(stat_res.st_mode)
assert stat.S_ISREG(stat_res.st_mode) == 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
stat_file2 = os.stat(fname2)
print(stat_file2.st_ctime)
assert stat_file2.st_ctime > stat_res.st_ctime
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)
# 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
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)
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)
# 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)
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"
# 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
# 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.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.chmod
if os.name != "nt":
os.chmod(fname, 0o666)
assert oct(os.stat(fname).st_mode) == '0o100666'
# os.path
assert os.path.exists(fname) == True
assert os.path.exists("NO_SUCH_FILE") == False
assert os.path.isfile(fname) == True
assert os.path.isdir(folder) == True
assert os.path.isfile(folder) == False
assert os.path.isdir(fname) == False
# os.path
assert os.path.exists(fname) == True
assert os.path.exists("NO_SUCH_FILE") == False
assert os.path.isfile(fname) == True
assert os.path.isdir(folder) == True
assert os.path.isfile(folder) == False
assert os.path.isdir(fname) == False
assert os.path.basename(fname) == FILE_NAME
assert os.path.dirname(fname) == tmpdir
assert os.path.basename(fname) == FILE_NAME
assert os.path.dirname(fname) == tmpdir
with TestWithTempCurrentDir():
os.chdir(tmpdir)
assert os.getcwd() == os.path.realpath(tmpdir)
os.path.exists(FILE_NAME)
# os.get_blocking, os.set_blocking
# TODO: windows support should be added for below functions
# os.pipe,
# os.set_inheritable, os.get_inheritable,
if os.name != "nt":
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 TestWithTempCurrentDir():
os.chdir(tmpdir)
assert os.getcwd() == os.path.realpath(tmpdir)
os.path.exists(FILE_NAME)
# supports
assert isinstance(os.supports_fd, set)
@@ -290,68 +353,6 @@ if "win" not in sys.platform:
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))

24
vm/src/stdlib/os.rs
View File

@@ -989,15 +989,21 @@ fn os_set_blocking(fd: RawFd, blocking: bool, vm: &VirtualMachine) -> PyResult<(
fn os_pipe(vm: &VirtualMachine) -> PyResult<(RawFd, RawFd)> {
use nix::unistd::close;
use nix::unistd::pipe;
let (rfd, wfd) = pipe().map_err(|err| convert_nix_error(vm, err))?;
os_set_inheritable(rfd, false, vm)
.and_then(|_| os_set_inheritable(wfd, false, vm))
.or_else(|err| {
let _ = close(rfd);
let _ = close(wfd);
Err(err)
})?;
Ok((rfd, wfd))
match pipe() {
Err(err) => Err(convert_nix_error(vm, err)),
Ok(ok) => {
os_set_inheritable(ok.0, false, vm)
.and_then(|_| os_set_inheritable(ok.1, false, vm))
.or_else(|err| {
close(ok.0)
.and_then(|_| close(ok.1))
.or_else(|err| dbg!(Err(err)))
.ok();
Err(err)
})?;
Ok(ok)
}
}
}
// cfg from nix