| # 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 checkout.""" |
| |
| from __future__ import annotations |
| |
| import collections |
| import urllib |
| |
| from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb2 |
| from PB.recipe_modules.pigweed.checkout import options as checkout_options |
| from recipe_engine import post_process, recipe_test_api |
| |
| REPO = 'https://pigweed.googlesource.com/pigweed/pigweed' |
| MANIFEST_REPO = 'https://pigweed.googlesource.com/pigweed/manifest' |
| |
| DEFAULT_MANIFEST = """ |
| <?xml version="1.0" encoding="UTF-8"?> |
| <manifest> |
| <remote |
| name="default_remote" |
| fetch="sso://default" |
| /> |
| |
| <remote |
| name="pigweed_remote" |
| revision="main" |
| fetch=".." |
| review="https://pigweed.googlesource.com" |
| /> |
| |
| <remote |
| name="pigweed-internal_remote" |
| revision="main" |
| fetch="https://pigweed-internal.googlesource.com" |
| review="https://pigweed-internal.googlesource.com" |
| /> |
| |
| <remote |
| name="prefixed" |
| fetch="https://foo.googlesource.com/prefix" |
| review="https://foo.googlesource.com/prefix" |
| /> |
| |
| <default |
| remote="default_remote" |
| revision="main" |
| /> |
| |
| <project |
| name="default_name" |
| path="default_path" |
| /> |
| |
| <project |
| name="pigweed_name" |
| path="pigweed_path" |
| remote="pigweed_remote" |
| revision="main" |
| /> |
| |
| <project |
| name="pigweed-internal_name" |
| path="pigweed-internal_path" |
| remote="pigweed-internal_remote" |
| /> |
| |
| <project |
| name="pinned" |
| path="pinned" |
| remote="pigweed_remote" |
| revision="0123456789012345678901234567890123456789" |
| upstream="main" |
| /> |
| |
| <project |
| name="suffix" |
| path="prefix/suffix" |
| remote="prefixed" |
| /> |
| </manifest> |
| """.strip() |
| |
| |
| def _project( |
| project: str | None = None, |
| remote: str | None = None, |
| default: str | None = None, |
| ): |
| if project is not None: |
| return project |
| if remote is None: |
| assert default |
| return default |
| |
| assert remote |
| project = urllib.parse.urlparse(remote).path.lstrip('/') |
| if project.endswith('.git'): |
| project = name[0 : -len('.git')] # pragma: no cover |
| return project |
| |
| |
| def _name(name, project): |
| if name is not None: |
| return name |
| |
| name = project |
| if name.endswith('.git'): |
| name = name[0 : -len('.git')] # pragma: no cover |
| parts = name.split('/') |
| if parts[-1] == 'manifest': |
| parts.pop() |
| return parts[-1] |
| |
| |
| class CheckoutTestApi(recipe_test_api.RecipeTestApi): |
| """Test API for checkout.""" |
| |
| @property |
| def pigweed_repo(self): |
| return REPO |
| |
| @property |
| def pigweed_repo_dot_git(self): |
| return f'{REPO}.git' |
| |
| @property |
| def manifest_repo(self): |
| return MANIFEST_REPO |
| |
| def git_options( |
| self, |
| *, |
| remote=REPO, |
| branch='main', |
| use_trigger=True, |
| use_repo=False, |
| equivalent_remotes=(), |
| force_no_rebase=False, |
| match_branch=True, |
| manifest_file='', |
| root_subdirectory='', |
| initialize_submodules=True, |
| included_submodules=(), |
| excluded_submodules=(), |
| manifest_groups=(), |
| rewrites=(), |
| eligible_workspace_paths=(), |
| ): |
| props = checkout_options.Options() |
| props.remote = remote |
| props.branch = branch or 'main' |
| props.use_trigger = use_trigger |
| props.use_repo = use_repo |
| for remotes in equivalent_remotes: |
| remote_group = checkout_options.EquivalentRemotes() |
| remote_group.remotes.extend(remotes) |
| props.equivalent_remotes.extend([remote_group]) |
| props.force_no_rebase = force_no_rebase |
| props.match_branch = match_branch |
| props.manifest_file = manifest_file |
| props.root_subdirectory = root_subdirectory |
| props.initialize_submodules = initialize_submodules |
| props.included_submodules.extend(included_submodules) |
| props.excluded_submodules.extend(excluded_submodules) |
| props.manifest_groups.extend(manifest_groups) |
| props.rewrites.extend( |
| checkout_options.Rewrite(original=x, final=y) for x, y in rewrites |
| ) |
| props.eligible_workspace_paths.extend(eligible_workspace_paths) |
| return props |
| |
| def repo_options( |
| self, |
| remote=MANIFEST_REPO, |
| use_repo=True, |
| manifest_file='default.xml', |
| **kwargs, |
| ): |
| return self.git_options( |
| remote=remote, |
| use_repo=use_repo, |
| manifest_file=manifest_file, |
| **kwargs, |
| ) |
| |
| def ci_test_data( |
| self, |
| git_repo: str = REPO, |
| name: str | None = None, |
| branch: str | None = None, |
| **kwargs, |
| ): |
| project = _project(remote=git_repo) |
| if branch: |
| kwargs['git_ref'] = f'refs/heads/{branch}' |
| ret = self.m.buildbucket.ci_build(git_repo=git_repo, **kwargs) |
| if branch: |
| name = _name(name, project) |
| |
| ret += self.override_step_data( |
| f'checkout {name}.change data.process gitiles commit.number', |
| self.m.json.output( |
| [ |
| { |
| '_number': '1234', |
| 'branch': branch, |
| 'project': project or name, |
| } |
| ] |
| ), |
| ) |
| return ret |
| |
| def cl(self, host, project, change, patchset=1): |
| return common_pb2.GerritChange( |
| host=host, project=project, change=change, patchset=patchset |
| ) |
| |
| def try_test_data(self, git_repo=REPO, **kwargs): |
| return self.m.buildbucket.try_build(git_repo=git_repo, **kwargs) |
| |
| def manifest_test_data(self, name='pigweed', raw_xml=DEFAULT_MANIFEST): |
| return self.step_data( |
| f'checkout {name}.read manifest.read file', |
| self.m.file.read_text(raw_xml), |
| ) |
| |
| def cl_branch_parents( |
| self, |
| branch='main', |
| num_parents=1, |
| index=0, |
| name=None, |
| message='', |
| project=None, |
| remote=None, |
| ): |
| project = _project(project, remote, 'pigweed/pigweed') |
| name = _name(name, project) |
| |
| return self.override_step_data( |
| f'checkout {name}.change data.process gerrit changes.{index}.details', |
| self.m.json.output( |
| { |
| 'branch': branch, |
| 'current_revision': 'f' * 40, |
| 'revisions': { |
| 'f' |
| * 40: { |
| '_number': 4, |
| 'commit': { |
| 'parents': [None for _ in range(num_parents)], |
| 'message': message, |
| }, |
| }, |
| }, |
| 'project': project or name, |
| } |
| ), |
| ) |
| |
| def manifest_has_matching_branch(self, branch, name='pigweed'): |
| # The contents of the returned JSON data don't matter--they just need to |
| # evaluate to True when converted to bool. Gitiles returns an empty |
| # dictionary when the branch does not exist. |
| return self.m.git.get_remote_branch_head( |
| f'checkout {name}.manifest has branch.git ls-remote {branch}', |
| 'a' * 40, |
| ) |
| |
| def root_files(self, *files, name='pigweed'): |
| files = set(files) |
| files.add('.repo') |
| return self.step_data( |
| f'checkout {name}.ls', |
| stdout=self.m.raw_io.output_text( |
| ''.join((f'{x}\n' for x in sorted(files))) |
| ), |
| ) |
| |
| def submodule( |
| self, |
| path, |
| remote, |
| status='', |
| *, |
| hash='a' * 40, |
| initialized=True, |
| branch='main', |
| name=None, |
| describe='', |
| modified=False, |
| conflict=False, |
| url=None, |
| update=None, |
| ignore=None, |
| shallow=False, |
| fetchRecurseSubmodules=None, |
| ): |
| result = {} |
| result['path'] = path |
| result['remote'] = remote |
| result['hash'] = hash |
| result['initialized'] = initialized |
| result['branch'] = branch |
| result['name'] = name or path |
| result['describe'] = describe |
| result['modified'] = modified |
| result['conflict'] = conflict |
| result['url'] = url or remote |
| result['update'] = update |
| result['ignore'] = ignore |
| result['shallow'] = shallow |
| result['fetchRecurseSubmodules'] = fetchRecurseSubmodules |
| |
| if status == '-': |
| result['initialized'] = True |
| result['modified'] = False |
| result['conflict'] = False |
| elif status == '+': |
| result['initialized'] = True |
| result['modified'] = True |
| result['conflict'] = False |
| elif status == 'U': # pragma: no cover |
| # Including for completeness but this is not expected to be used. |
| result['initialized'] = True |
| result['modified'] = True |
| result['conflict'] = True |
| elif status == ' ': |
| result['initialized'] = False |
| result['modified'] = False |
| result['conflict'] = False |
| |
| return result |
| |
| def submodules( |
| self, |
| *submodules, |
| prefix='checkout pigweed.', |
| checkout_root='[START_DIR]/checkout', |
| ): |
| sub_data = {sub['path']: sub for sub in submodules} |
| |
| res = self.step_data( |
| f'{prefix}submodule status', |
| self.m.json.output(sub_data), |
| ) |
| |
| return res |
| |
| def all_changes_applied(self): |
| return self.some_changes_applied() + self.post_process( |
| post_process.DoesNotRunRE, '.*some changes were not applied.*' |
| ) |
| |
| def some_changes_applied(self): |
| return self.post_process( |
| post_process.DoesNotRunRE, '.*no changes were applied.*' |
| ) |
| |
| def no_changes_applied(self): |
| return self.post_process( |
| post_process.MustRunRE, '.*no changes were applied.*' |
| ) |
| |
| def change_applied(self, name): |
| return self.post_process( |
| post_process.DoesNotRunRE, f'.*failed to apply {name}.*' |
| ) + self.post_process( |
| post_process.MustRunRE, r'.*apply {}.git.*'.format(name) |
| ) |
| |
| def change_not_applied(self, name): |
| return self.post_process( |
| post_process.MustRunRE, f'.*failed to apply {name}.*' |
| ) + self.post_process( |
| post_process.DoesNotRunRE, r'.*apply {}.git.*'.format(name) |
| ) |
| |
| def included_submodule(self, name): |
| return self.post_process( |
| post_process.MustRunRE, r'.*including submodule {}'.format(name) |
| ) + self.post_process( |
| post_process.DoesNotRunRE, r'.*excluding submodule {}'.format(name) |
| ) |
| |
| def excluded_submodule(self, name): |
| return self.post_process( |
| post_process.DoesNotRunRE, r'.*including submodule {}'.format(name) |
| ) + self.post_process( |
| post_process.MustRunRE, r'.*excluding submodule {}'.format(name) |
| ) |
