pw_cli/py/pw_cli/requires.py

#!/usr/bin/env python3

# Copyright 2020 The Pigweed Authors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
"""Create transitive CLs for requirements on internal Gerrits.

This is only intended to be used by Googlers.

If the current CL needs to be tested alongside internal-project:1234 on an
internal project, but "internal-project" is something that can't be referenced
publicly, this automates creation of a CL on the pigweed-internal Gerrit that
references internal-project:1234 so the current commit effectively has a
requirement on internal-project:1234.

For more see http://go/pigweed-ci-cq-intro.
"""

import argparse
import json
import logging
from pathlib import Path
import re
import subprocess
import sys
import tempfile
import uuid

HELPER_GERRIT = 'pigweed-internal'
HELPER_PROJECT = 'requires-helper'
HELPER_REPO = 'sso://{}/{}'.format(HELPER_GERRIT, HELPER_PROJECT)

# Pass checks that look for "DO NOT ..." and block submission.
_DNS = ' '.join((
    'DO',
    'NOT',
    'SUBMIT',
))

# Subset of the output from pushing to Gerrit.
DEFAULT_OUTPUT = f'''
remote:
remote:   https://{HELPER_GERRIT}-review.git.corp.google.com/c/{HELPER_PROJECT}/+/123456789 {_DNS} [NEW]
remote:
'''.strip()

_LOG = logging.getLogger(__name__)


def parse_args() -> argparse.Namespace:
    """Creates an argument parser and parses arguments."""

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        'requirements',
        nargs='+',
        help='Requirements to be added ("<gerrit-name>:<cl-number>").',
    )
    parser.add_argument(
        '--no-push',
        dest='push',
        action='store_false',
        help=argparse.SUPPRESS,  # This option is only for debugging.
    )

    return parser.parse_args()


def _run_command(*args, **kwargs):
    kwargs.setdefault('capture_output', True)
    _LOG.debug('%s', args)
    _LOG.debug('%s', kwargs)
    res = subprocess.run(*args, **kwargs)
    _LOG.debug('%s', res.stdout)
    _LOG.debug('%s', res.stderr)
    res.check_returncode()
    return res


def check_status() -> bool:
    res = subprocess.run(['git', 'status'], capture_output=True)
    if res.returncode:
        _LOG.error('repository not clean, commit to suppress this warning')
        return False
    return True


def clone(requires_dir: Path) -> None:
    _LOG.info('cloning helper repository into %s', requires_dir)
    _run_command(['git', 'clone', HELPER_REPO, '.'], cwd=requires_dir)


def create_commit(requires_dir: Path, requirements) -> None:
    """Create a commit in the local tree with the given requirements."""
    change_id = str(uuid.uuid4()).replace('-', '00')
    _LOG.debug('change_id %s', change_id)

    reqs = []
    for req in requirements:
        gerrit_name, number = req.split(':', 1)
        reqs.append({'gerrit_name': gerrit_name, 'number': number})

    path = requires_dir / 'patches.json'
    _LOG.debug('path %s', path)
    with open(path, 'w') as outs:
        json.dump(reqs, outs)

    _run_command(['git', 'add', path], cwd=requires_dir)

    commit_message = [
        f'{_DNS} {change_id[0:10]}\n\n',
        '',
        f'Change-Id: I{change_id}',
    ]
    for req in requirements:
        commit_message.append(f'Requires: {req}')

    _LOG.debug('message %s', commit_message)
    _run_command(
        ['git', 'commit', '-m', '\n'.join(commit_message)],
        cwd=requires_dir,
    )

    # Not strictly necessary, only used for logging.
    _run_command(['git', 'show'], cwd=requires_dir)


def push_commit(requires_dir: Path, push=True) -> str:
    output = DEFAULT_OUTPUT
    if push:
        res = _run_command(
            ['git', 'push', HELPER_REPO, '+HEAD:refs/for/main'],
            cwd=requires_dir,
        )
        output = res.stderr.decode()

    _LOG.debug('output: %s', output)
    regex = re.compile(
        f'^\\s*remote:\\s*'
        f'https://{HELPER_GERRIT}-review.(?:git.corp.google|googlesource).com/'
        f'c/{HELPER_PROJECT}/\\+/(?P<num>\\d+)\\s+',
        re.MULTILINE,
    )
    _LOG.debug('regex %r', regex)
    match = regex.search(output)
    if not match:
        raise ValueError(f"invalid output from 'git push': {output}")
    change_num = match.group('num')
    _LOG.info('created %s change %s', HELPER_PROJECT, change_num)
    return f'{HELPER_GERRIT}:{change_num}'


def amend_existing_change(change: str) -> None:
    res = _run_command(['git', 'log', '-1', '--pretty=%B'])
    original = res.stdout.rstrip().decode()

    addition = f'Requires: {change}'
    _LOG.info('adding "%s" to current commit message', addition)
    message = '\n'.join((original, addition))
    _run_command(['git', 'commit', '--amend', '--message', message])


def run(requirements, push=True) -> int:
    """Entry point for requires."""

    if not check_status():
        return -1

    # Create directory for checking out helper repository.
    with tempfile.TemporaryDirectory() as requires_dir_str:
        requires_dir = Path(requires_dir_str)
        # Clone into helper repository.
        clone(requires_dir)
        # Make commit with requirements from command line.
        create_commit(requires_dir, requirements)
        # Push that commit and save its number.
        change = push_commit(requires_dir, push=push)
    # Add dependency on newly pushed commit on current commit.
    amend_existing_change(change)

    return 0


def main() -> int:
    return run(**vars(parse_args()))


if __name__ == '__main__':
    try:
        # If pw_cli is available, use it to initialize logs.
        from pw_cli import log

        log.install(logging.INFO)
    except ImportError:
        # If pw_cli isn't available, display log messages like a simple print.
        logging.basicConfig(format='%(message)s', level=logging.INFO)

    sys.exit(main())