fuchsia_upload.py

#!/usr/bin/env python3
"""Releases Fuchsia VIM3 firmware.

This has two parts:
  1. build and upload firmware prebuilts to CIPD
  2. publish source code to the open-source repo

We could potentially just develop in the open - there's nothing internal in this
codebase - except that we don't really have a good host at the moment for
developing GPL code publicly (http://b/213950490). So for now we just manually
flush code out on every release so that the CIPD binaries match the public
source code.

Intended for Googler use only - external users won't have the necessary tooling
or write-access to the CIPD/git hosts so won't be able to run this.
"""

import argparse
import json
import logging
import os
import shutil
import subprocess
import sys
import tempfile
import textwrap
from typing import Optional

_MY_DIR = os.path.dirname(__file__)

_LICENSE_SEPARATOR = ('\n===================================\n\n')
# Explicitly enumerate known license files so that if a new one comes in,
# the script will fail and notify us. Once the new file has been reviewed and
# approved we can add it here.
_KNOWN_LICENSE_FILES = ('bsd-2-clause.txt', 'bsd-3-clause.txt', 'eCos-2.0.txt',
                        'gpl-2.0.txt', 'ibm-pibs.txt', 'isc.txt',
                        'lgpl-2.0.txt', 'lgpl-2.1.txt')
# Files in the licenses dir that are informational only and don't need to be
# included in the license dump.
_INFORMATIONAL_LICENSE_FILES = ('Exceptions', 'README')

_BUILD_SCRIPT = os.path.join(_MY_DIR, "build_uboot_vim3_zircon.sh")

# u-boot.bin.unsigned is the combined BL2 + TPL format that goes in the eMMC
# boot0/boot1 partitions. Unclear why u-boot.bin.signed.encrypted doesn't work,
# but we don't need signatures for dev boards so we're OK with this.
_BUILD_ARTIFACT_NAME = "u-boot.bin.unsigned"
_BUILD_ARTIFACT_PATH = os.path.join(_MY_DIR, "build", _BUILD_ARTIFACT_NAME)

_REPO_NAME = "u-boot"
_UPSTREAM_BRANCH = "origin/vim3"

_CIPD_PACKAGE = "fuchsia/prebuilt/third_party/firmware/vim3"

# http://go/copybara-setup.
_COPYBARA_PATH = os.path.join("/", "google", "bin", "releases", "copybara",
                              "public", "copybara", "copybara")


def license_file_contents() -> str:
    """Creates and returns the combined license file contents.

    Raises:
        FileExistsError if an unexpected license file was found.
    """
    licenses = []
    license_dir = os.path.join(_MY_DIR, 'Licenses')

    def _add_license(name: str):
        with open(os.path.join(license_dir, name), 'r') as f:
            contents = f.read()
            # No need to duplicate licenses if they're the exact same.
            if contents not in licenses:
                licenses.append(contents)

    dir_contents = sorted(os.listdir(license_dir))
    for name in dir_contents:
        if name in _KNOWN_LICENSE_FILES:
            _add_license(name)
        elif name in _INFORMATIONAL_LICENSE_FILES:
            pass
        else:
            raise FileExistsError(f'Unknown u-boot license file: {name}')

    return _LICENSE_SEPARATOR.join(licenses)


def ensure_git_is_clean(dir, upstream_branch) -> str:
    """Checks if the given git repo is clean.

    A clean repo is one that is on the upstream branch and has no local changes.

    Args:
        dir: a path to the repo root dir.
        upstream_branch: upstream branch we should match.

    Raises:
        ValueError if the given repo is not clean.

    Returns:
        The current clean git revision.
    """

    def git(command):
        return subprocess.run(["git"] + command,
                              cwd=dir,
                              check=True,
                              capture_output=True,
                              text=True).stdout.strip()

    # Update the local repo.
    git(["fetch"])

    # Check the HEAD revision to look for committed changes.
    head_revision = git(["rev-parse", "HEAD"])
    upstream_revision = git(["rev-parse", upstream_branch])
    if head_revision != upstream_revision:
        raise ValueError(
            f"HEAD {head_revision} != upstream {upstream_revision}")

    # Check for any uncommitted file changes. This will print the name of
    # any uncommitted files, so empty string means no local changes.
    status = git(["status", "--porcelain"])
    if status:
        raise ValueError(f"Uncommitted local changes:\n{status}")

    return head_revision


def build_and_upload(revision: Optional[str], cipd_tool: str, dry_run: bool):
    """Builds and uploads the CIPD package.

    Args:
        revision: u-boot source revision, or None for ToT.
        cipd: path to the cipd executable.
        dry_run: True to create the package but not upload it.
    """
    subprocess.run([_BUILD_SCRIPT], check=True)

    # Source revision is only used in metadata for CIPD so it doesn't have
    # to be a real ref.
    if not revision:
        revision = "__local_dirty__"

    metadata_files = {
        "OWNERS":
            "tq-firmware-team@google.com",
        "manifest.json":
            json.dumps({_REPO_NAME: revision}, indent=2, sort_keys=True),
        "LICENSE":
            license_file_contents()
    }

    with tempfile.TemporaryDirectory() as temp_dir:
        for name, contents in metadata_files.items():
            with open(os.path.join(temp_dir, name), "w") as f:
                f.write(contents)

        shutil.copyfile(_BUILD_ARTIFACT_PATH,
                        os.path.join(temp_dir, _BUILD_ARTIFACT_NAME))

        upload_command = [
            cipd_tool, "create", "-name", _CIPD_PACKAGE, "-in", temp_dir,
            "-tag", f"{_REPO_NAME}:{revision}", "-install-mode", "copy"
        ]

        if dry_run:
            logging.info("Dry run only: `%s`", " ".join(upload_command))
            logging.info("Files can be examined at %s", temp_dir)
            input("Press Enter to exit")
        else:
            subprocess.run(upload_command, check=True)


def create_copybara_config(revision: Optional[str]):
    """Creates the Copybara config file contents.

    Args:
        revision: u-boot source revision to publish, or None for ToT.

    Returns:
        The Copybara config file contents.
    """
    # When dry-running with local changes, our local revision doesn't exist on
    # the host so copybara can't find it. In this case just use the tip of tree.
    if not revision:
        logging.warning("No source revision given, using 'vim3' ToT")
        revision = "vim3"

    # http://go/copybara-reference.
    contents = textwrap.dedent(f"""\
        core.workflow(
            name = "default",
            origin = git.origin(
                url = "sso://turquoise-internal/third_party/u-boot/",
                ref = "{revision}",
            ),
            destination = git.destination(
                url = "sso://third-party-mirror/u-boot/",
                push = "vim3",
            ),
            mode = "ITERATIVE",
            authoring = authoring.pass_thru(
                default = "Fuchsia firmware team <tq-firmware-team@google.com>"
            ),
        )
        """)

    return contents


def publish_source(revision: Optional[str], copybara_path: str,
                   last_rev: Optional[str], dry_run: bool,
                   push_justification: str):
    """Publishes this source to the public repo.

    Args:
        revision: u-boot source revision to publish, or None for ToT.
        copybara_path: path to the copybara executable.
        last_rev: last source revision that was pushed to destination. Use this
                  only to initialize the repo the first time so we can tell
                  Copybara where to start.
        dry_run: True to skip submitting anything.
        push_justification: BugID for push justification.

    Raises:
        subprocess.CalledProcessError if copybara fails.
    """
    config = create_copybara_config(revision)

    with tempfile.TemporaryDirectory() as temp_dir:
        config_path = os.path.join(temp_dir, "copy.bara.sky")

        with open(config_path, "w") as config_file:
            config_file.write(config)

        command = [copybara_path, config_path, "--git-push-option", "push-justification=" + push_justification]
        if last_rev:
            command += ["--last-rev", last_rev]
        if dry_run:
            command.append("--dry-run")
        logging.info("Copybara command: `%s`", ' '.join(command))

        # Let stdout/sterr through to the console since this may take a while.
        subprocess.run(command, check=True)


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)

    parser.add_argument("--cipd",
                        default="cipd",
                        help="Path to CIPD tool; looks on $PATH by default")
    parser.add_argument("--copybara",
                        default=_COPYBARA_PATH,
                        help=f"Path to copybara tool; default {_COPYBARA_PATH}")
    parser.add_argument("--dry-run",
                        action="store_true",
                        help="Don't upload anything")
    parser.add_argument(
        "--last-rev",
        help="The last source revision that was released."
        " Only use this on the first run to tell copybara where to start.")
    parser.add_argument(
        "--push-justification",
        default="b/302031093",
        help="BugID for git push justification.")

    return parser.parse_args()


def _main() -> int:
    logging.basicConfig(level=logging.INFO)
    args = _parse_args()

    try:
        revision = ensure_git_is_clean(_MY_DIR, _UPSTREAM_BRANCH)
    except ValueError as exception:
        if args.dry_run:
            logging.warning("%s", exception)
            logging.warning("Ignoring dirty local repo for dry run")
            revision = None
        else:
            raise

    publish_source(revision, args.copybara, args.last_rev, args.dry_run, args.push_justification)
    build_and_upload(revision, args.cipd, args.dry_run)

    return 0


if __name__ == '__main__':
    sys.exit(_main())