ComfyUI_ETK_LTXV_Timeline_Editor/tools/verify_manager_install.py

#!/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())