388 lines
12 KiB
Python
388 lines
12 KiB
Python
"""Deterministic, bounded updates for repository-owned version metadata."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
from dataclasses import dataclass
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import tempfile
|
|
import tomllib
|
|
|
|
|
|
MAX_VERSION_FILES = 64
|
|
MAX_VERSION_FILE_BYTES = 2 * 1024 * 1024
|
|
|
|
|
|
class VersionMetadataError(RuntimeError):
|
|
"""Raised when release version metadata cannot be changed safely."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class VersionFileMutation:
|
|
path: str
|
|
before: bytes
|
|
after: bytes
|
|
|
|
|
|
def version_metadata_mutations(
|
|
repo_path: Path,
|
|
*,
|
|
target_version: str,
|
|
) -> tuple[VersionFileMutation, ...]:
|
|
"""Render all recognized repository version files without writing them."""
|
|
|
|
version = target_version.removeprefix("v")
|
|
candidates: list[tuple[Path, str]] = []
|
|
if (repo_path / "pyproject.toml").is_file():
|
|
candidates.append((repo_path / "pyproject.toml", "pyproject"))
|
|
for package_name, lock_name in (
|
|
("package.json", "package-lock.json"),
|
|
("webui/package.json", "webui/package-lock.json"),
|
|
("webui/package.release.json", "webui/package-lock.release.json"),
|
|
):
|
|
package = repo_path / package_name
|
|
if not package.is_file():
|
|
continue
|
|
candidates.append((package, "package"))
|
|
lock = repo_path / lock_name
|
|
if lock.is_file():
|
|
candidates.append((lock, "lock"))
|
|
src = repo_path / "src"
|
|
if src.is_dir():
|
|
candidates.extend(
|
|
(path, "manifest")
|
|
for path in sorted(src.glob("**/backend/manifest.py"))
|
|
if path.is_file()
|
|
)
|
|
candidates.extend(
|
|
(path, "package_init")
|
|
for path in sorted(src.glob("*/__init__.py"))
|
|
if path.is_file()
|
|
)
|
|
if len(candidates) > MAX_VERSION_FILES:
|
|
raise VersionMetadataError("Repository version metadata exceeds its file bound.")
|
|
|
|
mutations: list[VersionFileMutation] = []
|
|
declarations = 0
|
|
for path, kind in candidates:
|
|
before = _read_bounded(path)
|
|
if kind == "pyproject":
|
|
after, found = _render_pyproject(before, version=version)
|
|
elif kind == "package":
|
|
after, found = _render_package_json(before, version=version)
|
|
elif kind == "lock":
|
|
after, found = _render_package_lock(before, version=version)
|
|
elif kind == "manifest":
|
|
after, found = _render_python_version(
|
|
before,
|
|
path=path,
|
|
target_name="ModuleManifest",
|
|
keyword_name="version",
|
|
version=version,
|
|
)
|
|
else:
|
|
after, found = _render_python_assignment(
|
|
before,
|
|
path=path,
|
|
assignment_name="__version__",
|
|
version=version,
|
|
)
|
|
if not found:
|
|
continue
|
|
declarations += 1
|
|
if before != after:
|
|
mutations.append(
|
|
VersionFileMutation(
|
|
path=path.relative_to(repo_path).as_posix(),
|
|
before=before,
|
|
after=after,
|
|
)
|
|
)
|
|
if declarations == 0:
|
|
raise VersionMetadataError(
|
|
"Repository has no supported version declaration to update."
|
|
)
|
|
return tuple(mutations)
|
|
|
|
|
|
def apply_version_metadata_mutations(
|
|
repo_path: Path,
|
|
*,
|
|
target_version: str,
|
|
) -> tuple[str, ...]:
|
|
"""Apply one deterministic version update, rolling back on write failure."""
|
|
|
|
mutations = version_metadata_mutations(
|
|
repo_path,
|
|
target_version=target_version,
|
|
)
|
|
written: list[VersionFileMutation] = []
|
|
try:
|
|
for mutation in mutations:
|
|
destination = repo_path / mutation.path
|
|
if _read_bounded(destination) != mutation.before:
|
|
raise VersionMetadataError(
|
|
f"Version metadata changed before update: {mutation.path}"
|
|
)
|
|
_atomic_write(destination, mutation.after)
|
|
written.append(mutation)
|
|
except Exception:
|
|
for mutation in reversed(written):
|
|
_atomic_write(repo_path / mutation.path, mutation.before)
|
|
raise
|
|
return tuple(mutation.path for mutation in mutations)
|
|
|
|
|
|
def version_metadata_paths(
|
|
repo_path: Path,
|
|
*,
|
|
target_version: str,
|
|
) -> tuple[str, ...]:
|
|
"""Return the exact paths a deterministic target-version update owns."""
|
|
|
|
return tuple(
|
|
mutation.path
|
|
for mutation in version_metadata_mutations(
|
|
repo_path,
|
|
target_version=target_version,
|
|
)
|
|
)
|
|
|
|
|
|
def _read_bounded(path: Path) -> bytes:
|
|
try:
|
|
metadata = path.lstat()
|
|
except OSError as exc:
|
|
raise VersionMetadataError(
|
|
f"Version metadata is unavailable: {path.name}"
|
|
) from exc
|
|
if not path.is_file() or path.is_symlink():
|
|
raise VersionMetadataError(
|
|
f"Version metadata must be a regular file: {path.name}"
|
|
)
|
|
if metadata.st_size > MAX_VERSION_FILE_BYTES:
|
|
raise VersionMetadataError(
|
|
f"Version metadata exceeds its size bound: {path.name}"
|
|
)
|
|
try:
|
|
return path.read_bytes()
|
|
except OSError as exc:
|
|
raise VersionMetadataError(
|
|
f"Version metadata could not be read: {path.name}"
|
|
) from exc
|
|
|
|
|
|
def _decode(payload: bytes, *, path: Path | None = None) -> str:
|
|
try:
|
|
return payload.decode("utf-8")
|
|
except UnicodeDecodeError as exc:
|
|
label = path.name if path is not None else "file"
|
|
raise VersionMetadataError(
|
|
f"Version metadata is not UTF-8: {label}"
|
|
) from exc
|
|
|
|
|
|
def _render_pyproject(payload: bytes, *, version: str) -> tuple[bytes, bool]:
|
|
text = _decode(payload)
|
|
try:
|
|
parsed = tomllib.loads(text)
|
|
except tomllib.TOMLDecodeError as exc:
|
|
raise VersionMetadataError("pyproject.toml is malformed.") from exc
|
|
project = parsed.get("project")
|
|
if not isinstance(project, dict) or not isinstance(project.get("version"), str):
|
|
return payload, False
|
|
section = re.search(
|
|
r"(?ms)^\[project\][^\n]*\n(?P<body>.*?)(?=^\[[^\n]+\]|\Z)",
|
|
text,
|
|
)
|
|
if section is None:
|
|
raise VersionMetadataError("pyproject.toml has no editable [project] block.")
|
|
body = section.group("body")
|
|
match = re.search(
|
|
r'(?m)^(?P<prefix>\s*version\s*=\s*)(?P<quote>["\'])(?P<value>[^"\']+)(?P=quote)(?P<suffix>\s*(?:#.*)?)$',
|
|
body,
|
|
)
|
|
if match is None:
|
|
raise VersionMetadataError(
|
|
"pyproject.toml project.version is not a supported literal."
|
|
)
|
|
start = section.start("body") + match.start("value")
|
|
end = section.start("body") + match.end("value")
|
|
rendered = f"{text[:start]}{version}{text[end:]}"
|
|
tomllib.loads(rendered)
|
|
return rendered.encode("utf-8"), True
|
|
|
|
|
|
def _render_package_json(payload: bytes, *, version: str) -> tuple[bytes, bool]:
|
|
data = _json_object(payload, label="package metadata")
|
|
if not isinstance(data.get("version"), str):
|
|
return payload, False
|
|
data["version"] = version
|
|
return _json_bytes(data), True
|
|
|
|
|
|
def _render_package_lock(payload: bytes, *, version: str) -> tuple[bytes, bool]:
|
|
data = _json_object(payload, label="package lock")
|
|
found = False
|
|
if isinstance(data.get("version"), str):
|
|
data["version"] = version
|
|
found = True
|
|
packages = data.get("packages")
|
|
root = packages.get("") if isinstance(packages, dict) else None
|
|
if isinstance(root, dict) and isinstance(root.get("version"), str):
|
|
root["version"] = version
|
|
found = True
|
|
return (_json_bytes(data), True) if found else (payload, False)
|
|
|
|
|
|
def _json_object(payload: bytes, *, label: str) -> dict[str, object]:
|
|
try:
|
|
value = json.loads(_decode(payload))
|
|
except json.JSONDecodeError as exc:
|
|
raise VersionMetadataError(f"{label.capitalize()} is malformed.") from exc
|
|
if not isinstance(value, dict):
|
|
raise VersionMetadataError(f"{label.capitalize()} must be an object.")
|
|
return value
|
|
|
|
|
|
def _json_bytes(value: dict[str, object]) -> bytes:
|
|
return (
|
|
json.dumps(value, indent=2, ensure_ascii=True, separators=(",", ": "))
|
|
+ "\n"
|
|
).encode("utf-8")
|
|
|
|
|
|
def _render_python_version(
|
|
payload: bytes,
|
|
*,
|
|
path: Path,
|
|
target_name: str,
|
|
keyword_name: str,
|
|
version: str,
|
|
) -> tuple[bytes, bool]:
|
|
text = _decode(payload, path=path)
|
|
try:
|
|
tree = ast.parse(text, filename=str(path))
|
|
except SyntaxError as exc:
|
|
raise VersionMetadataError(f"Python metadata is malformed: {path.name}") from exc
|
|
values: list[ast.Constant] = []
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Call) or _call_name(node.func) != target_name:
|
|
continue
|
|
for keyword in node.keywords:
|
|
if (
|
|
keyword.arg == keyword_name
|
|
and isinstance(keyword.value, ast.Constant)
|
|
and isinstance(keyword.value.value, str)
|
|
):
|
|
values.append(keyword.value)
|
|
if not values:
|
|
return payload, False
|
|
if len(values) != 1:
|
|
raise VersionMetadataError(
|
|
f"Python metadata has multiple {target_name}.{keyword_name} values: {path.name}"
|
|
)
|
|
return _replace_python_literal(text, values[0], version=version, path=path), True
|
|
|
|
|
|
def _render_python_assignment(
|
|
payload: bytes,
|
|
*,
|
|
path: Path,
|
|
assignment_name: str,
|
|
version: str,
|
|
) -> tuple[bytes, bool]:
|
|
text = _decode(payload, path=path)
|
|
try:
|
|
tree = ast.parse(text, filename=str(path))
|
|
except SyntaxError as exc:
|
|
raise VersionMetadataError(f"Python metadata is malformed: {path.name}") from exc
|
|
values: list[ast.Constant] = []
|
|
for node in tree.body:
|
|
if not isinstance(node, (ast.Assign, ast.AnnAssign)):
|
|
continue
|
|
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
|
|
value = node.value
|
|
if (
|
|
any(
|
|
isinstance(target, ast.Name) and target.id == assignment_name
|
|
for target in targets
|
|
)
|
|
and isinstance(value, ast.Constant)
|
|
and isinstance(value.value, str)
|
|
):
|
|
values.append(value)
|
|
if not values:
|
|
return payload, False
|
|
if len(values) != 1:
|
|
raise VersionMetadataError(
|
|
f"Python metadata has multiple {assignment_name} assignments: {path.name}"
|
|
)
|
|
return _replace_python_literal(text, values[0], version=version, path=path), True
|
|
|
|
|
|
def _replace_python_literal(
|
|
text: str,
|
|
node: ast.Constant,
|
|
*,
|
|
version: str,
|
|
path: Path,
|
|
) -> bytes:
|
|
if (
|
|
node.lineno != node.end_lineno
|
|
or node.end_col_offset is None
|
|
or node.col_offset < 0
|
|
):
|
|
raise VersionMetadataError(
|
|
f"Python version declaration must be a single-line literal: {path.name}"
|
|
)
|
|
lines = text.splitlines(keepends=True)
|
|
line = lines[node.lineno - 1]
|
|
lines[node.lineno - 1] = (
|
|
line[: node.col_offset]
|
|
+ json.dumps(version)
|
|
+ line[node.end_col_offset :]
|
|
)
|
|
rendered = "".join(lines)
|
|
ast.parse(rendered, filename=str(path))
|
|
return rendered.encode("utf-8")
|
|
|
|
|
|
def _call_name(node: ast.expr) -> str | None:
|
|
if isinstance(node, ast.Name):
|
|
return node.id
|
|
if isinstance(node, ast.Attribute):
|
|
return node.attr
|
|
return None
|
|
|
|
|
|
def _atomic_write(path: Path, payload: bytes) -> None:
|
|
metadata = path.stat()
|
|
descriptor, temporary_name = tempfile.mkstemp(
|
|
prefix=f".{path.name}.",
|
|
dir=path.parent,
|
|
)
|
|
temporary = Path(temporary_name)
|
|
try:
|
|
os.fchmod(descriptor, metadata.st_mode & 0o777)
|
|
with os.fdopen(descriptor, "wb") as handle:
|
|
handle.write(payload)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
temporary.replace(path)
|
|
directory = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
|
|
try:
|
|
os.fsync(directory)
|
|
finally:
|
|
os.close(directory)
|
|
except Exception:
|
|
try:
|
|
os.close(descriptor)
|
|
except OSError:
|
|
pass
|
|
temporary.unlink(missing_ok=True)
|
|
raise
|