226 lines
8.1 KiB
Python
226 lines
8.1 KiB
Python
"""Run the complete release gate from an anonymous clone at one exact commit."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PUBLIC_REPOSITORY = "https://repo.andraxion.net/administrator/DocForge2.git"
|
|
EXPECTED_ORIGINS = {
|
|
"forgejo@repo.andraxion.net:administrator/DocForge2.git",
|
|
PUBLIC_REPOSITORY,
|
|
PUBLIC_REPOSITORY.removesuffix(".git"),
|
|
}
|
|
COMMIT_PATTERN = re.compile(r"^[0-9a-f]{40}$")
|
|
COMMAND_TIMEOUT_SECONDS = 30 * 60
|
|
LEGACY_MIGRATION_TAG = "v1.0.0"
|
|
LEGACY_MIGRATION_TAG_REF = f"refs/tags/{LEGACY_MIGRATION_TAG}"
|
|
EXPECTED_LEGACY_TAG_OBJECT = "2d7d306a37da89f1c860c7f0be161c45386acf61"
|
|
EXPECTED_LEGACY_TAG_COMMIT = "593c173b453236a6872d0a4e88e7a51a67a21cde"
|
|
|
|
|
|
class FreshCloneError(RuntimeError):
|
|
"""The anonymous exact-commit release rehearsal failed."""
|
|
|
|
|
|
def _run(
|
|
arguments: list[str],
|
|
*,
|
|
cwd: Path,
|
|
environment: dict[str, str] | None = None,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
completed = subprocess.run(
|
|
arguments,
|
|
cwd=cwd,
|
|
env=environment,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=COMMAND_TIMEOUT_SECONDS,
|
|
)
|
|
if completed.returncode != 0:
|
|
diagnostic = (completed.stdout + completed.stderr)[-4000:]
|
|
raise FreshCloneError(f"Command failed ({' '.join(arguments)}):\n{diagnostic.rstrip()}")
|
|
return completed
|
|
|
|
|
|
def _head_commit() -> str:
|
|
commit = _run(["git", "rev-parse", "HEAD"], cwd=ROOT).stdout.strip()
|
|
if COMMIT_PATTERN.fullmatch(commit) is None:
|
|
raise FreshCloneError("Current release commit is invalid")
|
|
return commit
|
|
|
|
|
|
def _verify_source(commit: str) -> None:
|
|
if _head_commit() != commit:
|
|
raise FreshCloneError("Requested fresh-clone commit is not the current candidate")
|
|
if _run(["git", "status", "--porcelain"], cwd=ROOT).stdout:
|
|
raise FreshCloneError("Fresh-clone proof requires a clean source candidate")
|
|
origin = _run(["git", "remote", "get-url", "origin"], cwd=ROOT).stdout.strip()
|
|
if origin not in EXPECTED_ORIGINS:
|
|
raise FreshCloneError("Origin is not the public DocForge2 successor repository")
|
|
remote = _run(
|
|
["git", "ls-remote", PUBLIC_REPOSITORY, "refs/heads/main", "refs/heads/dev"],
|
|
cwd=ROOT,
|
|
).stdout
|
|
remote_commits = {line.split()[0] for line in remote.splitlines() if line.split()}
|
|
if commit not in remote_commits:
|
|
raise FreshCloneError("Release candidate is not published on main or dev")
|
|
|
|
|
|
def _digest(value: str) -> str:
|
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def validate_legacy_tag(*, object_type: str, tag_object: str, tag_commit: str) -> None:
|
|
if object_type != "tag":
|
|
raise FreshCloneError(f"{LEGACY_MIGRATION_TAG} is not an annotated tag")
|
|
if tag_object != EXPECTED_LEGACY_TAG_OBJECT:
|
|
raise FreshCloneError(
|
|
f"{LEGACY_MIGRATION_TAG} tag object does not match the frozen release"
|
|
)
|
|
if tag_commit != EXPECTED_LEGACY_TAG_COMMIT:
|
|
raise FreshCloneError(f"{LEGACY_MIGRATION_TAG} commit does not match the frozen release")
|
|
|
|
|
|
def obtain_legacy_tag(
|
|
clone: Path,
|
|
) -> tuple[subprocess.CompletedProcess[str], str, str]:
|
|
fetch = _run(
|
|
[
|
|
"git",
|
|
"fetch",
|
|
"--no-tags",
|
|
"origin",
|
|
f"{LEGACY_MIGRATION_TAG_REF}:{LEGACY_MIGRATION_TAG_REF}",
|
|
],
|
|
cwd=clone,
|
|
)
|
|
object_type = _run(
|
|
["git", "cat-file", "-t", LEGACY_MIGRATION_TAG_REF],
|
|
cwd=clone,
|
|
).stdout.strip()
|
|
tag_object = _run(
|
|
["git", "rev-parse", LEGACY_MIGRATION_TAG_REF],
|
|
cwd=clone,
|
|
).stdout.strip()
|
|
tag_commit = _run(
|
|
["git", "rev-parse", f"{LEGACY_MIGRATION_TAG_REF}^{{commit}}"],
|
|
cwd=clone,
|
|
).stdout.strip()
|
|
validate_legacy_tag(
|
|
object_type=object_type,
|
|
tag_object=tag_object,
|
|
tag_commit=tag_commit,
|
|
)
|
|
return fetch, tag_object, tag_commit
|
|
|
|
|
|
def run_fresh_clone_gate(commit: str) -> dict[str, object]:
|
|
"""Clone the public successor anonymously and run its complete release gate."""
|
|
|
|
if COMMIT_PATTERN.fullmatch(commit) is None:
|
|
raise FreshCloneError("Fresh-clone commit must be one full lowercase SHA-1")
|
|
_verify_source(commit)
|
|
started = time.perf_counter_ns()
|
|
with tempfile.TemporaryDirectory(prefix="docforge-m5-fresh-clone-") as directory_name:
|
|
parent = Path(directory_name)
|
|
clone = parent / "DocForge2"
|
|
clone_result = _run(
|
|
["git", "clone", "--no-tags", PUBLIC_REPOSITORY, str(clone)],
|
|
cwd=parent,
|
|
)
|
|
_run(["git", "checkout", "--detach", commit], cwd=clone)
|
|
checked_out = _run(["git", "rev-parse", "HEAD"], cwd=clone).stdout.strip()
|
|
if checked_out != commit:
|
|
raise FreshCloneError("Anonymous clone did not check out the requested commit")
|
|
tag_fetch, tag_object, tag_commit = obtain_legacy_tag(clone)
|
|
fsck = _run(["git", "fsck", "--full"], cwd=clone)
|
|
environment = dict(os.environ)
|
|
environment["UV_LINK_MODE"] = "copy"
|
|
sync = _run(["uv", "sync", "--frozen", "--offline"], cwd=clone, environment=environment)
|
|
npm = _run(["npm", "ci", "--offline"], cwd=clone, environment=environment)
|
|
gate = _run(["make", "release-gate"], cwd=clone, environment=environment)
|
|
status = _run(["git", "status", "--porcelain"], cwd=clone).stdout
|
|
if status:
|
|
raise FreshCloneError("Release gate left the anonymous clone dirty")
|
|
version = _run(
|
|
[str(clone / ".venv/bin/python"), "-m", "docforge.cli", "--version"],
|
|
cwd=clone,
|
|
environment=environment,
|
|
).stdout.strip()
|
|
elapsed_ms = round((time.perf_counter_ns() - started) / 1_000_000, 3)
|
|
return {
|
|
"schema_version": 2,
|
|
"repository": PUBLIC_REPOSITORY,
|
|
"authentication": "anonymous_https",
|
|
"commit": commit,
|
|
"legacy_migration_tag": {
|
|
"name": LEGACY_MIGRATION_TAG,
|
|
"annotated_tag_object": tag_object,
|
|
"commit": tag_commit,
|
|
},
|
|
"clean_after_gate": True,
|
|
"version_surface": version,
|
|
"elapsed_ms": elapsed_ms,
|
|
"logs": {
|
|
"clone_sha256": _digest(clone_result.stdout + clone_result.stderr),
|
|
"legacy_tag_fetch_sha256": _digest(tag_fetch.stdout + tag_fetch.stderr),
|
|
"fsck_sha256": _digest(fsck.stdout + fsck.stderr),
|
|
"sync_sha256": _digest(sync.stdout + sync.stderr),
|
|
"npm_sha256": _digest(npm.stdout + npm.stderr),
|
|
"release_gate_sha256": _digest(gate.stdout + gate.stderr),
|
|
"release_gate_bytes": len((gate.stdout + gate.stderr).encode("utf-8")),
|
|
},
|
|
}
|
|
|
|
|
|
def _write_output(path: Path, payload: bytes) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
descriptor, temporary_name = tempfile.mkstemp(prefix=".m5-fresh-clone-", dir=path.parent)
|
|
temporary = Path(temporary_name)
|
|
try:
|
|
with os.fdopen(descriptor, "wb") as handle:
|
|
handle.write(payload)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
directory_fd = os.open(path.parent, os.O_RDONLY)
|
|
try:
|
|
os.fsync(directory_fd)
|
|
finally:
|
|
os.close(directory_fd)
|
|
except Exception:
|
|
temporary.unlink(missing_ok=True)
|
|
raise
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--commit")
|
|
parser.add_argument("--output", type=Path)
|
|
arguments = parser.parse_args()
|
|
try:
|
|
commit = arguments.commit or _head_commit()
|
|
evidence = run_fresh_clone_gate(commit)
|
|
except (FreshCloneError, OSError, subprocess.SubprocessError) as error:
|
|
print(str(error), file=sys.stderr)
|
|
return 2
|
|
payload = json.dumps(evidence, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n"
|
|
if arguments.output is not None:
|
|
_write_output(arguments.output, payload)
|
|
print(payload.decode("utf-8"), end="")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|