# Copyright 2021 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.
"""Assorted checks that do not require checking out code."""

import functools
import re

from PB.recipes.pigweed.static_checks import InputProperties
from recipe_engine import post_process

DEPS = [
    'fuchsia/gerrit',
    'fuchsia/status_check',
    'pigweed/checkout',
    'pigweed/util',
    'recipe_engine/buildbucket',
    'recipe_engine/cq',
    'recipe_engine/json',
    'recipe_engine/path',
    'recipe_engine/properties',
    'recipe_engine/step',
]

PROPERTIES = InputProperties

PYTHON_VERSION_COMPATIBILITY = "PY3"


def nest_step(func):
    @functools.wraps(func)
    def wrapper(api, *args, **kwargs):
        name = func.__name__.strip('_').replace('_', ' ')
        with api.step.nest(name):
            return func(api, *args, **kwargs)

    return wrapper


@nest_step
def _check_docs(api, details, comments, doc_file_extensions=()):
    """Check that docs are included in the CL.

    If doc_file_extensions is set make sure at least one document change is in
    the CL. Can be overridden by adding "No-Docs-Update-Reason:" to the commit
    message, a Gerrit comment, or a patchset description.
    """
    if not doc_file_extensions:
        with api.step.nest('doc changes not required'):
            return

    match = api.util.find_matching_comment(
        re.compile(r'No-Docs-Update-Reason: \w.*'), comments,
    )
    if match:
        with api.step.nest('found No-Docs-Update-Reason'):
            return

    with api.step.nest('checking changed files') as pres:
        current_revision = details['revisions'][details['current_revision']]
        files = tuple(current_revision['files'])
        pres.step_summary_text = '\n'.join(files)
        for path in files:
            if path.endswith(tuple(str(x) for x in doc_file_extensions)):
                with api.step.nest('found') as pres:
                    pres.step_summary_text = path
                    return

        # Only check that all files are OWNERS files if there are files. This
        # makes testing easier and should have no effect in production except
        # to fail some CLs that don't modify files.
        if files and all(api.path.basename(x) == 'OWNERS' for x in files):
            with api.step.nest('no non-OWNERS files'):
                return

    raise api.step.StepFailure(
        'No *.rst or *.md changes in CL and no exception applies. Add docs '
        "or explain why they're not needed in a Gerrit comment or in the "
        'commit message with "No-Docs-Update-Reason: <reason>".'
    )


@nest_step
def _check_requires(api, change, details, dry_run=False, allow_requires=True):
    """Check the format of "Requires:" lines and warn for common mistakes."""
    warnings = []

    requires_rx = re.compile(r'^\s*(Require(?:s|d|):).*', re.IGNORECASE)
    paragraphs = details['message'].split('\n\n')

    def _warn(warning):
        with api.step.nest('warning') as pres:
            pres.step_summary_text = warning
        warnings.append(warning)

    for paragraph in paragraphs[0:-1]:
        for line in paragraph.split('\n'):
            match = requires_rx.match(line)
            if match:
                _warn(
                    'Found "Requires:"-like line outside final '
                    'paragraph: {!r}. To pull in another change when '
                    'testing this change in CQ, move the "Requires:" '
                    'line to be adjacent to the "Change-Id:" line. '
                    'Example:\n\nRequires: pigweed:1234\nChange-Id: '
                    'I0123456789012345678901234567890123456789\n'.format(
                        match.group(0)
                    )
                )

    for line in paragraphs[-1].split('\n'):
        match = requires_rx.match(line)
        if match:
            if match.group(1) != 'Requires:':
                _warn(
                    'Found invalid "Requires:"-like line: {!r}. The line '
                    'should start with "Requires:" or be removed from the '
                    'final paragraph.'.format(match.group(0))
                )

        if not allow_requires and line.startswith('Requires:'):
            _warn(
                'Found "Requires:" line {!r} but they are not allowed for '
                'this repository.'.format(line)
            )

    if warnings and not dry_run:
        api.gerrit.set_review(
            'set review',
            str(change.change),
            host=change.host,
            message='\n\n'.join(warnings),
            revision=details['current_revision'],
            notify='OWNER',
        )


@nest_step
def _check_regexp(api, change, details, regexp, failure_message):
    if re.match(regexp, details['message']):
        with api.step.nest('matches') as pres:
            pres.step_summary_text = regexp
            return

    with api.step.nest('does not match') as pres:
        pres.step_summary_text = '{!r} does not match {!r}'.format(
            details['message'], regexp,
        )

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

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


@nest_step
def _check_dns(api, details):
    """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),
    ]

    for regex in regexes:
        match = regex.search(details['message'])
        if match:
            raise api.step.StepFailure(
                'found "{}" in commit message'.format(match.group(0))
            )


@nest_step
def _check_tested(api, details):
    """Check for a "Tested:" line in the commit message."""

    for line in details['message'].split('\n'):
        if line.lower().startswith('tested:'):
            with api.step.nest('found tested line in commit message') as pres:
                pres.step_summary_text = line
                return

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


def _check_readability(api, details, comments, readability):
    """Check that the readability label has been approved.

    Check that the readability label has been approved if files matching
    readability.file_extensions are in the CL.
    """

    with api.step.nest(readability.label):
        if api.cq.active and api.cq.run_mode != 'FULL_RUN':
            with api.step.nest('dry run'):
                return

        if readability.label not in details['labels']:
            with api.step.nest('no such label'):
                return

        # "approved" would mean +2 if values are [-2,+2].
        if 'approved' in details['labels'][readability.label]:
            with api.step.nest('label approved'):
                return

        # "recommended" would mean +1 if values are [-2,+2].
        if readability.allow_recommended:
            if 'recommended' in details['labels'][readability.label]:
                with api.step.nest('label recommended'):
                    return

        skip_line = 'Skip-{}-Justification'.format(readability.label)

        match = api.util.find_matching_comment(
            re.compile(r'{}: \w+'.format(skip_line)), comments,
        )
        if match:
            with api.step.nest('found {}'.format(skip_line)):
                return

        current_revision = details['revisions'][details['current_revision']]
        files = sorted(current_revision['files'])
        with api.step.nest('files') as pres:
            pres.step_summary_text = '\n'.join(files)
        for path in files:
            if path.endswith(tuple(readability.file_extensions)):
                raise api.step.StepFailure(
                    '{} not set but {} changed. If necessary this requirement '
                    'can be bypassed by adding a Gerrit comment like '
                    '"{}: <reason>"'.format(readability.label, path, skip_line,)
                )

        with api.step.nest('no matching files'):
            pass


def RunSteps(api, props):
    res = api.util.get_change_with_comments()
    change = res.change
    details = res.details
    comments = res.comments

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

    with api.step.nest('checking if revert') as pres:
        pres.step_summary_text = str(details['revert_of'])
        if details['revert_of']:
            with api.step.nest('ignored'):
                return

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

    _check_docs(api, details, comments, props.doc_extensions)

    _check_requires(
        api,
        change,
        details,
        dry_run=props.dry_run,
        allow_requires=not props.forbid_requires,
    )

    if props.require_tested:
        _check_tested(api, details)

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

    _check_dns(api, details)

    for readability in props.readability:
        _check_readability(api, details, comments, readability)


def change_details(
    api,
    msg,
    description='',
    files=(),
    owner='nobody@example.com',
    revert_of=0,
    votes=None,
):
    labels = {}
    if votes:
        for k, v in votes.items():
            labels[k] = {}
            if v is not None:
                assert v in "approved recommended disliked rejected".split()
                labels[k][v] = True

    return api.step_data(
        'change details',
        api.json.output(
            {
                'current_revision': 'HASH',
                'revisions': {
                    'HASH': {
                        '_number': 1,
                        'commit': {
                            'subject': msg.split('\n')[0],
                            'message': msg,
                            'parents': [{'commit': 'PARENT',}],
                        },
                        'files': {x: {} for x in files},
                        'description': description,
                    }
                },
                'owner': {'email': owner},
                'revert_of': revert_of,
                'labels': labels,
            },
        ),
    )


def must_run(api, step):
    return api.post_process(post_process.MustRun, step)


def requires_tests(api):
    def check_num_warnings(n):
        if not n:
            return api.post_process(
                post_process.DoesNotRun, 'check requires.warning'
            )

        else:
            # Make sure the first warning is found.
            result = must_run(api, 'check requires.warning')

            # Make sure the nth warning is also found.
            if n > 1:
                result += must_run(api, 'check requires.warning ({})'.format(n))

            # Make sure the nth warning is the last one.
            result += api.post_process(
                post_process.DoesNotRun,
                'check requires.warning ({})'.format(n + 1),
            )

            return result

    def requires_test(name, msg, warnings=0, status='success', **kwargs):
        res = api.status_check.test('requires.{}'.format(name), status=status)
        res += api.properties(**kwargs)
        res += api.checkout.try_test_data()
        res += change_details(api, msg)
        res += check_num_warnings(warnings)
        return res

    yield requires_test('no-requires', '')

    yield requires_test('valid', 'Foo\n\nRequires: foo:123')

    yield requires_test(
        'forbid', 'Foo\n\nRequires: foo:123', forbid_requires=True, warnings=1
    )

    yield requires_test(
        'first-paragraph', 'Requires: foo:123\n\nChange-Id: I1', warnings=1,
    )
    yield requires_test(
        'middle-paragraph',
        'Foo\n\nRequires: foo:123\n\nChange-Id: I1',
        warnings=1,
    )

    yield requires_test('require', 'Foo\n\nRequire: foo:123', warnings=1)
    yield requires_test('required', 'Foo\n\nRequired: foo:123', warnings=1)
    yield requires_test('case', 'Foo\n\nrEqUiReS: foo:123', warnings=1)

    yield requires_test(
        'multiple-warnings',
        'Foo\n\nRequires: foo:123\n\nrEqUiReS: foo:123',
        warnings=2,
    )


def tested_tests(api):
    def test(name, msg, status='success', require_tested=True, **kwargs):
        res = api.status_check.test('tested.{}'.format(name), status=status)
        res += api.properties(require_tested=require_tested, **kwargs)
        res += api.checkout.try_test_data()
        res += change_details(api, msg)
        return res

    def tested_passes():
        return must_run(api, 'check tested.found tested line in commit message')

    def owner_ignored():
        return must_run(api, 'checking owner account.ignored')

    yield (test('tested-pass', 'Tested: maybe',) + tested_passes())

    yield test('tested-fail', 'Test: nope', status='failure')

    yield (
        test(
            'tested-skip',
            'Test: nope',
            ignored_accounts=['nobody@example.com'],
        )
        + owner_ignored()
    )


def docs_tests(api):
    def test(name, msg='', status='success', properties=None, **kwargs):
        final_props = {'doc_extensions': ['.rst', '.md']}
        if properties:
            final_props.update(properties)
        return (
            api.status_check.test('docs.{}'.format(name), status=status)
            + api.properties(**final_props)
            + change_details(api, msg=msg, **kwargs)
        )

    def comment(with_exception):
        message = 'Unrelated comment'
        if with_exception:
            message = 'No-Docs-Update-Reason: laziness'

        return api.util.change_comment(message, prefix='')

    yield (
        test('no_doc_changes', status='failure')
        + comment(with_exception=False)
        + api.checkout.try_test_data()
    )

    yield (
        test('has_doc_changes', files=('foo.rst',))
        + api.checkout.try_test_data()
        + must_run(api, 'check docs.checking changed files.found')
    )

    yield (
        test(
            'no_doc_changes_properties',
            status='failure',
            files=('foo.rst',),
            properties={'doc_extensions': ['.gn']},
        )
        + api.checkout.try_test_data()
    )

    yield (
        test(
            'has_doc_changes_properties',
            files=('foo.gn',),
            properties={'doc_extensions': ['.gn']},
        )
        + api.checkout.try_test_data()
        + must_run(api, 'check docs.checking changed files.found')
    )

    yield (
        test('commit-message', msg='No-Docs-Update-Reason: foo')
        + api.checkout.try_test_data()
        + must_run(api, 'check docs.checking comments.found')
    )

    yield (
        test('owners-files', files=('foo/OWNERS', 'OWNERS'))
        + api.checkout.try_test_data()
        + must_run(api, 'check docs.checking changed files.no non-OWNERS files')
    )

    yield (
        test('gerrit-comment')
        + api.checkout.try_test_data()
        + comment(with_exception=True)
        + must_run(api, 'check docs.checking comments.found')
    )

    yield (
        test(
            'owner',
            owner='roller@example.com',
            properties={'ignored_accounts': ['roller@example.com']},
        )
        + api.checkout.try_test_data()
        + must_run(api, 'checking owner account.ignored')
    )

    yield (
        test('revert', revert_of=123456)
        + api.checkout.try_test_data()
        + must_run(api, 'checking if revert.ignored')
    )


def readability_tests(api):
    def test(name, msg='', status='success', **kwargs):
        final_props = {
            'readability': [
                {
                    'label': 'Readability',
                    'file_extensions': (
                        '.c .cc .cpp .cxx .C .h .hh .hpp .hxx .H'.split()
                    ),
                    'allow_recommended': True,
                },
            ],
        }

        return (
            api.status_check.test('readability.{}'.format(name), status=status)
            + api.properties(**final_props)
            + change_details(api, msg=msg, **kwargs)
            + api.cq(api.cq.FULL_RUN)
        )

    def comment(message='unrelated'):
        return api.util.change_comment(message, prefix='')

    yield (
        test('no_cpp_changes', votes={'Readability': None})
        + api.checkout.try_test_data()
        + must_run(api, 'Readability.no matching files')
    )

    yield (
        test(
            'has_cpp_changes_no_vote',
            files=('foo.cpp',),
            votes={'Readability': None},
            status='failure',
        )
        + api.checkout.try_test_data()
        + comment()
    )

    yield (
        test(
            'has_cpp_changes_approved',
            files=('foo.cpp',),
            votes={'Readability': 'approved'},
        )
        + api.checkout.try_test_data()
        + must_run(api, 'Readability.label approved')
    )

    yield (
        test(
            'has_cpp_changes_recommended',
            files=('foo.cpp',),
            votes={'Readability': 'recommended'},
        )
        + api.checkout.try_test_data()
        + must_run(api, 'Readability.label recommended')
    )

    yield (
        test(
            'has_cpp_changes_justification',
            files=('foo.cpp',),
            votes={'Readability': None},
        )
        + api.checkout.try_test_data()
        + comment('Skip-Readability-Justification: because')
        + must_run(api, 'Readability.found Skip-Readability-Justification')
    )

    yield (
        test('has_cpp_changes_no_label', files=('foo.cpp',))
        + api.checkout.try_test_data()
        + must_run(api, 'Readability.no such label')
    )

    yield (
        test('dry_run', files=('foo.cpp',))
        + api.cq(api.cq.DRY_RUN)
        + api.checkout.try_test_data()
        + must_run(api, 'Readability.dry run')
    )


def regexp_tests(api):
    def test(name, msg, regexp, failure_msg='', status='success', **kwargs):
        res = api.status_check.test('regexp.{}'.format(name), status=status)
        res += api.properties(
            commit_message_regexp=regexp,
            commit_message_regexp_failure_message=failure_msg,
        )
        res += api.checkout.try_test_data()
        res += change_details(api, msg)
        return res

    yield (
        test(
            'pass',
            'Bug: pwbug/99999999999999999999999999999999999999999999999999999',
            regexp='.*Bug: pwbug/\d+.*',
        )
        + must_run(api, 'check regexp.matches')
    )

    yield (
        test('fail', 'message', regexp='.*Bug: pwbug/\d+.*', status='failure',)
        + must_run(api, 'check regexp.does not match')
    )

    yield (
        test(
            'fail-msg',
            'message',
            regexp='.*Bug: pwbug/\d+.*',
            failure_msg='Did not reference a bug.',
            status='failure',
        )
        + must_run(api, 'check regexp.does not match')
        + must_run(api, 'check regexp.more details')
    )


def GenTests(api):
    for test in requires_tests(api):
        yield test

    for test in tested_tests(api):
        yield test

    for test in docs_tests(api):
        yield test

    for test in readability_tests(api):
        yield test

    for test in regexp_tests(api):
        yield test

    yield (
        api.status_check.test('do-not-submit', status='failure')
        + api.buildbucket.try_build()
        + change_details(api, 'Do_NoT-sUbMiT')
    )