| # 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. |
| """Update an Android Repo Tool project.""" |
| |
| from __future__ import annotations |
| |
| import collections |
| import dataclasses |
| import json |
| import re |
| from typing import Sequence, TYPE_CHECKING |
| import urllib |
| import xml.etree.ElementTree |
| |
| from PB.recipe_modules.pigweed.checkout.options import ( |
| Options as CheckoutOptions, |
| ) |
| from recipe_engine import recipe_api |
| |
| if TYPE_CHECKING: # pragma: no cover |
| from recipe_engine import config_types |
| from RECIPE_MODULES.pigweed.checkout import api as checkout_api |
| from RECIPE_MODULES.pigweed.roll_util import api as roll_util_api |
| from PB.recipe_modules.pigweed.repo_roll import Project |
| |
| |
| class _TreeBuilder(xml.etree.ElementTree.TreeBuilder): |
| def comment(self, data): |
| self.start(xml.etree.ElementTree.Comment, {}) |
| self.data(data) |
| self.end(xml.etree.ElementTree.Comment) |
| |
| |
| @dataclasses.dataclass |
| class RevisionChange: |
| dir: config_types.Path |
| old: str |
| new: str |
| direction: roll_util_api.Direction |
| |
| |
| class RepoRollApi(recipe_api.RecipeApi): |
| """Update an Android Repo Tool project.""" |
| |
| def is_branch(self, revision): |
| """Return True if revision appears to be a branch name.""" |
| if re.search(r'^[0-9a-fA-F]{40}$', revision): |
| return False |
| return not revision.startswith('refs/tags/') |
| |
| # ElementTree orders attributes differently in Python 2 and 3, but if given |
| # attributes in a specific order it preserves that order. |
| def _order_attributes(self, root): |
| for el in root.iter(): |
| if len(el.attrib) > 1: |
| new_attrib = sorted(el.attrib.items()) |
| el.attrib.clear() |
| el.attrib.update(new_attrib) |
| |
| def update_project( |
| self, |
| checkout: checkout_api.CheckoutContext, |
| project: Project, |
| ) -> list[roll_util_api.Roll]: |
| new_revision = None |
| |
| # Try to get new_revision from the trigger. |
| bb_remote = None |
| commit = self.m.buildbucket.build.input.gitiles_commit |
| if commit and commit.project: |
| new_revision = commit.id |
| host = commit.host |
| bb_remote = f'https://{host}/{commit.project}' |
| |
| tree = _TreeBuilder() |
| parser = xml.etree.ElementTree.XMLParser(target=tree) |
| parser.feed( |
| self.m.file.read_text('read manifest', checkout.manifest_path) |
| ) |
| parser.close() |
| root = tree.close() |
| |
| defaults = {} |
| for default in root.findall('default'): |
| defaults.update(default.attrib) |
| |
| remotes = {} |
| for rem in root.findall('remote'): |
| remotes[rem.attrib['name']] = rem.attrib |
| |
| # Search for path_to_update and check the repository it refers to |
| # matches the triggering repository. This check is mostly a config check |
| # and shouldn't fail in real runs unless somebody moves around project |
| # paths in the manifest without updating the builder definition. |
| paths_found = [] |
| proj = None |
| proj_attrib = {} |
| |
| for proj in root.findall('project'): |
| # Sometimes projects don't define something in which case we need to |
| # fall back on the remote specified by the project or the default |
| # for the entire manifest. |
| proj_attrib = {} |
| proj_attrib.update(defaults) |
| remote_name = proj.attrib.get( |
| 'remote', defaults.get('remote', None) |
| ) |
| assert remote_name |
| proj_attrib.update(remotes[remote_name]) |
| proj_attrib.update(proj.attrib) |
| proj_attrib['fetch'] = self.m.sso.sso_to_https(proj_attrib['fetch']) |
| |
| # Apparently if path is left off name is used. This is mildly |
| # infuriating, but easy to work around. |
| if 'path' not in proj_attrib: |
| proj_attrib['path'] = proj_attrib['name'] |
| |
| paths_found.append(proj_attrib['path']) |
| if proj_attrib['path'] == project.path_to_update: |
| fetch_host = proj_attrib['fetch'].strip('/') |
| if fetch_host.startswith('..'): |
| fetch_host = urllib.parse.urljoin( |
| checkout.options.remote, |
| fetch_host, |
| ) |
| |
| manifest_remote = ( |
| fetch_host.rstrip('/') |
| + '/' |
| + proj_attrib['name'].strip('/') |
| ) |
| if bb_remote and not checkout.remotes_equivalent( |
| manifest_remote, bb_remote |
| ): |
| raise self.m.step.StepFailure( |
| f"repo paths don't match: {manifest_remote!r} from " |
| f'manifest and {bb_remote!r} from buildbucket' |
| ) |
| break |
| |
| # Reset proj_attrib to None if this entry wasn't a match, so if this |
| # is the last iteration the condition a few lines down will work. |
| proj_attrib = {} |
| |
| if not proj_attrib: |
| raise self.m.step.StepFailure( |
| 'cannot find "{}" in manifest (found {})'.format( |
| project.path_to_update, |
| ', '.join('"{}"'.format(x) for x in paths_found), |
| ) |
| ) |
| |
| if 'upstream' in proj_attrib: |
| proj_branch = proj_attrib['upstream'] |
| elif 'revision' in proj_attrib and self.is_branch( |
| proj_attrib['revision'] |
| ): |
| proj_branch = proj_attrib['revision'] |
| else: |
| proj_branch = 'main' |
| |
| # If we still don't have a revision then it wasn't in the trigger. |
| # (Perhaps this was manually triggered.) In this case we need to |
| # determine the latest revision of this repository. The use_trigger flag |
| # should have no effect but using it to be explicit. Checking out even |
| # if not needed so commit messages can be collected later. |
| proj_dir = self.m.path.start_dir / 'project' |
| proj_checkout = self.m.checkout( |
| CheckoutOptions( |
| remote=manifest_remote, branch=proj_branch, use_trigger=False |
| ), |
| root=proj_dir, |
| ) |
| |
| new_revision = self.m.git_roll_util.resolve_new_revision( |
| manifest_remote, |
| proj_branch, |
| checkout.remotes_equivalent, |
| ) |
| |
| # If upstream not set we may be transitioning from tracking a branch to |
| # rolling. In that case set upstream to be the revision, but only if the |
| # revision appears to be a branch. |
| if 'upstream' not in proj_attrib: |
| if self.is_branch(proj_attrib['revision']): |
| # Explicitly update both proj.attrib and proj_attrib to minimize |
| # confusion if these statements are moved around later. |
| proj.attrib['upstream'] = proj_attrib['revision'] |
| proj_attrib['upstream'] = proj.attrib['upstream'] |
| else: |
| raise self.m.step.StepFailure( |
| 'upstream not set and revision is not a branch, aborting' |
| ) |
| |
| old_revision = proj_attrib['revision'] |
| # Explicitly update both proj.attrib and proj_attrib to minimize |
| # confusion. |
| proj_attrib['revision'] = proj.attrib['revision'] = new_revision |
| |
| direction = self.m.roll_util.Direction.FORWARD |
| if not self.is_branch(old_revision): |
| direction = self.m.roll_util.get_roll_direction( |
| proj_dir, old_revision, new_revision |
| ) |
| |
| if not self.m.roll_util.can_roll(direction): |
| self.m.roll_util.skip_roll_step( |
| manifest_remote, old_revision, new_revision |
| ) |
| return [] |
| |
| self._order_attributes(root) |
| |
| self.m.file.write_text( |
| 'write manifest', |
| checkout.manifest_path, |
| '<?xml version="1.0" encoding="UTF-8"?>\n{}\n'.format( |
| xml.etree.ElementTree.tostring(root).decode(), |
| ), |
| ) |
| |
| return [ |
| self.m.roll_util.create_roll( |
| project_name=project.path_to_update, |
| old_revision=old_revision, |
| new_revision=new_revision, |
| proj_dir=proj_dir, |
| direction=direction, |
| ), |
| ] |
