| # Copyright 2024 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 __future__ import annotations |
| |
| import configparser |
| import dataclasses |
| import io |
| import re |
| from typing import TYPE_CHECKING |
| |
| from PB.recipe_modules.pigweed.checkout.options import ( |
| Options as CheckoutOptions, |
| ) |
| from recipe_engine import recipe_api |
| |
| if TYPE_CHECKING: # pragma: no cover |
| from typing import Generator |
| from PB.recipe_modules.pigweed.txt_roll.txt_entry import TxtEntry |
| from RECIPE_MODULES.pigweed.checkout import api as checkout_api |
| from RECIPE_MODULES.pigweed.roll_util import api as roll_util_api |
| |
| |
| class TxtRollApi(recipe_api.RecipeApi): |
| |
| def update( |
| self, |
| checkout: checkout_api.CheckoutContext, |
| txt_entry: TxtEntry, |
| ) -> list[roll_util_api.Roll]: |
| branch = txt_entry.branch or 'main' |
| |
| new_revision = self.m.git_roll_util.resolve_new_revision( |
| txt_entry.remote, |
| branch, |
| checkout.remotes_equivalent, |
| ) |
| |
| project_dir = self.m.path.start_dir / 'project' |
| |
| project_checkout = self.m.checkout( |
| CheckoutOptions( |
| remote=txt_entry.remote, |
| branch=branch, |
| use_trigger=True, |
| ), |
| root=project_dir, |
| ) |
| |
| # In case new_revision is a branch name we need to retrieve the hash it |
| # resolves to. |
| if not re.search(r'^[0-9a-f]{40}$', new_revision): |
| new_revision = self.m.checkout.get_revision( |
| project_dir, 'get new revision', test_data='2' * 40 |
| ) |
| |
| full_txt_path = checkout.root / txt_entry.path |
| |
| old_revision = self.m.file.read_text( |
| 'read old revision', |
| full_txt_path, |
| test_data='1' * 40, |
| ).strip() |
| |
| self.m.file.write_text( |
| 'write new revision', full_txt_path, f'{new_revision}\n' |
| ) |
| |
| direction = self.m.roll_util.get_roll_direction( |
| project_dir, old_revision, new_revision |
| ) |
| |
| # If the primary roll is not necessary or is backwards we can exit |
| # immediately. |
| if not self.m.roll_util.can_roll(direction): |
| self.m.roll_util.skip_roll_step( |
| txt_entry.remote, |
| old_revision, |
| new_revision, |
| ) |
| return [] |
| |
| with self.m.step.nest('txt_path') as pres: |
| pres.step_summary_text = repr(txt_entry.path) |
| |
| return [ |
| self.m.roll_util.create_roll( |
| project_name=txt_entry.path, |
| old_revision=old_revision, |
| new_revision=new_revision, |
| proj_dir=project_dir, |
| direction=direction, |
| ), |
| ] |