| # Copyright 2026 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. |
| """Roll-related utility functions.""" |
| |
| from __future__ import annotations |
| |
| from collections.abc import Sequence |
| |
| from recipe_engine import recipe_api |
| |
| |
| class RollUtilApi(recipe_api.RecipeApi): |
| def on_last_iteration( |
| self, |
| poll_interval_secs: int, |
| *, |
| debug: bool = False, |
| ) -> bool: |
| remaining = self._secs_remaining_in_build() |
| if debug: |
| pres = self.m.step.empty('secs remaining in build').presentation |
| pres.step_summary_text = f'{remaining}' |
| |
| pres = self.m.step.empty('poll interval secs').presentation |
| pres.step_summary_text = f'{poll_interval_secs}' |
| |
| return remaining < poll_interval_secs * 3 // 2 |
| |
| def _secs_remaining_in_build(self) -> float: |
| current_time = self.m.time.time() |
| elapsed_time = ( |
| current_time - self.m.buildbucket.build.start_time.seconds |
| ) |
| remaining = ( |
| self.m.buildbucket.build.execution_timeout.seconds - elapsed_time |
| ) |
| |
| # If a test isn't fully specified we might get some inconsistent |
| # results. Ignore those cases. |
| if remaining < -1337000000: |
| remaining = 10000 |
| |
| return remaining |
| |
| def update_roll_comments( |
| self, |
| lines: Sequence[str | None], |
| matching_line_number: int, |
| whitespace_prefix: str, |
| comment_prefix, |
| ) -> list[str]: |
| # Remove lines immediately preceding this line that start with "ROLL:". |
| for i in range( |
| matching_line_number - 1, |
| max(matching_line_number - 10, 0), |
| -1, |
| ): |
| if lines[i] and lines[i].startswith( |
| f'{whitespace_prefix}{comment_prefix} ROLL:' |
| ): |
| lines[i] = None |
| else: |
| break |
| |
| metadata_prefix = f'{whitespace_prefix}{comment_prefix} ROLL: ' |
| return [ |
| f'{metadata_prefix}Warning: this is automatically updated by', |
| f'{metadata_prefix}{self.m.buildbucket.build_url()}', |
| ] |