blob: ed1345e0cd1817bcd29799054a5d2915dc488133 [file] [log] [blame]
# 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 properties(self, **kwargs):
return {'$pigweed/cq_deps': kwargs}
def lookup_cl(self, commit_hash, number=None, prefix='', found=True):
if not found:
return self.override_step_data(
f'{prefix}resolve CL deps.number {commit_hash}',
self.m.json.output(None),
)
return self.override_step_data(
f'{prefix}resolve CL deps.number {commit_hash}',
self.m.json.output([{'_number': number}] if number else []),
)
def _step_name(self, prefix, name, n):
return '{}resolve CL deps.details {}{}'.format(
prefix, name, ' ({})'.format(n) if n and n > 1 else '',
)
def details(
self,
name,
message='',
status='NEW',
project='project',
patchset=1,
prefix='',
commit_hash='HASH',
parent=None,
n=None,
):
number = int(name.split(':')[1])
if not parent:
parent = f'parent-{number}'
assert status in ('NEW', 'MERGED', 'ABANDONED')
return self.override_step_data(
self._step_name(prefix, name, n),
self.m.json.output(
{
'_number': number,
'current_revision': commit_hash,
'project': project,
'revisions': {
commit_hash: {
'_number': patchset,
'commit': {
'message': message,
'parents': [{'commit': parent,}],
},
}
},
'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='', attempts=2):
data = []
for i in range(0, attempts):
data.append(
self.override_step_data(
self._step_name(prefix, name, i + 1),
self.m.json.output({}),
retcode=1,
)
)
return sum(data[1:], data[0])
def has_deps(self, *names):
results = []
for name in names:
results.append(
self.post_process(post_process.MustRun, f'final deps.{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, f'final deps.{name}',
)
)
return sum(results[1:], results[0])
def assert_disabled(self, prefix=''):
return self.post_process(
post_process.MustRun,
f'{prefix}resolve CL deps.dependency processing disabled',
)