recipes/tokendb_check.py

# 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.
"""Check that lines were not removed from the token database."""

import re

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

DEPS = [
    'fuchsia/git',
    'fuchsia/status_check',
    'pigweed/checkout',
    'pigweed/util',
    'recipe_engine/context',
    'recipe_engine/properties',
    'recipe_engine/raw_io',
    'recipe_engine/step',
]

PROPERTIES = InputProperties


def _read_tokens(api, path, revision):
    """Read token hashes from a given path at a specific revision."""
    step = api.git(
        'show {}'.format(revision),
        'show',
        '{}:{}'.format(revision, path),
        stdout=api.raw_io.output(),
    )
    step.presentation.logs['stdout'] = step.stdout
    token_lines = step.stdout.strip().split(b'\n')
    return frozenset(x.split(b',')[0] for x in token_lines if x.strip())


def RunSteps(api, props):
    if not props.tokendb_paths:
        raise api.step.StepFailure('no tokendb_paths property')

    # TODO(mohrr) Remove nesting. Apparently the gerrit module doesn't work as
    # a top-level step. Change the next line to 'if True:' to reproduce.
    with api.step.nest('gerrit'):
        res = api.util.get_change_with_comments()

    match = api.util.find_matching_comment(
        re.compile(r'Token-Database-Removal-Reason: \w.*'), res.comments,
    )
    if match:
        return

    checkout = api.checkout(props.checkout_options)
    with api.context(cwd=checkout.root):
        kwargs = {
            'stdout': api.raw_io.output_text(),
            'step_test_data': lambda: api.raw_io.test_api.stream_output_text(
                ''
            ),
        }

        step = api.git(
            'git show --numstat',
            'show',
            '--numstat',
            '--pretty=format:',
            '--',
            *props.tokendb_paths,
            **kwargs,
        )
        step.presentation.logs["stdout"] = step.stdout
        lines = step.stdout.strip().split('\n')

        # Each line of output looks like this:
        # $ADDED $REMOVED $PATH
        for line in lines:
            if not line:
                continue

            _, removed, path = line.split()
            with api.step.nest(str(path)):
                if not int(removed):
                    continue

                prev_tokens = _read_tokens(api, path, 'HEAD~1')
                curr_tokens = _read_tokens(api, path, 'HEAD')

                if not prev_tokens.issubset(curr_tokens):
                    with api.step.nest('removed tokens') as pres:
                        pres.step_summary_text = ', '.join(
                            x.decode() for x in prev_tokens - curr_tokens
                        )
                    raise api.step.StepFailure(
                        'Lines are not allowed to be removed from token '
                        "database {}. If there's a good reason to remove them "
                        'post a Gerrit comment explaining why that starts with '
                        '"Token-Database-Removal-Reason: ".'.format(path)
                    )


def GenTests(api):
    def diff(path, added, removed):
        return '{} {} {}'.format(added, removed, path)

    def tokens(path, old_contents, new_contents):
        return api.step_data(
            f'{path}.show HEAD~1', stdout=api.raw_io.output(old_contents)
        ) + api.step_data(
            f'{path}.show HEAD', stdout=api.raw_io.output(new_contents)
        )

    def test(name, paths=(), diffs=None, status='success', comment=None):
        res = api.status_check.test(name, status=status)
        res += api.properties(tokendb_paths=list(paths))
        res += api.checkout.try_test_data()
        res += api.properties(**api.checkout.git_properties())

        if diffs is not None:
            commit_summary = '\n'.join(diffs)
            res += api.step_data(
                'git show --numstat',
                api.raw_io.stream_output_text(commit_summary),
            )

        if comment:
            res += api.util.change_comment(comment)

        return res

    yield test('no-props', status='failure')

    yield test('no-change', paths=['token.db'], diffs=[])

    yield test('addition', paths=['token.db'], diffs=[diff('token.db', 1, 0)])

    yield (
        test(
            'removal',
            paths=['token.db'],
            diffs=[diff('token.db', 0, 1)],
            status='failure',
        )
        + tokens('token.db', b'1234,\n2345,\n', b'2345,\n')
    )

    yield test(
        'comment',
        paths=['token.db'],
        comment='Token-Database-Removal-Reason: because',
    )

    yield (
        test('update', paths=['token.db'], diffs=[diff('token.db', 1, 1)])
        + tokens('token.db', b'1234,old\n2345,foo\n', b'1234,new\n2345,foo\n')
    )