shac.star

# 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.

def _get_affected_python_files(ctx):
    files = [f for f in ctx.scm.affected_files() if f.endswith(".py")]
    exclusions = [
        # recipes.py is vendored from the recipe engine and should be ignored.
        "recipes.py",
        # cleanup_deps.py and renumber_proto_fields.py are vendored from the
        # upstream Fuchsia recipes repo as-is and should not be modified.
        "scripts/cleanup_deps.py",
        "scripts/renumber_proto_fields.py",
    ]
    return [f for f in files if f not in exclusions]

def ruff_format(ctx):
    tools_dir = _install_tools(ctx)

    py_files = _get_affected_python_files(ctx)
    if not py_files:
        return

    original_contents = {}

    # First run `ruff check --fix --select=I` to sort imports, which `ruff
    # format` doesn't support.
    isort_procs = {}
    for filepath in py_files:
        original = str(ctx.io.read_file(filepath))
        original_contents[filepath] = original
        isort_procs[filepath] = ctx.os.exec(
            [
                tools_dir + "/ruff",
                "check",
                "--fix",
                # Only check import sorting.
                "--select=I",
                "--stdin-filename",
                filepath,
                "-",
            ],
            stdin = original,
        )

    # Then run `ruff format`.
    format_procs = {}
    for filepath, proc in isort_procs.items():
        res = proc.wait()
        format_procs[filepath] = ctx.os.exec(
            [
                tools_dir + "/ruff",
                "format",
                "--stdin-filename",
                filepath,
                "-",
            ],
            stdin = res.stdout,
        )

    for filepath, proc in format_procs.items():
        res = proc.wait()
        original = original_contents[filepath]
        if res.stdout != original:
            ctx.emit.finding(
                filepath = filepath,
                level = "error",
                replacements = [res.stdout],
            )

def ruff_lint(ctx):
    tools_dir = _install_tools(ctx)

    py_files = _get_affected_python_files(ctx)
    if not py_files:
        return

    res = ctx.os.exec(
        [
            tools_dir + "/ruff",
            "check",
            "--output-format",
            "json",
            "--show-fixes",
        ] +
        py_files,
        ok_retcodes = [0, 1],
    ).wait()

    for finding in json.decode(res.stdout):
        filepath = finding["filename"]
        if filepath.startswith(ctx.scm.root):
            filepath = filepath[len(ctx.scm.root) + 1:]

        replacements = None
        if finding.get("fix") and len(finding["fix"].get("edits", [])) == 1:
            edit = finding["fix"]["edits"][0]
            start_loc, end_loc = edit["location"], edit["end_location"]
            replacements = [edit["content"]]
        else:
            start_loc, end_loc = finding["location"], finding["end_location"]

        ctx.emit.finding(
            filepath = filepath,
            message = "%s: %s" % (finding["code"], finding["message"]),
            level = "error",
            line = start_loc["row"],
            col = start_loc["column"],
            end_line = end_loc["row"],
            end_col = end_loc["column"],
            replacements = replacements,
        )

def proto_format(ctx):
    tools_dir = _install_tools(ctx)

    procs = []
    for p in ctx.scm.affected_files():
        if not p.endswith(".proto"):
            continue

        # TODO(olivernewman): Use a single `buf format` invocation and the
        # --diff option once shac knows how to parse diffs.
        cmd = [
            tools_dir + "/buf",
            "format",
            "--exit-code",
            p,
        ]
        procs.append((p, ctx.os.exec(cmd, ok_retcodes = [0, 100])))

    for p, proc in procs:
        res = proc.wait()
        if res.retcode == 100:
            ctx.emit.finding(
                filepath = p,
                level = "error",
                replacements = [res.stdout],
            )

def proto_field_numbering(ctx):
    """Checks that recipe property protobuf files have clean field numbers.

    Recipe property protos need not subscribe to normal protobuf maintenance
    conventions such as never deleting fields or changing field numbers, because
    they are only used to de/serialize JSON-encoded protos, never binary-encoded
    protos.

    So it's safe (and required, for code cleanliness) to keep proto fields
    monotonically increasing with no jumps.

    Args:
        ctx: A ctx instance.
    """
    for f in ctx.scm.affected_files():
        # The recipe_proto directory contains protos that may be used in other
        # places than just recipe properties, so we cannot safely renumber their
        # fields.
        if not f.endswith(".proto") or f.startswith("recipe_proto/"):
            continue
        res = ctx.os.exec(
            [
                "python3",
                "scripts/renumber_proto_fields.py",
                "--dry-run",
                f,
            ],
        ).wait()
        if res.stdout != str(ctx.io.read_file(f)):
            ctx.emit.finding(
                filepath = f,
                level = "error",
                replacements = [res.stdout],
            )

def _install_tools(ctx):
    install_dir = ctx.scm.root + "/.tools"
    ctx.os.exec(
        [ctx.scm.root + "/scripts/install-shac-tools.sh", install_dir],
        allow_network = True,
    ).wait()
    return install_dir

def check_deps(ctx):
    res = ctx.os.exec(
        [
            "python3",
            "scripts/cleanup_deps.py",
            "--check",
            "--json-output",
            "-",
        ],
        ok_retcodes = [0, 65],
    ).wait()
    if res.retcode == 65:
        for file in json.decode(res.stdout):
            # TODO(olivernewman): Parse the diff so fixes can be applied with
            # `shac fix`.
            ctx.emit.finding(
                filepath = file,
                message = "DEPS are malformatted. Run ./scripts/cleanup_deps.py to fix.",
                level = "error",
            )

shac.register_check(shac.check(ruff_format, formatter = True))
shac.register_check(shac.check(proto_format, formatter = True))
shac.register_check(shac.check(proto_field_numbering, formatter = True))

shac.register_check(ruff_lint)
shac.register_check(check_deps)