xref: /external/pigweed/pw_build/py/pw_build/project_builder_presubmit_runner.py

# Copyright 2023 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.
"""pw_build.project_builder_presubmit_runner"""

from __future__ import annotations

import argparse
import fnmatch
import logging
from pathlib import Path


import pw_cli.log
from pw_cli.arguments import (
    print_completions_for_option,
    add_tab_complete_arguments,
)
from pw_presubmit.presubmit import (
    Program,
    Programs,
    Presubmit,
    PresubmitContext,
    Check,
    fetch_file_lists,
)
import pw_presubmit.pigweed_presubmit
from pw_presubmit.build import GnGenNinja, gn_args
from pw_presubmit.presubmit_context import get_check_traces, PresubmitCheckTrace
from pw_presubmit.tools import file_summary

# pw_watch is not required by pw_build, this is an optional feature.
try:
    from pw_watch.argparser import (  # type: ignore
        add_parser_arguments as add_watch_arguments,
    )
    from pw_watch.watch import run_watch, watch_setup  # type: ignore
    from pw_watch.watch_app import WatchAppPrefs  # type: ignore

    PW_WATCH_AVAILABLE = True
except ImportError:
    PW_WATCH_AVAILABLE = False

from pw_build.project_builder import (
    ProjectBuilder,
    run_builds,
    ASCII_CHARSET,
    EMOJI_CHARSET,
)
from pw_build.build_recipe import (
    BuildCommand,
    BuildRecipe,
    UnknownBuildSystem,
    create_build_recipes,
    should_gn_gen,
)
from pw_build.project_builder_argparse import add_project_builder_arguments
from pw_build.project_builder_prefs import ProjectBuilderPrefs


_COLOR = pw_cli.color.colors()
_LOG = logging.getLogger('pw_build')


class PresubmitTraceAnnotationError(Exception):
    """Exception for malformed PresubmitCheckTrace annotations."""


def _pw_package_install_command(package_name: str) -> BuildCommand:
    return BuildCommand(
        command=[
            'pw',
            '--no-banner',
            'package',
            'install',
            package_name,
        ],
    )


def _pw_package_install_to_build_command(
    trace: PresubmitCheckTrace,
) -> BuildCommand:
    """Returns a BuildCommand from a PresubmitCheckTrace."""
    package_name = trace.call_annotation.get('pw_package_install', None)
    if package_name is None:
        raise PresubmitTraceAnnotationError(
            'Missing "pw_package_install" value.'
        )

    return _pw_package_install_command(package_name)


def _bazel_command_args_to_build_commands(
    trace: PresubmitCheckTrace,
) -> list[BuildCommand]:
    """Returns a list of BuildCommands based on a bazel PresubmitCheckTrace."""
    build_steps: list[BuildCommand] = []

    if not 'bazel' in trace.args:
        return build_steps

    bazel_command = list(arg for arg in trace.args if not arg.startswith('--'))
    bazel_options = list(
        arg for arg in trace.args if arg.startswith('--') and arg != '--'
    )
    # Check for `bazel build` or `bazel test`
    if not (
        bazel_command[0].endswith('bazel')
        and bazel_command[1] in ['build', 'test']
    ):
        raise UnknownBuildSystem(
            f'Unable to parse bazel command:\n  {trace.args}'
        )

    bazel_subcommand = bazel_command[1]
    bazel_targets = bazel_command[2:]
    if bazel_subcommand == 'build':
        build_steps.append(
            BuildCommand(
                build_system_command='bazel',
                build_system_extra_args=['build'] + bazel_options,
                targets=bazel_targets,
            )
        )
    if bazel_subcommand == 'test':
        build_steps.append(
            BuildCommand(
                build_system_command='bazel',
                build_system_extra_args=['test'] + bazel_options,
                targets=bazel_targets,
            )
        )
    return build_steps


def _presubmit_trace_to_build_commands(
    ctx: PresubmitContext,
    presubmit_step: Check,
) -> list[BuildCommand]:
    """Convert a presubmit step to a list of BuildCommands.

    Specifically, this handles the following types of PresubmitCheckTraces:

    - pw package installs
    - gn gen followed by ninja
    - bazel commands

    If none of the specific scenarios listed above are found the command args
    are passed along to BuildCommand as is.

    Returns:
      List of BuildCommands representing each command found in the
      presubmit_step traces.
    """
    build_steps: list[BuildCommand] = []

    presubmit_step(ctx)

    step_traces = get_check_traces(ctx)

    for trace in step_traces:
        trace_args = list(trace.args)
        # Check for ninja -t graph command and skip it
        if trace_args[0].endswith('ninja'):
            try:
                dash_t_index = trace_args.index('-t')
                graph_index = trace_args.index('graph')
                if graph_index == dash_t_index + 1:
                    # This trace has -t graph, skip it.
                    continue
            except ValueError:
                # '-t graph' was not found
                pass

        if 'pw_package_install' in trace.call_annotation:
            build_steps.append(_pw_package_install_to_build_command(trace))
            continue

        if 'bazel' in trace.args:
            build_steps.extend(_bazel_command_args_to_build_commands(trace))
            continue

        # Check for gn gen or pw-wrap-ninja
        transformed_args = []
        pw_wrap_ninja_found = False
        gn_found = False
        gn_gen_found = False

        for arg in trace.args:
            # Check for a 'gn gen' command
            if arg == 'gn':
                gn_found = True
            if arg == 'gen' and gn_found:
                gn_gen_found = True

            # Check for pw-wrap-ninja, pw build doesn't use this.
            if arg == 'pw-wrap-ninja':
                # Use ninja instead
                transformed_args.append('ninja')
                pw_wrap_ninja_found = True
                continue
            # Remove --log-actions if pw-wrap-ninja was found. This is a
            # non-standard ninja arg.
            if pw_wrap_ninja_found and arg == '--log-actions':
                continue
            transformed_args.append(str(arg))

        if gn_gen_found:
            # Run the command with run_if=should_gn_gen
            build_steps.append(
                BuildCommand(run_if=should_gn_gen, command=transformed_args)
            )
        else:
            # Run the command as is.
            build_steps.append(BuildCommand(command=transformed_args))

    return build_steps


def presubmit_build_recipe(  # pylint: disable=too-many-locals
    repo_root: Path,
    presubmit_out_dir: Path,
    package_root: Path,
    presubmit_step: Check,
    all_files: list[Path],
    modified_files: list[Path],
) -> BuildRecipe | None:
    """Construct a BuildRecipe from a pw_presubmit step."""
    out_dir = presubmit_out_dir / presubmit_step.name

    ctx = PresubmitContext(
        root=repo_root,
        repos=(repo_root,),
        output_dir=out_dir,
        failure_summary_log=out_dir / 'failure-summary.log',
        paths=tuple(modified_files),
        all_paths=tuple(all_files),
        package_root=package_root,
        luci=None,
        override_gn_args={},
        num_jobs=None,
        continue_after_build_error=True,
        _failed=False,
        format_options=pw_presubmit.presubmit.FormatOptions.load(),
        dry_run=True,
    )

    presubmit_instance = Presubmit(
        root=repo_root,
        repos=(repo_root,),
        output_directory=out_dir,
        paths=modified_files,
        all_paths=all_files,
        package_root=package_root,
        override_gn_args={},
        continue_after_build_error=True,
        rng_seed=1,
        full=False,
    )

    program = Program('', [presubmit_step])
    checks = list(presubmit_instance.apply_filters(program))
    if not checks:
        _LOG.warning('')
        _LOG.warning(
            'Step "%s" is not required for the current set of modified files.',
            presubmit_step.name,
        )
        _LOG.warning('')
        return None

    try:
        ctx.paths = tuple(checks[0].paths)
    except IndexError:
        raise PresubmitTraceAnnotationError(
            'Missing pw_presubmit.presubmit.Check for presubmit step:\n'
            + repr(presubmit_step)
        )

    if isinstance(presubmit_step, GnGenNinja):
        # GnGenNinja is directly translatable to a BuildRecipe.
        selected_gn_args = {
            name: value(ctx) if callable(value) else value
            for name, value in presubmit_step.gn_args.items()
        }

        return BuildRecipe(
            build_dir=out_dir,
            title=presubmit_step.name,
            steps=[
                _pw_package_install_command(name)
                for name in presubmit_step._packages  # pylint: disable=protected-access
            ]
            + [
                BuildCommand(
                    run_if=should_gn_gen,
                    command=[
                        'gn',
                        'gen',
                        str(out_dir),
                        gn_args(**selected_gn_args),
                    ],
                ),
                BuildCommand(
                    build_system_command='ninja',
                    targets=presubmit_step.ninja_targets,
                ),
            ],
        )

    # Unknown type of presubmit, use dry-run to capture subprocess traces.
    build_steps = _presubmit_trace_to_build_commands(ctx, presubmit_step)

    out_dir.mkdir(parents=True, exist_ok=True)

    return BuildRecipe(
        build_dir=out_dir,
        title=presubmit_step.name,
        steps=build_steps,
    )


def get_parser(
    presubmit_programs: Programs | None = None,
    build_recipes: list[BuildRecipe] | None = None,
) -> argparse.ArgumentParser:
    """Setup argparse for pw_build.project_builder and optionally pw_watch."""
    parser = argparse.ArgumentParser(
        prog='pw build',
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    if PW_WATCH_AVAILABLE:
        parser = add_watch_arguments(parser)
    else:
        parser = add_project_builder_arguments(parser)

    if build_recipes is not None:

        def build_recipe_argparse_type(arg: str) -> list[BuildRecipe]:
            """Return a list of matching presubmit steps."""
            assert build_recipes
            all_recipe_names = list(
                recipe.display_name for recipe in build_recipes
            )
            filtered_names = fnmatch.filter(all_recipe_names, arg)

            if not filtered_names:
                recipe_name_str = '\n'.join(sorted(all_recipe_names))
                raise argparse.ArgumentTypeError(
                    f'"{arg}" does not match the name of a recipe.\n\n'
                    f'Valid Recipes:\n{recipe_name_str}'
                )

            return list(
                recipe
                for recipe in build_recipes
                if recipe.display_name in filtered_names
            )

        parser.add_argument(
            '-r',
            '--recipe',
            action='extend',
            default=[],
            help=(
                'Run a build recipe. Include an asterix to match more than one '
                "name. For example: --recipe 'gn_*'"
            ),
            type=build_recipe_argparse_type,
        )

    if presubmit_programs is not None:
        # Add presubmit step arguments.
        all_steps = presubmit_programs.all_steps()

        def presubmit_step_argparse_type(arg: str) -> list[Check]:
            """Return a list of matching presubmit steps."""
            filtered_step_names = fnmatch.filter(all_steps.keys(), arg)

            if not filtered_step_names:
                all_step_names = '\n'.join(sorted(all_steps.keys()))
                raise argparse.ArgumentTypeError(
                    f'"{arg}" does not match the name of a presubmit step.\n\n'
                    f'Valid Steps:\n{all_step_names}'
                )

            return list(all_steps[name] for name in filtered_step_names)

        parser.add_argument(
            '-s',
            '--step',
            action='extend',
            default=[],
            help=(
                'Run presubmit step. Include an asterix to match more than one '
                "step name. For example: --step '*_format'"
            ),
            type=presubmit_step_argparse_type,
        )

    if build_recipes or presubmit_programs:
        parser.add_argument(
            '-l',
            '--list',
            action='store_true',
            default=False,
            help=('List all known build recipes and presubmit steps.'),
        )

    if build_recipes:
        parser.add_argument(
            '--all',
            action='store_true',
            default=False,
            help=('Run all known build recipes.'),
        )

    parser.add_argument(
        '--progress-bars',
        action=argparse.BooleanOptionalAction,
        default=True,
        help='Show progress bars in the terminal.',
    )

    parser.add_argument(
        '--log-build-steps',
        action=argparse.BooleanOptionalAction,
        help='Show ninja build step log lines in output.',
    )

    if PW_WATCH_AVAILABLE:
        parser.add_argument(
            '-w',
            '--watch',
            action='store_true',
            help='Use pw_watch to monitor changes.',
            default=False,
        )

    parser.add_argument(
        '-b',
        '--base',
        help=(
            'Git revision to diff for changed files. This is used for '
            'presubmit steps.'
        ),
    )

    parser = add_tab_complete_arguments(parser)

    parser.add_argument(
        '--tab-complete-recipe',
        nargs='?',
        help='Print tab completions for the supplied recipe name.',
    )

    parser.add_argument(
        '--tab-complete-presubmit-step',
        nargs='?',
        help='Print tab completions for the supplied presubmit name.',
    )

    return parser


def _get_prefs(
    args: argparse.Namespace,
) -> ProjectBuilderPrefs | WatchAppPrefs:
    """Load either WatchAppPrefs or ProjectBuilderPrefs.

    Applies the command line args to the correct prefs class.

    Returns:
      A WatchAppPrefs instance if pw_watch is importable, ProjectBuilderPrefs
      otherwise.
    """
    prefs: ProjectBuilderPrefs | WatchAppPrefs
    if PW_WATCH_AVAILABLE:
        prefs = WatchAppPrefs(load_argparse_arguments=add_watch_arguments)
        prefs.apply_command_line_args(args)
    else:
        prefs = ProjectBuilderPrefs(
            load_argparse_arguments=add_project_builder_arguments
        )
        prefs.apply_command_line_args(args)
    return prefs


def load_presubmit_build_recipes(
    presubmit_programs: Programs,
    presubmit_steps: list[Check],
    repo_root: Path,
    presubmit_out_dir: Path,
    package_root: Path,
    all_files: list[Path],
    modified_files: list[Path],
    default_presubmit_step_names: list[str] | None = None,
) -> list[BuildRecipe]:
    """Convert selected presubmit steps into a list of BuildRecipes."""
    # Use the default presubmit if no other steps or command line out
    # directories are provided.
    if len(presubmit_steps) == 0 and default_presubmit_step_names:
        default_steps = list(
            check
            for name, check in presubmit_programs.all_steps().items()
            if name in default_presubmit_step_names
        )
        presubmit_steps = default_steps

    presubmit_recipes: list[BuildRecipe] = []

    for step in presubmit_steps:
        build_recipe = presubmit_build_recipe(
            repo_root,
            presubmit_out_dir,
            package_root,
            step,
            all_files,
            modified_files,
        )
        if build_recipe:
            presubmit_recipes.append(build_recipe)

    return presubmit_recipes


def _tab_complete_recipe(
    build_recipes: list[BuildRecipe],
    text: str = '',
) -> None:
    for name in sorted(recipe.display_name for recipe in build_recipes):
        if name.startswith(text):
            print(name)


def _tab_complete_presubmit_step(
    presubmit_programs: Programs,
    text: str = '',
) -> None:
    for name in sorted(presubmit_programs.all_steps().keys()):
        if name.startswith(text):
            print(name)


def _list_steps_and_recipes(
    presubmit_programs: Programs | None = None,
    build_recipes: list[BuildRecipe] | None = None,
) -> None:
    if presubmit_programs:
        _LOG.info('Presubmit steps:')
        print()
        for name in sorted(presubmit_programs.all_steps().keys()):
            print(name)
        print()
    if build_recipes:
        _LOG.info('Build recipes:')
        print()
        for name in sorted(recipe.display_name for recipe in build_recipes):
            print(name)
        print()


def _print_usage_help(
    presubmit_programs: Programs | None = None,
    build_recipes: list[BuildRecipe] | None = None,
) -> None:
    """Print usage examples with known presubmits and build recipes."""

    def print_pw_build(
        option: str, arg: str | None = None, end: str = '\n'
    ) -> None:
        print(
            ' '.join(
                [
                    'pw build',
                    _COLOR.cyan(option),
                    _COLOR.yellow(arg) if arg else '',
                ]
            ),
            end=end,
        )

    if presubmit_programs:
        print(_COLOR.green('All presubmit steps:'))
        for name in sorted(presubmit_programs.all_steps().keys()):
            print_pw_build('--step', name)
    if build_recipes:
        if presubmit_programs:
            # Add a blank line separator
            print()
        print(_COLOR.green('All build recipes:'))
        for name in sorted(recipe.display_name for recipe in build_recipes):
            print_pw_build('--recipe', name)

        print()
        print(
            _COLOR.green(
                'Recipe and step names may use wildcards and be repeated:'
            )
        )
        print_pw_build('--recipe', '"default_*"', end=' ')
        print(
            _COLOR.cyan('--step'),
            _COLOR.yellow('step1'),
            _COLOR.cyan('--step'),
            _COLOR.yellow('step2'),
        )
        print()
        print(_COLOR.green('Run all build recipes:'))
        print_pw_build('--all')
        print()
        print(_COLOR.green('For more help please run:'))
        print_pw_build('--help')


def main(
    presubmit_programs: Programs | None = None,
    default_presubmit_step_names: list[str] | None = None,
    build_recipes: list[BuildRecipe] | None = None,
    default_build_recipe_names: list[str] | None = None,
    repo_root: Path | None = None,
    presubmit_out_dir: Path | None = None,
    package_root: Path | None = None,
    default_root_logfile: Path = Path('out/build.txt'),
    force_pw_watch: bool = False,
) -> int:
    """Build upstream Pigweed presubmit steps."""
    # pylint: disable=too-many-locals,too-many-branches
    parser = get_parser(presubmit_programs, build_recipes)
    args = parser.parse_args()

    if args.tab_complete_option is not None:
        print_completions_for_option(
            parser,
            text=args.tab_complete_option,
            tab_completion_format=args.tab_complete_format,
        )
        return 0

    log_level = logging.DEBUG if args.debug_logging else logging.INFO

    pw_cli.log.install(
        level=log_level,
        use_color=args.colors,
        # Hide the date from the timestamp
        time_format='%H:%M:%S',
    )

    pw_env = pw_cli.env.pigweed_environment()
    if pw_env.PW_EMOJI:
        charset = EMOJI_CHARSET
    else:
        charset = ASCII_CHARSET

    if args.tab_complete_recipe is not None:
        if build_recipes:
            _tab_complete_recipe(build_recipes, text=args.tab_complete_recipe)
        # Must exit if there are no build_recipes.
        return 0

    if args.tab_complete_presubmit_step is not None:
        if presubmit_programs:
            _tab_complete_presubmit_step(
                presubmit_programs, text=args.tab_complete_presubmit_step
            )
        # Must exit if there are no presubmit_programs.
        return 0

    # List valid steps + recipes.
    if hasattr(args, 'list') and args.list:
        _list_steps_and_recipes(presubmit_programs, build_recipes)
        return 0

    command_line_dash_c_recipes: list[BuildRecipe] = []
    # If -C out directories are provided add them to the recipes list.
    if args.build_directories:
        prefs = _get_prefs(args)
        command_line_dash_c_recipes = create_build_recipes(prefs)

    if repo_root is None:
        repo_root = pw_env.PW_PROJECT_ROOT
    if presubmit_out_dir is None:
        presubmit_out_dir = repo_root / 'out/presubmit'
    if package_root is None:
        package_root = pw_env.PW_PACKAGE_ROOT

    all_files: list[Path]
    modified_files: list[Path]

    all_files, modified_files = fetch_file_lists(
        root=repo_root,
        repo=repo_root,
        pathspecs=[],
        base=args.base,
    )

    # Log modified file summary just like pw_presubmit if using --base.
    if args.base:
        _LOG.info(
            'Running steps that apply to modified files since "%s":', args.base
        )
        _LOG.info('')
        for line in file_summary(
            mf.relative_to(repo_root) for mf in modified_files
        ):
            _LOG.info(line)
        _LOG.info('')

    selected_build_recipes: list[BuildRecipe] = []
    if build_recipes:
        if hasattr(args, 'recipe'):
            selected_build_recipes = args.recipe
        if not selected_build_recipes and default_build_recipe_names:
            selected_build_recipes = [
                recipe
                for recipe in build_recipes
                if recipe.display_name in default_build_recipe_names
            ]

    selected_presubmit_recipes: list[BuildRecipe] = []
    if presubmit_programs and hasattr(args, 'step'):
        selected_presubmit_recipes = load_presubmit_build_recipes(
            presubmit_programs,
            args.step,
            repo_root,
            presubmit_out_dir,
            package_root,
            all_files,
            modified_files,
            default_presubmit_step_names=default_presubmit_step_names,
        )

    # If no builds specifed on the command line print a useful help message:
    if (
        not selected_build_recipes
        and not command_line_dash_c_recipes
        and not selected_presubmit_recipes
        and not args.all
    ):
        _print_usage_help(presubmit_programs, build_recipes)
        return 1

    if build_recipes and args.all:
        selected_build_recipes = build_recipes

    # Run these builds in order:
    recipes_to_build = (
        # -C dirs
        command_line_dash_c_recipes
        # --step 'name'
        + selected_presubmit_recipes
        # --recipe 'name'
        + selected_build_recipes
    )

    # Always set separate build file logging.
    if not args.logfile:
        args.logfile = default_root_logfile
    if not args.separate_logfiles:
        args.separate_logfiles = True

    workers = 1
    if args.parallel:
        # If parallel is requested and parallel_workers is set to 0 run all
        # recipes in parallel. That is, use the number of recipes as the worker
        # count.
        if args.parallel_workers == 0:
            workers = len(recipes_to_build)
        else:
            workers = args.parallel_workers

    project_builder = ProjectBuilder(
        build_recipes=recipes_to_build,
        jobs=args.jobs,
        banners=args.banners,
        keep_going=args.keep_going,
        colors=args.colors,
        charset=charset,
        separate_build_file_logging=args.separate_logfiles,
        # If running builds in serial, send all sub build logs to the root log
        # window (or terminal).
        send_recipe_logs_to_root=(workers == 1),
        root_logfile=args.logfile,
        root_logger=_LOG,
        log_level=log_level,
        allow_progress_bars=args.progress_bars,
        log_build_steps=args.log_build_steps,
    )

    if project_builder.should_use_progress_bars():
        project_builder.use_stdout_proxy()

    if PW_WATCH_AVAILABLE and (
        force_pw_watch or (args.watch or args.fullscreen)
    ):
        event_handler, exclude_list = watch_setup(
            project_builder,
            parallel=args.parallel,
            parallel_workers=workers,
            fullscreen=args.fullscreen,
            logfile=args.logfile,
            separate_logfiles=args.separate_logfiles,
        )

        run_watch(
            event_handler,
            exclude_list,
            fullscreen=args.fullscreen,
        )
        return 0

    # One off build
    return run_builds(project_builder, workers)