Gate releases on public Manager clone
Verify Manager release install / verify-manager-install (push) Has been cancelled
Verify Manager release install / verify-manager-install (push) Has been cancelled
This commit is contained in:
Executable
+192
@@ -0,0 +1,192 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Verify the public repository exactly as ComfyUI Manager installs it."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
import tomllib
|
||||
|
||||
|
||||
class VerificationError(RuntimeError):
|
||||
"""Raised when release metadata cannot produce a clean Manager install."""
|
||||
|
||||
|
||||
def run(command: list[str], *, cwd: Path, env: dict[str, str]) -> str:
|
||||
completed = subprocess.run(
|
||||
command,
|
||||
cwd=cwd,
|
||||
env=env,
|
||||
check=True,
|
||||
text=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
)
|
||||
return completed.stdout.strip()
|
||||
|
||||
|
||||
def load_release_metadata(source: Path) -> tuple[str, str]:
|
||||
with (source / "pyproject.toml").open("rb") as handle:
|
||||
metadata = tomllib.load(handle)
|
||||
|
||||
project = metadata.get("project", {})
|
||||
version = str(project.get("version", "")).strip()
|
||||
repository = str(project.get("urls", {}).get("Repository", "")).strip()
|
||||
if not version:
|
||||
raise VerificationError("pyproject.toml is missing project.version")
|
||||
validate_public_repository_url(repository)
|
||||
return version, repository.rstrip("/")
|
||||
|
||||
|
||||
def validate_public_repository_url(repository: str) -> None:
|
||||
parsed = urlsplit(repository)
|
||||
if parsed.scheme != "https" or not parsed.netloc or not parsed.path.strip("/"):
|
||||
raise VerificationError(
|
||||
"project.urls.Repository must be a complete public HTTPS repository URL"
|
||||
)
|
||||
if parsed.username or parsed.password or parsed.query or parsed.fragment:
|
||||
raise VerificationError(
|
||||
"project.urls.Repository must not contain credentials, query parameters, or a fragment"
|
||||
)
|
||||
|
||||
|
||||
def manager_clone_command(repository: str, destination: Path) -> list[str]:
|
||||
return [
|
||||
"git",
|
||||
"clone",
|
||||
"-v",
|
||||
"--recursive",
|
||||
"--progress",
|
||||
"--",
|
||||
repository,
|
||||
str(destination),
|
||||
]
|
||||
|
||||
|
||||
def anonymous_git_environment(base: dict[str, str], home: Path) -> dict[str, str]:
|
||||
env = dict(base)
|
||||
for key in (
|
||||
"GIT_ASKPASS",
|
||||
"GIT_CONFIG_GLOBAL",
|
||||
"GIT_CONFIG_PARAMETERS",
|
||||
"GIT_SSH",
|
||||
"GIT_SSH_COMMAND",
|
||||
"SSH_ASKPASS",
|
||||
):
|
||||
env.pop(key, None)
|
||||
env.update(
|
||||
{
|
||||
"GIT_ASKPASS": "/bin/false",
|
||||
"GIT_CONFIG_GLOBAL": "/dev/null",
|
||||
"GIT_TERMINAL_PROMPT": "0",
|
||||
"HOME": str(home),
|
||||
"SSH_ASKPASS": "/bin/false",
|
||||
}
|
||||
)
|
||||
return env
|
||||
|
||||
|
||||
def verify_source_state(source: Path, version: str, env: dict[str, str]) -> str:
|
||||
status = run(
|
||||
["git", "status", "--porcelain", "--untracked-files=all"],
|
||||
cwd=source,
|
||||
env=env,
|
||||
)
|
||||
if status:
|
||||
raise VerificationError("release source worktree is not clean")
|
||||
head = run(["git", "rev-parse", "HEAD"], cwd=source, env=env)
|
||||
tag_commit = run(
|
||||
["git", "rev-list", "-n", "1", f"v{version}"], cwd=source, env=env
|
||||
)
|
||||
if tag_commit != head:
|
||||
raise VerificationError(f"tag v{version} does not resolve to source HEAD {head}")
|
||||
return head
|
||||
|
||||
|
||||
def verify_cloned_sources(clone: Path, scratch: Path, env: dict[str, str]) -> None:
|
||||
required = (
|
||||
"__init__.py",
|
||||
"LICENSE",
|
||||
"README.md",
|
||||
"pyproject.toml",
|
||||
"timeline/editor.py",
|
||||
"web/etk_ltxv_timeline_image_editor.js",
|
||||
)
|
||||
missing = [relative for relative in required if not (clone / relative).is_file()]
|
||||
if missing:
|
||||
raise VerificationError(f"public clone is missing release files: {missing}")
|
||||
|
||||
python_files = sorted(str(path) for path in clone.rglob("*.py"))
|
||||
compile_env = dict(env)
|
||||
compile_env["PYTHONPYCACHEPREFIX"] = str(scratch / "pycache")
|
||||
run([sys.executable, "-m", "py_compile", *python_files], cwd=clone, env=compile_env)
|
||||
|
||||
node = shutil.which("node")
|
||||
if not node:
|
||||
raise VerificationError("node is required to syntax-check shipped JavaScript")
|
||||
for path in sorted(clone.rglob("*.js")):
|
||||
run([node, "--check", str(path)], cwd=clone, env=env)
|
||||
|
||||
|
||||
def verify_manager_install(source: Path) -> tuple[str, str, str]:
|
||||
source = source.resolve()
|
||||
version, repository = load_release_metadata(source)
|
||||
with tempfile.TemporaryDirectory(prefix="etk-manager-release-") as temporary:
|
||||
scratch = Path(temporary)
|
||||
home = scratch / "anonymous-home"
|
||||
home.mkdir()
|
||||
env = anonymous_git_environment(os.environ, home)
|
||||
source_head = verify_source_state(source, version, env)
|
||||
|
||||
destination = scratch / "custom_nodes" / "etk-ltxv-timeline-editor"
|
||||
destination.parent.mkdir()
|
||||
run(manager_clone_command(repository, destination), cwd=scratch, env=env)
|
||||
|
||||
clone_head = run(["git", "rev-parse", "HEAD"], cwd=destination, env=env)
|
||||
if clone_head != source_head:
|
||||
raise VerificationError(
|
||||
f"public clone HEAD {clone_head} does not match release source {source_head}"
|
||||
)
|
||||
clone_version, clone_repository = load_release_metadata(destination)
|
||||
if (clone_version, clone_repository) != (version, repository):
|
||||
raise VerificationError("public clone release metadata differs from source")
|
||||
clone_tag = run(
|
||||
["git", "rev-list", "-n", "1", f"v{version}"],
|
||||
cwd=destination,
|
||||
env=env,
|
||||
)
|
||||
if clone_tag != clone_head:
|
||||
raise VerificationError(
|
||||
f"public clone tag v{version} does not resolve to clone HEAD {clone_head}"
|
||||
)
|
||||
verify_cloned_sources(destination, scratch, env)
|
||||
return version, repository, source_head
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument(
|
||||
"--source",
|
||||
type=Path,
|
||||
default=Path(__file__).resolve().parents[1],
|
||||
help="release repository to verify",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
version, repository, commit = verify_manager_install(args.source)
|
||||
except (OSError, subprocess.CalledProcessError, VerificationError) as exc:
|
||||
print(f"MANAGER INSTALL VERIFICATION FAILED: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
print(f"MANAGER INSTALL VERIFICATION PASSED: {repository} v{version} {commit}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user