#
# The high-throughput toolkit (httk)
# Copyright (C) 2012-2024 the httk AUTHORS
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Create the single orphan commit used by a generated documentation site."""
import os
import subprocess
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
__all__ = ["CommitSiteResult", "GitSiteError", "GitUnavailableError", "commit_site"]
[docs]
class GitSiteError(RuntimeError):
"""Raised when a generated site cannot be committed to a Git repository."""
[docs]
class GitUnavailableError(GitSiteError):
"""Raised when the ``git`` executable is not available."""
@dataclass(frozen=True)
[docs]
class CommitSiteResult:
"""Summary of a site commit operation."""
def _run_git(
repository: Path,
arguments: list[str],
*,
input_data: bytes | None = None,
check: bool = True,
environment: Mapping[str, str] | None = None,
) -> subprocess.CompletedProcess[bytes]:
sanitized_environment = {key: value for key, value in os.environ.items() if not key.startswith("GIT_")}
sanitized_environment["GIT_CONFIG_GLOBAL"] = os.devnull
sanitized_environment["GIT_CONFIG_SYSTEM"] = os.devnull
if environment is not None:
sanitized_environment.update(environment)
try:
result = subprocess.run(
["git", "-c", "core.hooksPath=/dev/null", *arguments],
cwd=repository,
input=input_data,
capture_output=True,
check=False,
env=sanitized_environment,
)
except FileNotFoundError as exc:
raise GitUnavailableError("git is not available on PATH") from exc
if check and result.returncode != 0:
detail = (result.stderr or result.stdout).decode(errors="replace").strip()
raise GitSiteError(f"git {' '.join(arguments)} failed ({result.returncode}): {detail}")
return result
def _repository_from(path: Path) -> Path:
result = _run_git(path, ["rev-parse", "--show-toplevel"], check=False)
if result.returncode != 0:
raise GitSiteError(f"{path} is not inside a Git repository")
return Path(result.stdout.decode().strip()).resolve()
def _resolve_repository(site: Path, repository: Path | None) -> Path:
if repository is not None:
root = repository.expanduser().resolve()
if not root.is_dir():
raise GitSiteError(f"Git repository directory does not exist: {root}")
return _repository_from(root)
try:
return _repository_from(site)
except GitSiteError as site_error:
try:
return _repository_from(Path.cwd().resolve())
except GitSiteError:
raise site_error
def _hash_blob(repository: Path, content: bytes) -> str:
result = _run_git(repository, ["hash-object", "-w", "--stdin"], input_data=content)
return result.stdout.decode().strip()
def _make_tree(repository: Path, directory: Path, *, root: bool = False) -> str:
entries: list[bytes] = []
for entry in sorted(os.scandir(directory), key=lambda item: item.name):
# A worktree has a .git file at its root. It is Git administration data,
# not part of the generated site tree.
if root and entry.name == ".git":
continue
path = Path(entry.path)
if entry.is_symlink():
mode = "120000"
kind = "blob"
object_id = _hash_blob(repository, os.fsencode(os.readlink(path)))
elif entry.is_dir(follow_symlinks=False):
mode = "040000"
kind = "tree"
object_id = _make_tree(repository, path)
elif entry.is_file(follow_symlinks=False):
mode = "100755" if os.access(path, os.X_OK) else "100644"
kind = "blob"
object_id = _hash_blob(repository, path.read_bytes())
else:
raise GitSiteError(f"unsupported site entry (not a regular file, directory, or symlink): {path}")
entries.append(f"{mode} {kind} {object_id}\t".encode() + os.fsencode(entry.name) + b"\0")
result = _run_git(repository, ["mktree", "-z"], input_data=b"".join(entries))
return result.stdout.decode().strip()
[docs]
def commit_site(
site_directory: str | Path,
branch: str,
message: str,
*,
repository: str | Path | None = None,
author_name: str | None = None,
author_email: str | None = None,
committer_name: str | None = None,
committer_email: str | None = None,
) -> CommitSiteResult:
"""Replace *branch* with one parentless commit containing *site_directory*.
The site is converted directly into Git objects, so the caller's index and
checked-out files are not changed. Ref leases for concurrent publishers
remain the caller's responsibility when pushing the resulting branch. Git
dates remain ambient, so commit IDs are intentionally not deterministic.
"""
site = Path(site_directory).expanduser().resolve()
if site.is_symlink() or not site.is_dir():
raise GitSiteError(f"site directory does not exist or is not a directory: {site}")
root = _resolve_repository(site, Path(repository) if repository is not None else None)
check = _run_git(root, ["check-ref-format", "--branch", branch], check=False)
if check.returncode != 0:
detail = (check.stderr or check.stdout).decode(errors="replace").strip()
raise GitSiteError(f"invalid branch name {branch!r}: {detail}")
tree = _make_tree(root, site, root=True)
identity = {
"GIT_AUTHOR_NAME": author_name if author_name is not None else "httk docs bot",
"GIT_AUTHOR_EMAIL": author_email if author_email is not None else "docs-bot@httk.org",
"GIT_COMMITTER_NAME": committer_name if committer_name is not None else "httk docs bot",
"GIT_COMMITTER_EMAIL": committer_email if committer_email is not None else "docs-bot@httk.org",
}
commit = (
_run_git(
root,
[
"commit-tree",
tree,
"-m",
message,
],
environment=identity,
)
.stdout.decode()
.strip()
)
_run_git(root, ["update-ref", f"refs/heads/{branch}", commit])
return CommitSiteResult(root, branch, commit, tree)