typing: add type hints to `mercurial.dirstatemap`

Somewhere since 10db46e128d4, pytype stopped being able to infer the type of the
`identity` field.  Fill in some obvious other hints along the way.

These hints caused pytype to flag a missing attribute:

    File "/mnt/c/Users/Matt/hg/mercurial/dirstatemap.py", line 714, in _v1_map:
          No attribute 'stat' on mercurial.windows.cachestat [attribute-error]
      In Union[Any, mercurial.posix.cachestat, mercurial.windows.cachestat]
    File "/mnt/c/Users/Matt/hg/mercurial/dirstatemap.py", line 715, in _v1_map:
          No attribute 'stat' on mercurial.windows.cachestat [attribute-error]
      In Union[Any, mercurial.posix.cachestat, mercurial.windows.cachestat]

In practice, the `identity` field is NOT replaced with None if it isn't
cacheable, so it's probably safer to just add the field and set it to None,
since that check is already in place on line 715.

mercurial/dirstatemap.py

@@ -1,849 +1,879
 # dirstatemap.py
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
+from typing import (
+    Optional,
+    TYPE_CHECKING,
+)
+
 from .i18n import _
 
 from . import (
     error,
     pathutil,
     policy,
     testing,
     txnutil,
+    typelib,
     util,
 )
 
 from .dirstateutils import (
     docket as docketmod,
     v2,
 )
 
+if TYPE_CHECKING:
+    from . import (
+        ui as uimod,
+    )
+
 parsers = policy.importmod('parsers')
 rustmod = policy.importrust('dirstate')
 
 propertycache = util.propertycache
 
 if rustmod is None:
     DirstateItem = parsers.DirstateItem
 else:
     DirstateItem = rustmod.DirstateItem
 
 rangemask = 0x7FFFFFFF
 
 WRITE_MODE_AUTO = 0
 WRITE_MODE_FORCE_NEW = 1
 WRITE_MODE_FORCE_APPEND = 2
 
 
 V2_MAX_READ_ATTEMPTS = 5
 
 
 class _dirstatemapcommon:
     """
     Methods that are identical for both implementations of the dirstatemap
     class, with and without Rust extensions enabled.
     """
 
+    _use_dirstate_v2: bool
+    _nodeconstants: typelib.NodeConstants
+    _ui: "uimod.ui"
+    _root: bytes
+    _filename: bytes
+    _nodelen: int
+    _dirtyparents: bool
+    _docket: Optional["docketmod.DirstateDocket"]
+    _write_mode: int
+    _pendingmode: Optional[bool]
+    identity: Optional[typelib.CacheStat]
+
     # please pytype
 
     _map = None
     copymap = None
 
-    def __init__(self, ui, opener, root, nodeconstants, use_dirstate_v2):
+    def __init__(
+        self,
+        ui: "uimod.ui",
+        opener,
+        root: bytes,
+        nodeconstants: typelib.NodeConstants,
+        use_dirstate_v2: bool,
+    ) -> None:
         self._use_dirstate_v2 = use_dirstate_v2
         self._nodeconstants = nodeconstants
         self._ui = ui
         self._opener = opener
         self._root = root
         self._filename = b'dirstate'
         self._nodelen = 20  # Also update Rust code when changing this!
         self._parents = None
         self._dirtyparents = False
         self._docket = None
         write_mode = ui.config(b"devel", b"dirstate.v2.data_update_mode")
         if write_mode == b"auto":
             self._write_mode = WRITE_MODE_AUTO
         elif write_mode == b"force-append":
             self._write_mode = WRITE_MODE_FORCE_APPEND
         elif write_mode == b"force-new":
             self._write_mode = WRITE_MODE_FORCE_NEW
         else:
             # unknown value, fallback to default
             self._write_mode = WRITE_MODE_AUTO
 
         # for consistent view between _pl() and _read() invocations
         self._pendingmode = None
 
-    def _set_identity(self):
+    def _set_identity(self) -> None:
         self.identity = self._get_current_identity()
 
-    def _get_current_identity(self):
+    def _get_current_identity(self) -> Optional[typelib.CacheStat]:
         try:
             return util.cachestat(self._opener.join(self._filename))
         except FileNotFoundError:
             return None
 
-    def may_need_refresh(self):
+    def may_need_refresh(self) -> bool:
         if 'identity' not in vars(self):
             # no existing identity, we need a refresh
             return True
         if self.identity is None:
             return True
         if not self.identity.cacheable():
             # We cannot trust the entry
             # XXX this is a problem on windows, NFS, or other inode less system
             return True
         current_identity = self._get_current_identity()
         if current_identity is None:
             return True
         if not current_identity.cacheable():
             # We cannot trust the entry
             # XXX this is a problem on windows, NFS, or other inode less system
             return True
         return current_identity != self.identity
 
-    def preload(self):
+    def preload(self) -> None:
         """Loads the underlying data, if it's not already loaded"""
         self._map
 
     def get(self, key, default=None):
         return self._map.get(key, default)
 
     def __len__(self):
         return len(self._map)
 
     def __iter__(self):
         return iter(self._map)
 
     def __contains__(self, key):
         return key in self._map
 
     def __getitem__(self, item):
         return self._map[item]
 
     ### disk interaction
 
     def _opendirstatefile(self):
         fp, mode = txnutil.trypending(self._root, self._opener, self._filename)
         if self._pendingmode is not None and self._pendingmode != mode:
             fp.close()
             raise error.Abort(
                 _(b'working directory state may be changed parallelly')
             )
         self._pendingmode = mode
         return fp
 
-    def _readdirstatefile(self, size=-1):
+    def _readdirstatefile(self, size: int = -1) -> bytes:
         try:
             with self._opendirstatefile() as fp:
                 return fp.read(size)
         except FileNotFoundError:
             # File doesn't exist, so the current state is empty
             return b''
 
     @property
-    def docket(self):
+    def docket(self) -> "docketmod.DirstateDocket":
         if not self._docket:
             if not self._use_dirstate_v2:
                 raise error.ProgrammingError(
                     b'dirstate only has a docket in v2 format'
                 )
             self._set_identity()
             data = self._readdirstatefile()
             if data == b'' or data.startswith(docketmod.V2_FORMAT_MARKER):
                 self._docket = docketmod.DirstateDocket.parse(
                     data, self._nodeconstants
                 )
             else:
                 raise error.CorruptedDirstate(b"dirstate is not in v2 format")
         return self._docket
 
     def _read_v2_data(self):
         data = None
         attempts = 0
         while attempts < V2_MAX_READ_ATTEMPTS:
             attempts += 1
             try:
                 # TODO: use mmap when possible
                 data = self._opener.read(self.docket.data_filename())
             except FileNotFoundError:
                 # read race detected between docket and data file
                 # reload the docket and retry
                 self._docket = None
         if data is None:
             assert attempts >= V2_MAX_READ_ATTEMPTS
             msg = b"dirstate read race happened %d times in a row"
             msg %= attempts
             raise error.Abort(msg)
         return self._opener.read(self.docket.data_filename())
 
     def write_v2_no_append(self, tr, st, meta, packed):
         try:
             old_docket = self.docket
         except error.CorruptedDirstate:
             # This means we've identified a dirstate-v1 file on-disk when we
             # were expecting a dirstate-v2 docket. We've managed to recover
             # from that unexpected situation, and now we want to write back a
             # dirstate-v2 file to make the on-disk situation right again.
             #
             # This shouldn't be triggered since `self.docket` is cached and
             # we would have called parents() or read() first, but it's here
             # just in case.
             old_docket = None
 
         new_docket = docketmod.DirstateDocket.with_new_uuid(
             self.parents(), len(packed), meta
         )
         if old_docket is not None and old_docket.uuid == new_docket.uuid:
             raise error.ProgrammingError(b'dirstate docket name collision')
         data_filename = new_docket.data_filename()
         self._opener.write(data_filename, packed)
         # tell the transaction that we are adding a new file
         if tr is not None:
             tr.addbackup(data_filename, location=b'plain')
         # Write the new docket after the new data file has been
         # written. Because `st` was opened with `atomictemp=True`,
         # the actual `.hg/dirstate` file is only affected on close.
         st.write(new_docket.serialize())
         st.close()
         # Remove the old data file after the new docket pointing to
         # the new data file was written.
         if old_docket is not None and old_docket.uuid:
             data_filename = old_docket.data_filename()
             if tr is not None:
                 tr.addbackup(data_filename, location=b'plain')
             unlink = lambda _tr=None: self._opener.unlink(data_filename)
             if tr:
                 category = b"dirstate-v2-clean-" + old_docket.uuid
                 tr.addpostclose(category, unlink)
             else:
                 unlink()
         self._docket = new_docket
 
     ### reading/setting parents
 
     def parents(self):
         if not self._parents:
             if self._use_dirstate_v2:
                 try:
                     self.docket
                 except error.CorruptedDirstate as e:
                     # fall back to dirstate-v1 if we fail to read v2
                     self._v1_parents(e)
                 else:
                     self._parents = self.docket.parents
             else:
                 self._v1_parents()
 
         return self._parents
 
     def _v1_parents(self, from_v2_exception=None):
         read_len = self._nodelen * 2
         st = self._readdirstatefile(read_len)
         l = len(st)
         if l == read_len:
             self._parents = (
                 st[: self._nodelen],
                 st[self._nodelen : 2 * self._nodelen],
             )
         elif l == 0:
             self._parents = (
                 self._nodeconstants.nullid,
                 self._nodeconstants.nullid,
             )
         else:
             hint = None
             if from_v2_exception is not None:
                 hint = _(b"falling back to dirstate-v1 from v2 also failed")
             raise error.Abort(
                 _(b'working directory state appears damaged!'), hint
             )
 
 
 class dirstatemap(_dirstatemapcommon):
     """Map encapsulating the dirstate's contents.
 
     The dirstate contains the following state:
 
     - `identity` is the identity of the dirstate file, which can be used to
       detect when changes have occurred to the dirstate file.
 
     - `parents` is a pair containing the parents of the working copy. The
       parents are updated by calling `setparents`.
 
     - the state map maps filenames to tuples of (state, mode, size, mtime),
       where state is a single character representing 'normal', 'added',
       'removed', or 'merged'. It is read by treating the dirstate as a
       dict.  File state is updated by calling various methods (see each
       documentation for details):
 
       - `reset_state`,
       - `set_tracked`
       - `set_untracked`
       - `set_clean`
       - `set_possibly_dirty`
 
     - `copymap` maps destination filenames to their source filename.
 
     The dirstate also provides the following views onto the state:
 
     - `filefoldmap` is a dict mapping normalized filenames to the denormalized
       form that they appear as in the dirstate.
 
     - `dirfoldmap` is a dict mapping normalized directory names to the
       denormalized form that they appear as in the dirstate.
     """
 
     ### Core data storage and access
 
     @propertycache
     def _map(self):
         self._map = {}
         self.read()
         return self._map
 
     @propertycache
     def copymap(self):
         self.copymap = {}
         self._map
         return self.copymap
 
     def clear(self):
         self._map.clear()
         self.copymap.clear()
         self.setparents(self._nodeconstants.nullid, self._nodeconstants.nullid)
         util.clearcachedproperty(self, b"_dirs")
         util.clearcachedproperty(self, b"_alldirs")
         util.clearcachedproperty(self, b"filefoldmap")
         util.clearcachedproperty(self, b"dirfoldmap")
 
     def items(self):
         return self._map.items()
 
     # forward for python2,3 compat
     iteritems = items
 
     def debug_iter(self, all):
         """
         Return an iterator of (filename, state, mode, size, mtime) tuples
 
         `all` is unused when Rust is not enabled
         """
         for filename, item in self.items():
             yield (filename, item.state, item.mode, item.size, item.mtime)
 
     def keys(self):
         return self._map.keys()
 
     ### reading/setting parents
 
     def setparents(self, p1, p2, fold_p2=False):
         self._parents = (p1, p2)
         self._dirtyparents = True
         copies = {}
         if fold_p2:
             for f, s in self._map.items():
                 # Discard "merged" markers when moving away from a merge state
                 if s.p2_info:
                     source = self.copymap.pop(f, None)
                     if source:
                         copies[f] = source
                     s.drop_merge_data()
         return copies
 
     ### disk interaction
 
     def read(self):
         testing.wait_on_cfg(self._ui, b'dirstate.pre-read-file')
         if self._use_dirstate_v2:
             try:
                 self.docket
             except error.CorruptedDirstate:
                 # fall back to dirstate-v1 if we fail to read v2
                 self._set_identity()
                 st = self._readdirstatefile()
             else:
                 if not self.docket.uuid:
                     return
                 testing.wait_on_cfg(self._ui, b'dirstate.post-docket-read-file')
                 st = self._read_v2_data()
         else:
             self._set_identity()
             st = self._readdirstatefile()
 
         if not st:
             return
 
         # TODO: adjust this estimate for dirstate-v2
         if hasattr(parsers, 'dict_new_presized'):
             # Make an estimate of the number of files in the dirstate based on
             # its size. This trades wasting some memory for avoiding costly
             # resizes. Each entry have a prefix of 17 bytes followed by one or
             # two path names. Studies on various large-scale real-world repositories
             # found 54 bytes a reasonable upper limit for the average path names.
             # Copy entries are ignored for the sake of this estimate.
             self._map = parsers.dict_new_presized(len(st) // 71)
 
         # Python's garbage collector triggers a GC each time a certain number
         # of container objects (the number being defined by
         # gc.get_threshold()) are allocated. parse_dirstate creates a tuple
         # for each file in the dirstate. The C version then immediately marks
         # them as not to be tracked by the collector. However, this has no
         # effect on when GCs are triggered, only on what objects the GC looks
         # into. This means that O(number of files) GCs are unavoidable.
         # Depending on when in the process's lifetime the dirstate is parsed,
         # this can get very expensive. As a workaround, disable GC while
         # parsing the dirstate.
         #
         # (we cannot decorate the function directly since it is in a C module)
         if self._use_dirstate_v2:
             try:
                 self.docket
             except error.CorruptedDirstate:
                 # fall back to dirstate-v1 if we fail to parse v2
                 parse_dirstate = util.nogc(parsers.parse_dirstate)
                 p = parse_dirstate(self._map, self.copymap, st)
             else:
                 p = self.docket.parents
                 meta = self.docket.tree_metadata
                 parse_dirstate = util.nogc(v2.parse_dirstate)
                 parse_dirstate(self._map, self.copymap, st, meta)
         else:
             parse_dirstate = util.nogc(parsers.parse_dirstate)
             p = parse_dirstate(self._map, self.copymap, st)
         if not self._dirtyparents:
             self.setparents(*p)
 
         # Avoid excess attribute lookups by fast pathing certain checks
         self.__contains__ = self._map.__contains__
         self.__getitem__ = self._map.__getitem__
         self.get = self._map.get
 
     def write(self, tr, st):
         if self._use_dirstate_v2:
             packed, meta = v2.pack_dirstate(self._map, self.copymap)
             self.write_v2_no_append(tr, st, meta, packed)
         else:
             packed = parsers.pack_dirstate(
                 self._map, self.copymap, self.parents()
             )
             st.write(packed)
             st.close()
         self._dirtyparents = False
 
     @propertycache
     def identity(self):
         self._map
         return self.identity
 
     ### code related to maintaining and accessing "extra" property
     # (e.g. "has_dir")
 
     def _dirs_incr(self, filename, old_entry=None):
         """increment the dirstate counter if applicable"""
         if (
             old_entry is None or old_entry.removed
         ) and "_dirs" in self.__dict__:
             self._dirs.addpath(filename)
         if old_entry is None and "_alldirs" in self.__dict__:
             self._alldirs.addpath(filename)
 
     def _dirs_decr(self, filename, old_entry=None, remove_variant=False):
         """decrement the dirstate counter if applicable"""
         if old_entry is not None:
             if "_dirs" in self.__dict__ and not old_entry.removed:
                 self._dirs.delpath(filename)
             if "_alldirs" in self.__dict__ and not remove_variant:
                 self._alldirs.delpath(filename)
         elif remove_variant and "_alldirs" in self.__dict__:
             self._alldirs.addpath(filename)
         if "filefoldmap" in self.__dict__:
             normed = util.normcase(filename)
             self.filefoldmap.pop(normed, None)
 
     @propertycache
     def filefoldmap(self):
         """Returns a dictionary mapping normalized case paths to their
         non-normalized versions.
         """
         try:
             makefilefoldmap = parsers.make_file_foldmap
         except AttributeError:
             pass
         else:
             return makefilefoldmap(
                 self._map, util.normcasespec, util.normcasefallback
             )
 
         f = {}
         normcase = util.normcase
         for name, s in self._map.items():
             if not s.removed:
                 f[normcase(name)] = name
         f[b'.'] = b'.'  # prevents useless util.fspath() invocation
         return f
 
     @propertycache
     def dirfoldmap(self):
         f = {}
         normcase = util.normcase
         for name in self._dirs:
             f[normcase(name)] = name
         return f
 
     def hastrackeddir(self, d):
         """
         Returns True if the dirstate contains a tracked (not removed) file
         in this directory.
         """
         return d in self._dirs
 
     def hasdir(self, d):
         """
         Returns True if the dirstate contains a file (tracked or removed)
         in this directory.
         """
         return d in self._alldirs
 
     @propertycache
     def _dirs(self):
         return pathutil.dirs(self._map, only_tracked=True)
 
     @propertycache
     def _alldirs(self):
         return pathutil.dirs(self._map)
 
     ### code related to manipulation of entries and copy-sources
 
     def reset_state(
         self,
         filename,
         wc_tracked=False,
         p1_tracked=False,
         p2_info=False,
         has_meaningful_mtime=True,
         parentfiledata=None,
     ):
         """Set a entry to a given state, diregarding all previous state
 
         This is to be used by the part of the dirstate API dedicated to
         adjusting the dirstate after a update/merge.
 
         note: calling this might result to no entry existing at all if the
         dirstate map does not see any point at having one for this file
         anymore.
         """
         # copy information are now outdated
         # (maybe new information should be in directly passed to this function)
         self.copymap.pop(filename, None)
 
         if not (p1_tracked or p2_info or wc_tracked):
             old_entry = self._map.get(filename)
             self._drop_entry(filename)
             self._dirs_decr(filename, old_entry=old_entry)
             return
 
         old_entry = self._map.get(filename)
         self._dirs_incr(filename, old_entry)
         entry = DirstateItem(
             wc_tracked=wc_tracked,
             p1_tracked=p1_tracked,
             p2_info=p2_info,
             has_meaningful_mtime=has_meaningful_mtime,
             parentfiledata=parentfiledata,
         )
         self._map[filename] = entry
 
     def set_tracked(self, filename):
         new = False
         entry = self.get(filename)
         if entry is None:
             self._dirs_incr(filename)
             entry = DirstateItem(
                 wc_tracked=True,
             )
 
             self._map[filename] = entry
             new = True
         elif not entry.tracked:
             self._dirs_incr(filename, entry)
             entry.set_tracked()
             self._refresh_entry(filename, entry)
             new = True
         else:
             # XXX This is probably overkill for more case, but we need this to
             # fully replace the `normallookup` call with `set_tracked` one.
             # Consider smoothing this in the future.
             entry.set_possibly_dirty()
             self._refresh_entry(filename, entry)
         return new
 
     def set_untracked(self, f):
         """Mark a file as no longer tracked in the dirstate map"""
         entry = self.get(f)
         if entry is None:
             return False
         else:
             self._dirs_decr(f, old_entry=entry, remove_variant=not entry.added)
             if not entry.p2_info:
                 self.copymap.pop(f, None)
             entry.set_untracked()
             self._refresh_entry(f, entry)
             return True
 
     def set_clean(self, filename, mode, size, mtime):
         """mark a file as back to a clean state"""
         entry = self[filename]
         size = size & rangemask
         entry.set_clean(mode, size, mtime)
         self._refresh_entry(filename, entry)
         self.copymap.pop(filename, None)
 
     def set_possibly_dirty(self, filename):
         """record that the current state of the file on disk is unknown"""
         entry = self[filename]
         entry.set_possibly_dirty()
         self._refresh_entry(filename, entry)
 
     def _refresh_entry(self, f, entry):
         """record updated state of an entry"""
         if not entry.any_tracked:
             self._map.pop(f, None)
 
     def _drop_entry(self, f):
         """remove any entry for file f
 
         This should also drop associated copy information
 
         The fact we actually need to drop it is the responsability of the caller
         """
         self._map.pop(f, None)
         self.copymap.pop(f, None)
 
 
 if rustmod is not None:
 
     class dirstatemap(_dirstatemapcommon):
         ### Core data storage and access
 
         @propertycache
         def _map(self):
             """
             Fills the Dirstatemap when called.
             """
             # ignore HG_PENDING because identity is used only for writing
             self._set_identity()
 
             testing.wait_on_cfg(self._ui, b'dirstate.pre-read-file')
             if self._use_dirstate_v2:
                 try:
                     self.docket
                 except error.CorruptedDirstate as e:
                     # fall back to dirstate-v1 if we fail to read v2
                     parents = self._v1_map(e)
                 else:
                     parents = self.docket.parents
                     inode = (
                         self.identity.stat.st_ino
                         if self.identity is not None
                         and self.identity.stat is not None
                         else None
                     )
                     testing.wait_on_cfg(
                         self._ui, b'dirstate.post-docket-read-file'
                     )
                     if not self.docket.uuid:
                         data = b''
                         self._map = rustmod.DirstateMap.new_empty()
                     else:
                         data = self._read_v2_data()
                         self._map = rustmod.DirstateMap.new_v2(
                             data,
                             self.docket.data_size,
                             self.docket.tree_metadata,
                             self.docket.uuid,
                             inode,
                         )
                     parents = self.docket.parents
             else:
                 parents = self._v1_map()
 
             if parents and not self._dirtyparents:
                 self.setparents(*parents)
 
             self.__contains__ = self._map.__contains__
             self.__getitem__ = self._map.__getitem__
             self.get = self._map.get
             return self._map
 
         def _v1_map(self, from_v2_exception=None):
             self._set_identity()
             inode = (
                 self.identity.stat.st_ino
                 if self.identity is not None and self.identity.stat is not None
                 else None
             )
             try:
                 self._map, parents = rustmod.DirstateMap.new_v1(
                     self._readdirstatefile(), inode
                 )
             except OSError as e:
                 if from_v2_exception is not None:
                     raise e from from_v2_exception
                 raise
             return parents
 
         @property
         def copymap(self):
             return self._map.copymap()
 
         def debug_iter(self, all):
             """
             Return an iterator of (filename, state, mode, size, mtime) tuples
 
             `all`: also include with `state == b' '` dirstate tree nodes that
             don't have an associated `DirstateItem`.
 
             """
             return self._map.debug_iter(all)
 
         def clear(self):
             self._map.clear()
             self.setparents(
                 self._nodeconstants.nullid, self._nodeconstants.nullid
             )
             util.clearcachedproperty(self, b"_dirs")
             util.clearcachedproperty(self, b"_alldirs")
             util.clearcachedproperty(self, b"dirfoldmap")
 
         def items(self):
             return self._map.items()
 
         # forward for python2,3 compat
         iteritems = items
 
         def keys(self):
             return iter(self._map)
 
         ### reading/setting parents
 
         def setparents(self, p1, p2, fold_p2=False):
             self._parents = (p1, p2)
             self._dirtyparents = True
             copies = {}
             if fold_p2:
                 copies = self._map.setparents_fixup()
             return copies
 
         ### disk interaction
 
         @propertycache
         def identity(self):
             self._map
             return self.identity
 
         def write(self, tr, st):
             if not self._use_dirstate_v2:
                 p1, p2 = self.parents()
                 packed = self._map.write_v1(p1, p2)
                 st.write(packed)
                 st.close()
                 self._dirtyparents = False
                 return
 
             write_mode = self._write_mode
             try:
                 docket = self.docket
             except error.CorruptedDirstate:
                 # fall back to dirstate-v1 if we fail to parse v2
                 docket = None
 
             # We can only append to an existing data file if there is one
             if docket is None or docket.uuid is None:
                 write_mode = WRITE_MODE_FORCE_NEW
             packed, meta, append = self._map.write_v2(write_mode)
             if append:
                 docket = self.docket
                 data_filename = docket.data_filename()
                 # We mark it for backup to make sure a future `hg rollback` (or
                 # `hg recover`?) call find the data it needs to restore a
                 # working repository.
                 #
                 # The backup can use a hardlink because the format is resistant
                 # to trailing "dead" data.
                 if tr is not None:
                     tr.addbackup(data_filename, location=b'plain')
                 with self._opener(data_filename, b'r+b') as fp:
                     fp.seek(docket.data_size)
                     assert fp.tell() == docket.data_size
                     written = fp.write(packed)
                     if written is not None:  # py2 may return None
                         assert written == len(packed), (written, len(packed))
                 docket.data_size += len(packed)
                 docket.parents = self.parents()
                 docket.tree_metadata = meta
                 st.write(docket.serialize())
                 st.close()
             else:
                 self.write_v2_no_append(tr, st, meta, packed)
             # Reload from the newly-written file
             util.clearcachedproperty(self, b"_map")
             self._dirtyparents = False
 
         ### code related to maintaining and accessing "extra" property
         # (e.g. "has_dir")
 
         @propertycache
         def filefoldmap(self):
             """Returns a dictionary mapping normalized case paths to their
             non-normalized versions.
             """
             return self._map.filefoldmapasdict()
 
         def hastrackeddir(self, d):
             return self._map.hastrackeddir(d)
 
         def hasdir(self, d):
             return self._map.hasdir(d)
 
         @propertycache
         def dirfoldmap(self):
             f = {}
             normcase = util.normcase
             for name in self._map.tracked_dirs():
                 f[normcase(name)] = name
             return f
 
         ### code related to manipulation of entries and copy-sources
 
         def set_tracked(self, f):
             return self._map.set_tracked(f)
 
         def set_untracked(self, f):
             return self._map.set_untracked(f)
 
         def set_clean(self, filename, mode, size, mtime):
             self._map.set_clean(filename, mode, size, mtime)
 
         def set_possibly_dirty(self, f):
             self._map.set_possibly_dirty(f)
 
         def reset_state(
             self,
             filename,
             wc_tracked=False,
             p1_tracked=False,
             p2_info=False,
             has_meaningful_mtime=True,
             parentfiledata=None,
         ):
             return self._map.reset_state(
                 filename,
                 wc_tracked,
                 p1_tracked,
                 p2_info,
                 has_meaningful_mtime,
                 parentfiledata,
             )

mercurial/posix.py

@@ -1,809 +1,811
 # posix.py - Posix utility function implementations for Mercurial
 #
 #  Copyright 2005-2009 Olivia Mackall <[e-mail]> and others
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 import errno
 import fcntl
 import getpass
 import grp
 import os
 import pwd
 import re
 import select
 import stat
 import sys
 import tempfile
 import typing
 import unicodedata
 
 from typing import (
     Any,
     AnyStr,
     Iterable,
     Iterator,
     List,
     Match,
     NoReturn,
     Optional,
     Sequence,
     Tuple,
     Union,
 )
 
 from .i18n import _
 from .pycompat import (
     open,
 )
 from . import (
     encoding,
     error,
     policy,
     pycompat,
 )
 
 osutil = policy.importmod('osutil')
 
 normpath = os.path.normpath
 samestat = os.path.samestat
 abspath = os.path.abspath  # re-exports
 
 try:
     oslink = os.link
 except AttributeError:
     # Some platforms build Python without os.link on systems that are
     # vaguely unix-like but don't have hardlink support. For those
     # poor souls, just say we tried and that it failed so we fall back
     # to copies.
     def oslink(src: bytes, dst: bytes) -> NoReturn:
         raise OSError(
             errno.EINVAL, b'hardlinks not supported: %s to %s' % (src, dst)
         )
 
 
 readlink = os.readlink
 unlink = os.unlink
 rename = os.rename
 removedirs = os.removedirs
 
 if typing.TYPE_CHECKING:
 
     def normpath(path: bytes) -> bytes:
         raise NotImplementedError
 
     def abspath(path: AnyStr) -> AnyStr:
         raise NotImplementedError
 
     def oslink(src: bytes, dst: bytes) -> None:
         raise NotImplementedError
 
     def readlink(path: bytes) -> bytes:
         raise NotImplementedError
 
     def unlink(path: bytes) -> None:
         raise NotImplementedError
 
     def rename(src: bytes, dst: bytes) -> None:
         raise NotImplementedError
 
     def removedirs(name: bytes) -> None:
         raise NotImplementedError
 
 
 expandglobs: bool = False
 
 umask: int = os.umask(0)
 os.umask(umask)
 
 posixfile = open
 
 
 def split(p: bytes) -> Tuple[bytes, bytes]:
     """Same as posixpath.split, but faster
 
     >>> import posixpath
     >>> for f in [b'/absolute/path/to/file',
     ...           b'relative/path/to/file',
     ...           b'file_alone',
     ...           b'path/to/directory/',
     ...           b'/multiple/path//separators',
     ...           b'/file_at_root',
     ...           b'///multiple_leading_separators_at_root',
     ...           b'']:
     ...     assert split(f) == posixpath.split(f), f
     """
     ht = p.rsplit(b'/', 1)
     if len(ht) == 1:
         return b'', p
     nh = ht[0].rstrip(b'/')
     if nh:
         return nh, ht[1]
     return ht[0] + b'/', ht[1]
 
 
 def openhardlinks() -> bool:
     '''return true if it is safe to hold open file handles to hardlinks'''
     return True
 
 
 def nlinks(name: bytes) -> int:
     '''return number of hardlinks for the given file'''
     return os.lstat(name).st_nlink
 
 
 def parsepatchoutput(output_line: bytes) -> bytes:
     """parses the output produced by patch and returns the filename"""
     pf = output_line[14:]
     if pycompat.sysplatform == b'OpenVMS':
         if pf[0] == b'`':
             pf = pf[1:-1]  # Remove the quotes
     else:
         if pf.startswith(b"'") and pf.endswith(b"'") and b" " in pf:
             pf = pf[1:-1]  # Remove the quotes
     return pf
 
 
 def sshargs(
     sshcmd: bytes, host: bytes, user: Optional[bytes], port: Optional[bytes]
 ) -> bytes:
     '''Build argument list for ssh'''
     args = user and (b"%s@%s" % (user, host)) or host
     if b'-' in args[:1]:
         raise error.Abort(
             _(b'illegal ssh hostname or username starting with -: %s') % args
         )
     args = shellquote(args)
     if port:
         args = b'-p %s %s' % (shellquote(port), args)
     return args
 
 
 def isexec(f: bytes) -> bool:
     """check whether a file is executable"""
     return os.lstat(f).st_mode & 0o100 != 0
 
 
 def setflags(f: bytes, l: bool, x: bool) -> None:
     st = os.lstat(f)
     s = st.st_mode
     if l:
         if not stat.S_ISLNK(s):
             # switch file to link
             with open(f, b'rb') as fp:
                 data = fp.read()
             unlink(f)
             try:
                 os.symlink(data, f)
             except OSError:
                 # failed to make a link, rewrite file
                 with open(f, b"wb") as fp:
                     fp.write(data)
 
         # no chmod needed at this point
         return
     if stat.S_ISLNK(s):
         # switch link to file
         data = os.readlink(f)
         unlink(f)
         with open(f, b"wb") as fp:
             fp.write(data)
         s = 0o666 & ~umask  # avoid restatting for chmod
 
     sx = s & 0o100
     if st.st_nlink > 1 and bool(x) != bool(sx):
         # the file is a hardlink, break it
         with open(f, b"rb") as fp:
             data = fp.read()
         unlink(f)
         with open(f, b"wb") as fp:
             fp.write(data)
 
     if x and not sx:
         # Turn on +x for every +r bit when making a file executable
         # and obey umask.
         os.chmod(f, s | (s & 0o444) >> 2 & ~umask)
     elif not x and sx:
         # Turn off all +x bits
         os.chmod(f, s & 0o666)
 
 
 def copymode(
     src: bytes,
     dst: bytes,
     mode: Optional[bytes] = None,
     enforcewritable: bool = False,
 ) -> None:
     """Copy the file mode from the file at path src to dst.
     If src doesn't exist, we're using mode instead. If mode is None, we're
     using umask."""
     try:
         st_mode = os.lstat(src).st_mode & 0o777
     except FileNotFoundError:
         st_mode = mode
         if st_mode is None:
             st_mode = ~umask
         st_mode &= 0o666
 
     new_mode = st_mode
 
     if enforcewritable:
         new_mode |= stat.S_IWUSR
 
     os.chmod(dst, new_mode)
 
 
 def checkexec(path: bytes) -> bool:
     """
     Check whether the given path is on a filesystem with UNIX-like exec flags
 
     Requires a directory (like /foo/.hg)
     """
 
     # VFAT on some Linux versions can flip mode but it doesn't persist
     # a FS remount. Frequently we can detect it if files are created
     # with exec bit on.
 
     try:
         EXECFLAGS = stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
         basedir = os.path.join(path, b'.hg')
         cachedir = os.path.join(basedir, b'wcache')
         storedir = os.path.join(basedir, b'store')
         if not os.path.exists(cachedir):
             try:
                 # we want to create the 'cache' directory, not the '.hg' one.
                 # Automatically creating '.hg' directory could silently spawn
                 # invalid Mercurial repositories. That seems like a bad idea.
                 os.mkdir(cachedir)
                 if os.path.exists(storedir):
                     copymode(storedir, cachedir)
                 else:
                     copymode(basedir, cachedir)
             except (IOError, OSError):
                 # we other fallback logic triggers
                 pass
         if os.path.isdir(cachedir):
             checkisexec = os.path.join(cachedir, b'checkisexec')
             checknoexec = os.path.join(cachedir, b'checknoexec')
 
             try:
                 m = os.stat(checkisexec).st_mode
             except FileNotFoundError:
                 # checkisexec does not exist - fall through ...
                 pass
             else:
                 # checkisexec exists, check if it actually is exec
                 if m & EXECFLAGS != 0:
                     # ensure checknoexec exists, check it isn't exec
                     try:
                         m = os.stat(checknoexec).st_mode
                     except FileNotFoundError:
                         open(checknoexec, b'w').close()  # might fail
                         m = os.stat(checknoexec).st_mode
                     if m & EXECFLAGS == 0:
                         # check-exec is exec and check-no-exec is not exec
                         return True
                     # checknoexec exists but is exec - delete it
                     unlink(checknoexec)
                 # checkisexec exists but is not exec - delete it
                 unlink(checkisexec)
 
             # check using one file, leave it as checkisexec
             checkdir = cachedir
         else:
             # check directly in path and don't leave checkisexec behind
             checkdir = path
             checkisexec = None
         fh, fn = pycompat.mkstemp(dir=checkdir, prefix=b'hg-checkexec-')
         try:
             os.close(fh)
             m = os.stat(fn).st_mode
             if m & EXECFLAGS == 0:
                 os.chmod(fn, m & 0o777 | EXECFLAGS)
                 if os.stat(fn).st_mode & EXECFLAGS != 0:
                     if checkisexec is not None:
                         os.rename(fn, checkisexec)
                         fn = None
                     return True
         finally:
             if fn is not None:
                 unlink(fn)
     except (IOError, OSError):
         # we don't care, the user probably won't be able to commit anyway
         return False
 
 
 def checklink(path: bytes) -> bool:
     """check whether the given path is on a symlink-capable filesystem"""
     # mktemp is not racy because symlink creation will fail if the
     # file already exists
     while True:
         cachedir = os.path.join(path, b'.hg', b'wcache')
         checklink = os.path.join(cachedir, b'checklink')
         # try fast path, read only
         if os.path.islink(checklink):
             return True
         if os.path.isdir(cachedir):
             checkdir = cachedir
         else:
             checkdir = path
             cachedir = None
         name = tempfile.mktemp(
             dir=pycompat.fsdecode(checkdir), prefix=r'checklink-'
         )
         name = pycompat.fsencode(name)
         try:
             fd = None
             if cachedir is None:
                 fd = pycompat.namedtempfile(
                     dir=checkdir, prefix=b'hg-checklink-'
                 )
                 target = os.path.basename(fd.name)
             else:
                 # create a fixed file to link to; doesn't matter if it
                 # already exists.
                 target = b'checklink-target'
                 try:
                     fullpath = os.path.join(cachedir, target)
                     open(fullpath, b'w').close()
                 except PermissionError:
                     # If we can't write to cachedir, just pretend
                     # that the fs is readonly and by association
                     # that the fs won't support symlinks. This
                     # seems like the least dangerous way to avoid
                     # data loss.
                     return False
             try:
                 os.symlink(target, name)
                 if cachedir is None:
                     unlink(name)
                 else:
                     try:
                         os.rename(name, checklink)
                     except OSError:
                         unlink(name)
                 return True
             except FileExistsError:
                 # link creation might race, try again
                 continue
             finally:
                 if fd is not None:
                     fd.close()
         except AttributeError:
             return False
         except OSError as inst:
             # sshfs might report failure while successfully creating the link
             if inst.errno == errno.EIO and os.path.exists(name):
                 unlink(name)
             return False
 
 
 def checkosfilename(path: bytes) -> Optional[bytes]:
     """Check that the base-relative path is a valid filename on this platform.
     Returns None if the path is ok, or a UI string describing the problem."""
     return None  # on posix platforms, every path is ok
 
 
 def getfsmountpoint(dirpath: bytes) -> Optional[bytes]:
     """Get the filesystem mount point from a directory (best-effort)
 
     Returns None if we are unsure. Raises OSError on ENOENT, EPERM, etc.
     """
     return getattr(osutil, 'getfsmountpoint', lambda x: None)(dirpath)
 
 
 def getfstype(dirpath: bytes) -> Optional[bytes]:
     """Get the filesystem type name from a directory (best-effort)
 
     Returns None if we are unsure. Raises OSError on ENOENT, EPERM, etc.
     """
     return getattr(osutil, 'getfstype', lambda x: None)(dirpath)
 
 
 def get_password() -> bytes:
     return encoding.strtolocal(getpass.getpass(''))
 
 
 def setbinary(fd) -> None:
     pass
 
 
 def pconvert(path: bytes) -> bytes:
     return path
 
 
 def localpath(path: bytes) -> bytes:
     return path
 
 
 def samefile(fpath1: bytes, fpath2: bytes) -> bool:
     """Returns whether path1 and path2 refer to the same file. This is only
     guaranteed to work for files, not directories."""
     return os.path.samefile(fpath1, fpath2)
 
 
 def samedevice(fpath1: bytes, fpath2: bytes) -> bool:
     """Returns whether fpath1 and fpath2 are on the same device. This is only
     guaranteed to work for files, not directories."""
     st1 = os.lstat(fpath1)
     st2 = os.lstat(fpath2)
     return st1.st_dev == st2.st_dev
 
 
 # os.path.normcase is a no-op, which doesn't help us on non-native filesystems
 def normcase(path: bytes) -> bytes:
     return path.lower()
 
 
 # what normcase does to ASCII strings
 normcasespec: int = encoding.normcasespecs.lower
 # fallback normcase function for non-ASCII strings
 normcasefallback = normcase
 
 if pycompat.isdarwin:
 
     def normcase(path: bytes) -> bytes:
         """
         Normalize a filename for OS X-compatible comparison:
         - escape-encode invalid characters
         - decompose to NFD
         - lowercase
         - omit ignored characters [200c-200f, 202a-202e, 206a-206f,feff]
 
         >>> normcase(b'UPPER')
         'upper'
         >>> normcase(b'Caf\\xc3\\xa9')
         'cafe\\xcc\\x81'
         >>> normcase(b'\\xc3\\x89')
         'e\\xcc\\x81'
         >>> normcase(b'\\xb8\\xca\\xc3\\xca\\xbe\\xc8.JPG') # issue3918
         '%b8%ca%c3\\xca\\xbe%c8.jpg'
         """
 
         try:
             return encoding.asciilower(path)  # exception for non-ASCII
         except UnicodeDecodeError:
             return normcasefallback(path)
 
     normcasespec = encoding.normcasespecs.lower
 
     def normcasefallback(path: bytes) -> bytes:
         try:
             u = path.decode('utf-8')
         except UnicodeDecodeError:
             # OS X percent-encodes any bytes that aren't valid utf-8
             s = b''
             pos = 0
             l = len(path)
             while pos < l:
                 try:
                     c = encoding.getutf8char(path, pos)
                     pos += len(c)
                 except ValueError:
                     c = b'%%%02X' % ord(path[pos : pos + 1])
                     pos += 1
                 s += c
 
             u = s.decode('utf-8')
 
         # Decompose then lowercase (HFS+ technote specifies lower)
         enc = unicodedata.normalize('NFD', u).lower().encode('utf-8')
         # drop HFS+ ignored characters
         return encoding.hfsignoreclean(enc)
 
 
 if pycompat.sysplatform == b'cygwin':
     # workaround for cygwin, in which mount point part of path is
     # treated as case sensitive, even though underlying NTFS is case
     # insensitive.
 
     # default mount points
     cygwinmountpoints = sorted(
         [
             b"/usr/bin",
             b"/usr/lib",
             b"/cygdrive",
         ],
         reverse=True,
     )
 
     # use upper-ing as normcase as same as NTFS workaround
     def normcase(path: bytes) -> bytes:
         pathlen = len(path)
         if (pathlen == 0) or (path[0] != pycompat.ossep):
             # treat as relative
             return encoding.upper(path)
 
         # to preserve case of mountpoint part
         for mp in cygwinmountpoints:
             if not path.startswith(mp):
                 continue
 
             mplen = len(mp)
             if mplen == pathlen:  # mount point itself
                 return mp
             if path[mplen] == pycompat.ossep:
                 return mp + encoding.upper(path[mplen:])
 
         return encoding.upper(path)
 
     normcasespec = encoding.normcasespecs.other
     normcasefallback = normcase
 
     # Cygwin translates native ACLs to POSIX permissions,
     # but these translations are not supported by native
     # tools, so the exec bit tends to be set erroneously.
     # Therefore, disable executable bit access on Cygwin.
     def checkexec(path: bytes) -> bool:
         return False
 
     # Similarly, Cygwin's symlink emulation is likely to create
     # problems when Mercurial is used from both Cygwin and native
     # Windows, with other native tools, or on shared volumes
     def checklink(path: bytes) -> bool:
         return False
 
 
 if pycompat.sysplatform == b'OpenVMS':
     # OpenVMS's symlink emulation is broken on some OpenVMS versions.
     def checklink(path: bytes) -> bool:
         return False
 
 
 _needsshellquote: Optional[Match[bytes]] = None
 
 
 def shellquote(s: bytes) -> bytes:
     if pycompat.sysplatform == b'OpenVMS':
         return b'"%s"' % s
     global _needsshellquote
     if _needsshellquote is None:
         _needsshellquote = re.compile(br'[^a-zA-Z0-9._/+-]').search
     if s and not _needsshellquote(s):
         # "s" shouldn't have to be quoted
         return s
     else:
         return b"'%s'" % s.replace(b"'", b"'\\''")
 
 
 def shellsplit(s: bytes) -> List[bytes]:
     """Parse a command string in POSIX shell way (best-effort)"""
     return pycompat.shlexsplit(s, posix=True)
 
 
 def testpid(pid: int) -> bool:
     '''return False if pid dead, True if running or not sure'''
     if pycompat.sysplatform == b'OpenVMS':
         return True
     try:
         os.kill(pid, 0)
         return True
     except OSError as inst:
         return inst.errno != errno.ESRCH
 
 
 def isowner(st: os.stat_result) -> bool:
     """Return True if the stat object st is from the current user."""
     return st.st_uid == os.getuid()
 
 
 def findexe(command: bytes) -> Optional[bytes]:
     """Find executable for command searching like which does.
     If command is a basename then PATH is searched for command.
     PATH isn't searched if command is an absolute or relative path.
     If command isn't found None is returned."""
     if pycompat.sysplatform == b'OpenVMS':
         return command
 
     def findexisting(executable: bytes) -> Optional[bytes]:
         b'Will return executable if existing file'
         if os.path.isfile(executable) and os.access(executable, os.X_OK):
             return executable
         return None
 
     if pycompat.ossep in command:
         return findexisting(command)
 
     if pycompat.sysplatform == b'plan9':
         return findexisting(os.path.join(b'/bin', command))
 
     for path in encoding.environ.get(b'PATH', b'').split(pycompat.ospathsep):
         executable = findexisting(os.path.join(path, command))
         if executable is not None:
             return executable
     return None
 
 
 def setsignalhandler() -> None:
     pass
 
 
 _wantedkinds = {stat.S_IFREG, stat.S_IFLNK}
 
 
 def statfiles(files: Sequence[bytes]) -> Iterator[Optional[os.stat_result]]:
     """Stat each file in files. Yield each stat, or None if a file does not
     exist or has a type we don't care about."""
     lstat = os.lstat
     getkind = stat.S_IFMT
     for nf in files:
         try:
             st = lstat(nf)
             if getkind(st.st_mode) not in _wantedkinds:
                 st = None
         except (FileNotFoundError, NotADirectoryError):
             st = None
         yield st
 
 
 def getuser() -> bytes:
     '''return name of current user'''
     return pycompat.fsencode(getpass.getuser())
 
 
 def username(uid: Optional[int] = None) -> Optional[bytes]:
     """Return the name of the user with the given uid.
 
     If uid is None, return the name of the current user."""
 
     if uid is None:
         uid = os.getuid()
     try:
         return pycompat.fsencode(pwd.getpwuid(uid)[0])
     except KeyError:
         return b'%d' % uid
 
 
 def groupname(gid: Optional[int] = None) -> Optional[bytes]:
     """Return the name of the group with the given gid.
 
     If gid is None, return the name of the current group."""
 
     if gid is None:
         gid = os.getgid()
     try:
         return pycompat.fsencode(grp.getgrgid(gid)[0])
     except KeyError:
         return pycompat.bytestr(gid)
 
 
 def groupmembers(name: bytes) -> List[bytes]:
     """Return the list of members of the group with the given
     name, KeyError if the group does not exist.
     """
     name = pycompat.fsdecode(name)
     return pycompat.rapply(pycompat.fsencode, list(grp.getgrnam(name).gr_mem))
 
 
 def spawndetached(args: List[bytes]) -> int:
     return os.spawnvp(os.P_NOWAIT | getattr(os, 'P_DETACH', 0), args[0], args)
 
 
 def gethgcmd():  # TODO: convert to bytes, like on Windows?
     return sys.argv[:1]
 
 
 def makedir(path: bytes, notindexed: bool) -> None:
     os.mkdir(path)
 
 
 def lookupreg(
     key: bytes,
     name: Optional[bytes] = None,
     scope: Optional[Union[int, Iterable[int]]] = None,
 ) -> Optional[bytes]:
     return None
 
 
 def hidewindow() -> None:
     """Hide current shell window.
 
     Used to hide the window opened when starting asynchronous
     child process under Windows, unneeded on other systems.
     """
     pass
 
 
 class cachestat:
+    stat: os.stat_result
+
     def __init__(self, path: bytes) -> None:
         self.stat = os.stat(path)
 
     def cacheable(self) -> bool:
         return bool(self.stat.st_ino)
 
     __hash__ = object.__hash__
 
     def __eq__(self, other: Any) -> bool:
         try:
             # Only dev, ino, size, mtime and atime are likely to change. Out
             # of these, we shouldn't compare atime but should compare the
             # rest. However, one of the other fields changing indicates
             # something fishy going on, so return False if anything but atime
             # changes.
             return (
                 self.stat.st_mode == other.stat.st_mode
                 and self.stat.st_ino == other.stat.st_ino
                 and self.stat.st_dev == other.stat.st_dev
                 and self.stat.st_nlink == other.stat.st_nlink
                 and self.stat.st_uid == other.stat.st_uid
                 and self.stat.st_gid == other.stat.st_gid
                 and self.stat.st_size == other.stat.st_size
                 and self.stat[stat.ST_MTIME] == other.stat[stat.ST_MTIME]
                 and self.stat[stat.ST_CTIME] == other.stat[stat.ST_CTIME]
             )
         except AttributeError:
             return False
 
     def __ne__(self, other: Any) -> bool:
         return not self == other
 
 
 def statislink(st: Optional[os.stat_result]) -> bool:
     '''check whether a stat result is a symlink'''
     return stat.S_ISLNK(st.st_mode) if st else False
 
 
 def statisexec(st: Optional[os.stat_result]) -> bool:
     '''check whether a stat result is an executable file'''
     return (st.st_mode & 0o100 != 0) if st else False
 
 
 def poll(fds):
     """block until something happens on any file descriptor
 
     This is a generic helper that will check for any activity
     (read, write.  exception) and return the list of touched files.
 
     In unsupported cases, it will raise a NotImplementedError"""
     try:
         res = select.select(fds, fds, fds)
     except ValueError:  # out of range file descriptor
         raise NotImplementedError()
     return sorted(list(set(sum(res, []))))
 
 
 def readpipe(pipe) -> bytes:
     """Read all available data from a pipe."""
     # We can't fstat() a pipe because Linux will always report 0.
     # So, we set the pipe to non-blocking mode and read everything
     # that's available.
     flags = fcntl.fcntl(pipe, fcntl.F_GETFL)
     flags |= os.O_NONBLOCK
     oldflags = fcntl.fcntl(pipe, fcntl.F_SETFL, flags)
 
     try:
         chunks = []
         while True:
             try:
                 s = pipe.read()
                 if not s:
                     break
                 chunks.append(s)
             except IOError:
                 break
 
         return b''.join(chunks)
     finally:
         fcntl.fcntl(pipe, fcntl.F_SETFL, oldflags)
 
 
 def bindunixsocket(sock, path: bytes) -> None:
     """Bind the UNIX domain socket to the specified path"""
     # use relative path instead of full path at bind() if possible, since
     # AF_UNIX path has very small length limit (107 chars) on common
     # platforms (see sys/un.h)
     dirname, basename = os.path.split(path)
     bakwdfd = None
 
     try:
         if dirname:
             bakwdfd = os.open(b'.', os.O_DIRECTORY)
             os.chdir(dirname)
         sock.bind(basename)
         if bakwdfd:
             os.fchdir(bakwdfd)
     finally:
         if bakwdfd:
             os.close(bakwdfd)

mercurial/typelib.py

@@ -1,28 +1,41
 # typelib.py - type hint aliases and support
 #
 # Copyright 2022 Matt Harbison <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 import typing
 
 # Note: this is slightly different from pycompat.TYPE_CHECKING, as using
 # pycompat causes the BinaryIO_Proxy type to be resolved to ``object`` when
 # used as the base class during a pytype run.
 TYPE_CHECKING = typing.TYPE_CHECKING
 
 
 # The BinaryIO class provides empty methods, which at runtime means that
 # ``__getattr__`` on the proxy classes won't get called for the methods that
 # should delegate to the internal object.  So to avoid runtime changes because
 # of the required typing inheritance, just use BinaryIO when typechecking, and
 # ``object`` otherwise.
 if TYPE_CHECKING:
     from typing import (
         BinaryIO,
+        Union,
+    )
+
+    from . import (
+        node,
+        posix,
+        windows,
     )
 
     BinaryIO_Proxy = BinaryIO
+    CacheStat = Union[posix.cachestat, windows.cachestat]
+    NodeConstants = node.sha1nodeconstants
 else:
+    from typing import Any
+
     BinaryIO_Proxy = object
+    CacheStat = Any
+    NodeConstants = Any

mercurial/windows.py

@@ -1,757 +1,759
 # windows.py - Windows utility function implementations for Mercurial
 #
 #  Copyright 2005-2009 Olivia Mackall <[e-mail]> and others
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 import errno
 import getpass
 import msvcrt  # pytype: disable=import-error
 import os
 import re
 import stat
 import string
 import sys
 import typing
 import winreg  # pytype: disable=import-error
 
 from typing import (
     AnyStr,
     BinaryIO,
     Iterable,
     Iterator,
     List,
     Mapping,
     NoReturn,
     Optional,
     Pattern,
     Sequence,
     Tuple,
     Union,
 )
 
 from .i18n import _
 from . import (
     encoding,
     error,
     policy,
     pycompat,
     typelib,
     win32,
 )
 
 
 osutil = policy.importmod('osutil')
 
 getfsmountpoint = win32.getvolumename
 getfstype = win32.getfstype
 getuser = win32.getuser
 hidewindow = win32.hidewindow
 makedir = win32.makedir
 nlinks = win32.nlinks
 oslink = win32.oslink
 samedevice = win32.samedevice
 samefile = win32.samefile
 setsignalhandler = win32.setsignalhandler
 spawndetached = win32.spawndetached
 split = os.path.split
 testpid = win32.testpid
 unlink = win32.unlink
 
 if typing.TYPE_CHECKING:
 
     def split(p: bytes) -> Tuple[bytes, bytes]:
         raise NotImplementedError
 
 
 umask: int = 0o022
 
 
 class mixedfilemodewrapper:
     """Wraps a file handle when it is opened in read/write mode.
 
     fopen() and fdopen() on Windows have a specific-to-Windows requirement
     that files opened with mode r+, w+, or a+ make a call to a file positioning
     function when switching between reads and writes. Without this extra call,
     Python will raise a not very intuitive "IOError: [Errno 0] Error."
 
     This class wraps posixfile instances when the file is opened in read/write
     mode and automatically adds checks or inserts appropriate file positioning
     calls when necessary.
     """
 
     OPNONE = 0
     OPREAD = 1
     OPWRITE = 2
 
     def __init__(self, fp):
         object.__setattr__(self, '_fp', fp)
         object.__setattr__(self, '_lastop', 0)
 
     def __enter__(self):
         self._fp.__enter__()
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self._fp.__exit__(exc_type, exc_val, exc_tb)
 
     def __getattr__(self, name):
         return getattr(self._fp, name)
 
     def __setattr__(self, name, value):
         return self._fp.__setattr__(name, value)
 
     def _noopseek(self):
         self._fp.seek(0, os.SEEK_CUR)
 
     def seek(self, *args, **kwargs):
         object.__setattr__(self, '_lastop', self.OPNONE)
         return self._fp.seek(*args, **kwargs)
 
     def write(self, d):
         if self._lastop == self.OPREAD:
             self._noopseek()
 
         object.__setattr__(self, '_lastop', self.OPWRITE)
         return self._fp.write(d)
 
     def writelines(self, *args, **kwargs):
         if self._lastop == self.OPREAD:
             self._noopeseek()
 
         object.__setattr__(self, '_lastop', self.OPWRITE)
         return self._fp.writelines(*args, **kwargs)
 
     def read(self, *args, **kwargs):
         if self._lastop == self.OPWRITE:
             self._noopseek()
 
         object.__setattr__(self, '_lastop', self.OPREAD)
         return self._fp.read(*args, **kwargs)
 
     def readline(self, *args, **kwargs):
         if self._lastop == self.OPWRITE:
             self._noopseek()
 
         object.__setattr__(self, '_lastop', self.OPREAD)
         return self._fp.readline(*args, **kwargs)
 
     def readlines(self, *args, **kwargs):
         if self._lastop == self.OPWRITE:
             self._noopseek()
 
         object.__setattr__(self, '_lastop', self.OPREAD)
         return self._fp.readlines(*args, **kwargs)
 
 
 class fdproxy:
     """Wraps osutil.posixfile() to override the name attribute to reflect the
     underlying file name.
     """
 
     def __init__(self, name, fp):
         self.name = name
         self._fp = fp
 
     def __enter__(self):
         self._fp.__enter__()
         # Return this wrapper for the context manager so that the name is
         # still available.
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
         self._fp.__exit__(exc_type, exc_value, traceback)
 
     def __iter__(self):
         return iter(self._fp)
 
     def __getattr__(self, name):
         return getattr(self._fp, name)
 
 
 def posixfile(name, mode=b'r', buffering=-1):
     '''Open a file with even more POSIX-like semantics'''
     try:
         fp = osutil.posixfile(name, mode, buffering)  # may raise WindowsError
 
         # PyFile_FromFd() ignores the name, and seems to report fp.name as the
         # underlying file descriptor.
         fp = fdproxy(name, fp)
 
         # The position when opening in append mode is implementation defined, so
         # make it consistent with other platforms, which position at EOF.
         if b'a' in mode:
             fp.seek(0, os.SEEK_END)
 
         if b'+' in mode:
             return mixedfilemodewrapper(fp)
 
         return fp
     except WindowsError as err:  # pytype: disable=name-error
         # convert to a friendlier exception
         raise IOError(
             err.errno, '%s: %s' % (encoding.strfromlocal(name), err.strerror)
         )
 
 
 # may be wrapped by win32mbcs extension
 listdir = osutil.listdir
 
 
 def get_password() -> bytes:
     """Prompt for password with echo off, using Windows getch().
 
     This shouldn't be called directly- use ``ui.getpass()`` instead, which
     checks if the session is interactive first.
     """
     pw = u""
     while True:
         c = msvcrt.getwch()  # pytype: disable=module-attr
         if c == u'\r' or c == u'\n':
             break
         if c == u'\003':
             raise KeyboardInterrupt
         if c == u'\b':
             pw = pw[:-1]
         else:
             pw = pw + c
     msvcrt.putwch(u'\r')  # pytype: disable=module-attr
     msvcrt.putwch(u'\n')  # pytype: disable=module-attr
     return encoding.unitolocal(pw)
 
 
 class winstdout(typelib.BinaryIO_Proxy):
     """Some files on Windows misbehave.
 
     When writing to a broken pipe, EINVAL instead of EPIPE may be raised.
 
     When writing too many bytes to a console at the same, a "Not enough space"
     error may happen. Python 3 already works around that.
     """
 
     def __init__(self, fp: BinaryIO):
         self.fp = fp
 
     def __getattr__(self, key):
         return getattr(self.fp, key)
 
     def close(self):
         try:
             self.fp.close()
         except IOError:
             pass
 
     def write(self, s):
         try:
             return self.fp.write(s)
         except IOError as inst:
             if inst.errno != 0 and not win32.lasterrorwaspipeerror(inst):
                 raise
             self.close()
             raise IOError(errno.EPIPE, 'Broken pipe')
 
     def flush(self):
         try:
             return self.fp.flush()
         except IOError as inst:
             if not win32.lasterrorwaspipeerror(inst):
                 raise
             raise IOError(errno.EPIPE, 'Broken pipe')
 
 
 def openhardlinks() -> bool:
     return True
 
 
 def parsepatchoutput(output_line: bytes) -> bytes:
     """parses the output produced by patch and returns the filename"""
     pf = output_line[14:]
     if pf[0] == b'`':
         pf = pf[1:-1]  # Remove the quotes
     return pf
 
 
 def sshargs(
     sshcmd: bytes, host: bytes, user: Optional[bytes], port: Optional[bytes]
 ) -> bytes:
     '''Build argument list for ssh or Plink'''
     pflag = b'plink' in sshcmd.lower() and b'-P' or b'-p'
     args = user and (b"%s@%s" % (user, host)) or host
     if args.startswith(b'-') or args.startswith(b'/'):
         raise error.Abort(
             _(b'illegal ssh hostname or username starting with - or /: %s')
             % args
         )
     args = shellquote(args)
     if port:
         args = b'%s %s %s' % (pflag, shellquote(port), args)
     return args
 
 
 def setflags(f: bytes, l: bool, x: bool) -> None:
     pass
 
 
 def copymode(
     src: bytes,
     dst: bytes,
     mode: Optional[bytes] = None,
     enforcewritable: bool = False,
 ) -> None:
     pass
 
 
 def checkexec(path: bytes) -> bool:
     return False
 
 
 def checklink(path: bytes) -> bool:
     return False
 
 
 def setbinary(fd) -> None:
     # When run without console, pipes may expose invalid
     # fileno(), usually set to -1.
     fno = getattr(fd, 'fileno', None)
     if fno is not None and fno() >= 0:
         msvcrt.setmode(fno(), os.O_BINARY)  # pytype: disable=module-attr
 
 
 def pconvert(path: bytes) -> bytes:
     return path.replace(pycompat.ossep, b'/')
 
 
 def localpath(path: bytes) -> bytes:
     return path.replace(b'/', b'\\')
 
 
 def normpath(path: bytes) -> bytes:
     return pconvert(os.path.normpath(path))
 
 
 def normcase(path: bytes) -> bytes:
     return encoding.upper(path)  # NTFS compares via upper()
 
 
 DRIVE_RE_B: Pattern[bytes] = re.compile(b'^[a-z]:')
 DRIVE_RE_S: Pattern[str] = re.compile('^[a-z]:')
 
 
 # TODO: why is this accepting str?
 def abspath(path: AnyStr) -> AnyStr:
     abs_path = os.path.abspath(path)  # re-exports
     # Python on Windows is inconsistent regarding the capitalization of drive
     # letter and this cause issue with various path comparison along the way.
     # So we normalize the drive later to upper case here.
     #
     # See https://bugs.python.org/issue40368 for and example of this hell.
     if isinstance(abs_path, bytes):
         if DRIVE_RE_B.match(abs_path):
             abs_path = abs_path[0:1].upper() + abs_path[1:]
     elif DRIVE_RE_S.match(abs_path):
         abs_path = abs_path[0:1].upper() + abs_path[1:]
     return abs_path
 
 
 # see posix.py for definitions
 normcasespec: int = encoding.normcasespecs.upper
 normcasefallback = encoding.upperfallback
 
 
 def samestat(s1: os.stat_result, s2: os.stat_result) -> bool:
     return False
 
 
 def shelltocmdexe(path: bytes, env: Mapping[bytes, bytes]) -> bytes:
     r"""Convert shell variables in the form $var and ${var} inside ``path``
     to %var% form.  Existing Windows style variables are left unchanged.
 
     The variables are limited to the given environment.  Unknown variables are
     left unchanged.
 
     >>> e = {b'var1': b'v1', b'var2': b'v2', b'var3': b'v3'}
     >>> # Only valid values are expanded
     >>> shelltocmdexe(b'cmd $var1 ${var2} %var3% $missing ${missing} %missing%',
     ...               e)
     'cmd %var1% %var2% %var3% $missing ${missing} %missing%'
     >>> # Single quote prevents expansion, as does \$ escaping
     >>> shelltocmdexe(b"cmd '$var1 ${var2} %var3%' \$var1 \${var2} \\", e)
     'cmd "$var1 ${var2} %var3%" $var1 ${var2} \\'
     >>> # $$ is not special. %% is not special either, but can be the end and
     >>> # start of consecutive variables
     >>> shelltocmdexe(b"cmd $$ %% %var1%%var2%", e)
     'cmd $$ %% %var1%%var2%'
     >>> # No double substitution
     >>> shelltocmdexe(b"$var1 %var1%", {b'var1': b'%var2%', b'var2': b'boom'})
     '%var1% %var1%'
     >>> # Tilde expansion
     >>> shelltocmdexe(b"~/dir ~\dir2 ~tmpfile \~/", {})
     '%USERPROFILE%/dir %USERPROFILE%\\dir2 ~tmpfile ~/'
     """
     if not any(c in path for c in b"$'~"):
         return path
 
     varchars = pycompat.sysbytes(string.ascii_letters + string.digits) + b'_-'
 
     res = b''
     index = 0
     pathlen = len(path)
     while index < pathlen:
         c = path[index : index + 1]
         if c == b'\'':  # no expansion within single quotes
             path = path[index + 1 :]
             pathlen = len(path)
             try:
                 index = path.index(b'\'')
                 res += b'"' + path[:index] + b'"'
             except ValueError:
                 res += c + path
                 index = pathlen - 1
         elif c == b'%':  # variable
             path = path[index + 1 :]
             pathlen = len(path)
             try:
                 index = path.index(b'%')
             except ValueError:
                 res += b'%' + path
                 index = pathlen - 1
             else:
                 var = path[:index]
                 res += b'%' + var + b'%'
         elif c == b'$':  # variable
             if path[index + 1 : index + 2] == b'{':
                 path = path[index + 2 :]
                 pathlen = len(path)
                 try:
                     index = path.index(b'}')
                     var = path[:index]
 
                     # See below for why empty variables are handled specially
                     if env.get(var, b'') != b'':
                         res += b'%' + var + b'%'
                     else:
                         res += b'${' + var + b'}'
                 except ValueError:
                     res += b'${' + path
                     index = pathlen - 1
             else:
                 var = b''
                 index += 1
                 c = path[index : index + 1]
                 while c != b'' and c in varchars:
                     var += c
                     index += 1
                     c = path[index : index + 1]
                 # Some variables (like HG_OLDNODE) may be defined, but have an
                 # empty value.  Those need to be skipped because when spawning
                 # cmd.exe to run the hook, it doesn't replace %VAR% for an empty
                 # VAR, and that really confuses things like revset expressions.
                 # OTOH, if it's left in Unix format and the hook runs sh.exe, it
                 # will substitute to an empty string, and everything is happy.
                 if env.get(var, b'') != b'':
                     res += b'%' + var + b'%'
                 else:
                     res += b'$' + var
 
                 if c != b'':
                     index -= 1
         elif (
             c == b'~'
             and index + 1 < pathlen
             and path[index + 1 : index + 2] in (b'\\', b'/')
         ):
             res += b"%USERPROFILE%"
         elif (
             c == b'\\'
             and index + 1 < pathlen
             and path[index + 1 : index + 2] in (b'$', b'~')
         ):
             # Skip '\', but only if it is escaping $ or ~
             res += path[index + 1 : index + 2]
             index += 1
         else:
             res += c
 
         index += 1
     return res
 
 
 # A sequence of backslashes is special iff it precedes a double quote:
 # - if there's an even number of backslashes, the double quote is not
 #   quoted (i.e. it ends the quoted region)
 # - if there's an odd number of backslashes, the double quote is quoted
 # - in both cases, every pair of backslashes is unquoted into a single
 #   backslash
 # (See http://msdn2.microsoft.com/en-us/library/a1y7w461.aspx )
 # So, to quote a string, we must surround it in double quotes, double
 # the number of backslashes that precede double quotes and add another
 # backslash before every double quote (being careful with the double
 # quote we've appended to the end)
 _quotere: Optional[Pattern[bytes]] = None
 _needsshellquote = None
 
 
 def shellquote(s: bytes) -> bytes:
     r"""
     >>> shellquote(br'C:\Users\xyz')
     '"C:\\Users\\xyz"'
     >>> shellquote(br'C:\Users\xyz/mixed')
     '"C:\\Users\\xyz/mixed"'
     >>> # Would be safe not to quote too, since it is all double backslashes
     >>> shellquote(br'C:\\Users\\xyz')
     '"C:\\\\Users\\\\xyz"'
     >>> # But this must be quoted
     >>> shellquote(br'C:\\Users\\xyz/abc')
     '"C:\\\\Users\\\\xyz/abc"'
     """
     global _quotere
     if _quotere is None:
         _quotere = re.compile(br'(\\*)("|\\$)')
     global _needsshellquote
     if _needsshellquote is None:
         # ":" is also treated as "safe character", because it is used as a part
         # of path name on Windows.  "\" is also part of a path name, but isn't
         # safe because shlex.split() (kind of) treats it as an escape char and
         # drops it.  It will leave the next character, even if it is another
         # "\".
         _needsshellquote = re.compile(br'[^a-zA-Z0-9._:/-]').search
     if s and not _needsshellquote(s) and not _quotere.search(s):
         # "s" shouldn't have to be quoted
         return s
     return b'"%s"' % _quotere.sub(br'\1\1\\\2', s)
 
 
 def _unquote(s: bytes) -> bytes:
     if s.startswith(b'"') and s.endswith(b'"'):
         return s[1:-1]
     return s
 
 
 def shellsplit(s: bytes) -> List[bytes]:
     """Parse a command string in cmd.exe way (best-effort)"""
     return pycompat.maplist(_unquote, pycompat.shlexsplit(s, posix=False))
 
 
 # if you change this stub into a real check, please try to implement the
 # username and groupname functions above, too.
 def isowner(st: os.stat_result) -> bool:
     return True
 
 
 def findexe(command: bytes) -> Optional[bytes]:
     """Find executable for command searching like cmd.exe does.
     If command is a basename then PATH is searched for command.
     PATH isn't searched if command is an absolute or relative path.
     An extension from PATHEXT is found and added if not present.
     If command isn't found None is returned."""
     pathext = encoding.environ.get(b'PATHEXT', b'.COM;.EXE;.BAT;.CMD')
     pathexts = [ext for ext in pathext.lower().split(pycompat.ospathsep)]
     if os.path.splitext(command)[1].lower() in pathexts:
         pathexts = [b'']
 
     def findexisting(pathcommand: bytes) -> Optional[bytes]:
         """Will append extension (if needed) and return existing file"""
         for ext in pathexts:
             executable = pathcommand + ext
             if os.path.exists(executable):
                 return executable
         return None
 
     if pycompat.ossep in command:
         return findexisting(command)
 
     for path in encoding.environ.get(b'PATH', b'').split(pycompat.ospathsep):
         executable = findexisting(os.path.join(path, command))
         if executable is not None:
             return executable
     return findexisting(os.path.expanduser(os.path.expandvars(command)))
 
 
 _wantedkinds = {stat.S_IFREG, stat.S_IFLNK}
 
 
 def statfiles(files: Sequence[bytes]) -> Iterator[Optional[os.stat_result]]:
     """Stat each file in files. Yield each stat, or None if a file
     does not exist or has a type we don't care about.
 
     Cluster and cache stat per directory to minimize number of OS stat calls."""
     dircache = {}  # dirname -> filename -> status | None if file does not exist
     getkind = stat.S_IFMT
     for nf in files:
         nf = normcase(nf)
         dir, base = os.path.split(nf)
         if not dir:
             dir = b'.'
         cache = dircache.get(dir, None)
         if cache is None:
             try:
                 dmap = {
                     normcase(n): s
                     for n, k, s in listdir(dir, True)
                     if getkind(s.st_mode) in _wantedkinds
                 }
             except (FileNotFoundError, NotADirectoryError):
                 dmap = {}
             cache = dircache.setdefault(dir, dmap)
         yield cache.get(base, None)
 
 
 def username(uid: Optional[int] = None) -> Optional[bytes]:
     """Return the name of the user with the given uid.
 
     If uid is None, return the name of the current user."""
     if not uid:
         try:
             return pycompat.fsencode(getpass.getuser())
         except ModuleNotFoundError:
             # getpass.getuser() checks for a few environment variables first,
             # but if those aren't set, imports pwd and calls getpwuid(), none of
             # which exists on Windows.
             pass
     return None
 
 
 def groupname(gid: Optional[int] = None) -> Optional[bytes]:
     """Return the name of the group with the given gid.
 
     If gid is None, return the name of the current group."""
     return None
 
 
 def readlink(pathname: bytes) -> bytes:
     path = pycompat.fsdecode(pathname)
     try:
         link = os.readlink(path)
     except ValueError as e:
         # On py2, os.readlink() raises an AttributeError since it is
         # unsupported.  On py3, reading a non-link raises a ValueError.  Simply
         # treat this as the error the locking code has been expecting up to now
         # until an effort can be made to enable symlink support on Windows.
         raise AttributeError(e)
     return pycompat.fsencode(link)
 
 
 def removedirs(name: bytes) -> None:
     """special version of os.removedirs that does not remove symlinked
     directories or junction points if they actually contain files"""
     if listdir(name):
         return
     os.rmdir(name)
     head, tail = os.path.split(name)
     if not tail:
         head, tail = os.path.split(head)
     while head and tail:
         try:
             if listdir(head):
                 return
             os.rmdir(head)
         except (ValueError, OSError):
             break
         head, tail = os.path.split(head)
 
 
 def rename(src: bytes, dst: bytes) -> None:
     '''atomically rename file src to dst, replacing dst if it exists'''
     try:
         os.rename(src, dst)
     except FileExistsError:
         unlink(dst)
         os.rename(src, dst)
 
 
 def gethgcmd() -> List[bytes]:
     return [encoding.strtolocal(arg) for arg in [sys.executable] + sys.argv[:1]]
 
 
 def groupmembers(name: bytes) -> List[bytes]:
     # Don't support groups on Windows for now
     raise KeyError
 
 
 def isexec(f: bytes) -> bool:
     return False
 
 
 class cachestat:
+    stat: Optional[os.stat_result]
+
     def __init__(self, path: bytes) -> None:
-        pass
+        self.stat = None
 
     def cacheable(self) -> bool:
         return False
 
 
 def lookupreg(
     key: bytes,
     valname: Optional[bytes] = None,
     scope: Optional[Union[int, Iterable[int]]] = None,
 ) -> Optional[bytes]:
     """Look up a key/value name in the Windows registry.
 
     valname: value name. If unspecified, the default value for the key
     is used.
     scope: optionally specify scope for registry lookup, this can be
     a sequence of scopes to look up in order. Default (CURRENT_USER,
     LOCAL_MACHINE).
     """
     if scope is None:
         # pytype: disable=module-attr
         scope = (winreg.HKEY_CURRENT_USER, winreg.HKEY_LOCAL_MACHINE)
         # pytype: enable=module-attr
     elif not isinstance(scope, (list, tuple)):
         scope = (scope,)
     for s in scope:
         try:
             # pytype: disable=module-attr
             with winreg.OpenKey(s, encoding.strfromlocal(key)) as hkey:
                 # pytype: enable=module-attr
                 name = None
                 if valname is not None:
                     name = encoding.strfromlocal(valname)
                 # pytype: disable=module-attr
                 val = winreg.QueryValueEx(hkey, name)[0]
                 # pytype: enable=module-attr
 
                 # never let a Unicode string escape into the wild
                 return encoding.unitolocal(val)
         except EnvironmentError:
             pass
 
 
 expandglobs: bool = True
 
 
 def statislink(st: Optional[os.stat_result]) -> bool:
     '''check whether a stat result is a symlink'''
     return False
 
 
 def statisexec(st: Optional[os.stat_result]) -> bool:
     '''check whether a stat result is an executable file'''
     return False
 
 
 def poll(fds) -> List:
     # see posix.py for description
     raise NotImplementedError()
 
 
 def readpipe(pipe) -> bytes:
     """Read all available data from a pipe."""
     chunks = []
     while True:
         size = win32.peekpipe(pipe)
         if not size:
             break
 
         s = pipe.read(size)
         if not s:
             break
         chunks.append(s)
 
     return b''.join(chunks)
 
 
 def bindunixsocket(sock, path: bytes) -> NoReturn:
     raise NotImplementedError('unsupported platform')