gh-118507 : Refactor `nt._path_is*` to improve applicability for other cases (GH-118755)

(cherry picked from commit b64182550f)

Co-authored-by: Nice Zombies <nineteendo19d0@gmail.com>
This commit is contained in:
Miss Islington (bot) 2024-05-22 02:29:34 +02:00 committed by GitHub
parent f371565169
commit f15fbe9991
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 528 additions and 440 deletions

View File

@ -288,21 +288,6 @@ def dirname(p):
return split(p)[0]
# Is a path a junction?
if hasattr(os.stat_result, 'st_reparse_tag'):
def isjunction(path):
"""Test whether a path is a junction"""
try:
st = os.lstat(path)
except (OSError, ValueError, AttributeError):
return False
return st.st_reparse_tag == stat.IO_REPARSE_TAG_MOUNT_POINT
else:
# Use genericpath.isjunction as imported above
pass
# Is a path a mount point?
# Any drive letter root (eg c:\)
# Any share UNC (eg \\server\share)
@ -911,13 +896,15 @@ def commonpath(paths):
try:
# The isdir(), isfile(), islink() and exists() implementations in
# genericpath use os.stat(). This is overkill on Windows. Use simpler
# The isdir(), isfile(), islink(), exists() and lexists() implementations
# in genericpath use os.stat(). This is overkill on Windows. Use simpler
# builtin functions if they are available.
from nt import _path_isdir as isdir
from nt import _path_isfile as isfile
from nt import _path_islink as islink
from nt import _path_isjunction as isjunction
from nt import _path_exists as exists
from nt import _path_lexists as lexists
except ImportError:
# Use genericpath.* as imported above
pass

View File

@ -135,6 +135,9 @@ class GenericTest:
self.assertIs(self.pathmodule.exists(filename), False)
self.assertIs(self.pathmodule.exists(bfilename), False)
self.assertIs(self.pathmodule.lexists(filename), False)
self.assertIs(self.pathmodule.lexists(bfilename), False)
create_file(filename)
self.assertIs(self.pathmodule.exists(filename), True)
@ -145,14 +148,13 @@ class GenericTest:
self.assertIs(self.pathmodule.exists(filename + '\x00'), False)
self.assertIs(self.pathmodule.exists(bfilename + b'\x00'), False)
if self.pathmodule is not genericpath:
self.assertIs(self.pathmodule.lexists(filename), True)
self.assertIs(self.pathmodule.lexists(bfilename), True)
self.assertIs(self.pathmodule.lexists(filename), True)
self.assertIs(self.pathmodule.lexists(bfilename), True)
self.assertIs(self.pathmodule.lexists(filename + '\udfff'), False)
self.assertIs(self.pathmodule.lexists(bfilename + b'\xff'), False)
self.assertIs(self.pathmodule.lexists(filename + '\x00'), False)
self.assertIs(self.pathmodule.lexists(bfilename + b'\x00'), False)
self.assertIs(self.pathmodule.lexists(filename + '\udfff'), False)
self.assertIs(self.pathmodule.lexists(bfilename + b'\xff'), False)
self.assertIs(self.pathmodule.lexists(filename + '\x00'), False)
self.assertIs(self.pathmodule.lexists(bfilename + b'\x00'), False)
@unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
@unittest.skipIf(is_emscripten, "Emscripten pipe fds have no stat")

View File

@ -1095,6 +1095,27 @@ class TestNtpath(NtpathTestCase):
raise unittest.SkipTest('SystemDrive is not defined or malformed')
self.assertFalse(os.path.isfile('\\\\.\\' + drive))
@unittest.skipUnless(hasattr(os, 'pipe'), "need os.pipe()")
def test_isfile_anonymous_pipe(self):
pr, pw = os.pipe()
try:
self.assertFalse(ntpath.isfile(pr))
finally:
os.close(pr)
os.close(pw)
@unittest.skipIf(sys.platform != 'win32', "windows only")
def test_isfile_named_pipe(self):
import _winapi
named_pipe = f'//./PIPE/python_isfile_test_{os.getpid()}'
h = _winapi.CreateNamedPipe(named_pipe,
_winapi.PIPE_ACCESS_INBOUND,
0, 1, 0, 0, 0, 0)
try:
self.assertFalse(ntpath.isfile(named_pipe))
finally:
_winapi.CloseHandle(h)
@unittest.skipIf(sys.platform != 'win32', "windows only")
def test_con_device(self):
self.assertFalse(os.path.isfile(r"\\.\CON"))
@ -1114,8 +1135,12 @@ class TestNtpath(NtpathTestCase):
self.assertFalse(inspect.isfunction(os.path.isfile))
self.assertTrue(os.path.islink is nt._path_islink)
self.assertFalse(inspect.isfunction(os.path.islink))
self.assertTrue(os.path.isjunction is nt._path_isjunction)
self.assertFalse(inspect.isfunction(os.path.isjunction))
self.assertTrue(os.path.exists is nt._path_exists)
self.assertFalse(inspect.isfunction(os.path.exists))
self.assertTrue(os.path.lexists is nt._path_lexists)
self.assertFalse(inspect.isfunction(os.path.lexists))
@unittest.skipIf(os.name != 'nt', "Dev Drives only exist on Win32")
def test_isdevdrive(self):

View File

@ -0,0 +1 @@
Speedup :func:`os.path.isjunction` and :func:`os.path.lexists` on Windows with a native implementation.

View File

@ -2014,6 +2014,70 @@ exit:
#if defined(MS_WINDOWS)
PyDoc_STRVAR(os__path_exists__doc__,
"_path_exists($module, path, /)\n"
"--\n"
"\n"
"Test whether a path exists. Returns False for broken symbolic links.");
#define OS__PATH_EXISTS_METHODDEF \
{"_path_exists", (PyCFunction)os__path_exists, METH_O, os__path_exists__doc__},
static int
os__path_exists_impl(PyObject *module, PyObject *path);
static PyObject *
os__path_exists(PyObject *module, PyObject *path)
{
PyObject *return_value = NULL;
int _return_value;
_return_value = os__path_exists_impl(module, path);
if ((_return_value == -1) && PyErr_Occurred()) {
goto exit;
}
return_value = PyBool_FromLong((long)_return_value);
exit:
return return_value;
}
#endif /* defined(MS_WINDOWS) */
#if defined(MS_WINDOWS)
PyDoc_STRVAR(os__path_lexists__doc__,
"_path_lexists($module, path, /)\n"
"--\n"
"\n"
"Test whether a path exists. Returns True for broken symbolic links.");
#define OS__PATH_LEXISTS_METHODDEF \
{"_path_lexists", (PyCFunction)os__path_lexists, METH_O, os__path_lexists__doc__},
static int
os__path_lexists_impl(PyObject *module, PyObject *path);
static PyObject *
os__path_lexists(PyObject *module, PyObject *path)
{
PyObject *return_value = NULL;
int _return_value;
_return_value = os__path_lexists_impl(module, path);
if ((_return_value == -1) && PyErr_Occurred()) {
goto exit;
}
return_value = PyBool_FromLong((long)_return_value);
exit:
return return_value;
}
#endif /* defined(MS_WINDOWS) */
#if defined(MS_WINDOWS)
PyDoc_STRVAR(os__path_isdir__doc__,
"_path_isdir($module, /, s)\n"
"--\n"
@ -2023,8 +2087,8 @@ PyDoc_STRVAR(os__path_isdir__doc__,
#define OS__PATH_ISDIR_METHODDEF \
{"_path_isdir", _PyCFunction_CAST(os__path_isdir), METH_FASTCALL|METH_KEYWORDS, os__path_isdir__doc__},
static PyObject *
os__path_isdir_impl(PyObject *module, PyObject *s);
static int
os__path_isdir_impl(PyObject *module, PyObject *path);
static PyObject *
os__path_isdir(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames)
@ -2056,14 +2120,19 @@ os__path_isdir(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObje
};
#undef KWTUPLE
PyObject *argsbuf[1];
PyObject *s;
PyObject *path;
int _return_value;
args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser, 1, 1, 0, argsbuf);
if (!args) {
goto exit;
}
s = args[0];
return_value = os__path_isdir_impl(module, s);
path = args[0];
_return_value = os__path_isdir_impl(module, path);
if ((_return_value == -1) && PyErr_Occurred()) {
goto exit;
}
return_value = PyBool_FromLong((long)_return_value);
exit:
return return_value;
@ -2082,7 +2151,7 @@ PyDoc_STRVAR(os__path_isfile__doc__,
#define OS__PATH_ISFILE_METHODDEF \
{"_path_isfile", _PyCFunction_CAST(os__path_isfile), METH_FASTCALL|METH_KEYWORDS, os__path_isfile__doc__},
static PyObject *
static int
os__path_isfile_impl(PyObject *module, PyObject *path);
static PyObject *
@ -2116,72 +2185,18 @@ os__path_isfile(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObj
#undef KWTUPLE
PyObject *argsbuf[1];
PyObject *path;
int _return_value;
args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser, 1, 1, 0, argsbuf);
if (!args) {
goto exit;
}
path = args[0];
return_value = os__path_isfile_impl(module, path);
exit:
return return_value;
}
#endif /* defined(MS_WINDOWS) */
#if defined(MS_WINDOWS)
PyDoc_STRVAR(os__path_exists__doc__,
"_path_exists($module, /, path)\n"
"--\n"
"\n"
"Test whether a path exists. Returns False for broken symbolic links");
#define OS__PATH_EXISTS_METHODDEF \
{"_path_exists", _PyCFunction_CAST(os__path_exists), METH_FASTCALL|METH_KEYWORDS, os__path_exists__doc__},
static PyObject *
os__path_exists_impl(PyObject *module, PyObject *path);
static PyObject *
os__path_exists(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames)
{
PyObject *return_value = NULL;
#if defined(Py_BUILD_CORE) && !defined(Py_BUILD_CORE_MODULE)
#define NUM_KEYWORDS 1
static struct {
PyGC_Head _this_is_not_used;
PyObject_VAR_HEAD
PyObject *ob_item[NUM_KEYWORDS];
} _kwtuple = {
.ob_base = PyVarObject_HEAD_INIT(&PyTuple_Type, NUM_KEYWORDS)
.ob_item = { &_Py_ID(path), },
};
#undef NUM_KEYWORDS
#define KWTUPLE (&_kwtuple.ob_base.ob_base)
#else // !Py_BUILD_CORE
# define KWTUPLE NULL
#endif // !Py_BUILD_CORE
static const char * const _keywords[] = {"path", NULL};
static _PyArg_Parser _parser = {
.keywords = _keywords,
.fname = "_path_exists",
.kwtuple = KWTUPLE,
};
#undef KWTUPLE
PyObject *argsbuf[1];
PyObject *path;
args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser, 1, 1, 0, argsbuf);
if (!args) {
_return_value = os__path_isfile_impl(module, path);
if ((_return_value == -1) && PyErr_Occurred()) {
goto exit;
}
path = args[0];
return_value = os__path_exists_impl(module, path);
return_value = PyBool_FromLong((long)_return_value);
exit:
return return_value;
@ -2200,7 +2215,7 @@ PyDoc_STRVAR(os__path_islink__doc__,
#define OS__PATH_ISLINK_METHODDEF \
{"_path_islink", _PyCFunction_CAST(os__path_islink), METH_FASTCALL|METH_KEYWORDS, os__path_islink__doc__},
static PyObject *
static int
os__path_islink_impl(PyObject *module, PyObject *path);
static PyObject *
@ -2234,13 +2249,82 @@ os__path_islink(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObj
#undef KWTUPLE
PyObject *argsbuf[1];
PyObject *path;
int _return_value;
args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser, 1, 1, 0, argsbuf);
if (!args) {
goto exit;
}
path = args[0];
return_value = os__path_islink_impl(module, path);
_return_value = os__path_islink_impl(module, path);
if ((_return_value == -1) && PyErr_Occurred()) {
goto exit;
}
return_value = PyBool_FromLong((long)_return_value);
exit:
return return_value;
}
#endif /* defined(MS_WINDOWS) */
#if defined(MS_WINDOWS)
PyDoc_STRVAR(os__path_isjunction__doc__,
"_path_isjunction($module, /, path)\n"
"--\n"
"\n"
"Test whether a path is a junction");
#define OS__PATH_ISJUNCTION_METHODDEF \
{"_path_isjunction", _PyCFunction_CAST(os__path_isjunction), METH_FASTCALL|METH_KEYWORDS, os__path_isjunction__doc__},
static int
os__path_isjunction_impl(PyObject *module, PyObject *path);
static PyObject *
os__path_isjunction(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames)
{
PyObject *return_value = NULL;
#if defined(Py_BUILD_CORE) && !defined(Py_BUILD_CORE_MODULE)
#define NUM_KEYWORDS 1
static struct {
PyGC_Head _this_is_not_used;
PyObject_VAR_HEAD
PyObject *ob_item[NUM_KEYWORDS];
} _kwtuple = {
.ob_base = PyVarObject_HEAD_INIT(&PyTuple_Type, NUM_KEYWORDS)
.ob_item = { &_Py_ID(path), },
};
#undef NUM_KEYWORDS
#define KWTUPLE (&_kwtuple.ob_base.ob_base)
#else // !Py_BUILD_CORE
# define KWTUPLE NULL
#endif // !Py_BUILD_CORE
static const char * const _keywords[] = {"path", NULL};
static _PyArg_Parser _parser = {
.keywords = _keywords,
.fname = "_path_isjunction",
.kwtuple = KWTUPLE,
};
#undef KWTUPLE
PyObject *argsbuf[1];
PyObject *path;
int _return_value;
args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser, 1, 1, 0, argsbuf);
if (!args) {
goto exit;
}
path = args[0];
_return_value = os__path_isjunction_impl(module, path);
if ((_return_value == -1) && PyErr_Occurred()) {
goto exit;
}
return_value = PyBool_FromLong((long)_return_value);
exit:
return return_value;
@ -12097,6 +12181,14 @@ os__supports_virtual_terminal(PyObject *module, PyObject *Py_UNUSED(ignored))
#define OS__PATH_SPLITROOT_METHODDEF
#endif /* !defined(OS__PATH_SPLITROOT_METHODDEF) */
#ifndef OS__PATH_EXISTS_METHODDEF
#define OS__PATH_EXISTS_METHODDEF
#endif /* !defined(OS__PATH_EXISTS_METHODDEF) */
#ifndef OS__PATH_LEXISTS_METHODDEF
#define OS__PATH_LEXISTS_METHODDEF
#endif /* !defined(OS__PATH_LEXISTS_METHODDEF) */
#ifndef OS__PATH_ISDIR_METHODDEF
#define OS__PATH_ISDIR_METHODDEF
#endif /* !defined(OS__PATH_ISDIR_METHODDEF) */
@ -12105,14 +12197,14 @@ os__supports_virtual_terminal(PyObject *module, PyObject *Py_UNUSED(ignored))
#define OS__PATH_ISFILE_METHODDEF
#endif /* !defined(OS__PATH_ISFILE_METHODDEF) */
#ifndef OS__PATH_EXISTS_METHODDEF
#define OS__PATH_EXISTS_METHODDEF
#endif /* !defined(OS__PATH_EXISTS_METHODDEF) */
#ifndef OS__PATH_ISLINK_METHODDEF
#define OS__PATH_ISLINK_METHODDEF
#endif /* !defined(OS__PATH_ISLINK_METHODDEF) */
#ifndef OS__PATH_ISJUNCTION_METHODDEF
#define OS__PATH_ISJUNCTION_METHODDEF
#endif /* !defined(OS__PATH_ISJUNCTION_METHODDEF) */
#ifndef OS_NICE_METHODDEF
#define OS_NICE_METHODDEF
#endif /* !defined(OS_NICE_METHODDEF) */
@ -12660,4 +12752,4 @@ os__supports_virtual_terminal(PyObject *module, PyObject *Py_UNUSED(ignored))
#ifndef OS__SUPPORTS_VIRTUAL_TERMINAL_METHODDEF
#define OS__SUPPORTS_VIRTUAL_TERMINAL_METHODDEF
#endif /* !defined(OS__SUPPORTS_VIRTUAL_TERMINAL_METHODDEF) */
/*[clinic end generated code: output=c4698b47007cd6eb input=a9049054013a1b77]*/
/*[clinic end generated code: output=af5074c4ce4b19f1 input=a9049054013a1b77]*/

View File

@ -5088,384 +5088,363 @@ os__path_splitroot_impl(PyObject *module, path_t *path)
}
/*[clinic input]
os._path_isdir
#define PY_IFREG 1 // Regular file
#define PY_IFDIR 2 // Directory
#define PY_IFLNK 4 // Symlink
#define PY_IFMNT 8 // Mount Point (junction)
#define PY_IFLRP 16 // Link Reparse Point (name-surrogate, symlink, junction)
#define PY_IFRRP 32 // Regular Reparse Point
s: 'O'
static inline BOOL
_testInfo(DWORD attributes, DWORD reparseTag, BOOL diskDevice, int testedType)
{
switch (testedType) {
case PY_IFREG:
return diskDevice && attributes &&
!(attributes & FILE_ATTRIBUTE_DIRECTORY);
case PY_IFDIR:
return attributes & FILE_ATTRIBUTE_DIRECTORY;
case PY_IFLNK:
return (attributes & FILE_ATTRIBUTE_REPARSE_POINT) &&
reparseTag == IO_REPARSE_TAG_SYMLINK;
case PY_IFMNT:
return (attributes & FILE_ATTRIBUTE_REPARSE_POINT) &&
reparseTag == IO_REPARSE_TAG_MOUNT_POINT;
case PY_IFLRP:
return (attributes & FILE_ATTRIBUTE_REPARSE_POINT) &&
IsReparseTagNameSurrogate(reparseTag);
case PY_IFRRP:
return (attributes & FILE_ATTRIBUTE_REPARSE_POINT) &&
reparseTag && !IsReparseTagNameSurrogate(reparseTag);
}
return FALSE;
}
static BOOL
_testFileTypeByHandle(HANDLE hfile, int testedType, BOOL diskOnly)
{
assert(testedType == PY_IFREG || testedType == PY_IFDIR ||
testedType == PY_IFLNK || testedType == PY_IFMNT ||
testedType == PY_IFLRP || testedType == PY_IFRRP);
BOOL diskDevice = GetFileType(hfile) == FILE_TYPE_DISK;
if (diskOnly && !diskDevice) {
return FALSE;
}
if (testedType != PY_IFREG && testedType != PY_IFDIR) {
FILE_ATTRIBUTE_TAG_INFO info;
return GetFileInformationByHandleEx(hfile, FileAttributeTagInfo, &info,
sizeof(info)) &&
_testInfo(info.FileAttributes, info.ReparseTag, diskDevice,
testedType);
}
FILE_BASIC_INFO info;
return GetFileInformationByHandleEx(hfile, FileBasicInfo, &info,
sizeof(info)) &&
_testInfo(info.FileAttributes, 0, diskDevice, testedType);
}
static BOOL
_testFileTypeByName(LPCWSTR path, int testedType)
{
assert(testedType == PY_IFREG || testedType == PY_IFDIR ||
testedType == PY_IFLNK || testedType == PY_IFMNT ||
testedType == PY_IFLRP || testedType == PY_IFRRP);
FILE_STAT_BASIC_INFORMATION info;
if (_Py_GetFileInformationByName(path, FileStatBasicByNameInfo, &info,
sizeof(info)))
{
BOOL diskDevice = info.DeviceType == FILE_DEVICE_DISK ||
info.DeviceType == FILE_DEVICE_VIRTUAL_DISK ||
info.DeviceType == FILE_DEVICE_CD_ROM;
BOOL result = _testInfo(info.FileAttributes, info.ReparseTag,
diskDevice, testedType);
if (!result || (testedType != PY_IFREG && testedType != PY_IFDIR) ||
!(info.FileAttributes & FILE_ATTRIBUTE_REPARSE_POINT))
{
return result;
}
}
else if (_Py_GetFileInformationByName_ErrorIsTrustworthy(
GetLastError()))
{
return FALSE;
}
DWORD flags = FILE_FLAG_BACKUP_SEMANTICS;
if (testedType != PY_IFREG && testedType != PY_IFDIR) {
flags |= FILE_FLAG_OPEN_REPARSE_POINT;
}
HANDLE hfile = CreateFileW(path, FILE_READ_ATTRIBUTES, 0, NULL,
OPEN_EXISTING, flags, NULL);
if (hfile != INVALID_HANDLE_VALUE) {
BOOL result = _testFileTypeByHandle(hfile, testedType, FALSE);
CloseHandle(hfile);
return result;
}
switch (GetLastError()) {
case ERROR_ACCESS_DENIED:
case ERROR_SHARING_VIOLATION:
case ERROR_CANT_ACCESS_FILE:
case ERROR_INVALID_PARAMETER:
int rc;
STRUCT_STAT st;
if (testedType == PY_IFREG || testedType == PY_IFDIR) {
rc = STAT(path, &st);
}
else {
// PY_IFRRP is not generally supported in this case, except for
// unhandled reparse points such as IO_REPARSE_TAG_APPEXECLINK.
rc = LSTAT(path, &st);
}
if (!rc) {
return _testInfo(st.st_file_attributes, st.st_reparse_tag,
st.st_mode & S_IFREG, testedType);
}
}
return FALSE;
}
static BOOL
_testFileExistsByName(LPCWSTR path, BOOL followLinks)
{
FILE_STAT_BASIC_INFORMATION info;
if (_Py_GetFileInformationByName(path, FileStatBasicByNameInfo, &info,
sizeof(info)))
{
if (!(info.FileAttributes & FILE_ATTRIBUTE_REPARSE_POINT) ||
!followLinks && IsReparseTagNameSurrogate(info.ReparseTag))
{
return TRUE;
}
}
else if (_Py_GetFileInformationByName_ErrorIsTrustworthy(
GetLastError()))
{
return FALSE;
}
DWORD flags = FILE_FLAG_BACKUP_SEMANTICS;
if (!followLinks) {
flags |= FILE_FLAG_OPEN_REPARSE_POINT;
}
HANDLE hfile = CreateFileW(path, FILE_READ_ATTRIBUTES, 0, NULL,
OPEN_EXISTING, flags, NULL);
if (hfile != INVALID_HANDLE_VALUE) {
if (followLinks) {
CloseHandle(hfile);
return TRUE;
}
// Regular Reparse Points (PY_IFRRP) have to be traversed.
BOOL result = _testFileTypeByHandle(hfile, PY_IFRRP, FALSE);
CloseHandle(hfile);
if (!result) {
return TRUE;
}
hfile = CreateFileW(path, FILE_READ_ATTRIBUTES, 0, NULL, OPEN_EXISTING,
FILE_FLAG_BACKUP_SEMANTICS, NULL);
if (hfile != INVALID_HANDLE_VALUE) {
CloseHandle(hfile);
return TRUE;
}
}
switch (GetLastError()) {
case ERROR_ACCESS_DENIED:
case ERROR_SHARING_VIOLATION:
case ERROR_CANT_ACCESS_FILE:
case ERROR_INVALID_PARAMETER:
STRUCT_STAT _st;
return followLinks ? !STAT(path, &_st): !LSTAT(path, &_st);
}
return FALSE;
}
static int
_testFileExists(path_t *_path, PyObject *path, BOOL followLinks)
{
BOOL result = FALSE;
if (!path_converter(path, _path)) {
path_cleanup(_path);
if (PyErr_ExceptionMatches(PyExc_ValueError)) {
PyErr_Clear();
return FALSE;
}
return -1;
}
Py_BEGIN_ALLOW_THREADS
if (_path->fd != -1) {
HANDLE hfile = _Py_get_osfhandle_noraise(_path->fd);
if (hfile != INVALID_HANDLE_VALUE) {
if (GetFileType(hfile) != FILE_TYPE_UNKNOWN || !GetLastError()) {
result = TRUE;
}
}
}
else if (_path->wide) {
result = _testFileExistsByName(_path->wide, followLinks);
}
Py_END_ALLOW_THREADS
path_cleanup(_path);
return result;
}
static int
_testFileType(path_t *_path, PyObject *path, int testedType)
{
BOOL result = FALSE;
if (!path_converter(path, _path)) {
path_cleanup(_path);
if (PyErr_ExceptionMatches(PyExc_ValueError)) {
PyErr_Clear();
return FALSE;
}
return -1;
}
Py_BEGIN_ALLOW_THREADS
if (_path->fd != -1) {
HANDLE hfile = _Py_get_osfhandle_noraise(_path->fd);
if (hfile != INVALID_HANDLE_VALUE) {
result = _testFileTypeByHandle(hfile, testedType, TRUE);
}
}
else if (_path->wide) {
result = _testFileTypeByName(_path->wide, testedType);
}
Py_END_ALLOW_THREADS
path_cleanup(_path);
return result;
}
/*[clinic input]
os._path_exists -> bool
path: object
/
Test whether a path exists. Returns False for broken symbolic links.
[clinic start generated code]*/
static int
os__path_exists_impl(PyObject *module, PyObject *path)
/*[clinic end generated code: output=8f784b3abf9f8588 input=2777da15bc4ba5a3]*/
{
path_t _path = PATH_T_INITIALIZE("_path_exists", "path", 0, 1);
return _testFileExists(&_path, path, TRUE);
}
/*[clinic input]
os._path_lexists -> bool
path: object
/
Test whether a path exists. Returns True for broken symbolic links.
[clinic start generated code]*/
static int
os__path_lexists_impl(PyObject *module, PyObject *path)
/*[clinic end generated code: output=fec4a91cf4ffccf1 input=8843d4d6d4e7c779]*/
{
path_t _path = PATH_T_INITIALIZE("_path_lexists", "path", 0, 1);
return _testFileExists(&_path, path, FALSE);
}
/*[clinic input]
os._path_isdir -> bool
s as path: object
Return true if the pathname refers to an existing directory.
[clinic start generated code]*/
static PyObject *
os__path_isdir_impl(PyObject *module, PyObject *s)
/*[clinic end generated code: output=9d87ab3c8b8a4e61 input=c17f7ef21d22d64e]*/
static int
os__path_isdir_impl(PyObject *module, PyObject *path)
/*[clinic end generated code: output=0504fd403f369701 input=2cb54dd97eb970f7]*/
{
HANDLE hfile;
BOOL close_file = TRUE;
FILE_BASIC_INFO info;
path_t _path = PATH_T_INITIALIZE("isdir", "s", 0, 1);
int result;
BOOL slow_path = TRUE;
FILE_STAT_BASIC_INFORMATION statInfo;
if (!path_converter(s, &_path)) {
path_cleanup(&_path);
if (PyErr_ExceptionMatches(PyExc_ValueError)) {
PyErr_Clear();
Py_RETURN_FALSE;
}
return NULL;
}
Py_BEGIN_ALLOW_THREADS
if (_path.wide) {
if (_Py_GetFileInformationByName(_path.wide, FileStatBasicByNameInfo,
&statInfo, sizeof(statInfo))) {
if (!(statInfo.FileAttributes & FILE_ATTRIBUTE_REPARSE_POINT)) {
slow_path = FALSE;
result = statInfo.FileAttributes & FILE_ATTRIBUTE_DIRECTORY;
} else if (!(statInfo.FileAttributes & FILE_ATTRIBUTE_DIRECTORY)) {
slow_path = FALSE;
result = 0;
}
} else if (_Py_GetFileInformationByName_ErrorIsTrustworthy(GetLastError())) {
slow_path = FALSE;
result = 0;
}
}
if (slow_path) {
if (_path.fd != -1) {
hfile = _Py_get_osfhandle_noraise(_path.fd);
close_file = FALSE;
}
else {
hfile = CreateFileW(_path.wide, FILE_READ_ATTRIBUTES, 0, NULL,
OPEN_EXISTING, FILE_FLAG_BACKUP_SEMANTICS, NULL);
}
if (hfile != INVALID_HANDLE_VALUE) {
if (GetFileInformationByHandleEx(hfile, FileBasicInfo, &info,
sizeof(info)))
{
result = info.FileAttributes & FILE_ATTRIBUTE_DIRECTORY;
}
else {
result = 0;
}
if (close_file) {
CloseHandle(hfile);
}
}
else {
STRUCT_STAT st;
switch (GetLastError()) {
case ERROR_ACCESS_DENIED:
case ERROR_SHARING_VIOLATION:
case ERROR_CANT_ACCESS_FILE:
case ERROR_INVALID_PARAMETER:
if (STAT(_path.wide, &st)) {
result = 0;
}
else {
result = S_ISDIR(st.st_mode);
}
break;
default:
result = 0;
}
}
}
Py_END_ALLOW_THREADS
path_cleanup(&_path);
if (result) {
Py_RETURN_TRUE;
}
Py_RETURN_FALSE;
path_t _path = PATH_T_INITIALIZE("_path_isdir", "s", 0, 1);
return _testFileType(&_path, path, PY_IFDIR);
}
/*[clinic input]
os._path_isfile
os._path_isfile -> bool
path: 'O'
path: object
Test whether a path is a regular file
[clinic start generated code]*/
static PyObject *
static int
os__path_isfile_impl(PyObject *module, PyObject *path)
/*[clinic end generated code: output=2394ed7c4b5cfd85 input=de22d74960ade365]*/
/*[clinic end generated code: output=b40d620efe5a896f input=54b428a310debaea]*/
{
HANDLE hfile;
BOOL close_file = TRUE;
FILE_BASIC_INFO info;
path_t _path = PATH_T_INITIALIZE("isfile", "path", 0, 1);
int result;
BOOL slow_path = TRUE;
FILE_STAT_BASIC_INFORMATION statInfo;
if (!path_converter(path, &_path)) {
path_cleanup(&_path);
if (PyErr_ExceptionMatches(PyExc_ValueError)) {
PyErr_Clear();
Py_RETURN_FALSE;
}
return NULL;
}
Py_BEGIN_ALLOW_THREADS
if (_path.wide) {
if (_Py_GetFileInformationByName(_path.wide, FileStatBasicByNameInfo,
&statInfo, sizeof(statInfo))) {
if (!(statInfo.FileAttributes & FILE_ATTRIBUTE_REPARSE_POINT)) {
slow_path = FALSE;
result = !(statInfo.FileAttributes & FILE_ATTRIBUTE_DIRECTORY);
} else if (statInfo.FileAttributes & FILE_ATTRIBUTE_DIRECTORY) {
slow_path = FALSE;
result = 0;
}
} else if (_Py_GetFileInformationByName_ErrorIsTrustworthy(GetLastError())) {
slow_path = FALSE;
result = 0;
}
}
if (slow_path) {
if (_path.fd != -1) {
hfile = _Py_get_osfhandle_noraise(_path.fd);
close_file = FALSE;
}
else {
hfile = CreateFileW(_path.wide, FILE_READ_ATTRIBUTES, 0, NULL,
OPEN_EXISTING, FILE_FLAG_BACKUP_SEMANTICS, NULL);
}
if (hfile != INVALID_HANDLE_VALUE) {
if (GetFileInformationByHandleEx(hfile, FileBasicInfo, &info,
sizeof(info)))
{
result = !(info.FileAttributes & FILE_ATTRIBUTE_DIRECTORY);
}
else {
result = 0;
}
if (close_file) {
CloseHandle(hfile);
}
}
else {
STRUCT_STAT st;
switch (GetLastError()) {
case ERROR_ACCESS_DENIED:
case ERROR_SHARING_VIOLATION:
case ERROR_CANT_ACCESS_FILE:
case ERROR_INVALID_PARAMETER:
if (STAT(_path.wide, &st)) {
result = 0;
}
else {
result = S_ISREG(st.st_mode);
}
break;
default:
result = 0;
}
}
}
Py_END_ALLOW_THREADS
path_cleanup(&_path);
if (result) {
Py_RETURN_TRUE;
}
Py_RETURN_FALSE;
path_t _path = PATH_T_INITIALIZE("_path_isfile", "path", 0, 1);
return _testFileType(&_path, path, PY_IFREG);
}
/*[clinic input]
os._path_exists
os._path_islink -> bool
path: 'O'
Test whether a path exists. Returns False for broken symbolic links
[clinic start generated code]*/
static PyObject *
os__path_exists_impl(PyObject *module, PyObject *path)
/*[clinic end generated code: output=f508c3b35e13a249 input=380f77cdfa0f7ae8]*/
{
HANDLE hfile;
BOOL close_file = TRUE;
path_t _path = PATH_T_INITIALIZE("exists", "path", 0, 1);
int result;
BOOL slow_path = TRUE;
FILE_STAT_BASIC_INFORMATION statInfo;
if (!path_converter(path, &_path)) {
path_cleanup(&_path);
if (PyErr_ExceptionMatches(PyExc_ValueError)) {
PyErr_Clear();
Py_RETURN_FALSE;
}
return NULL;
}
Py_BEGIN_ALLOW_THREADS
if (_path.wide) {
if (_Py_GetFileInformationByName(_path.wide, FileStatBasicByNameInfo,
&statInfo, sizeof(statInfo))) {
if (!(statInfo.FileAttributes & FILE_ATTRIBUTE_REPARSE_POINT)) {
slow_path = FALSE;
result = 1;
}
} else if (_Py_GetFileInformationByName_ErrorIsTrustworthy(GetLastError())) {
slow_path = FALSE;
result = 0;
}
}
if (slow_path) {
if (_path.fd != -1) {
hfile = _Py_get_osfhandle_noraise(_path.fd);
close_file = FALSE;
}
else {
hfile = CreateFileW(_path.wide, FILE_READ_ATTRIBUTES, 0, NULL,
OPEN_EXISTING, FILE_FLAG_BACKUP_SEMANTICS, NULL);
}
if (hfile != INVALID_HANDLE_VALUE) {
result = 1;
if (close_file) {
CloseHandle(hfile);
}
}
else {
STRUCT_STAT st;
switch (GetLastError()) {
case ERROR_ACCESS_DENIED:
case ERROR_SHARING_VIOLATION:
case ERROR_CANT_ACCESS_FILE:
case ERROR_INVALID_PARAMETER:
if (STAT(_path.wide, &st)) {
result = 0;
}
else {
result = 1;
}
break;
default:
result = 0;
}
}
}
Py_END_ALLOW_THREADS
path_cleanup(&_path);
if (result) {
Py_RETURN_TRUE;
}
Py_RETURN_FALSE;
}
/*[clinic input]
os._path_islink
path: 'O'
path: object
Test whether a path is a symbolic link
[clinic start generated code]*/
static PyObject *
static int
os__path_islink_impl(PyObject *module, PyObject *path)
/*[clinic end generated code: output=6d8640b1a390c054 input=38a3cb937ccf59bf]*/
/*[clinic end generated code: output=9d0cf8e4c640dfe6 input=b71fed60b9b2cd73]*/
{
HANDLE hfile;
BOOL close_file = TRUE;
FILE_ATTRIBUTE_TAG_INFO info;
path_t _path = PATH_T_INITIALIZE("islink", "path", 0, 1);
int result;
BOOL slow_path = TRUE;
FILE_STAT_BASIC_INFORMATION statInfo;
if (!path_converter(path, &_path)) {
path_cleanup(&_path);
if (PyErr_ExceptionMatches(PyExc_ValueError)) {
PyErr_Clear();
Py_RETURN_FALSE;
}
return NULL;
}
Py_BEGIN_ALLOW_THREADS
if (_path.wide) {
if (_Py_GetFileInformationByName(_path.wide, FileStatBasicByNameInfo,
&statInfo, sizeof(statInfo))) {
slow_path = FALSE;
if (statInfo.FileAttributes & FILE_ATTRIBUTE_REPARSE_POINT) {
result = (statInfo.ReparseTag == IO_REPARSE_TAG_SYMLINK);
}
else {
result = 0;
}
} else if (_Py_GetFileInformationByName_ErrorIsTrustworthy(GetLastError())) {
slow_path = FALSE;
result = 0;
}
}
if (slow_path) {
if (_path.fd != -1) {
hfile = _Py_get_osfhandle_noraise(_path.fd);
close_file = FALSE;
}
else {
hfile = CreateFileW(_path.wide, FILE_READ_ATTRIBUTES, 0, NULL,
OPEN_EXISTING,
FILE_FLAG_OPEN_REPARSE_POINT | FILE_FLAG_BACKUP_SEMANTICS,
NULL);
}
if (hfile != INVALID_HANDLE_VALUE) {
if (GetFileInformationByHandleEx(hfile, FileAttributeTagInfo, &info,
sizeof(info)))
{
result = (info.ReparseTag == IO_REPARSE_TAG_SYMLINK);
}
else {
result = 0;
}
if (close_file) {
CloseHandle(hfile);
}
}
else {
STRUCT_STAT st;
switch (GetLastError()) {
case ERROR_ACCESS_DENIED:
case ERROR_SHARING_VIOLATION:
case ERROR_CANT_ACCESS_FILE:
case ERROR_INVALID_PARAMETER:
if (LSTAT(_path.wide, &st)) {
result = 0;
}
else {
result = S_ISLNK(st.st_mode);
}
break;
default:
result = 0;
}
}
}
Py_END_ALLOW_THREADS
path_cleanup(&_path);
if (result) {
Py_RETURN_TRUE;
}
Py_RETURN_FALSE;
path_t _path = PATH_T_INITIALIZE("_path_islink", "path", 0, 1);
return _testFileType(&_path, path, PY_IFLNK);
}
/*[clinic input]
os._path_isjunction -> bool
path: object
Test whether a path is a junction
[clinic start generated code]*/
static int
os__path_isjunction_impl(PyObject *module, PyObject *path)
/*[clinic end generated code: output=f1d51682a077654d input=103ccedcdb714f11]*/
{
path_t _path = PATH_T_INITIALIZE("_path_isjunction", "path", 0, 1);
return _testFileType(&_path, path, PY_IFMNT);
}
#undef PY_IFREG
#undef PY_IFDIR
#undef PY_IFLNK
#undef PY_IFMNT
#undef PY_IFLRP
#undef PY_IFRRP
#endif /* MS_WINDOWS */
@ -16921,7 +16900,9 @@ static PyMethodDef posix_methods[] = {
OS__PATH_ISDIR_METHODDEF
OS__PATH_ISFILE_METHODDEF
OS__PATH_ISLINK_METHODDEF
OS__PATH_ISJUNCTION_METHODDEF
OS__PATH_EXISTS_METHODDEF
OS__PATH_LEXISTS_METHODDEF
OS__SUPPORTS_VIRTUAL_TERMINAL_METHODDEF
{NULL, NULL} /* Sentinel */