GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing raw.githubusercontent.com — raw.githubusercontent.com # Copyright (C) 2008, 2009 Michael Trier ([email protected]) and contributors
#
# This module is part of GitPython and is released under the
# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/
import sys
__all__ = [
"stream_copy",
"join_path",
"to_native_path_linux",
"join_path_native",
"Stats",
"IterableObj",
"IterableList",
"BlockingLockFile",
"LockFile",
"Actor",
"get_user_id",
"assure_directory_exists",
"RemoteProgress",
"CallableRemoteProgress",
"rmtree",
"unbare_repo",
]
if sys.platform == "win32":
__all__.append("to_native_path_windows")
import contextlib
import getpass
import logging
import ntpath
import os
import os.path as osp
import platform
import re
import shutil
import stat
import subprocess
import time
from abc import abstractmethod
from functools import wraps
from pathlib import Path
# typing ---------------------------------------------------------
from typing import (
TYPE_CHECKING,
Any,
AnyStr,
Callable,
Dict,
Generator,
Iterator,
List,
Optional,
Pattern,
Sequence,
Tuple,
Type,
TypeVar,
Union,
cast,
overload,
)
from urllib.parse import urlsplit, urlunsplit
# NOTE: Unused imports can be improved now that CI testing has fully resumed. Some of
# these be used indirectly through other GitPython modules, which avoids having to write
# gitdb all the time in their imports. They are not in __all__, at least currently,
# because they could be removed or changed at any time, and so should not be considered
# conceptually public to code outside GitPython. Linters of course do not like it.
from gitdb.util import (
LazyMixin, # noqa: F401
LockedFD, # noqa: F401
bin_to_hex, # noqa: F401
file_contents_ro, # noqa: F401
file_contents_ro_filepath, # noqa: F401
hex_to_bin, # noqa: F401
)
if TYPE_CHECKING:
from git.cmd import Git
from git.config import GitConfigParser, SectionConstraint
from git.remote import Remote
from git.repo.base import Repo
from git.compat import safe_decode
from git.types import (
HSH_TD,
Files_TD,
Has_id_attribute,
Literal,
PathLike,
Protocol,
SupportsIndex,
SupportsRead,
SupportsWrite,
Total_TD,
runtime_checkable,
)
# ---------------------------------------------------------------------
T_IterableObj = TypeVar("T_IterableObj", bound=Union["IterableObj", "Has_id_attribute"], covariant=True)
# So IterableList[Head] is subtype of IterableList[IterableObj].
T_Actor = TypeVar("T_Actor", bound="Actor")
_logger = logging.getLogger(__name__)
def to_hex_sha(sha: Union[str, bytes]) -> Union[str, bytes]:
"""Normalize legacy SHA-1 or SHA-256 IDs; use ``bin_to_hex`` for raw IDs."""
if len(sha) in (40, 64):
hex_to_bin(sha) # Validate already hexadecimal input.
return sha
if isinstance(sha, bytes) and len(sha) in (20, 32):
return bin_to_hex(sha)
raise ValueError("Expected a binary or hexadecimal SHA-1/SHA-256 object ID")
def to_bin_sha(sha: Union[str, bytes]) -> bytes:
"""Normalize legacy SHA-1 or SHA-256 IDs; use ``hex_to_bin`` for hex IDs."""
if isinstance(sha, bytes) and len(sha) in (20, 32):
return sha
if len(sha) in (40, 64):
return hex_to_bin(sha)
raise ValueError("Expected a binary or hexadecimal SHA-1/SHA-256 object ID")
# { Utility Methods
T = TypeVar("T")
def unbare_repo(func: Callable[..., T]) -> Callable[..., T]:
"""Methods with this decorator raise :exc:`~git.exc.InvalidGitRepositoryError` if
they encounter a bare repository."""
from .exc import InvalidGitRepositoryError
@wraps(func)
def wrapper(self: "Remote", *args: Any, **kwargs: Any) -> T:
if self.repo.bare:
raise InvalidGitRepositoryError("Method '%s' cannot operate on bare repositories" % func.__name__)
# END bare method
return func(self, *args, **kwargs)
# END wrapper
return wrapper
@contextlib.contextmanager
def cwd(new_dir: PathLike) -> Generator[PathLike, None, None]:
"""Context manager to temporarily change directory.
This is similar to :func:`contextlib.chdir` introduced in Python 3.11, but the
context manager object returned by a single call to this function is not reentrant.
"""
old_dir = os.getcwd()
os.chdir(new_dir)
try:
yield new_dir
finally:
os.chdir(old_dir)
@contextlib.contextmanager
def patch_env(name: str, value: str) -> Generator[None, None, None]:
"""Context manager to temporarily patch an environment variable."""
old_value = os.getenv(name)
os.environ[name] = value
try:
yield
finally:
if old_value is None:
os.environ.pop(name, None)
else:
os.environ[name] = old_value
def rmtree(path: PathLike) -> None:
"""Remove the given directory tree recursively.
:note:
We use :func:`shutil.rmtree` but adjust its behaviour to see whether files that
couldn't be deleted are read-only. Windows will not remove them in that case.
"""
def handler(function: Callable[[str], Any], path: str, _excinfo: Any) -> None:
"""Callback for :func:`shutil.rmtree`.
This works as either a ``onexc`` or ``onerror`` style callback.
"""
# Is the error an access error?
os.chmod(path, stat.S_IWUSR)
function(path)
if sys.platform != "win32":
shutil.rmtree(path)
elif sys.version_info >= (3, 12):
shutil.rmtree(path, onexc=handler)
else:
shutil.rmtree(path, onerror=handler)
def rmfile(path: PathLike) -> None:
"""Ensure file deleted also on *Windows* where read-only files need special
treatment."""
if osp.isfile(path):
if sys.platform == "win32":
os.chmod(path, stat.S_IWUSR)
os.remove(path)
def stream_copy(source: SupportsRead[AnyStr], destination: SupportsWrite[AnyStr], chunk_size: int = 512 * 1024) -> int:
"""Copy all data from the `source` stream into the `destination` stream in chunks
of size `chunk_size`.
Keep reading after short reads until the source returns an empty chunk.
:return:
Number of bytes written
"""
br = 0
while True:
chunk = source.read(chunk_size)
destination.write(chunk)
br += len(chunk)
if not chunk:
break
# END reading output stream
return br
def join_path(a: PathLike, *p: PathLike) -> PathLike:
R"""Join path tokens together similar to osp.join, but always use ``/`` instead of
possibly ``\`` on Windows.
Later tokens are treated as relative, even when they start with ``/``.
"""
path = os.fspath(a)
for b in p:
b = os.fspath(b).lstrip("/")
if not b:
continue
if path == "" or path.endswith("/"):
path += b
else:
path += "/" + b
# END for each path token to add
return path
if sys.platform == "win32":
def to_native_path_windows(path: PathLike) -> str:
path = os.fspath(path)
return path.replace("/", "\\")
def to_native_path_linux(path: PathLike) -> str:
path = os.fspath(path)
return path.replace("\\", "/")
to_native_path = to_native_path_windows
else:
# No need for any work on Linux.
def to_native_path_linux(path: PathLike) -> str:
return os.fspath(path)
to_native_path = to_native_path_linux
def join_path_native(a: PathLike, *p: PathLike) -> PathLike:
R"""Like :func:`join_path`, but makes sure an OS native path is returned.
This is only needed to play it safe on Windows and to ensure nice paths that only
use ``\``.
"""
return to_native_path(join_path(a, *p))
def _is_path_rooted(path: PathLike) -> bool:
r"""Whether ``path`` has a root, including one encoded in a UNC drive.
On Windows, ``\directory`` is rooted on the current drive without being
absolute, while ``C:\directory`` has both a drive and a root. In contrast,
``directory`` and the drive-relative ``C:directory`` have no root.
UNC paths are rooted: ``\\server\share`` stores the share in the drive
returned by :func:`os.path.splitdrive`, while ``\\server\share\directory``
additionally has a rooted tail.
On POSIX, which has no drive concept, this simply distinguishes absolute
paths such as ``/directory`` from relative paths such as ``directory``.
"""
drive, tail = osp.splitdrive(os.fspath(path))
separators = (os.sep,) if os.altsep is None else (os.sep, os.altsep)
return tail.startswith(separators) or drive.startswith(separators)
def _to_relative_path(root: PathLike, path: PathLike) -> str:
r"""Return a normalized Git-style path confined to ``root``.
A Windows path such as ``\directory`` is rooted but not absolute. Resolve it
against the drive of ``root`` rather than treating it as relative to ``root``.
Drive-relative paths such as ``C:directory`` are rejected because their meaning
depends on process-global per-drive state.
For example, with ``root`` set to ``C:\repo`` on Windows:
* ``directory\file`` -> ``directory/file``
* ``directory\`` -> ``directory/``
* ``C:\repo\directory\file`` -> ``directory/file``
* ``\repo\directory\file`` -> ``directory/file``
* ``C:directory\file`` -> :exc:`ValueError`
* ``C:\other\file`` -> :exc:`ValueError`
On POSIX, ``/repo/directory/file`` under ``/repo`` similarly becomes
``directory/file``. A trailing separator is preserved as a Git-style ``/``.
"""
path_str = os.fspath(path)
if not path_str:
return path_str
drive, _tail = osp.splitdrive(path_str)
rooted = _is_path_rooted(path_str)
if drive and not rooted:
raise ValueError("Drive-relative path %r is not supported" % path_str)
root_abs = osp.abspath(os.fspath(root))
path_abs = osp.abspath(osp.join(root_abs, path_str))
try:
common_path = osp.commonpath([root_abs, path_abs])
except ValueError as e:
raise ValueError("Path %r is not in repository at %r" % (path_str, root_abs)) from e
if common_path != root_abs:
raise ValueError("Path %r is not in repository at %r" % (path_str, root_abs))
relative_path = to_native_path_linux(osp.relpath(path_abs, root_abs))
separators = (os.sep,) if os.altsep is None else (os.sep, os.altsep)
if path_str.endswith(separators) and relative_path != "." and not relative_path.endswith("/"):
relative_path += "/"
return relative_path
_HFS_IGNORABLES = str.maketrans(
"", "", "\u200c\u200d\u200e\u200f\u202a\u202b\u202c\u202d\u202e\u206a\u206b\u206c\u206d\u206e\u206f\ufeff"
)
# Match Git's is_ntfs_dotgitmodules in path.c on a lowercased, trimmed name.
# Besides gitmod~1..4, fallback aliases have exactly eight ASCII characters:
# a shrinking prefix of "gi7eba", "~", and digits with no leading zero.
# Explicit digit counts avoid accepting shorter/longer names or Unicode digits.
_NTFS_DOTGITMODULES_SHORT_NAME = re.compile(
r"(?:gitmod~[1-4]|gi7eba~[1-9]|gi7eb~[1-9][0-9]|gi7e~[1-9][0-9]{2}|"
r"gi7~[1-9][0-9]{3}|gi~[1-9][0-9]{4}|g~[1-9][0-9]{5}|~[1-9][0-9]{6})"
)
def _validate_repo_path(path: PathLike, mode: Union[int, None] = None) -> None:
"""Reject unsafe tree/index paths without normalizing away their components.
Protect Git metadata aliases on NTFS and HFS even when writing on another
platform. Other POSIX filename characters, including newlines, remain valid.
:param mode:
Mode of the index or tree entry the path belongs to, where one is known.
Git refuses a symbolic link that aliases ``.gitmodules``, since the
submodule configuration would then be read through the link, so that
name is only rejected once the mode says the entry is a link.
"""
name = os.fspath(path)
if not name or "\0" in name or ntpath.splitdrive(name)[0] or name.startswith("/"):
raise ValueError("Invalid repository path %r" % name)
if os.name == "nt" and "\\" in name:
raise ValueError("Index paths must use '/' separators: %r" % name)
for component in name.split("/"):
if component in ("", ".", ".."):
raise ValueError("Invalid repository path %r" % name)
# NTFS recognizes backslashes and alternate data streams in metadata names.
for part in component.split("\\"):
ntfs_name = part.split(":", 1)[0].rstrip(" .").lower()
hfs_name = part.translate(_HFS_IGNORABLES).lower()
if ntfs_name in (".git", "git~1") or hfs_name == ".git":
raise ValueError("Repository path aliases Git metadata: %r" % name)
if mode is not None and stat.S_ISLNK(mode):
if (
ntfs_name == ".gitmodules"
or hfs_name == ".gitmodules"
or _NTFS_DOTGITMODULES_SHORT_NAME.fullmatch(ntfs_name) is not None
):
raise ValueError("Symbolic link aliases the submodule configuration: %r" % name)
def assure_directory_exists(path: PathLike, is_file: bool = False) -> bool:
"""Make sure that the directory pointed to by path exists.
:param is_file:
If ``True``, `path` is assumed to be a file and handled correctly.
Otherwise it must be a directory.
:return:
``True`` if the directory was created, ``False`` if it already existed.
"""
if is_file:
path = osp.dirname(path) or osp.curdir
# END handle file
if not osp.isdir(path):
os.makedirs(path, exist_ok=True)
return True
return False
def _get_exe_extensions() -> Sequence[str]:
PATHEXT = os.environ.get("PATHEXT", None)
if PATHEXT:
return tuple(p.upper() for p in PATHEXT.split(os.pathsep))
elif sys.platform == "win32":
return (".BAT", ".COM", ".EXE")
else:
return ()
def py_where(program: str, path: Optional[PathLike] = None) -> List[str]:
"""Perform a path search to assist :func:`is_cygwin_git`.
This is not robust for general use. It is an implementation detail of
:func:`is_cygwin_git`. When a search following all shell rules is needed,
:func:`shutil.which` can be used instead.
:note:
Neither this function nor :func:`shutil.which` will predict the effect of an
executable search on a native Windows system due to a :class:`subprocess.Popen`
call without ``shell=True``, because shell and non-shell executable search on
Windows differ considerably.
"""
# From: http://stackoverflow.com/a/377028/548792
winprog_exts = _get_exe_extensions()
def is_exec(fpath: str) -> bool:
return (
osp.isfile(fpath)
and os.access(fpath, os.X_OK)
and (
sys.platform != "win32" or not winprog_exts or any(fpath.upper().endswith(ext) for ext in winprog_exts)
)
)
progs = []
if not path:
path = os.environ["PATH"]
for folder in os.fspath(path).split(os.pathsep):
folder = folder.strip('"')
if folder:
exe_path = osp.join(folder, program)
for f in [exe_path] + ["%s%s" % (exe_path, e) for e in winprog_exts]:
if is_exec(f):
progs.append(f)
return progs
def _cygexpath(drive: Optional[str], path: str) -> str:
if osp.isabs(path) and not drive:
# Invoked from `cygpath()` directly with `D:Apps\123`?
# It's an error, leave it alone just slashes)
p = path # convert to str if AnyPath given
else:
p = path and osp.normpath(osp.expanduser(path))
if osp.isabs(p):
if drive:
# Confusing, maybe a remote system should expand vars.
p = path
else:
p = cygpath(p)
elif drive:
p = "/proc/cygdrive/%s/%s" % (drive.lower(), p)
p_str = os.fspath(p) # ensure it is a str and not AnyPath
return p_str.replace("\\", "/")
_cygpath_parsers: Tuple[Tuple[Pattern[str], Callable[..., str], bool], ...] = (
# See: https://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx
# and: https://www.cygwin.com/cygwin-ug-net/using.html#unc-paths
(
re.compile(r"\\\\\?\\UNC\\([^\\]+)\\([^\\]+)(?:\\(.*))?"),
(lambda server, share, rest_path: "//%s/%s/%s" % (server, share, rest_path.replace("\\", "/"))),
False,
),
(re.compile(r"\\\\\?\\(\w):[/\\](.*)"), (_cygexpath), False),
(re.compile(r"(\w):[/\\](.*)"), (_cygexpath), False),
(re.compile(r"file:(.*)", re.I), (lambda rest_path: rest_path), True),
(re.compile(r"(\w{2,}:.*)"), (lambda url: url), False), # remote URL, do nothing
)
def cygpath(path: str) -> str:
"""Use :meth:`git.cmd.Git.polish_url` instead, that works on any environment."""
path = os.fspath(path) # Ensure is str and not AnyPath.
# Fix to use Paths when 3.5 dropped. Or to be just str if only for URLs?
if not path.startswith(("/cygdrive", "//", "/proc/cygdrive")):
for regex, parser, recurse in _cygpath_parsers:
match = regex.match(path)
if match:
if parser is _cygexpath:
path = parser(*match.groups())
else:
path = parser(*match.groups())
if recurse:
path = cygpath(path)
break
else:
path = _cygexpath(None, path)
return path
_decygpath_regex = re.compile(r"(?:/proc)?/cygdrive/(\w)(/.*)?")
def decygpath(path: PathLike) -> str:
path = os.fspath(path)
m = _decygpath_regex.match(path)
if m:
drive, rest_path = m.groups()
path = "%s:%s" % (drive.upper(), rest_path or "")
return path.replace("/", "\\")
#: Store boolean flags denoting if a specific Git executable
#: is from a Cygwin installation. TODO: use @functools.lru_cache(user_function)
_is_cygwin_cache: Dict[str, Optional[bool]] = {}
def _is_cygwin_git(git_executable: str) -> bool:
is_cygwin = _is_cygwin_cache.get(git_executable) # type: Optional[bool]
if is_cygwin is None:
is_cygwin = False
try:
git_dir = osp.dirname(git_executable)
if not git_dir:
res = py_where(git_executable)
git_dir = osp.dirname(res[0]) if res else ""
# Just a name given, not a real path.
uname_cmd = osp.join(git_dir, "uname")
if not (Path(uname_cmd).is_file() and os.access(uname_cmd, os.X_OK)):
_logger.debug(f"Failed checking if running in CYGWIN: {uname_cmd} is not an executable")
_is_cygwin_cache[git_executable] = is_cygwin
return is_cygwin
process = subprocess.Popen([uname_cmd], stdout=subprocess.PIPE, universal_newlines=True)
uname_out, _ = process.communicate()
# retcode = process.poll()
is_cygwin = "CYGWIN" in uname_out
except Exception as ex:
_logger.debug("Failed checking if running in CYGWIN due to: %r", ex)
_is_cygwin_cache[git_executable] = is_cygwin
return is_cygwin
@overload
def is_cygwin_git(git_executable: None) -> Literal[False]: ...
@overload
def is_cygwin_git(git_executable: PathLike) -> bool: ...
def is_cygwin_git(git_executable: Union[None, PathLike]) -> bool:
_logger.debug(f"{sys.platform=}, {git_executable=}")
if sys.platform != "cygwin":
return False
elif git_executable is None:
return False
else:
return _is_cygwin_git(str(git_executable))
def get_user_id() -> str:
""":return: String identifying the currently active system user as ``name@node``"""
return "%s@%s" % (getpass.getuser(), platform.node())
def finalize_process(proc: Union["subprocess.Popen[Any]", "Git.AutoInterrupt"], **kwargs: Any) -> None:
"""Wait for the process (clone, fetch, pull or push) and handle its errors
accordingly."""
# TODO: No close proc-streams??
proc.wait(**kwargs)
@overload
def expand_path(p: None) -> None: ...
@overload
def expand_path(p: PathLike) -> Optional[PathLike]:
# TODO: Support for Python 3.5 has been dropped, so these overloads can be improved.
...
def expand_path(p: Union[None, PathLike]) -> Optional[PathLike]:
if p is None:
return None
try:
expanded_path = osp.expanduser(os.fspath(p))
return osp.normpath(osp.abspath(expanded_path))
except Exception:
return None
def _redact_url_credentials(text: str) -> Optional[str]:
"""Return `text` with the userinfo of a URL redacted, or ``None`` if it has none."""
try:
url = urlsplit(text)
except ValueError:
# This is not a valid URL.
return None
if url.password is None and url.username is None:
return None
# Match urllib.parse's userinfo boundary. Keeping the raw hostinfo
# preserves hostname case, IPv6 brackets, and port formatting.
_, _, hostinfo = url.netloc.rpartition("@")
redacted = "*****:*****" if url.password is not None else "*****"
return urlunsplit(url._replace(netloc=f"{redacted}@{hostinfo}"))
def remove_password_if_present(cmdline: Sequence[str]) -> List[str]:
"""Redact credentials in URLs and HTTP Authorization extra headers in a command line.
URLs are also redacted in the key and value of a configuration assignment, such
as ``http.proxy= |
| You are visitor 01402520 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |