recipe_modules/cq_deps/test_api.py

# Copyright 2019 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.
"""Test API for cq_deps."""

from recipe_engine import post_process, recipe_test_api


class CqDepsTestApi(recipe_test_api.RecipeTestApi):
    def _step_name(self, prefix, name, n):
        return '{}resolve CL deps.details {}{}'.format(
            prefix, name, ' ({})'.format(n) if n else '',
        )

    def details(
        self,
        name,
        message='',
        status='NEW',
        project='project',
        patchset=1,
        prefix='',
        retcode=0,
        n=None,
    ):
        return self.override_step_data(
            self._step_name(prefix, name, n),
            self.m.json.output(
                {
                    '_number': int(name.split(':')[1]),
                    'current_revision': 'HASH',
                    'project': project,
                    'revisions': {
                        'HASH': {
                            '_number': patchset,
                            'commit': {'message': message},
                        }
                    },
                    'status': status,
                }
            ),
        )

    def transient(self, name, prefix='', n=None):
        return self.override_step_data(
            self._step_name(prefix, name, n),
            self.m.json.output({'transient': True}),
            retcode=1,
        )

    def forbidden(self, name, prefix='', n=None):
        return self.override_step_data(
            self._step_name(prefix, name, n), self.m.json.output({}), retcode=1,
        )

    def has_deps(self, *names):
        results = []
        for name in names:
            results.append(
                self.post_process(
                    post_process.MustRun, 'final deps.{}'.format(name),
                )
            )

        return sum(results[1:], results[0])

    def lacks_deps(self, *names):
        results = []
        for name in names:
            results.append(
                self.post_process(
                    post_process.DoesNotRun, 'final deps.{}'.format(name),
                )
            )

        return sum(results[1:], results[0])