Josh-sync init script

josh-sync-init.py

import os
import subprocess
from typing import Optional

JOSH_SYNC_BINARY = "/projects/personal/rust/josh-sync/target/debug/rustc-josh-sync"
FORK_USER = "kobzol"

MAINREPO_DIR = "/projects/personal/rust/rust"
MAINREPO_URL = "https://github.com/rust-lang/rust"
MAINREPO_FORK = f"{FORK_USER}/rust"
MAINREPO_FORK_URL = f"https://github.com/{MAINREPO_FORK}"
MAINREPO_SYNC_BRANCH = "clif-push"
MAINREPO_LOCAL_DEFAULT_BRANCH = f"{FORK_USER}-main"
MAINREPO_UNRELATED_FILE_TO_CHANGE = "README.md"

SUBREPO = "rustc_codegen_cranelift"
SUBREPO_URL = f"https://github.com/rust-lang/{SUBREPO}"
SUBREPO_PATH = f"compiler/{SUBREPO}"
SUBREPO_DIR = "/projects/personal/rust/rustc_codegen_cranelift"
SUBREPO_FORK_URL = f"https://github.com/{FORK_USER}/{SUBREPO}"
SUBREPO_DEFAULT_BRANCH = "main"
SUBREPO_LOCAL_DEFAULT_BRANCH = f"{FORK_USER}-{SUBREPO_DEFAULT_BRANCH}"
SUBREPO_SYNC_BRANCH = "clif-sync"
SUBREPO_FILE_TO_CHANGE = "triagebot.toml"
SUBREPO_FILE_TO_CHANGE2 = ".gitignore"

FILTER = f"""
repo = "{SUBREPO}"
filter = ":~(history=\\"keep-trivial-merges,no-splice\\")[:rev(<=bb351cd4ddf59d8a9b5c086a15e41bc6a1c05b5f:prefix={SUBREPO_PATH},<=8925ea358a0f265ca61026aadc7ecc506c545cbe:SQUASH)]:/{SUBREPO_PATH}"
filter-version = 2
"""


def reset_fork_default_branches():
    input("Synchronizing fork state")
    run(["git", "fetch", MAINREPO_URL], at=MAINREPO_DIR)
    run(["git", "push", MAINREPO_FORK_URL, "FETCH_HEAD:main", "--force"], at=MAINREPO_DIR)
    run(["git", "fetch", SUBREPO_URL])
    run(["git", "push", SUBREPO_FORK_URL, f"FETCH_HEAD:main", "--force"])


def step01_pull():
    input("Step 01: first pull into subrepo")
    checkout_branch_to(
        branch=SUBREPO_SYNC_BRANCH,
        remote=SUBREPO_URL,
        remote_branch=SUBREPO_DEFAULT_BRANCH,
    )

    with open("rust-version", "w"):
        pass
    with open("josh-sync.toml", "w") as f:
        f.write(FILTER.lstrip())
    run(["git", "add", "rust-version", "josh-sync.toml"])
    run(["git", "commit", "-m", "Initial josh-sync configuration"])

    pull_and_open_pr(title="Initial josh pull")


def step02_push():
    input("Step 02: first push from subrepo")
    checkout_branch_to(
        branch=SUBREPO_LOCAL_DEFAULT_BRANCH,
        remote=SUBREPO_FORK_URL,
        remote_branch=SUBREPO_DEFAULT_BRANCH
    )

    run(["git", "push", MAINREPO_FORK_URL, f":{MAINREPO_SYNC_BRANCH}", "--force"], at=MAINREPO_DIR)
    run([JOSH_SYNC_BINARY, "push", MAINREPO_SYNC_BRANCH, FORK_USER])
    input("Push finished, press enter to submit a PR against your mainrepo fork.")
    open_pr(repo=MAINREPO_FORK, title="Initial josh push")


def step03_make_changes_mainrepo():
    input("Step 03: make changes in mainrepo")

    def checkout():
        checkout_branch_to(
            branch=MAINREPO_LOCAL_DEFAULT_BRANCH,
            remote=MAINREPO_FORK_URL,
            at=MAINREPO_DIR
        )

    checkout()
    # Make an unrelated mainrepo change
    append_to_file(MAINREPO_UNRELATED_FILE_TO_CHANGE, "unrelated change in mainrepo",
                   msg="[rustc] change mainrepo file",
                   at=MAINREPO_DIR)
    open_pr(MAINREPO_FORK_URL, "Unrelated change", at=MAINREPO_DIR)
    checkout()

    # Make a related subrepo change
    append_to_file(os.path.join(SUBREPO_PATH, SUBREPO_FILE_TO_CHANGE), "subrepo change in mainrepo",
                   msg="[rustc] change subrepo file",
                   at=MAINREPO_DIR)
    open_pr(MAINREPO_FORK_URL, "Subrepo related change", at=MAINREPO_DIR)
    checkout()


def step04_make_changes_subrepo():
    input("Step 04: make changes in subrepo")

    checkout_branch_to(
        branch=SUBREPO_LOCAL_DEFAULT_BRANCH,
        remote=SUBREPO_FORK_URL,
        remote_branch=SUBREPO_DEFAULT_BRANCH
    )
    append_to_file(SUBREPO_FILE_TO_CHANGE2, content="subrepo change",
                   msg="[subrepo] change subrepo file")
    open_pr(SUBREPO_FORK_URL, title="Subrepo change")


def step05_pull_changes():
    input("Step 05: second pull in subrepo")

    checkout_branch_to(
        branch=SUBREPO_LOCAL_DEFAULT_BRANCH,
        remote=SUBREPO_FORK_URL,
        remote_branch=SUBREPO_DEFAULT_BRANCH
    )
    pull_and_open_pr("Second josh pull")


def step06_push_changes():
    input("Step 06: second push from subrepo")

    checkout_branch_to(
        branch=SUBREPO_LOCAL_DEFAULT_BRANCH,
        remote=SUBREPO_FORK_URL,
        remote_branch=SUBREPO_DEFAULT_BRANCH
    )

    run(["git", "push", MAINREPO_FORK_URL, f":{MAINREPO_SYNC_BRANCH}", "--force"], at=MAINREPO_DIR)
    run([JOSH_SYNC_BINARY, "push", MAINREPO_SYNC_BRANCH, FORK_USER])
    input("Push finished, press enter to submit a PR against your mainrepo fork.")
    open_pr(repo=MAINREPO_FORK, title="Second josh push")


def checkout_branch_to(branch: str, remote: str, remote_branch: str = "main",
                       at: Optional[str] = None):
    run(["git", "branch", "-D", branch], allow_fail=True, at=at)
    run(["git", "checkout", "-b", branch], at=at)
    run(["git", "fetch", remote, remote_branch], at=at)
    run(["git", "reset", "--hard", "FETCH_HEAD"], at=at)


def pull_and_open_pr(title: str):
    run([JOSH_SYNC_BINARY, "pull"], input="n\n")
    input("Pull finished, press enter to submit a PR against your subrepo fork.")
    run(["git", "push", SUBREPO_FORK_URL, f":{SUBREPO_SYNC_BRANCH}"], allow_fail=True)
    open_pr(SUBREPO_FORK_URL, title, at=SUBREPO_DIR)


def open_pr(repo: str, title: str, at: Optional[str] = None):
    run(["gh", "pr", "create", "--repo", repo, "--title", title, "--body", title], at=at)
    input("Opened a PR, now go merge it.")


def append_to_file(file: str, content: str, msg: str, at: Optional[str] = None):
    if at is not None:
        file = os.path.join(at, file)
    print(f"Modifying file `{file}`")
    with open(file, "a") as f:
        f.write(f"{content}\n")
    run(["git", "add", file], at=at)
    run(["git", "commit", "-m", msg], at=at)


def run(args: list[str], at: Optional[str] = None, input: Optional[str] = None, allow_fail=False):
    if at is None:
        at = SUBREPO_DIR
    if input is not None:
        input = input.encode("utf8")
    print(f"Running `{" ".join(args)}` at `{at}`")
    subprocess.run(args, cwd=at, input=input, check=not allow_fail)


if __name__ == "__main__":
    reset_fork_default_branches()
    step01_pull()
    step02_push()
    step03_make_changes_mainrepo()
    step04_make_changes_subrepo()
    step05_pull_changes()
    step06_push_changes()
    print("Finished. If everything passed, you should be good to go. Now migrate to josh!")