| # SPDX-License-Identifier: GPL-2.0+ |
| # |
| # Copyright 2025 Simon Glass <sjg@chromium.org> |
| # |
| """Helper functions for handling the 'series' subcommand |
| """ |
| |
| import asyncio |
| from collections import OrderedDict, defaultdict, namedtuple |
| from datetime import datetime |
| import hashlib |
| import os |
| import re |
| import sys |
| import time |
| from types import SimpleNamespace |
| |
| import aiohttp |
| import pygit2 |
| from pygit2.enums import CheckoutStrategy |
| |
| from u_boot_pylib import gitutil |
| from u_boot_pylib import terminal |
| from u_boot_pylib import tout |
| |
| from patman import patchstream |
| from patman.database import Database, Pcommit, SerVer |
| from patman import patchwork |
| from patman.series import Series |
| from patman import status |
| |
| |
| # Tag to use for Change IDs |
| CHANGE_ID_TAG = 'Change-Id' |
| |
| # Length of hash to display |
| HASH_LEN = 10 |
| |
| # Shorter version of some states, to save horizontal space |
| SHORTEN_STATE = { |
| 'handled-elsewhere': 'elsewhere', |
| 'awaiting-upstream': 'awaiting', |
| 'not-applicable': 'n/a', |
| 'changes-requested': 'changes', |
| } |
| |
| # Summary info returned from Cseries.link_auto_all() |
| AUTOLINK = namedtuple('autolink', 'name,version,link,desc,result') |
| |
| |
| def oid(oid_val): |
| """Convert a hash string into a shortened hash |
| |
| The number of hex digits git uses for showing hashes depends on the size of |
| the repo. For the purposes of showing hashes to the user in lists, we use a |
| fixed value for now |
| |
| Args: |
| str or Pygit2.oid: Hash value to shorten |
| |
| Return: |
| str: Shortened hash |
| """ |
| return str(oid_val)[:HASH_LEN] |
| |
| |
| def split_name_version(in_name): |
| """Split a branch name into its series name and its version |
| |
| For example: |
| 'series' returns ('series', 1) |
| 'series3' returns ('series', 3) |
| Args: |
| in_name (str): Name to parse |
| |
| Return: |
| tuple: |
| str: series name |
| int: series version, or None if there is none in in_name |
| """ |
| m_ver = re.match(r'([^0-9]*)(\d*)', in_name) |
| version = None |
| if m_ver: |
| name = m_ver.group(1) |
| if m_ver.group(2): |
| version = int(m_ver.group(2)) |
| else: |
| name = in_name |
| return name, version |
| |
| |
| class CseriesHelper: |
| """Helper functions for Cseries |
| |
| This class handles database read/write as well as operations in a git |
| directory to update series information. |
| """ |
| def __init__(self, topdir=None, colour=terminal.COLOR_IF_TERMINAL): |
| """Set up a new CseriesHelper |
| |
| Args: |
| topdir (str): Top-level directory of the repo |
| colour (terminal.enum): Whether to enable ANSI colour or not |
| |
| Properties: |
| gitdir (str): Git directory (typically topdir + '/.git') |
| db (Database): Database handler |
| col (terminal.Colour): Colour object |
| _fake_time (float): Holds the current fake time for tests, in |
| seconds |
| _fake_sleep (func): Function provided by a test; called to fake a |
| 'time.sleep()' call and take whatever action it wants to take. |
| The only argument is the (Float) time to sleep for; it returns |
| nothing |
| loop (asyncio event loop): Loop used for Patchwork operations |
| """ |
| self.topdir = topdir |
| self.gitdir = None |
| self.db = None |
| self.col = terminal.Color(colour) |
| self._fake_time = None |
| self._fake_sleep = None |
| self.fake_now = None |
| self.loop = asyncio.get_event_loop() |
| |
| def open_database(self): |
| """Open the database ready for use""" |
| if not self.topdir: |
| self.topdir = gitutil.get_top_level() |
| if not self.topdir: |
| raise ValueError('No git repo detected in current directory') |
| self.gitdir = os.path.join(self.topdir, '.git') |
| fname = f'{self.topdir}/.patman.db' |
| |
| # For the first instance, start it up with the expected schema |
| self.db, is_new = Database.get_instance(fname) |
| if is_new: |
| self.db.start() |
| else: |
| # If a previous test has already checked the schema, just open it |
| self.db.open_it() |
| |
| def close_database(self): |
| """Close the database""" |
| if self.db: |
| self.db.close() |
| |
| def commit(self): |
| """Commit changes to the database""" |
| self.db.commit() |
| |
| def rollback(self): |
| """Roll back changes to the database""" |
| self.db.rollback() |
| |
| def set_fake_time(self, fake_sleep): |
| """Setup the fake timer |
| |
| Args: |
| fake_sleep (func(float)): Function to call to fake a sleep |
| """ |
| self._fake_time = 0 |
| self._fake_sleep = fake_sleep |
| |
| def inc_fake_time(self, inc_s): |
| """Increment the fake time |
| |
| Args: |
| inc_s (float): Amount to increment the fake time by |
| """ |
| self._fake_time += inc_s |
| |
| def get_time(self): |
| """Get the current time, fake or real |
| |
| This function should always be used to read the time so that faking the |
| time works correctly in tests. |
| |
| Return: |
| float: Fake time, if time is being faked, else real time |
| """ |
| if self._fake_time is not None: |
| return self._fake_time |
| return time.monotonic() |
| |
| def sleep(self, time_s): |
| """Sleep for a while |
| |
| This function should always be used to sleep so that faking the time |
| works correctly in tests. |
| |
| Args: |
| time_s (float): Amount of seconds to sleep for |
| """ |
| print(f'Sleeping for {time_s} seconds') |
| if self._fake_time is not None: |
| self._fake_sleep(time_s) |
| else: |
| time.sleep(time_s) |
| |
| def get_now(self): |
| """Get the time now |
| |
| This function should always be used to read the datetime, so that |
| faking the time works correctly in tests |
| |
| Return: |
| DateTime object |
| """ |
| if self.fake_now: |
| return self.fake_now |
| return datetime.now() |
| |
| def get_ser_ver_list(self): |
| """Get a list of patchwork entries from the database |
| |
| Return: |
| list of SER_VER |
| """ |
| return self.db.ser_ver_get_list() |
| |
| def get_ser_ver_dict(self): |
| """Get a dict of patchwork entries from the database |
| |
| Return: dict contain all records: |
| key (int): ser_ver id |
| value (SER_VER): Information about one ser_ver record |
| """ |
| svlist = self.get_ser_ver_list() |
| svdict = {} |
| for sver in svlist: |
| svdict[sver.idnum] = sver |
| return svdict |
| |
| def get_upstream_dict(self): |
| """Get a list of upstream entries from the database |
| |
| Return: |
| OrderedDict: |
| key (str): upstream name |
| value (str): url |
| """ |
| return self.db.upstream_get_dict() |
| |
| def get_pcommit_dict(self, find_svid=None): |
| """Get a dict of pcommits entries from the database |
| |
| Args: |
| find_svid (int): If not None, finds the records associated with a |
| particular series and version |
| |
| Return: |
| OrderedDict: |
| key (int): record ID if find_svid is None, else seq |
| value (PCOMMIT): record data |
| """ |
| pcdict = OrderedDict() |
| for rec in self.db.pcommit_get_list(find_svid): |
| if find_svid is not None: |
| pcdict[rec.seq] = rec |
| else: |
| pcdict[rec.idnum] = rec |
| return pcdict |
| |
| def _get_series_info(self, idnum): |
| """Get information for a series from the database |
| |
| Args: |
| idnum (int): Series ID to look up |
| |
| Return: tuple: |
| str: Series name |
| str: Series description |
| |
| Raises: |
| ValueError: Series is not found |
| """ |
| return self.db.series_get_info(idnum) |
| |
| def prep_series(self, name, end=None): |
| """Prepare to work with a series |
| |
| Args: |
| name (str): Branch name with version appended, e.g. 'fix2' |
| end (str or None): Commit to end at, e.g. 'my_branch~16'. Only |
| commits up to that are processed. None to process commits up to |
| the upstream branch |
| |
| Return: tuple: |
| str: Series name, e.g. 'fix' |
| Series: Collected series information, including name |
| int: Version number, e.g. 2 |
| str: Message to show |
| """ |
| ser, version = self._parse_series_and_version(name, None) |
| if not name: |
| name = self._get_branch_name(ser.name, version) |
| |
| # First check we have a branch with this name |
| if not gitutil.check_branch(name, git_dir=self.gitdir): |
| raise ValueError(f"No branch named '{name}'") |
| |
| count = gitutil.count_commits_to_branch(name, self.gitdir, end) |
| if not count: |
| raise ValueError('Cannot detect branch automatically: ' |
| 'Perhaps use -U <upstream-commit> ?') |
| |
| series = patchstream.get_metadata(name, 0, count, git_dir=self.gitdir) |
| self._copy_db_fields_to(series, ser) |
| msg = None |
| if end: |
| repo = pygit2.init_repository(self.gitdir) |
| target = repo.revparse_single(end) |
| first_line = target.message.splitlines()[0] |
| msg = f'Ending before {oid(target.id)} {first_line}' |
| |
| return name, series, version, msg |
| |
| def _copy_db_fields_to(self, series, in_series): |
| """Copy over fields used by Cseries from one series to another |
| |
| This copes desc, idnum and name |
| |
| Args: |
| series (Series): Series to copy to |
| in_series (Series): Series to copy from |
| """ |
| series.desc = in_series.desc |
| series.idnum = in_series.idnum |
| series.name = in_series.name |
| |
| def _handle_mark(self, branch_name, in_series, version, mark, |
| allow_unmarked, force_version, dry_run): |
| """Handle marking a series, checking for unmarked commits, etc. |
| |
| Args: |
| branch_name (str): Name of branch to sync, or None for current one |
| in_series (Series): Series object |
| version (int): branch version, e.g. 2 for 'mychange2' |
| mark (bool): True to mark each commit with a change ID |
| allow_unmarked (str): True to not require each commit to be marked |
| force_version (bool): True if ignore a Series-version tag that |
| doesn't match its branch name |
| dry_run (bool): True to do a dry run |
| |
| Returns: |
| Series: New series object, if the series was marked; |
| copy_db_fields_to() is used to copy fields over |
| |
| Raises: |
| ValueError: Series being unmarked when it should be marked, etc. |
| """ |
| series = in_series |
| if 'version' in series and int(series.version) != version: |
| msg = (f"Series name '{branch_name}' suggests version {version} " |
| f"but Series-version tag indicates {series.version}") |
| if not force_version: |
| raise ValueError(msg + ' (see --force-version)') |
| |
| tout.warning(msg) |
| tout.warning(f'Updating Series-version tag to version {version}') |
| self.update_series(branch_name, series, int(series.version), |
| new_name=None, dry_run=dry_run, |
| add_vers=version) |
| |
| # Collect the commits again, as the hashes have changed |
| series = patchstream.get_metadata(branch_name, 0, |
| len(series.commits), |
| git_dir=self.gitdir) |
| self._copy_db_fields_to(series, in_series) |
| |
| if mark: |
| add_oid = self._mark_series(branch_name, series, dry_run=dry_run) |
| |
| # Collect the commits again, as the hashes have changed |
| series = patchstream.get_metadata(add_oid, 0, len(series.commits), |
| git_dir=self.gitdir) |
| self._copy_db_fields_to(series, in_series) |
| |
| bad_count = 0 |
| for commit in series.commits: |
| if not commit.change_id: |
| bad_count += 1 |
| if bad_count and not allow_unmarked: |
| raise ValueError( |
| f'{bad_count} commit(s) are unmarked; please use -m or -M') |
| |
| return series |
| |
| def _add_series_commits(self, series, svid): |
| """Add a commits from a series into the database |
| |
| Args: |
| series (Series): Series containing commits to add |
| svid (int): ser_ver-table ID to use for each commit |
| """ |
| to_add = [Pcommit(None, seq, commit.subject, None, commit.change_id, |
| None, None, None) |
| for seq, commit in enumerate(series.commits)] |
| |
| self.db.pcommit_add_list(svid, to_add) |
| |
| def get_series_by_name(self, name, include_archived=False): |
| """Get a Series object from the database by name |
| |
| Args: |
| name (str): Name of series to get |
| include_archived (bool): True to search in archives series |
| |
| Return: |
| Series: Object containing series info, or None if none |
| """ |
| idnum = self.db.series_find_by_name(name, include_archived) |
| if not idnum: |
| return None |
| name, desc = self.db.series_get_info(idnum) |
| |
| return Series.from_fields(idnum, name, desc) |
| |
| def _get_branch_name(self, name, version): |
| """Get the branch name for a particular version |
| |
| Args: |
| name (str): Base name of branch |
| version (int): Version number to use |
| """ |
| return name + (f'{version}' if version > 1 else '') |
| |
| def _ensure_version(self, ser, version): |
| """Ensure that a version exists in a series |
| |
| Args: |
| ser (Series): Series information, with idnum and name used here |
| version (int): Version to check |
| |
| Returns: |
| list of int: List of versions |
| """ |
| versions = self._get_version_list(ser.idnum) |
| if version not in versions: |
| raise ValueError( |
| f"Series '{ser.name}' does not have a version {version}") |
| return versions |
| |
| def _set_link(self, ser_id, name, version, link, update_commit, |
| dry_run=False): |
| """Add / update a series-links link for a series |
| |
| Args: |
| ser_id (int): Series ID number |
| name (str): Series name (used to find the branch) |
| version (int): Version number (used to update the database) |
| link (str): Patchwork link-string for the series |
| update_commit (bool): True to update the current commit with the |
| link |
| dry_run (bool): True to do a dry run |
| |
| Return: |
| bool: True if the database was update, False if the ser_id or |
| version was not found |
| """ |
| if update_commit: |
| branch_name = self._get_branch_name(name, version) |
| _, ser, max_vers, _ = self.prep_series(branch_name) |
| self.update_series(branch_name, ser, max_vers, add_vers=version, |
| dry_run=dry_run, add_link=link) |
| if link is None: |
| link = '' |
| updated = 1 if self.db.ser_ver_set_link(ser_id, version, link) else 0 |
| if dry_run: |
| self.rollback() |
| else: |
| self.commit() |
| |
| return updated |
| |
| def _get_autolink_dict(self, sdict, link_all_versions): |
| """Get a dict of ser_vers to fetch, along with their patchwork links |
| |
| Note that this returns items that already have links, as well as those |
| without links |
| |
| Args: |
| sdict: |
| key: series ID |
| value: Series with idnum, name and desc filled out |
| link_all_versions (bool): True to sync all versions of a series, |
| False to sync only the latest version |
| |
| Return: tuple: |
| dict: |
| key (int): svid |
| value (tuple): |
| int: series ID |
| str: series name |
| int: series version |
| str: patchwork link for the series, or None if none |
| desc: cover-letter name / series description |
| """ |
| svdict = self.get_ser_ver_dict() |
| to_fetch = {} |
| |
| if link_all_versions: |
| for svinfo in self.get_ser_ver_list(): |
| ser = sdict[svinfo.series_id] |
| |
| pwc = self.get_pcommit_dict(svinfo.idnum) |
| count = len(pwc) |
| branch = self._join_name_version(ser.name, svinfo.version) |
| series = patchstream.get_metadata(branch, 0, count, |
| git_dir=self.gitdir) |
| self._copy_db_fields_to(series, ser) |
| |
| to_fetch[svinfo.idnum] = (svinfo.series_id, series.name, |
| svinfo.version, svinfo.link, series) |
| else: |
| # Find the maximum version for each series |
| max_vers = self._series_all_max_versions() |
| |
| # Get a list of links to fetch |
| for svid, ser_id, version in max_vers: |
| svinfo = svdict[svid] |
| ser = sdict[ser_id] |
| |
| pwc = self.get_pcommit_dict(svid) |
| count = len(pwc) |
| branch = self._join_name_version(ser.name, version) |
| series = patchstream.get_metadata(branch, 0, count, |
| git_dir=self.gitdir) |
| self._copy_db_fields_to(series, ser) |
| |
| to_fetch[svid] = (ser_id, series.name, version, svinfo.link, |
| series) |
| return to_fetch |
| |
| def _get_version_list(self, idnum): |
| """Get a list of the versions available for a series |
| |
| Args: |
| idnum (int): ID of series to look up |
| |
| Return: |
| str: List of versions |
| """ |
| if idnum is None: |
| raise ValueError('Unknown series idnum') |
| return self.db.series_get_version_list(idnum) |
| |
| def _join_name_version(self, in_name, version): |
| """Convert a series name plus a version into a branch name |
| |
| For example: |
| ('series', 1) returns 'series' |
| ('series', 3) returns 'series3' |
| |
| Args: |
| in_name (str): Series name |
| version (int): Version number |
| |
| Return: |
| str: associated branch name |
| """ |
| if version == 1: |
| return in_name |
| return f'{in_name}{version}' |
| |
| def _parse_series(self, name, include_archived=False): |
| """Parse the name of a series, or detect it from the current branch |
| |
| Args: |
| name (str or None): name of series |
| include_archived (bool): True to search in archives series |
| |
| Return: |
| Series: New object with the name set; idnum is also set if the |
| series exists in the database |
| """ |
| if not name: |
| name = gitutil.get_branch(self.gitdir) |
| name, _ = split_name_version(name) |
| ser = self.get_series_by_name(name, include_archived) |
| if not ser: |
| ser = Series() |
| ser.name = name |
| return ser |
| |
| def _parse_series_and_version(self, in_name, in_version): |
| """Parse name and version of a series, or detect from current branch |
| |
| Figures out the name from in_name, or if that is None, from the current |
| branch. |
| |
| Uses the version in_version, or if that is None, uses the int at the |
| end of the name (e.g. 'series' is version 1, 'series4' is version 4) |
| |
| Args: |
| in_name (str or None): name of series |
| in_version (str or None): version of series |
| |
| Return: |
| tuple: |
| Series: New object with the name set; idnum is also set if the |
| series exists in the database |
| int: Series version-number detected from the name |
| (e.g. 'fred' is version 1, 'fred2' is version 2) |
| """ |
| name = in_name |
| if not name: |
| name = gitutil.get_branch(self.gitdir) |
| if not name: |
| raise ValueError('No branch detected: please use -s <series>') |
| name, version = split_name_version(name) |
| if not name: |
| raise ValueError(f"Series name '{in_name}' cannot be a number, " |
| f"use '<name><version>'") |
| if in_version: |
| if version and version != in_version: |
| tout.warning( |
| f"Version mismatch: -V has {in_version} but branch name " |
| f'indicates {version}') |
| version = in_version |
| if not version: |
| version = 1 |
| if version > 99: |
| raise ValueError(f"Version {version} exceeds 99") |
| ser = self.get_series_by_name(name) |
| if not ser: |
| ser = Series() |
| ser.name = name |
| return ser, version |
| |
| def _series_get_version_stats(self, idnum, vers): |
| """Get the stats for a series |
| |
| Args: |
| idnum (int): ID number of series to process |
| vers (int): Version number to process |
| |
| Return: |
| tuple: |
| str: Status string, '<accepted>/<count>' |
| OrderedDict: |
| key (int): record ID if find_svid is None, else seq |
| value (PCOMMIT): record data |
| """ |
| svid, link = self._get_series_svid_link(idnum, vers) |
| pwc = self.get_pcommit_dict(svid) |
| count = len(pwc.values()) |
| if link: |
| accepted = 0 |
| for pcm in pwc.values(): |
| accepted += pcm.state == 'accepted' |
| else: |
| accepted = '-' |
| return f'{accepted}/{count}', pwc |
| |
| def get_series_svid(self, series_id, version): |
| """Get the patchwork ID of a series version |
| |
| Args: |
| series_id (int): id of the series to look up |
| version (int): version number to look up |
| |
| Return: |
| str: link found |
| |
| Raises: |
| ValueError: No matching series found |
| """ |
| return self._get_series_svid_link(series_id, version)[0] |
| |
| def _get_series_svid_link(self, series_id, version): |
| """Get the patchwork ID of a series version |
| |
| Args: |
| series_id (int): series ID to look up |
| version (int): version number to look up |
| |
| Return: |
| tuple: |
| int: record id |
| str: link |
| """ |
| recs = self.get_ser_ver(series_id, version) |
| return recs.idnum, recs.link |
| |
| def get_ser_ver(self, series_id, version): |
| """Get the patchwork details for a series version |
| |
| Args: |
| series_id (int): series ID to look up |
| version (int): version number to look up |
| |
| Return: |
| SER_VER: Requested information |
| |
| Raises: |
| ValueError: There is no matching idnum/version |
| """ |
| return self.db.ser_ver_get_for_series(series_id, version) |
| |
| def _prepare_process(self, name, count, new_name=None, quiet=False): |
| """Get ready to process all commits in a branch |
| |
| Args: |
| name (str): Name of the branch to process |
| count (int): Number of commits |
| new_name (str or None): New name, if a new branch is to be created |
| quiet (bool): True to avoid output (used for testing) |
| |
| Return: tuple: |
| pygit2.repo: Repo to use |
| pygit2.oid: Upstream commit, onto which commits should be added |
| Pygit2.branch: Original branch, for later use |
| str: (Possibly new) name of branch to process |
| list of Commit: commits to process, in order |
| pygit2.Reference: Original head before processing started |
| """ |
| upstream_guess = gitutil.get_upstream(self.gitdir, name)[0] |
| |
| tout.debug(f"_process_series name '{name}' new_name '{new_name}' " |
| f"upstream_guess '{upstream_guess}'") |
| dirty = gitutil.check_dirty(self.gitdir, self.topdir) |
| if dirty: |
| raise ValueError( |
| f"Modified files exist: use 'git status' to check: " |
| f'{dirty[:5]}') |
| repo = pygit2.init_repository(self.gitdir) |
| |
| commit = None |
| upstream_name = None |
| if upstream_guess: |
| try: |
| upstream = repo.lookup_reference(upstream_guess) |
| upstream_name = upstream.name |
| commit = upstream.peel(pygit2.enums.ObjectType.COMMIT) |
| except KeyError: |
| pass |
| except pygit2.repository.InvalidSpecError as exc: |
| print(f"Error '{exc}'") |
| if not upstream_name: |
| upstream_name = f'{name}~{count}' |
| commit = repo.revparse_single(upstream_name) |
| |
| branch = repo.lookup_branch(name) |
| if not quiet: |
| tout.info( |
| f'Checking out upstream commit {upstream_name}: ' |
| f'{oid(commit.oid)}') |
| |
| old_head = repo.head |
| if old_head.shorthand == name: |
| old_head = None |
| else: |
| old_head = repo.head |
| |
| if new_name: |
| name = new_name |
| repo.set_head(commit.oid) |
| |
| commits = [] |
| cmt = repo.get(branch.target) |
| for _ in range(count): |
| commits.append(cmt) |
| cmt = cmt.parents[0] |
| |
| return (repo, repo.head, branch, name, commit, list(reversed(commits)), |
| old_head) |
| |
| def _pick_commit(self, repo, cmt): |
| """Apply a commit to the source tree, without committing it |
| |
| _prepare_process() must be called before starting to pick commits |
| |
| This function must be called before _finish_commit() |
| |
| Note that this uses a cherry-pick method, creating a new tree_id each |
| time, so can make source-code changes |
| |
| Args: |
| repo (pygit2.repo): Repo to use |
| cmt (Commit): Commit to apply |
| |
| Return: tuple: |
| tree_id (pygit2.oid): Oid of index with source-changes applied |
| commit (pygit2.oid): Old commit being cherry-picked |
| """ |
| tout.detail(f"- adding {oid(cmt.hash)} {cmt}") |
| repo.cherrypick(cmt.hash) |
| if repo.index.conflicts: |
| raise ValueError('Conflicts detected') |
| |
| tree_id = repo.index.write_tree() |
| cherry = repo.get(cmt.hash) |
| tout.detail(f"cherry {oid(cherry.oid)}") |
| return tree_id, cherry |
| |
| def _finish_commit(self, repo, tree_id, commit, cur, msg=None): |
| """Complete a commit |
| |
| This must be called after _pick_commit(). |
| |
| Args: |
| repo (pygit2.repo): Repo to use |
| tree_id (pygit2.oid): Oid of index with source-changes applied; if |
| None then the existing commit.tree_id is used |
| commit (pygit2.oid): Old commit being cherry-picked |
| cur (pygit2.reference): Reference to parent to use for the commit |
| msg (str): Commit subject and message; None to use commit.message |
| """ |
| if msg is None: |
| msg = commit.message |
| if not tree_id: |
| tree_id = commit.tree_id |
| repo.create_commit('HEAD', commit.author, commit.committer, |
| msg, tree_id, [cur.target]) |
| return repo.head |
| |
| def _finish_process(self, repo, branch, name, cur, old_head, new_name=None, |
| switch=False, dry_run=False, quiet=False): |
| """Finish processing commits |
| |
| Args: |
| repo (pygit2.repo): Repo to use |
| branch (pygit2.branch): Branch returned by _prepare_process() |
| name (str): Name of the branch to process |
| new_name (str or None): New name, if a new branch is being created |
| switch (bool): True to switch to the new branch after processing; |
| otherwise HEAD remains at the original branch, as amended |
| dry_run (bool): True to do a dry run, restoring the original tree |
| afterwards |
| quiet (bool): True to avoid output (used for testing) |
| |
| Return: |
| pygit2.reference: Final commit after everything is completed |
| """ |
| repo.state_cleanup() |
| |
| # Update the branch |
| target = repo.revparse_single('HEAD') |
| if not quiet: |
| tout.info(f'Updating branch {name} from {oid(branch.target)} to ' |
| f'{str(target.oid)[:HASH_LEN]}') |
| if dry_run: |
| if new_name: |
| repo.head.set_target(branch.target) |
| else: |
| branch_oid = branch.peel(pygit2.enums.ObjectType.COMMIT).oid |
| repo.head.set_target(branch_oid) |
| repo.head.set_target(branch.target) |
| repo.set_head(branch.name) |
| else: |
| if new_name: |
| new_branch = repo.branches.create(new_name, target) |
| if branch.upstream: |
| new_branch.upstream = branch.upstream |
| branch = new_branch |
| else: |
| branch.set_target(cur.target) |
| repo.set_head(branch.name) |
| if old_head: |
| if not switch: |
| repo.set_head(old_head.name) |
| return target |
| |
| def make_change_id(self, commit): |
| """Make a Change ID for a commit |
| |
| This is similar to the gerrit script: |
| git var GIT_COMMITTER_IDENT ; echo "$refhash" ; cat "README"; } |
| | git hash-object --stdin) |
| |
| Args: |
| commit (pygit2.commit): Commit to process |
| |
| Return: |
| Change ID in hex format |
| """ |
| sig = commit.committer |
| val = hashlib.sha1() |
| to_hash = f'{sig.name} <{sig.email}> {sig.time} {sig.offset}' |
| val.update(to_hash.encode('utf-8')) |
| val.update(str(commit.tree_id).encode('utf-8')) |
| val.update(commit.message.encode('utf-8')) |
| return val.hexdigest() |
| |
| def _filter_commits(self, name, series, seq_to_drop): |
| """Filter commits to drop one |
| |
| This function rebases the current branch, dropping a single commit, |
| thus changing the resulting code in the tree. |
| |
| Args: |
| name (str): Name of the branch to process |
| series (Series): Series object |
| seq_to_drop (int): Commit sequence to drop; commits are numbered |
| from 0, which is the one after the upstream branch, to |
| count - 1 |
| """ |
| count = len(series.commits) |
| (repo, cur, branch, name, commit, _, _) = self._prepare_process( |
| name, count, quiet=True) |
| repo.checkout_tree(commit, strategy=CheckoutStrategy.FORCE | |
| CheckoutStrategy.RECREATE_MISSING) |
| repo.set_head(commit.oid) |
| for seq, cmt in enumerate(series.commits): |
| if seq != seq_to_drop: |
| tree_id, cherry = self._pick_commit(repo, cmt) |
| cur = self._finish_commit(repo, tree_id, cherry, cur) |
| self._finish_process(repo, branch, name, cur, None, quiet=True) |
| |
| def process_series(self, name, series, new_name=None, switch=False, |
| dry_run=False): |
| """Rewrite a series commit messages, leaving code alone |
| |
| This uses a 'vals' namespace to pass things to the controlling |
| function. |
| |
| Each time _process_series() yields, it sets up: |
| commit (Commit): The pygit2 commit that is being processed |
| msg (str): Commit message, which can be modified |
| info (str): Initially empty; the controlling function can add a |
| short message here which will be shown to the user |
| final (bool): True if this is the last commit to apply |
| seq (int): Current sequence number in the commits to apply (0,,n-1) |
| |
| It also sets git HEAD at the commit before this commit being |
| processed |
| |
| The function can change msg and info, e.g. to add or remove tags from |
| the commit. |
| |
| Args: |
| name (str): Name of the branch to process |
| series (Series): Series object |
| new_name (str or None): New name, if a new branch is to be created |
| switch (bool): True to switch to the new branch after processing; |
| otherwise HEAD remains at the original branch, as amended |
| dry_run (bool): True to do a dry run, restoring the original tree |
| afterwards |
| |
| Return: |
| pygit.oid: oid of the new branch |
| """ |
| count = len(series.commits) |
| repo, cur, branch, name, _, commits, old_head = self._prepare_process( |
| name, count, new_name) |
| vals = SimpleNamespace() |
| vals.final = False |
| tout.info(f"Processing {count} commits from branch '{name}'") |
| |
| # Record the message lines |
| lines = [] |
| for seq, cmt in enumerate(series.commits): |
| commit = commits[seq] |
| vals.commit = commit |
| vals.msg = commit.message |
| vals.info = '' |
| vals.final = seq == len(series.commits) - 1 |
| vals.seq = seq |
| yield vals |
| |
| cur = self._finish_commit(repo, None, commit, cur, vals.msg) |
| lines.append([vals.info.strip(), |
| f'{oid(cmt.hash)} as {oid(cur.target)} {cmt}']) |
| |
| max_len = max(len(info) for info, rest in lines) + 1 |
| for info, rest in lines: |
| if info: |
| info += ':' |
| tout.info(f'- {info.ljust(max_len)} {rest}') |
| target = self._finish_process(repo, branch, name, cur, old_head, |
| new_name, switch, dry_run) |
| vals.oid = target.oid |
| |
| def _mark_series(self, name, series, dry_run=False): |
| """Mark a series with Change-Id tags |
| |
| Args: |
| name (str): Name of the series to mark |
| series (Series): Series object |
| dry_run (bool): True to do a dry run, restoring the original tree |
| afterwards |
| |
| Return: |
| pygit.oid: oid of the new branch |
| """ |
| vals = None |
| for vals in self.process_series(name, series, dry_run=dry_run): |
| if CHANGE_ID_TAG not in vals.msg: |
| change_id = self.make_change_id(vals.commit) |
| vals.msg = vals.msg + f'\n{CHANGE_ID_TAG}: {change_id}' |
| tout.detail(" - adding mark") |
| vals.info = 'marked' |
| else: |
| vals.info = 'has mark' |
| |
| return vals.oid |
| |
| def update_series(self, branch_name, series, max_vers, new_name=None, |
| dry_run=False, add_vers=None, add_link=None, |
| add_rtags=None, switch=False): |
| """Rewrite a series to update the Series-version/Series-links lines |
| |
| This updates the series in git; it does not update the database |
| |
| Args: |
| branch_name (str): Name of the branch to process |
| series (Series): Series object |
| max_vers (int): Version number of the series being updated |
| new_name (str or None): New name, if a new branch is to be created |
| dry_run (bool): True to do a dry run, restoring the original tree |
| afterwards |
| add_vers (int or None): Version number to add to the series, if any |
| add_link (str or None): Link to add to the series, if any |
| add_rtags (list of dict): List of review tags to add, one item for |
| each commit, each a dict: |
| key: Response tag (e.g. 'Reviewed-by') |
| value: Set of people who gave that response, each a name/email |
| string |
| switch (bool): True to switch to the new branch after processing; |
| otherwise HEAD remains at the original branch, as amended |
| |
| Return: |
| pygit.oid: oid of the new branch |
| """ |
| def _do_version(): |
| if add_vers: |
| if add_vers == 1: |
| vals.info += f'rm v{add_vers} ' |
| else: |
| vals.info += f'add v{add_vers} ' |
| out.append(f'Series-version: {add_vers}') |
| |
| def _do_links(new_links): |
| if add_link: |
| if 'add' not in vals.info: |
| vals.info += 'add ' |
| vals.info += f"links '{new_links}' " |
| else: |
| vals.info += f"upd links '{new_links}' " |
| out.append(f'Series-links: {new_links}') |
| |
| added_version = False |
| added_link = False |
| for vals in self.process_series(branch_name, series, new_name, switch, |
| dry_run): |
| out = [] |
| for line in vals.msg.splitlines(): |
| m_ver = re.match('Series-version:(.*)', line) |
| m_links = re.match('Series-links:(.*)', line) |
| if m_ver and add_vers: |
| if ('version' in series and |
| int(series.version) != max_vers): |
| tout.warning( |
| f'Branch {branch_name}: Series-version tag ' |
| f'{series.version} does not match expected ' |
| f'version {max_vers}') |
| _do_version() |
| added_version = True |
| elif m_links: |
| links = series.get_links(m_links.group(1), max_vers) |
| if add_link: |
| links[max_vers] = add_link |
| _do_links(series.build_links(links)) |
| added_link = True |
| else: |
| out.append(line) |
| if vals.final: |
| if not added_version and add_vers and add_vers > 1: |
| _do_version() |
| if not added_link and add_link: |
| _do_links(f'{max_vers}:{add_link}') |
| |
| vals.msg = '\n'.join(out) + '\n' |
| if add_rtags and add_rtags[vals.seq]: |
| lines = [] |
| for tag, people in add_rtags[vals.seq].items(): |
| for who in people: |
| lines.append(f'{tag}: {who}') |
| vals.msg = patchstream.insert_tags(vals.msg.rstrip(), |
| sorted(lines)) |
| vals.info += (f'added {len(lines)} ' |
| f"tag{'' if len(lines) == 1 else 's'}") |
| |
| def _build_col(self, state, prefix='', base_str=None): |
| """Build a patch-state string with colour |
| |
| Args: |
| state (str): State to colourise (also indicates the colour to use) |
| prefix (str): Prefix string to also colourise |
| base_str (str or None): String to show instead of state, or None to |
| show state |
| |
| Return: |
| str: String with ANSI colour characters |
| """ |
| bright = True |
| if state == 'accepted': |
| col = self.col.GREEN |
| elif state == 'awaiting-upstream': |
| bright = False |
| col = self.col.GREEN |
| elif state in ['changes-requested']: |
| col = self.col.CYAN |
| elif state in ['rejected', 'deferred', 'not-applicable', 'superseded', |
| 'handled-elsewhere']: |
| col = self.col.RED |
| elif not state: |
| state = 'unknown' |
| col = self.col.MAGENTA |
| else: |
| # under-review, rfc, needs-review-ack |
| col = self.col.WHITE |
| out = base_str or SHORTEN_STATE.get(state, state) |
| pad = ' ' * (10 - len(out)) |
| col_state = self.col.build(col, prefix + out, bright) |
| return col_state, pad |
| |
| def _get_patches(self, series, version): |
| """Get a Series object containing the patches in a series |
| |
| Args: |
| series (str): Name of series to use, or None to use current branch |
| version (int): Version number, or None to detect from name |
| |
| Return: tuple: |
| str: Name of branch, e.g. 'mary2' |
| Series: Series object containing the commits and idnum, desc, name |
| int: Version number of series, e.g. 2 |
| OrderedDict: |
| key (int): record ID if find_svid is None, else seq |
| value (PCOMMIT): record data |
| str: series name (for this version) |
| str: patchwork link |
| str: cover_id |
| int: cover_num_comments |
| """ |
| ser, version = self._parse_series_and_version(series, version) |
| if not ser.idnum: |
| raise ValueError(f"Unknown series '{series}'") |
| self._ensure_version(ser, version) |
| svinfo = self.get_ser_ver(ser.idnum, version) |
| pwc = self.get_pcommit_dict(svinfo.idnum) |
| |
| count = len(pwc) |
| branch = self._join_name_version(ser.name, version) |
| series = patchstream.get_metadata(branch, 0, count, |
| git_dir=self.gitdir) |
| self._copy_db_fields_to(series, ser) |
| |
| return (branch, series, version, pwc, svinfo.name, svinfo.link, |
| svinfo.cover_id, svinfo.cover_num_comments) |
| |
| def _list_patches(self, branch, pwc, series, desc, cover_id, num_comments, |
| show_commit, show_patch, list_patches, state_totals): |
| """List patches along with optional status info |
| |
| Args: |
| branch (str): Branch name if self.show_progress |
| pwc (dict): pcommit records: |
| key (int): seq |
| value (PCOMMIT): Record from database |
| series (Series): Series to show, or None to just use the database |
| desc (str): Series title |
| cover_id (int): Cover-letter ID |
| num_comments (int): The number of comments on the cover letter |
| show_commit (bool): True to show the commit and diffstate |
| show_patch (bool): True to show the patch |
| list_patches (bool): True to list all patches for each series, |
| False to just show the series summary on a single line |
| state_totals (dict): Holds totals for each state across all patches |
| key (str): state name |
| value (int): Number of patches in that state |
| |
| Return: |
| bool: True if OK, False if any commit subjects don't match their |
| patchwork subjects |
| """ |
| lines = [] |
| states = defaultdict(int) |
| count = len(pwc) |
| ok = True |
| for seq, item in enumerate(pwc.values()): |
| if series: |
| cmt = series.commits[seq] |
| if cmt.subject != item.subject: |
| ok = False |
| |
| col_state, pad = self._build_col(item.state) |
| patch_id = item.patch_id if item.patch_id else '' |
| if item.num_comments: |
| comments = str(item.num_comments) |
| elif item.num_comments is None: |
| comments = '-' |
| else: |
| comments = '' |
| |
| if show_commit or show_patch: |
| subject = self.col.build(self.col.BLACK, item.subject, |
| bright=False, back=self.col.YELLOW) |
| else: |
| subject = item.subject |
| |
| line = (f'{seq:3} {col_state}{pad} {comments.rjust(3)} ' |
| f'{patch_id:7} {oid(cmt.hash)} {subject}') |
| lines.append(line) |
| states[item.state] += 1 |
| out = '' |
| for state, freq in states.items(): |
| out += ' ' + self._build_col(state, f'{freq}:')[0] |
| state_totals[state] += freq |
| name = '' |
| if not list_patches: |
| name = desc or series.desc |
| name = self.col.build(self.col.YELLOW, name[:41].ljust(41)) |
| if not ok: |
| out = '*' + out[1:] |
| print(f"{branch:16} {name} {len(pwc):5} {out}") |
| return ok |
| print(f"Branch '{branch}' (total {len(pwc)}):{out}{name}") |
| |
| print(self.col.build( |
| self.col.MAGENTA, |
| f"Seq State Com PatchId {'Commit'.ljust(HASH_LEN)} Subject")) |
| |
| comments = '' if num_comments is None else str(num_comments) |
| if desc or comments or cover_id: |
| cov = 'Cov' if cover_id else '' |
| print(self.col.build( |
| self.col.WHITE, |
| f"{cov:14} {comments.rjust(3)} {cover_id or '':7} " |
| f'{desc or series.desc}', |
| bright=False)) |
| for seq in range(count): |
| line = lines[seq] |
| print(line) |
| if show_commit or show_patch: |
| print() |
| cmt = series.commits[seq] if series else '' |
| msg = gitutil.show_commit( |
| cmt.hash, show_commit, True, show_patch, |
| colour=self.col.enabled(), git_dir=self.gitdir) |
| sys.stdout.write(msg) |
| if seq != count - 1: |
| print() |
| print() |
| |
| return ok |
| |
| def _find_matched_commit(self, commits, pcm): |
| """Find a commit in a list of possible matches |
| |
| Args: |
| commits (dict of Commit): Possible matches |
| key (int): sequence number of patch (from 0) |
| value (Commit): Commit object |
| pcm (PCOMMIT): Patch to check |
| |
| Return: |
| int: Sequence number of matching commit, or None if not found |
| """ |
| for seq, cmt in commits.items(): |
| tout.debug(f"- match subject: '{cmt.subject}'") |
| if pcm.subject == cmt.subject: |
| return seq |
| return None |
| |
| def _find_matched_patch(self, patches, cmt): |
| """Find a patch in a list of possible matches |
| |
| Args: |
| patches: dict of ossible matches |
| key (int): sequence number of patch |
| value (PCOMMIT): patch |
| cmt (Commit): Commit to check |
| |
| Return: |
| int: Sequence number of matching patch, or None if not found |
| """ |
| for seq, pcm in patches.items(): |
| tout.debug(f"- match subject: '{pcm.subject}'") |
| if cmt.subject == pcm.subject: |
| return seq |
| return None |
| |
| def _sync_one(self, svid, series_name, version, show_comments, |
| show_cover_comments, gather_tags, cover, patches, dry_run): |
| """Sync one series to the database |
| |
| Args: |
| svid (int): Ser/ver ID |
| cover (dict or None): Cover letter from patchwork, with keys: |
| id (int): Cover-letter ID in patchwork |
| num_comments (int): Number of comments |
| name (str): Cover-letter name |
| patches (list of Patch): Patches in the series |
| """ |
| pwc = self.get_pcommit_dict(svid) |
| if gather_tags: |
| count = len(pwc) |
| branch = self._join_name_version(series_name, version) |
| series = patchstream.get_metadata(branch, 0, count, |
| git_dir=self.gitdir) |
| |
| _, new_rtag_list = status.do_show_status( |
| series, cover, patches, show_comments, show_cover_comments, |
| self.col, warnings_on_stderr=False) |
| self.update_series(branch, series, version, None, dry_run, |
| add_rtags=new_rtag_list) |
| |
| updated = 0 |
| for seq, item in enumerate(pwc.values()): |
| if seq >= len(patches): |
| continue |
| patch = patches[seq] |
| if patch.id: |
| if self.db.pcommit_update( |
| Pcommit(item.idnum, seq, None, None, None, patch.state, |
| patch.id, len(patch.comments))): |
| updated += 1 |
| if cover: |
| info = SerVer(svid, None, None, None, cover.id, |
| cover.num_comments, cover.name, None) |
| else: |
| info = SerVer(svid, None, None, None, None, None, patches[0].name, |
| None) |
| self.db.ser_ver_set_info(info) |
| |
| return updated, 1 if cover else 0 |
| |
| async def _gather(self, pwork, link, show_cover_comments): |
| """Sync the series status from patchwork |
| |
| Creates a new client sesion and calls _sync() |
| |
| Args: |
| pwork (Patchwork): Patchwork object to use |
| link (str): Patchwork link for the series |
| show_cover_comments (bool): True to show the comments on the cover |
| letter |
| |
| Return: tuple: |
| COVER object, or None if none or not read_cover_comments |
| list of PATCH objects |
| """ |
| async with aiohttp.ClientSession() as client: |
| return await pwork.series_get_state(client, link, True, |
| show_cover_comments) |
| |
| def _get_fetch_dict(self, sync_all_versions): |
| """Get a dict of ser_vers to fetch, along with their patchwork links |
| |
| Args: |
| sync_all_versions (bool): True to sync all versions of a series, |
| False to sync only the latest version |
| |
| Return: tuple: |
| dict: things to fetch |
| key (int): svid |
| value (str): patchwork link for the series |
| int: number of series which are missing a link |
| """ |
| missing = 0 |
| svdict = self.get_ser_ver_dict() |
| sdict = self.db.series_get_dict_by_id() |
| to_fetch = {} |
| |
| if sync_all_versions: |
| for svinfo in self.get_ser_ver_list(): |
| ser_ver = svdict[svinfo.idnum] |
| if svinfo.link: |
| to_fetch[svinfo.idnum] = patchwork.STATE_REQ( |
| svinfo.link, svinfo.series_id, |
| sdict[svinfo.series_id].name, svinfo.version, False, |
| False) |
| else: |
| missing += 1 |
| else: |
| # Find the maximum version for each series |
| max_vers = self._series_all_max_versions() |
| |
| # Get a list of links to fetch |
| for svid, series_id, version in max_vers: |
| ser_ver = svdict[svid] |
| if series_id not in sdict: |
| # skip archived item |
| continue |
| if ser_ver.link: |
| to_fetch[svid] = patchwork.STATE_REQ( |
| ser_ver.link, series_id, sdict[series_id].name, |
| version, False, False) |
| else: |
| missing += 1 |
| |
| # order by series name, version |
| ordered = OrderedDict() |
| for svid in sorted( |
| to_fetch, |
| key=lambda k: (to_fetch[k].series_name, to_fetch[k].version)): |
| sync = to_fetch[svid] |
| ordered[svid] = sync |
| |
| return ordered, missing |
| |
| async def _sync_all(self, client, pwork, to_fetch): |
| """Sync all series status from patchwork |
| |
| Args: |
| pwork (Patchwork): Patchwork object to use |
| sync_all_versions (bool): True to sync all versions of a series, |
| False to sync only the latest version |
| gather_tags (bool): True to gather review/test tags |
| |
| Return: list of tuple: |
| COVER object, or None if none or not read_cover_comments |
| list of PATCH objects |
| """ |
| with pwork.collect_stats() as stats: |
| tasks = [pwork.series_get_state(client, sync.link, True, True) |
| for sync in to_fetch.values() if sync.link] |
| result = await asyncio.gather(*tasks) |
| return result, stats.request_count |
| |
| async def _do_series_sync_all(self, pwork, to_fetch): |
| async with aiohttp.ClientSession() as client: |
| return await self._sync_all(client, pwork, to_fetch) |
| |
| def _progress_one(self, ser, show_all_versions, list_patches, |
| state_totals): |
| """Show progress information for all versions in a series |
| |
| Args: |
| ser (Series): Series to use |
| show_all_versions (bool): True to show all versions of a series, |
| False to show only the final version |
| list_patches (bool): True to list all patches for each series, |
| False to just show the series summary on a single line |
| state_totals (dict): Holds totals for each state across all patches |
| key (str): state name |
| value (int): Number of patches in that state |
| |
| Return: tuple |
| int: Number of series shown |
| int: Number of patches shown |
| int: Number of version which need a 'scan' |
| """ |
| max_vers = self._series_max_version(ser.idnum) |
| name, desc = self._get_series_info(ser.idnum) |
| coloured = self.col.build(self.col.BLACK, desc, bright=False, |
| back=self.col.YELLOW) |
| versions = self._get_version_list(ser.idnum) |
| vstr = list(map(str, versions)) |
| |
| if list_patches: |
| print(f"{name}: {coloured} (versions: {' '.join(vstr)})") |
| add_blank_line = False |
| total_series = 0 |
| total_patches = 0 |
| need_scan = 0 |
| for ver in versions: |
| if not show_all_versions and ver != max_vers: |
| continue |
| if add_blank_line: |
| print() |
| _, pwc = self._series_get_version_stats(ser.idnum, ver) |
| count = len(pwc) |
| branch = self._join_name_version(ser.name, ver) |
| series = patchstream.get_metadata(branch, 0, count, |
| git_dir=self.gitdir) |
| svinfo = self.get_ser_ver(ser.idnum, ver) |
| self._copy_db_fields_to(series, ser) |
| |
| ok = self._list_patches( |
| branch, pwc, series, svinfo.name, svinfo.cover_id, |
| svinfo.cover_num_comments, False, False, list_patches, |
| state_totals) |
| if not ok: |
| need_scan += 1 |
| add_blank_line = list_patches |
| total_series += 1 |
| total_patches += count |
| return total_series, total_patches, need_scan |
| |
| def _summary_one(self, ser): |
| """Show summary information for the latest version in a series |
| |
| Args: |
| series (str): Name of series to use, or None to show progress for |
| all series |
| """ |
| max_vers = self._series_max_version(ser.idnum) |
| name, desc = self._get_series_info(ser.idnum) |
| stats, pwc = self._series_get_version_stats(ser.idnum, max_vers) |
| states = {x.state for x in pwc.values()} |
| state = 'accepted' |
| for val in ['awaiting-upstream', 'changes-requested', 'rejected', |
| 'deferred', 'not-applicable', 'superseded', |
| 'handled-elsewhere']: |
| if val in states: |
| state = val |
| state_str, pad = self._build_col(state, base_str=name) |
| print(f"{state_str}{pad} {stats.rjust(6)} {desc}") |
| |
| def _series_max_version(self, idnum): |
| """Find the latest version of a series |
| |
| Args: |
| idnum (int): Series ID to look up |
| |
| Return: |
| int: maximum version |
| """ |
| return self.db.series_get_max_version(idnum) |
| |
| def _series_all_max_versions(self): |
| """Find the latest version of all series |
| |
| Return: list of: |
| int: ser_ver ID |
| int: series ID |
| int: Maximum version |
| """ |
| return self.db.series_get_all_max_versions() |
