get_non_upstreamed_commits.py - cos/scripts - Git at Google

 #!/usr/bin/env python3
 #
 # Copyright 2024 The ChromiumOS Authors
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.

 # This script is a slightly fancier version of `git cherry`, which can be used
 # to find commits which have been applied locally but haven't been applied
 # upstream.
 #
 # Like `git cherry`, this script takes arguments for an upstream, a head, and
 # a limit. If the head argument is omitted, HEAD is used.
 #
 # The script will do the following:
 # 1.  Get all of the commits which are present between limit and head.
 # 2.  For each commit:
 #     1.  Check if the commit is present upstream by:
 #         1.  Checking if a commit with an identical hash is present upstream.
 #         2.  Checking if a commit with the exact same contents is present
 #             upstream.
 #         3.  Checking if the there are upstream commit hashes in the commit
 #             message.
 #         4.  Checking for an upstream commit with the exact same subject line.
 #     2.  Output the commit hash, the upstream commit hash (if present),
 #         whether an exact content match was found, and the commit subject.
 #
 # The output will be a tab-separated table in commit order.
 #
 # In order for the upstream subject lookups to be fast, the commit history
 # between limit and upstream is read and the subjects are put into a hashmap.
 # This default <limit>..<upstream> commit range should almost always be fine,
 # but a different revision range can be passed via the
 # subject-map-revision-range argument if it's expected that there are relevant
 # commits elsewhere.
 #
 # Usage:
 #   If you were upgrading from kernel version 6.1 to 6.6, you could run this
 #   script as follows:
 #
 #   python3 get_non_upstreamed_commits.py \
 #     --upstream cos/upstream-6.6 \
 #     --head cos-6.6 \
 #     --limit cos/upstream-6.6 > commits.tsv
 #
 #   Using the abbreviated arguments:
 #
 #   python3 get_non_upstreamed_commits.py \
 #     -u cos/upstream-6.6 -h cos-6.6 -l cos/upstream-6.6 > commits.tsv

 import argparse
 import dataclasses
 import logging
 import re
 from subprocess import PIPE, Popen, run
 import sys
 from typing import Mapping, Optional

 COMMIT_SUBJECT_RE = re.compile(r'(.*?) (.*)')
 UPSTREAM_COMMIT_PATTERN_1 = re.compile('Upstream commit.*?([a-f0-9]+)')
 UPSTREAM_COMMIT_PATTERN_2 = re.compile(r'UPSTREAM\(([a-f0-9]+)\)')


 @dataclasses.dataclass
 class Commit:
   """Class representing information about a git commit."""

   _hash: str
   subject: str
   message: str
   has_exact_content_match: bool

   def get_upstream_commit_hashes(
       self,
       subject_to_commit_map: Mapping[str, list[str]],
   ) -> list[str]:
     """Gets the hashes of any upstream commits which "match" this one.

     A commit is determined to "match" this one if it is referred to as an
     upstream commit in the body of this commit, or if it has the same subject
     as this commit.

     Args:
       subject_to_commit_map: A mapping from commit subjects to a list of hashes
         for commits with that subject.

     Returns:
       A list of all commit hashes which match this one.
     """
     logging.info(
         f'Getting upstream hashes for commit {self._hash} ({self.subject})'
     )
     if (m := UPSTREAM_COMMIT_PATTERN_1.search(self.message)) is not None:
       logging.info('Matched upstream pattern 1')
       return [m.group(1)]
     elif (m := UPSTREAM_COMMIT_PATTERN_2.search(self.message)) is not None:
       logging.info('Matched upstream pattern 2')
       return [m.group(1)]
     elif subject_to_commit_map:
       logging.info(f'Checking subject: {self.subject}')
       return subject_to_commit_map.get(self.subject, [])
     else:
       return []


 def git_cherry(upstream: str, head: str, limit: str) -> list[Commit]:
   """Gets commits corresponding to the output of `git cherry -v upstream head limit`

   See https://git-scm.com/docs/git-cherry for more information.
   """
   commits = []

   with Popen(
       ['git', 'cherry', '-v', upstream, head, limit], stdout=PIPE
   ) as commit_lines:
     for line in commit_lines.stdout:
       try:
         line = line.decode('utf-8').strip()
       except Exception as e:
         logging.warning(e)
         continue

       has_exact_content_match_str, _hash, subject = line.split(maxsplit=2)
       has_exact_content_match = has_exact_content_match_str == '-'
       message = _get_commit_message(_hash)

       commit = Commit(
           _hash=_hash,
           subject=subject,
           message=message,
           has_exact_content_match=has_exact_content_match,
       )

       commits.append(commit)

   return commits


 def _get_commit_message(commit_hash: str) -> str:
   return run(
       ['git', 'log', '--format=%B', '-n', '1', commit_hash], stdout=PIPE
   ).stdout.decode('utf-8')


 def get_subject_to_commit_map(revision_range: str) -> dict[str, list[str]]:
   """Gets the map from commit subjects to their hashes for a revision range.

   Args:
     revision_range: The range of commits whose subjects will be loaded. The
       interpretation is the same as `git log`, i.e., <commit1>..<commit2>
       will load the commits between commit1 and commit2.

   Returns:
     A dict from commit subjects to lists of hashes of commits with that
     subject. Most subjects will only have one corresponding hash, but
     sometimes two unrelated commits may have the same subject.
   """
   mapping = {}
   with Popen(
       ['git', 'log', '--format=%H %s', revision_range], stdout=PIPE
   ) as commit_lines:
     for line in commit_lines.stdout:
       try:
         line = line.decode('utf-8').strip()
       except Exception as e:
         logging.warning(e)
         continue

       if (m := COMMIT_SUBJECT_RE.match(line)) is not None:
         commit, subject = m.groups()
         # A subject can have multiple commits, so we return them all.
         commits = mapping.get(subject, [])
         commits.append(commit)
         mapping[subject] = commits
   logging.info(f'mapped {len(mapping)} commit subjects')
   return mapping


 if __name__ == '__main__':
   logging.basicConfig(level=logging.INFO)

   parser = argparse.ArgumentParser(
       prog='get-non-upstreamed-commits',
       description=(
           "Finds commits which have been applied locally but haven't been"
           " applied upstream"
       ),
   )
   parser.add_argument('-u', '--upstream', type=str, required=True)
   parser.add_argument('--head', type=str, default='HEAD', required=False)
   parser.add_argument('-l', '--limit', type=str, required=True)
   parser.add_argument(
     '-s', '--subject-map-revision-range', type=str, required=False
   )
   args = parser.parse_args()

   logging.info('Finding COS-specific commits...')
   commits = git_cherry(args.upstream, args.head, args.limit)
   logging.info(f'Done. Found {len(commits)} commits.')

   logging.info('Getting map from commit subjects to commit hashes...')
   subject_map_revision_range = (
     args.subject_map_revision_range or f'{args.limit}..{args.upstream}'
   )
   subject_to_commit_map = get_subject_to_commit_map(subject_map_revision_range)
   logging.info('Done.')

   # Find the upstream commits for each commit, if present, and output the
   # results as a tab-separated table.
   sys.stdout.write(f'Hash\tUpstream Hash\tHas Exact Content Match\tSubject\n')
   for commit in commits:
     upstream_hashes = commit.get_upstream_commit_hashes(subject_to_commit_map)
     if len(upstream_hashes) > 1:
       logging.warning(
           f'Found multiple upstream commits for commit {commit._hash}:'
           f' {commit.subject}'
       )
     if not upstream_hashes:
       # We ouptut the string 'None' if we didn't find any matching commits.
       upstream_hashes = ['None']
     for upstream_hash in upstream_hashes:
       sys.stdout.write(
           f'{commit._hash}\t{upstream_hash}\t{commit.has_exact_content_match}\t{commit.subject}\n'
       )
	#!/usr/bin/env python3
	#
	# Copyright 2024 The ChromiumOS Authors
	# Use of this source code is governed by a BSD-style license that can be
	# found in the LICENSE file.

	# This script is a slightly fancier version of `git cherry`, which can be used
	# to find commits which have been applied locally but haven't been applied
	# upstream.
	#
	# Like `git cherry`, this script takes arguments for an upstream, a head, and
	# a limit. If the head argument is omitted, HEAD is used.
	#
	# The script will do the following:
	# 1. Get all of the commits which are present between limit and head.
	# 2. For each commit:
	# 1. Check if the commit is present upstream by:
	# 1. Checking if a commit with an identical hash is present upstream.
	# 2. Checking if a commit with the exact same contents is present
	# upstream.
	# 3. Checking if the there are upstream commit hashes in the commit
	# message.
	# 4. Checking for an upstream commit with the exact same subject line.
	# 2. Output the commit hash, the upstream commit hash (if present),
	# whether an exact content match was found, and the commit subject.
	#
	# The output will be a tab-separated table in commit order.
	#
	# In order for the upstream subject lookups to be fast, the commit history
	# between limit and upstream is read and the subjects are put into a hashmap.
	# This default <limit>..<upstream> commit range should almost always be fine,
	# but a different revision range can be passed via the
	# subject-map-revision-range argument if it's expected that there are relevant
	# commits elsewhere.
	#
	# Usage:
	# If you were upgrading from kernel version 6.1 to 6.6, you could run this
	# script as follows:
	#
	# python3 get_non_upstreamed_commits.py \
	# --upstream cos/upstream-6.6 \
	# --head cos-6.6 \
	# --limit cos/upstream-6.6 > commits.tsv
	#
	# Using the abbreviated arguments:
	#
	# python3 get_non_upstreamed_commits.py \
	# -u cos/upstream-6.6 -h cos-6.6 -l cos/upstream-6.6 > commits.tsv

	import argparse
	import dataclasses
	import logging
	import re
	from subprocess import PIPE, Popen, run
	import sys
	from typing import Mapping, Optional

	COMMIT_SUBJECT_RE = re.compile(r'(.?) (.)')
	UPSTREAM_COMMIT_PATTERN_1 = re.compile('Upstream commit.*?([a-f0-9]+)')
	UPSTREAM_COMMIT_PATTERN_2 = re.compile(r'UPSTREAM\(([a-f0-9]+)\)')


	@dataclasses.dataclass
	class Commit:
	"""Class representing information about a git commit."""

	_hash: str
	subject: str
	message: str
	has_exact_content_match: bool

	def get_upstream_commit_hashes(
	self,
	subject_to_commit_map: Mapping[str, list[str]],
	) -> list[str]:
	"""Gets the hashes of any upstream commits which "match" this one.

	A commit is determined to "match" this one if it is referred to as an
	upstream commit in the body of this commit, or if it has the same subject
	as this commit.

	Args:
	subject_to_commit_map: A mapping from commit subjects to a list of hashes
	for commits with that subject.

	Returns:
	A list of all commit hashes which match this one.
	"""
	logging.info(
	f'Getting upstream hashes for commit {self._hash} ({self.subject})'
	)
	if (m := UPSTREAM_COMMIT_PATTERN_1.search(self.message)) is not None:
	logging.info('Matched upstream pattern 1')
	return [m.group(1)]
	elif (m := UPSTREAM_COMMIT_PATTERN_2.search(self.message)) is not None:
	logging.info('Matched upstream pattern 2')
	return [m.group(1)]
	elif subject_to_commit_map:
	logging.info(f'Checking subject: {self.subject}')
	return subject_to_commit_map.get(self.subject, [])
	else:
	return []


	def git_cherry(upstream: str, head: str, limit: str) -> list[Commit]:
	"""Gets commits corresponding to the output of `git cherry -v upstream head limit`

	See https://git-scm.com/docs/git-cherry for more information.
	"""
	commits = []

	with Popen(
	['git', 'cherry', '-v', upstream, head, limit], stdout=PIPE
	) as commit_lines:
	for line in commit_lines.stdout:
	try:
	line = line.decode('utf-8').strip()
	except Exception as e:
	logging.warning(e)
	continue

	has_exact_content_match_str, _hash, subject = line.split(maxsplit=2)
	has_exact_content_match = has_exact_content_match_str == '-'
	message = _get_commit_message(_hash)

	commit = Commit(
	_hash=_hash,
	subject=subject,
	message=message,
	has_exact_content_match=has_exact_content_match,
	)

	commits.append(commit)

	return commits


	def _get_commit_message(commit_hash: str) -> str:
	return run(
	['git', 'log', '--format=%B', '-n', '1', commit_hash], stdout=PIPE
	).stdout.decode('utf-8')


	def get_subject_to_commit_map(revision_range: str) -> dict[str, list[str]]:
	"""Gets the map from commit subjects to their hashes for a revision range.

	Args:
	revision_range: The range of commits whose subjects will be loaded. The
	interpretation is the same as `git log`, i.e., <commit1>..<commit2>
	will load the commits between commit1 and commit2.

	Returns:
	A dict from commit subjects to lists of hashes of commits with that
	subject. Most subjects will only have one corresponding hash, but
	sometimes two unrelated commits may have the same subject.
	"""
	mapping = {}
	with Popen(
	['git', 'log', '--format=%H %s', revision_range], stdout=PIPE
	) as commit_lines:
	for line in commit_lines.stdout:
	try:
	line = line.decode('utf-8').strip()
	except Exception as e:
	logging.warning(e)
	continue

	if (m := COMMIT_SUBJECT_RE.match(line)) is not None:
	commit, subject = m.groups()
	# A subject can have multiple commits, so we return them all.
	commits = mapping.get(subject, [])
	commits.append(commit)
	mapping[subject] = commits
	logging.info(f'mapped {len(mapping)} commit subjects')
	return mapping


	if __name__ == '__main__':
	logging.basicConfig(level=logging.INFO)

	parser = argparse.ArgumentParser(
	prog='get-non-upstreamed-commits',
	description=(
	"Finds commits which have been applied locally but haven't been"
	" applied upstream"
	),
	)
	parser.add_argument('-u', '--upstream', type=str, required=True)
	parser.add_argument('--head', type=str, default='HEAD', required=False)
	parser.add_argument('-l', '--limit', type=str, required=True)
	parser.add_argument(
	'-s', '--subject-map-revision-range', type=str, required=False
	)
	args = parser.parse_args()

	logging.info('Finding COS-specific commits...')
	commits = git_cherry(args.upstream, args.head, args.limit)
	logging.info(f'Done. Found {len(commits)} commits.')

	logging.info('Getting map from commit subjects to commit hashes...')
	subject_map_revision_range = (
	args.subject_map_revision_range or f'{args.limit}..{args.upstream}'
	)
	subject_to_commit_map = get_subject_to_commit_map(subject_map_revision_range)
	logging.info('Done.')

	# Find the upstream commits for each commit, if present, and output the
	# results as a tab-separated table.
	sys.stdout.write(f'Hash\tUpstream Hash\tHas Exact Content Match\tSubject\n')
	for commit in commits:
	upstream_hashes = commit.get_upstream_commit_hashes(subject_to_commit_map)
	if len(upstream_hashes) > 1:
	logging.warning(
	f'Found multiple upstream commits for commit {commit._hash}:'
	f' {commit.subject}'
	)
	if not upstream_hashes:
	# We ouptut the string 'None' if we didn't find any matching commits.
	upstream_hashes = ['None']
	for upstream_hash in upstream_hashes:
	sys.stdout.write(
	f'{commit._hash}\t{upstream_hash}\t{commit.has_exact_content_match}\t{commit.subject}\n'
	)