Merge pull request #1822 from palaviv/glob

Add glob module
This commit is contained in:
Noah authored and GitHub committed 2020-03-21 09:37:48 -05:00
commit 44bd11fd49
4 files changed
+615 -40

No files matched your search

Vendored
+171
View File
@@ -0,0 +1,171 @@
"""Filename globbing utility."""
import os
import re
import fnmatch
__all__ = ["glob", "iglob", "escape"]
def glob(pathname, *, recursive=False):
"""Return a list of paths matching a pathname pattern.
The pattern may contain simple shell-style wildcards a la
fnmatch. However, unlike fnmatch, filenames starting with a
dot are special cases that are not matched by '*' and '?'
patterns.
If recursive is true, the pattern '**' will match any files and
zero or more directories and subdirectories.
"""
return list(iglob(pathname, recursive=recursive))
def iglob(pathname, *, recursive=False):
"""Return an iterator which yields the paths matching a pathname pattern.
The pattern may contain simple shell-style wildcards a la
fnmatch. However, unlike fnmatch, filenames starting with a
dot are special cases that are not matched by '*' and '?'
patterns.
If recursive is true, the pattern '**' will match any files and
zero or more directories and subdirectories.
"""
it = _iglob(pathname, recursive, False)
if recursive and _isrecursive(pathname):
s = next(it) # skip empty string
assert not s
return it
def _iglob(pathname, recursive, dironly):
dirname, basename = os.path.split(pathname)
if not has_magic(pathname):
assert not dironly
if basename:
if os.path.lexists(pathname):
yield pathname
else:
# Patterns ending with a slash should match only directories
if os.path.isdir(dirname):
yield pathname
return
if not dirname:
if recursive and _isrecursive(basename):
yield from _glob2(dirname, basename, dironly)
else:
yield from _glob1(dirname, basename, dironly)
return
# `os.path.split()` returns the argument itself as a dirname if it is a
# drive or UNC path. Prevent an infinite recursion if a drive or UNC path
# contains magic characters (i.e. r'\\?\C:').
if dirname != pathname and has_magic(dirname):
dirs = _iglob(dirname, recursive, True)
else:
dirs = [dirname]
if has_magic(basename):
if recursive and _isrecursive(basename):
glob_in_dir = _glob2
else:
glob_in_dir = _glob1
else:
glob_in_dir = _glob0
for dirname in dirs:
for name in glob_in_dir(dirname, basename, dironly):
yield os.path.join(dirname, name)
# These 2 helper functions non-recursively glob inside a literal directory.
# They return a list of basenames. _glob1 accepts a pattern while _glob0
# takes a literal basename (so it only has to check for its existence).
def _glob1(dirname, pattern, dironly):
names = list(_iterdir(dirname, dironly))
if not _ishidden(pattern):
names = (x for x in names if not _ishidden(x))
return fnmatch.filter(names, pattern)
def _glob0(dirname, basename, dironly):
if not basename:
# `os.path.split()` returns an empty basename for paths ending with a
# directory separator. 'q*x/' should match only directories.
if os.path.isdir(dirname):
return [basename]
else:
if os.path.lexists(os.path.join(dirname, basename)):
return [basename]
return []
# Following functions are not public but can be used by third-party code.
def glob0(dirname, pattern):
return _glob0(dirname, pattern, False)
def glob1(dirname, pattern):
return _glob1(dirname, pattern, False)
# This helper function recursively yields relative pathnames inside a literal
# directory.
def _glob2(dirname, pattern, dironly):
assert _isrecursive(pattern)
yield pattern[:0]
yield from _rlistdir(dirname, dironly)
# If dironly is false, yields all file names inside a directory.
# If dironly is true, yields only directory names.
def _iterdir(dirname, dironly):
if not dirname:
if isinstance(dirname, bytes):
dirname = bytes(os.curdir, 'ASCII')
else:
dirname = os.curdir
try:
with os.scandir(dirname) as it:
for entry in it:
try:
if not dironly or entry.is_dir():
yield entry.name
except OSError:
pass
except OSError:
return
# Recursively yields relative pathnames inside a literal directory.
def _rlistdir(dirname, dironly):
names = list(_iterdir(dirname, dironly))
for x in names:
if not _ishidden(x):
yield x
path = os.path.join(dirname, x) if dirname else x
for y in _rlistdir(path, dironly):
yield os.path.join(x, y)
magic_check = re.compile('([*?[])')
magic_check_bytes = re.compile(b'([*?[])')
def has_magic(s):
if isinstance(s, bytes):
match = magic_check_bytes.search(s)
else:
match = magic_check.search(s)
return match is not None
def _ishidden(path):
return path[0] in ('.', b'.'[0])
def _isrecursive(pattern):
if isinstance(pattern, bytes):
return pattern == b'**'
else:
return pattern == '**'
def escape(pathname):
"""Escape all special characters.
"""
# Escaping is done by wrapping any of "*?[" between square brackets.
# Metacharacters do not work in the drive part and shouldn't be escaped.
drive, pathname = os.path.splitdrive(pathname)
if isinstance(pathname, bytes):
pathname = magic_check_bytes.sub(br'[\1]', pathname)
else:
pathname = magic_check.sub(r'[\1]', pathname)
return drive + pathname
+318
View File
@@ -0,0 +1,318 @@
import glob
import os
import shutil
import sys
import unittest
from test.support import (TESTFN, skip_unless_symlink,
can_symlink, create_empty_file, change_cwd)
class GlobTests(unittest.TestCase):
def norm(self, *parts):
return os.path.normpath(os.path.join(self.tempdir, *parts))
def joins(self, *tuples):
return [os.path.join(self.tempdir, *parts) for parts in tuples]
def mktemp(self, *parts):
filename = self.norm(*parts)
base, file = os.path.split(filename)
if not os.path.exists(base):
os.makedirs(base)
create_empty_file(filename)
def setUp(self):
self.tempdir = TESTFN + "_dir"
self.mktemp('a', 'D')
self.mktemp('aab', 'F')
self.mktemp('.aa', 'G')
self.mktemp('.bb', 'H')
self.mktemp('aaa', 'zzzF')
self.mktemp('ZZZ')
self.mktemp('EF')
self.mktemp('a', 'bcd', 'EF')
self.mktemp('a', 'bcd', 'efg', 'ha')
if can_symlink():
os.symlink(self.norm('broken'), self.norm('sym1'))
os.symlink('broken', self.norm('sym2'))
os.symlink(os.path.join('a', 'bcd'), self.norm('sym3'))
def tearDown(self):
shutil.rmtree(self.tempdir)
def glob(self, *parts, **kwargs):
if len(parts) == 1:
pattern = parts[0]
else:
pattern = os.path.join(*parts)
p = os.path.join(self.tempdir, pattern)
res = glob.glob(p, **kwargs)
self.assertCountEqual(glob.iglob(p, **kwargs), res)
bres = [os.fsencode(x) for x in res]
self.assertCountEqual(glob.glob(os.fsencode(p), **kwargs), bres)
self.assertCountEqual(glob.iglob(os.fsencode(p), **kwargs), bres)
return res
def assertSequencesEqual_noorder(self, l1, l2):
l1 = list(l1)
l2 = list(l2)
self.assertEqual(set(l1), set(l2))
self.assertEqual(sorted(l1), sorted(l2))
def test_glob_literal(self):
eq = self.assertSequencesEqual_noorder
eq(self.glob('a'), [self.norm('a')])
eq(self.glob('a', 'D'), [self.norm('a', 'D')])
eq(self.glob('aab'), [self.norm('aab')])
eq(self.glob('zymurgy'), [])
res = glob.glob('*')
self.assertEqual({type(r) for r in res}, {str})
res = glob.glob(os.path.join(os.curdir, '*'))
self.assertEqual({type(r) for r in res}, {str})
res = glob.glob(b'*')
self.assertEqual({type(r) for r in res}, {bytes})
res = glob.glob(os.path.join(os.fsencode(os.curdir), b'*'))
self.assertEqual({type(r) for r in res}, {bytes})
def test_glob_one_directory(self):
eq = self.assertSequencesEqual_noorder
eq(self.glob('a*'), map(self.norm, ['a', 'aab', 'aaa']))
eq(self.glob('*a'), map(self.norm, ['a', 'aaa']))
eq(self.glob('.*'), map(self.norm, ['.aa', '.bb']))
eq(self.glob('?aa'), map(self.norm, ['aaa']))
eq(self.glob('aa?'), map(self.norm, ['aaa', 'aab']))
eq(self.glob('aa[ab]'), map(self.norm, ['aaa', 'aab']))
eq(self.glob('*q'), [])
def test_glob_nested_directory(self):
eq = self.assertSequencesEqual_noorder
if os.path.normcase("abCD") == "abCD":
# case-sensitive filesystem
eq(self.glob('a', 'bcd', 'E*'), [self.norm('a', 'bcd', 'EF')])
else:
# case insensitive filesystem
eq(self.glob('a', 'bcd', 'E*'), [self.norm('a', 'bcd', 'EF'),
self.norm('a', 'bcd', 'efg')])
eq(self.glob('a', 'bcd', '*g'), [self.norm('a', 'bcd', 'efg')])
def test_glob_directory_names(self):
eq = self.assertSequencesEqual_noorder
eq(self.glob('*', 'D'), [self.norm('a', 'D')])
eq(self.glob('*', '*a'), [])
eq(self.glob('a', '*', '*', '*a'),
[self.norm('a', 'bcd', 'efg', 'ha')])
eq(self.glob('?a?', '*F'), [self.norm('aaa', 'zzzF'),
self.norm('aab', 'F')])
def test_glob_directory_with_trailing_slash(self):
# Patterns ending with a slash shouldn't match non-dirs
res = glob.glob(self.norm('Z*Z') + os.sep)
self.assertEqual(res, [])
res = glob.glob(self.norm('ZZZ') + os.sep)
self.assertEqual(res, [])
# When there is a wildcard pattern which ends with os.sep, glob()
# doesn't blow up.
res = glob.glob(self.norm('aa*') + os.sep)
self.assertEqual(len(res), 2)
# either of these results is reasonable
self.assertIn(set(res), [
{self.norm('aaa'), self.norm('aab')},
{self.norm('aaa') + os.sep, self.norm('aab') + os.sep},
])
def test_glob_bytes_directory_with_trailing_slash(self):
# Same as test_glob_directory_with_trailing_slash, but with a
# bytes argument.
res = glob.glob(os.fsencode(self.norm('Z*Z') + os.sep))
self.assertEqual(res, [])
res = glob.glob(os.fsencode(self.norm('ZZZ') + os.sep))
self.assertEqual(res, [])
res = glob.glob(os.fsencode(self.norm('aa*') + os.sep))
self.assertEqual(len(res), 2)
# either of these results is reasonable
self.assertIn(set(res), [
{os.fsencode(self.norm('aaa')),
os.fsencode(self.norm('aab'))},
{os.fsencode(self.norm('aaa') + os.sep),
os.fsencode(self.norm('aab') + os.sep)},
])
@skip_unless_symlink
def test_glob_symlinks(self):
eq = self.assertSequencesEqual_noorder
eq(self.glob('sym3'), [self.norm('sym3')])
eq(self.glob('sym3', '*'), [self.norm('sym3', 'EF'),
self.norm('sym3', 'efg')])
self.assertIn(self.glob('sym3' + os.sep),
[[self.norm('sym3')], [self.norm('sym3') + os.sep]])
eq(self.glob('*', '*F'),
[self.norm('aaa', 'zzzF'),
self.norm('aab', 'F'), self.norm('sym3', 'EF')])
@skip_unless_symlink
def test_glob_broken_symlinks(self):
eq = self.assertSequencesEqual_noorder
eq(self.glob('sym*'), [self.norm('sym1'), self.norm('sym2'),
self.norm('sym3')])
eq(self.glob('sym1'), [self.norm('sym1')])
eq(self.glob('sym2'), [self.norm('sym2')])
@unittest.skipUnless(sys.platform == "win32", "Win32 specific test")
def test_glob_magic_in_drive(self):
eq = self.assertSequencesEqual_noorder
eq(glob.glob('*:'), [])
eq(glob.glob(b'*:'), [])
eq(glob.glob('?:'), [])
eq(glob.glob(b'?:'), [])
eq(glob.glob('\\\\?\\c:\\'), ['\\\\?\\c:\\'])
eq(glob.glob(b'\\\\?\\c:\\'), [b'\\\\?\\c:\\'])
eq(glob.glob('\\\\*\\*\\'), [])
eq(glob.glob(b'\\\\*\\*\\'), [])
def check_escape(self, arg, expected):
self.assertEqual(glob.escape(arg), expected)
self.assertEqual(glob.escape(os.fsencode(arg)), os.fsencode(expected))
@unittest.skip("TODO: RUSTPYTHON")
def test_escape(self):
check = self.check_escape
check('abc', 'abc')
check('[', '[[]')
check('?', '[?]')
check('*', '[*]')
check('[[_/*?*/_]]', '[[][[]_/[*][?][*]/_]]')
check('/[[_/*?*/_]]/', '/[[][[]_/[*][?][*]/_]]/')
@unittest.skipUnless(sys.platform == "win32", "Win32 specific test")
def test_escape_windows(self):
check = self.check_escape
check('?:?', '?:[?]')
check('*:*', '*:[*]')
check(r'\\?\c:\?', r'\\?\c:\[?]')
check(r'\\*\*\*', r'\\*\*\[*]')
check('//?/c:/?', '//?/c:/[?]')
check('//*/*/*', '//*/*/[*]')
def rglob(self, *parts, **kwargs):
return self.glob(*parts, recursive=True, **kwargs)
def test_recursive_glob(self):
eq = self.assertSequencesEqual_noorder
full = [('EF',), ('ZZZ',),
('a',), ('a', 'D'),
('a', 'bcd'),
('a', 'bcd', 'EF'),
('a', 'bcd', 'efg'),
('a', 'bcd', 'efg', 'ha'),
('aaa',), ('aaa', 'zzzF'),
('aab',), ('aab', 'F'),
]
if can_symlink():
full += [('sym1',), ('sym2',),
('sym3',),
('sym3', 'EF'),
('sym3', 'efg'),
('sym3', 'efg', 'ha'),
]
eq(self.rglob('**'), self.joins(('',), *full))
eq(self.rglob(os.curdir, '**'),
self.joins((os.curdir, ''), *((os.curdir,) + i for i in full)))
dirs = [('a', ''), ('a', 'bcd', ''), ('a', 'bcd', 'efg', ''),
('aaa', ''), ('aab', '')]
if can_symlink():
dirs += [('sym3', ''), ('sym3', 'efg', '')]
eq(self.rglob('**', ''), self.joins(('',), *dirs))
eq(self.rglob('a', '**'), self.joins(
('a', ''), ('a', 'D'), ('a', 'bcd'), ('a', 'bcd', 'EF'),
('a', 'bcd', 'efg'), ('a', 'bcd', 'efg', 'ha')))
eq(self.rglob('a**'), self.joins(('a',), ('aaa',), ('aab',)))
expect = [('a', 'bcd', 'EF'), ('EF',)]
if can_symlink():
expect += [('sym3', 'EF')]
eq(self.rglob('**', 'EF'), self.joins(*expect))
expect = [('a', 'bcd', 'EF'), ('aaa', 'zzzF'), ('aab', 'F'), ('EF',)]
if can_symlink():
expect += [('sym3', 'EF')]
eq(self.rglob('**', '*F'), self.joins(*expect))
eq(self.rglob('**', '*F', ''), [])
eq(self.rglob('**', 'bcd', '*'), self.joins(
('a', 'bcd', 'EF'), ('a', 'bcd', 'efg')))
eq(self.rglob('a', '**', 'bcd'), self.joins(('a', 'bcd')))
with change_cwd(self.tempdir):
join = os.path.join
eq(glob.glob('**', recursive=True), [join(*i) for i in full])
eq(glob.glob(join('**', ''), recursive=True),
[join(*i) for i in dirs])
eq(glob.glob(join('**', '*'), recursive=True),
[join(*i) for i in full])
eq(glob.glob(join(os.curdir, '**'), recursive=True),
[join(os.curdir, '')] + [join(os.curdir, *i) for i in full])
eq(glob.glob(join(os.curdir, '**', ''), recursive=True),
[join(os.curdir, '')] + [join(os.curdir, *i) for i in dirs])
eq(glob.glob(join(os.curdir, '**', '*'), recursive=True),
[join(os.curdir, *i) for i in full])
eq(glob.glob(join('**','zz*F'), recursive=True),
[join('aaa', 'zzzF')])
eq(glob.glob('**zz*F', recursive=True), [])
expect = [join('a', 'bcd', 'EF'), 'EF']
if can_symlink():
expect += [join('sym3', 'EF')]
eq(glob.glob(join('**', 'EF'), recursive=True), expect)
@skip_unless_symlink
class SymlinkLoopGlobTests(unittest.TestCase):
def test_selflink(self):
tempdir = TESTFN + "_dir"
os.makedirs(tempdir)
self.addCleanup(shutil.rmtree, tempdir)
with change_cwd(tempdir):
os.makedirs('dir')
create_empty_file(os.path.join('dir', 'file'))
os.symlink(os.curdir, os.path.join('dir', 'link'))
results = glob.glob('**', recursive=True)
self.assertEqual(len(results), len(set(results)))
results = set(results)
depth = 0
while results:
path = os.path.join(*(['dir'] + ['link'] * depth))
self.assertIn(path, results)
results.remove(path)
if not results:
break
path = os.path.join(path, 'file')
self.assertIn(path, results)
results.remove(path)
depth += 1
results = glob.glob(os.path.join('**', 'file'), recursive=True)
self.assertEqual(len(results), len(set(results)))
results = set(results)
depth = 0
while results:
path = os.path.join(*(['dir'] + ['link'] * depth + ['file']))
self.assertIn(path, results)
results.remove(path)
depth += 1
results = glob.glob(os.path.join('**', ''), recursive=True)
self.assertEqual(len(results), len(set(results)))
results = set(results)
depth = 0
while results:
path = os.path.join(*(['dir'] + ['link'] * depth + ['']))
self.assertIn(path, results)
results.remove(path)
depth += 1
if __name__ == "__main__":
unittest.main()
+17
View File
@@ -206,6 +206,8 @@ with TestWithTempDir() as tmpdir:
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
@@ -382,6 +384,21 @@ with TestWithTempDir() as tmpdir:
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:
+109 -40
View File
@@ -29,15 +29,15 @@ use super::errno::errors;
use crate::exceptions::PyBaseExceptionRef;
use crate::function::{IntoPyNativeFunc, OptionalArg, PyFuncArgs};
use crate::obj::objbyteinner::PyBytesLike;
use crate::obj::objbytes::PyBytesRef;
use crate::obj::objbytes::{PyBytes, PyBytesRef};
use crate::obj::objdict::PyDictRef;
use crate::obj::objint::PyIntRef;
use crate::obj::objiter;
use crate::obj::objset::PySet;
use crate::obj::objstr::PyStringRef;
use crate::obj::objstr::{self, PyString, PyStringRef};
use crate::obj::objtype::{self, PyClassRef};
use crate::pyobject::{
Either, ItemProtocol, PyClassImpl, PyObjectRef, PyRef, PyResult, PyValue, TryIntoRef,
Either, ItemProtocol, PyClassImpl, PyObjectRef, PyRef, PyResult, PyValue, TryFromObject,
TypeProtocol,
};
use crate::vm::VirtualMachine;
@@ -93,7 +93,61 @@ pub fn raw_file_number(handle: File) -> i64 {
unimplemented!();
}
fn make_path(_vm: &VirtualMachine, path: PyStringRef, dir_fd: &DirFd) -> PyStringRef {
#[derive(Debug, Copy, Clone)]
enum OutputMode {
String,
Bytes,
}
fn output_by_mode(val: String, mode: OutputMode, vm: &VirtualMachine) -> PyObjectRef {
match mode {
OutputMode::String => vm.ctx.new_str(val),
OutputMode::Bytes => vm.ctx.new_bytes(val.as_bytes().to_vec()),
}
}
struct PyPathLike {
path: String,
mode: OutputMode,
}
impl PyPathLike {
fn new_str(path: String) -> Self {
PyPathLike {
path,
mode: OutputMode::String,
}
}
}
impl TryFromObject for PyPathLike {
fn try_from_object(vm: &VirtualMachine, obj: PyObjectRef) -> PyResult<Self> {
// TODO: Support Path object
match_class!(match obj.clone() {
l @ PyString => {
Ok(PyPathLike {
path: l.as_str().to_owned(),
mode: OutputMode::String,
})
}
i @ PyBytes => {
let path = objstr::clone_value(&vm.call_method(i.as_object(), "decode", vec![])?);
Ok(PyPathLike {
path,
mode: OutputMode::Bytes,
})
}
_ => {
Err(vm.new_type_error(format!(
"path object need to be string or bytes not {}",
obj.class()
)))
}
})
}
}
fn make_path<'a>(_vm: &VirtualMachine, path: &'a str, dir_fd: &DirFd) -> &'a str {
if dir_fd.dir_fd.is_some() {
unimplemented!();
} else {
@@ -124,7 +178,7 @@ pub fn os_open(
let dir_fd = DirFd {
dir_fd: dir_fd.into_option(),
};
let fname = make_path(vm, name, &dir_fd);
let fname = make_path(vm, name.as_str(), &dir_fd);
let mut options = OpenOptions::new();
@@ -159,7 +213,7 @@ pub fn os_open(
options.custom_flags(flags);
let handle = options
.open(fname.as_str())
.open(fname)
.map_err(|err| convert_io_error(vm, err))?;
Ok(raw_file_number(handle))
@@ -389,8 +443,8 @@ fn os_write(fd: i64, data: PyBytesLike, vm: &VirtualMachine) -> PyResult {
}
fn os_remove(path: PyStringRef, dir_fd: DirFd, vm: &VirtualMachine) -> PyResult<()> {
let path = make_path(vm, path, &dir_fd);
fs::remove_file(path.as_str()).map_err(|err| convert_io_error(vm, err))
let path = make_path(vm, path.as_str(), &dir_fd);
fs::remove_file(path).map_err(|err| convert_io_error(vm, err))
}
fn os_mkdir(
@@ -399,8 +453,8 @@ fn os_mkdir(
dir_fd: DirFd,
vm: &VirtualMachine,
) -> PyResult<()> {
let path = make_path(vm, path, &dir_fd);
fs::create_dir(path.as_str()).map_err(|err| convert_io_error(vm, err))
let path = make_path(vm, path.as_str(), &dir_fd);
fs::create_dir(path).map_err(|err| convert_io_error(vm, err))
}
fn os_mkdirs(path: PyStringRef, vm: &VirtualMachine) -> PyResult<()> {
@@ -408,15 +462,19 @@ fn os_mkdirs(path: PyStringRef, vm: &VirtualMachine) -> PyResult<()> {
}
fn os_rmdir(path: PyStringRef, dir_fd: DirFd, vm: &VirtualMachine) -> PyResult<()> {
let path = make_path(vm, path, &dir_fd);
fs::remove_dir(path.as_str()).map_err(|err| convert_io_error(vm, err))
let path = make_path(vm, path.as_str(), &dir_fd);
fs::remove_dir(path).map_err(|err| convert_io_error(vm, err))
}
fn os_listdir(path: PyStringRef, vm: &VirtualMachine) -> PyResult {
let res: PyResult<Vec<PyObjectRef>> = fs::read_dir(path.as_str())
fn os_listdir(path: PyPathLike, vm: &VirtualMachine) -> PyResult {
let res: PyResult<Vec<PyObjectRef>> = fs::read_dir(&path.path)
.map_err(|err| convert_io_error(vm, err))?
.map(|entry| match entry {
Ok(path) => Ok(vm.ctx.new_str(path.file_name().into_string().unwrap())),
Ok(entry_path) => Ok(output_by_mode(
entry_path.file_name().into_string().unwrap(),
path.mode,
vm,
)),
Err(s) => Err(convert_io_error(vm, s)),
})
.collect();
@@ -492,8 +550,8 @@ fn _os_environ(vm: &VirtualMachine) -> PyDictRef {
}
fn os_readlink(path: PyStringRef, dir_fd: DirFd, vm: &VirtualMachine) -> PyResult {
let path = make_path(vm, path, &dir_fd);
let path = fs::read_link(path.as_str()).map_err(|err| convert_io_error(vm, err))?;
let path = make_path(vm, path.as_str(), &dir_fd);
let path = fs::read_link(path).map_err(|err| convert_io_error(vm, err))?;
let path = path.into_os_string().into_string().map_err(|_osstr| {
vm.new_unicode_decode_error("Can't convert OS path to valid UTF-8 string".into())
})?;
@@ -503,6 +561,7 @@ fn os_readlink(path: PyStringRef, dir_fd: DirFd, vm: &VirtualMachine) -> PyResul
#[derive(Debug)]
struct DirEntry {
entry: fs::DirEntry,
mode: OutputMode,
}
type DirEntryRef = PyRef<DirEntry>;
@@ -526,12 +585,14 @@ struct FollowSymlinks {
}
impl DirEntryRef {
fn name(self) -> String {
self.entry.file_name().into_string().unwrap()
fn name(self, vm: &VirtualMachine) -> PyObjectRef {
let file_name = self.entry.file_name().into_string().unwrap();
output_by_mode(file_name, self.mode, vm)
}
fn path(self) -> String {
self.entry.path().to_str().unwrap().to_owned()
fn path(self, vm: &VirtualMachine) -> PyObjectRef {
let path = self.entry.path().to_str().unwrap().to_owned();
output_by_mode(path, self.mode, vm)
}
#[allow(clippy::match_bool)]
@@ -575,7 +636,9 @@ impl DirEntryRef {
fn stat(self, dir_fd: DirFd, follow_symlinks: FollowSymlinks, vm: &VirtualMachine) -> PyResult {
os_stat(
Either::A(self.path().try_into_ref(vm)?),
Either::A(PyPathLike::new_str(
self.entry.path().to_str().unwrap().to_owned(),
)),
dir_fd,
follow_symlinks,
vm,
@@ -588,6 +651,7 @@ impl DirEntryRef {
struct ScandirIterator {
entries: RefCell<fs::ReadDir>,
exhausted: Cell<bool>,
mode: OutputMode,
}
impl PyValue for ScandirIterator {
@@ -606,7 +670,12 @@ impl ScandirIterator {
match self.entries.borrow_mut().next() {
Some(entry) => match entry {
Ok(entry) => Ok(DirEntry { entry }.into_ref(vm).into_object()),
Ok(entry) => Ok(DirEntry {
entry,
mode: self.mode,
}
.into_ref(vm)
.into_object()),
Err(s) => Err(convert_io_error(vm, s)),
},
None => {
@@ -637,16 +706,17 @@ impl ScandirIterator {
}
}
fn os_scandir(path: OptionalArg<PyStringRef>, vm: &VirtualMachine) -> PyResult {
fn os_scandir(path: OptionalArg<PyPathLike>, vm: &VirtualMachine) -> PyResult {
let path = match path {
OptionalArg::Present(ref path) => path.as_str(),
OptionalArg::Missing => ".",
OptionalArg::Present(path) => path,
OptionalArg::Missing => PyPathLike::new_str(".".to_owned()),
};
match fs::read_dir(path) {
match fs::read_dir(path.path) {
Ok(iter) => Ok(ScandirIterator {
entries: RefCell::new(iter),
exhausted: Cell::new(false),
mode: path.mode,
}
.into_ref(vm)
.into_object()),
@@ -697,7 +767,7 @@ fn to_seconds_from_nanos(secs: i64, nanos: i64) -> f64 {
#[cfg(unix)]
fn os_stat(
file: Either<PyStringRef, i64>,
file: Either<PyPathLike, i64>,
dir_fd: DirFd,
follow_symlinks: FollowSymlinks,
vm: &VirtualMachine,
@@ -714,8 +784,7 @@ fn os_stat(
let get_stats = move || -> io::Result<PyObjectRef> {
let meta = match file {
Either::A(path) => {
let path = make_path(vm, path, &dir_fd);
let path = path.as_str();
let path = make_path(vm, &path.path, &dir_fd);
if follow_symlinks.follow_symlinks {
fs::metadata(path)?
} else {
@@ -771,7 +840,7 @@ fn attributes_to_mode(attr: u32) -> u32 {
#[cfg(windows)]
fn os_stat(
file: Either<PyStringRef, i64>,
file: Either<PyPathLike, i64>,
_dir_fd: DirFd, // TODO: error
follow_symlinks: FollowSymlinks,
vm: &VirtualMachine,
@@ -781,8 +850,8 @@ fn os_stat(
let get_stats = move || -> io::Result<PyObjectRef> {
let meta = match file {
Either::A(path) => match follow_symlinks.follow_symlinks {
true => fs::metadata(path.as_str())?,
false => fs::symlink_metadata(path.as_str())?,
true => fs::metadata(path.path)?,
false => fs::symlink_metadata(path.path)?,
},
Either::B(fno) => {
let f = rust_file(fno);
@@ -825,7 +894,7 @@ fn os_stat(
unimplemented!();
}
fn os_lstat(file: Either<PyStringRef, i64>, dir_fd: DirFd, vm: &VirtualMachine) -> PyResult {
fn os_lstat(file: Either<PyPathLike, i64>, dir_fd: DirFd, vm: &VirtualMachine) -> PyResult {
os_stat(
file,
dir_fd,
@@ -844,8 +913,8 @@ fn os_symlink(
vm: &VirtualMachine,
) -> PyResult<()> {
use std::os::unix::fs as unix_fs;
let dst = make_path(vm, dst, &dir_fd);
unix_fs::symlink(src.as_str(), dst.as_str()).map_err(|err| convert_io_error(vm, err))
let dst = make_path(vm, dst.as_str(), &dir_fd);
unix_fs::symlink(src.as_str(), dst).map_err(|err| convert_io_error(vm, err))
}
#[cfg(windows)]
@@ -1005,16 +1074,16 @@ fn os_chmod(
vm: &VirtualMachine,
) -> PyResult<()> {
use std::os::unix::fs::PermissionsExt;
let path = make_path(vm, path, &dir_fd);
let path = make_path(vm, path.as_str(), &dir_fd);
let metadata = if follow_symlinks.follow_symlinks {
fs::metadata(path.as_str())
fs::metadata(path)
} else {
fs::symlink_metadata(path.as_str())
fs::symlink_metadata(path)
};
let meta = metadata.map_err(|err| convert_io_error(vm, err))?;
let mut permissions = meta.permissions();
permissions.set_mode(mode);
fs::set_permissions(path.as_str(), permissions).map_err(|err| convert_io_error(vm, err))?;
fs::set_permissions(path, permissions).map_err(|err| convert_io_error(vm, err))?;
Ok(())
}