blob: 79ef0c59da4f93dfb2bc6cd3664b5b81414dbd1a [file] [log] [blame]
# Copyright 2020 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.
from recipe_engine import post_process, recipe_test_api
class RollUtilTestApi(recipe_test_api.RecipeTestApi):
def properties(
self, labels_to_set=None, labels_to_wait_on=None, commit_divider=None,
):
props = {}
for label, value in (labels_to_set or {}).items():
props.setdefault('labels_to_set', [])
props['labels_to_set'].append({'label': label, 'value': value})
if labels_to_wait_on:
props['labels_to_wait_on'] = list(labels_to_wait_on)
if commit_divider:
props['commit_divider'] = commit_divider
return self.m.properties(**{'$pigweed/roll_util': props})
def commit(
self, commit_hash, message=None, author='author@example.com', name=None,
):
if not message:
message = """
module: Short commit summary
Longer commit details.
Bug: 123
Change-Id: I{commit_hash}
Reviewed-On: https://pigweed-review.googlesource.com/c/project/+/{commit_hash:.5}
""".strip().format(
commit_hash=commit_hash
)
if name is None:
name = author.split('@')[0]
return '\n'.join((commit_hash, name, author, message))
def format_prefix(self, p):
if not p:
return p
return '{}.'.format(p.rstrip('.'))
def commit_data(self, name, *commits, prefix=''):
assert isinstance(name, str)
return self.step_data(
'{}{}.git log'.format(prefix, name),
stdout=self.m.raw_io.output_text('\0'.join(commits)),
)
def cancelled(self):
return self.post_process(post_process.MustRunRE, '.*cancelling roll.*')
def not_cancelled(self):
return self.post_process(
post_process.DoesNotRunRE, '.*cancelling roll.*'
)
def noop_roll(self, prefix='', name='get roll direction'):
prefix = self.format_prefix(prefix)
return self.step_data(
'{}{}.is forward'.format(prefix, name), retcode=0
) + self.step_data('{}{}.is backward'.format(prefix, name), retcode=0)
def forward_roll(self, prefix='', name='get roll direction'):
prefix = self.format_prefix(prefix)
return self.step_data(
'{}{}.is forward'.format(prefix, name), retcode=0
) + self.step_data('{}{}.is backward'.format(prefix, name), retcode=1)
def backward_roll(self, prefix='', name='get roll direction'):
prefix = self.format_prefix(prefix)
return self.step_data(
'{}{}.is forward'.format(prefix, name), retcode=1
) + self.step_data('{}{}.is backward'.format(prefix, name), retcode=0)
def rebased_roll(self, prefix='', name='get roll direction'):
prefix = self.format_prefix(prefix)
return self.step_data(
'{}{}.is forward'.format(prefix, name), retcode=1
) + self.step_data('{}{}.is backward'.format(prefix, name), retcode=1)