# Copyright 2026 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.
"""Performs static checks on Gerrit changes without needing a full checkout.

This recipe module is designed to be run in a CI/CQ environment to validate
Gerrit changes against a set of configurable rules. These checks primarily focus
on the metadata of a change, such as the commit message and file modifications,
rather than the code's functional correctness.

Key checks performed by this module include:
  - Commit message regex: Validates the commit message against a custom regular
    expression.
  - "Do Not Submit" (DNS) / "Work In Progress" (WIP): Checks for markers like
    "DO NOT SUBMIT" or "WIP" in the commit message and can optionally mark the
    change as WIP on Gerrit.
  - "Tested:" line: Verifies the presence of a "Tested:" line in the commit
    message.
  - Readability labels: Automatically applies a "trivial" readability label
    (e.g., "Readability-Trivial") if the changes to specific file types are
    below a certain line count threshold.

The behavior of these checks is controlled by the `InputProperties` defined in
the `recipes.pigweed.static_checks.InputProperties` protocol buffer message.
The module interacts with Gerrit to post review comments, set labels, or
modify the change status (e.g., marking as WIP) based on the check results.
It also handles common scenarios like ignoring checks for revert changes,
merge commits, or changes made by allowlisted accounts.
"""

from __future__ import annotations

import functools
import re
from typing import Any

from recipe_engine import recipe_api

from PB.recipes.pigweed.static_checks import InputProperties
from PB.recipes.pigweed.static_checks import Readability as ReadabilityPb

from RECIPE_MODULES.pigweed.util import api as util_api


def nest_step(func: Any) -> Any:
    @functools.wraps(func)
    def wrapper(self: StaticChecksApi, *args: Any, **kwargs: Any) -> Any:
        name = func.__name__.strip('_').replace('_', ' ')
        with self.m.step.nest(name):
            return func(self, *args, **kwargs)

    return wrapper


class StaticChecksApi(recipe_api.RecipeApi):
    """API for performing static checks on Gerrit changes."""

    InputProperties = InputProperties
    Readability = ReadabilityPb

    @nest_step
    def _check_regexp(
        self,
        change: util_api.ChangeWithComments,
        details: dict[str, Any],
        regexp: re.Pattern | str,
        failure_message: str,
    ) -> None:
        _ = change  # Unused.

        if re.match(regexp, details['message']):
            with self.m.step.nest('matches') as pres:
                pres.step_summary_text = str(regexp)
                return

        with self.m.step.nest('does not match') as pres:
            pres.step_summary_text = (
                f'{details["message"]!r} does not match {regexp!r}'
            )

        if failure_message:
            with self.m.step.nest('more details') as details_pres:
                details_pres.step_summary_text = failure_message
                details_pres.status = 'FAILURE'

        raise self.m.step.StepFailure('commit message regexp match failed')

    @nest_step
    def _check_dns(
        self,
        details: dict[str, Any],
        *,
        dry_run: bool,
    ) -> None:
        """Check that the commit message doesn't contain "do not submit"."""
        regexes = [
            re.compile(r'do.not.(submit|com+it|merge)', re.IGNORECASE),
            re.compile(r'\bWIP\b', re.IGNORECASE),
            re.compile(r'\bwork.in.progres+\b', re.IGNORECASE),
        ]

        match: re.Match | None = None
        for regex in regexes:
            if match := regex.search(details['message']):
                break
        else:
            return

        # We only get here if the current change has matched "work in progress",
        # "do not commit", or similar.

        if self.m.cv.active and self.m.cv.run_mode == self.m.cv.FULL_RUN:
            raise self.m.step.StepFailure(
                f'found "{match.group(0)}" in commit message'
            )

        if not details.get('work_in_progress', False):
            if dry_run:  # pragma: no cover
                self.m.step.empty('would mark change as wip, but in dry run')
            else:
                try:
                    self.m.gerrit.wip('wip', details['_number'])
                except self.m.step.StepFailure:  # pragma: no cover
                    pass

    @nest_step
    def _check_tested(self, details: dict[str, Any]) -> None:
        """Check for a "Tested:" line in the commit message."""
        for line in details['message'].split('\n'):
            if line.lower().startswith('tested:'):
                with self.m.step.nest(
                    'found tested line in commit message'
                ) as pres:
                    pres.step_summary_text = line
                    return

        raise self.m.step.StepFailure(
            'could not find "Tested:" line in commit message'
        )

    def _check_readability(
        self,
        details: dict[str, Any],
        readability: ReadabilityPb,
    ) -> None:
        """Set the trivial readability label if the CL is trivial."""
        extensions = tuple(str(x) for x in readability.file_extensions)

        with self.m.step.nest(readability.trivial_label):
            if readability.trivial_label not in details['labels']:
                with self.m.step.nest('trivial label not present for repo'):
                    return

            current_revision = details['revisions'][details['current_revision']]
            changed_lines = 0
            changed_files = 0
            for filename, entry in current_revision['files'].items():
                # TODO: b/240290899 - Maybe don't count deleted files?
                if filename.endswith(extensions):
                    changed_files += 1
                    changed_lines += entry.get('lines_deleted', 0)
                    changed_lines += entry.get('lines_inserted', 0)

            with self.m.step.nest(f'changed files {changed_files}'):
                pass

            with self.m.step.nest(f'changed lines {changed_lines}'):
                pass

            with self.m.step.nest(
                f'threshold {readability.line_number_threshold}'
            ):
                pass

            if (
                changed_files
                and changed_lines < readability.line_number_threshold
            ):
                self.m.gerrit.set_review(
                    'set label',
                    str(details['_number']),
                    labels={readability.trivial_label: 1},
                    notify='OWNER',
                )

    def __call__(self, props: InputProperties) -> None:
        res = self.m.util.get_change_with_comments()
        change = res.change
        details = res.details

        self.m.cv.set_do_not_retry_build()

        # Make it just a little easier to retrieve current commit message later.
        current_revision = details['revisions'][details['current_revision']]
        details['message'] = current_revision['commit']['message']

        with self.m.step.nest('checking if revert') as pres:
            pres.step_summary_text = str(details['revert_of'])
            if details['revert_of']:
                with self.m.step.nest('ignored'):  # pragma: no cover
                    return

        with self.m.step.nest('checking if merge') as pres:
            parents = [
                x['commit'] for x in current_revision['commit']['parents']
            ]
            pres.step_summary_text = str(parents)
            if len(parents) > 1:
                with self.m.step.nest('ignored'):  # pragma: no cover
                    return

        with self.m.step.nest('checking owner account') as pres:
            owner = details['owner']['email']
            pres.step_summary_text = owner
            if owner in props.ignored_accounts:
                with self.m.step.nest('ignored'):
                    return

        if props.require_tested:
            self._check_tested(details)

        if props.commit_message_regexp:
            self._check_regexp(
                change,
                details,
                str(props.commit_message_regexp),
                str(props.commit_message_regexp_failure_message),
            )

        self._check_dns(details, dry_run=props.dry_run)

        for readability in props.readability:
            self._check_readability(details, readability)