recipe_modules/cq_deps/api.py - infra/recipes - Git at Google

 # 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.
 """Find dependencies on pending CLs."""

 import base64
 import json
 import re
 from typing import Any, Generator, Literal, Sequence

 import attrs
 from PB.recipe_modules.pigweed.cq_deps.properties import InputProperties
 from recipe_engine import recipe_api

 _PATCHES_FILE = 'patches.json'


 @attrs.define
 class Change:
     """Details of a matching change.

     This is meant to match the GerritChange message defined at
     https://source.chromium.org/chromium/infra/infra/+/HEAD:go/src/go.chromium.org/luci/buildbucket/proto/common.proto?q=GerritChange
     A few fields have been added for use by this module.
     """

     gerrit_name: str
     change: int = attrs.field(converter=int)
     project: str | None = None
     patchset: str | None = None
     status: str | None = None
     commit: str | None = None

     @property
     def host(self) -> str:
         return f'{self.gerrit_name}-review.googlesource.com'

     @property
     def link(self) -> str:
         return f'https://{self.host}/{self.change}'

     @property
     def name(self) -> str:
         return f'{self.gerrit_name}:{self.change}'

     @property
     def remote(self) -> str:
         return f'https://{self.gerrit_name}.googlesource.com/{self.project}'

     @property
     def gerrit_url(self) -> str:
         return f'https://{self.gerrit_name}-review.googlesource.com/c/{self.change}'

     def __str__(self) -> str:
         return self.name


 @attrs.define
 class CqDepsResult:
     """Result of resolve(), containing references that were found or not."""

     resolved: tuple[Change]
     unresolved: tuple[Change]


 ChangeState = Literal['NEW', 'MERGED', 'ABANDONED']


 class CqDepsApi(recipe_api.RecipeApi):
     """Find dependencies on pending CLs."""

     ChangeState = ChangeState
     Change = Change
     Result = CqDepsResult

     def __init__(self, props: InputProperties, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._details = {}
         self._patches_json_enabled: bool = not props.patches_json_disabled
         self._topics_enabled: bool = props.topics_enabled

     def _lookup_cl(
         self, gerrit_name: str, commit: str
     ) -> list[dict[str, Any]] | None:
         query_results = self.m.gerrit.change_query(
             f'number {commit}',
             f'commit:{commit}',
             host=f'{gerrit_name}-review.googlesource.com',
             max_attempts=2,
             timeout=30,
             # The strange value for _number is to ensure different hashes result
             # in different CL numbers by default. This way we don't accidentally
             # create a cycle by specifying a fixed default value.
             test_data=self.m.json.test_api.output(
                 [
                     {
                         '_number': sum(ord(x) for x in commit),
                     }
                 ]
             ),
         ).json.output
         if query_results:
             return query_results[0]
         return None

     def resolve(
         self,
         gerrit_name: str,
         number: int | str,
         topic: str | None = None,
     ) -> CqDepsResult:
         """Recursively resolve dependencies of a change.

         Resolve dependencies of a change. Depending on the module properties,
         this might be by parsing a 'patches.json' file in the change or by
         looking up other changes in the same topic.

         The two systems are evaluated independently. Topics of patches found via
         'patches.json' are not evaluated, and 'patches.json' files on other
         changes in the same topic are not evaluated.

         Args:
             gerrit_name: Name of Gerrit host of initial change.
             number: Gerrit number or git commit hash of initial change.
             topic: Gerrit topic containing this the change, if any.

         Returns resolved Change objects (excluding the given change) and
         unresolved Change objects (which are not completely populated).
         """

         with self.m.step.nest('resolve CL deps', status='last') as pres:
             with self.m.context(infra_steps=True):
                 if isinstance(number, str):
                     cl = self._lookup_cl(gerrit_name, number)
                     if not cl:
                         return CqDepsResult((), ())
                     number: int = cl['_number']

                 seen: set[str] = set()
                 deps: dict[str, Change] = {}
                 unresolved: list[str] = []
                 initial_change = Change(gerrit_name, number)
                 to_process: list[str] = [initial_change]

                 # Want all initial details calls to come from this step level so
                 # their step names are predictable. Further calls can be made
                 # anywhere and will hit the cache.
                 _ = self._change_details(initial_change)

                 if self._patches_json_enabled:
                     while to_process:
                         current: Change = to_process.pop()
                         deps[current.name] = current
                         seen.add(current.name)

                         for dep in self._resolve(current):
                             if dep.name not in deps:
                                 details = self._change_details(dep)
                                 if not details:
                                     unresolved.append(dep)
                                     continue
                                 if details['status'] != 'NEW':
                                     continue

                                 if dep.name not in seen:
                                     to_process.append(dep)
                                 deps[dep.name] = dep

                 if not topic:
                     self.m.step.empty('no topic')

                 else:
                     if self._topics_enabled:
                         assert '"' not in topic, f'invalid topic: {topic!r}'
                         changes = self.m.gerrit.change_query(
                             f'topic {topic}',
                             f'topic:"{topic}" status:new',
                             host=f'{gerrit_name}-review.googlesource.com',
                             max_attempts=2,
                             timeout=30,
                             # Always include the current change as well as the
                             # "next" one. We would only get here if there was a
                             # topic set, so always assume there's another change
                             # in the topic in testing.
                             test_data=self.m.json.test_api.output(
                                 [
                                     {
                                         '_number': number,
                                     },
                                     {
                                         '_number': number + 1,
                                     },
                                 ]
                             ),
                         ).json.output

                         for change in changes:
                             dep = Change(gerrit_name, change['_number'])
                             deps[dep.name] = dep

                     else:
                         self.m.step.empty('topic set but topics not enabled')

                 if initial_change.name in deps:
                     del deps[initial_change.name]

                 for dep in deps.values():
                     pres.links[dep.name] = dep.link

                 # Make sure the last step passes because _change_details might
                 # fail for an individual CL in ways we don't care about.
                 with self.m.step.nest('pass'):
                     pass

                 return CqDepsResult(deps.values(), unresolved)

     def _parse_deps(self, change: Change) -> Generator[Change, None, None]:
         """Parse dependencies from patches.json in a change.

         For a change with the following patches.json file,

             [
                 {"gerrit_name": "pigweed", "number": 1001},
                 {"gerrit_name": "pigweed", "number": 1002},
                 {"gerrit_name": "other", "number": 1003}
             ]

         the return value would look like

             [
                 Change("pigweed", 1001),
                 Change("pigweed", 1002),
                 Change("other", 1003),
             ]

         Args:
             change (Change): Change object from which to retrieve dependencies

         Returns a list of direct dependencies of the given change.
         """

         details = self._change_details(change)
         if not details:
             return []  # pragma: no cover
         host = self.m.gerrit.normalize_host(change.host)
         host = host.replace('-review.', '.')
         current_revision = details['revisions'][details['current_revision']]

         # If there is no _PATCHES_FILE in the CL, don't try to retrieve it.
         if _PATCHES_FILE not in current_revision.get('files', ()):
             return []

         # If _PATCHES_FILE is being deleted, don't try to retrieve it.
         if current_revision['files'][_PATCHES_FILE].get('status', 'M') == 'D':
             return []  # pragma: no cover

         cl = details['_number']
         ps = current_revision['_number']

         raw_data = self.m.gitiles.fetch(
             host,
             details['project'],
             _PATCHES_FILE,
             'refs/changes/{:02}/{}/{}'.format(cl % 100, cl, ps),
             step_name=f'fetch {_PATCHES_FILE}',
         )

         if not raw_data:
             raise self.m.step.InfraFailure(
                 f'{_PATCHES_FILE} is in CL but failed to retrieve it'
             )

         deps = json.loads(raw_data)

         for dep in deps:
             yield Change(dep['gerrit_name'], dep['number'])

     def _resolve(self, change: Change) -> list[Change]:
         """Resolve dependencies of the given change.

         Retrieve and parse the patches.json file of the given change. Does not
         attempt to verify if these dependent changes are accessible or even
         valid.

         Args:
             change (Change): Change to resolve dependencies for.

         Returns a list of changes on which the given change depends.
         """

         with self.m.step.nest(f'resolve deps for {change}') as pres:
             deps = []
             for dep in self._parse_deps(change):
                 deps.append(dep)
                 pres.links[dep.name] = dep.link
             if not deps:
                 pres.step_summary_text = 'no dependencies'
             return deps

     def _change_details(
         self, change: Change, *, test_status: ChangeState = 'NEW'
     ) -> dict[str, Any]:
         if change.name in self._details:
             return self._details[change.name]

         try:
             step = self.m.gerrit.change_details(
                 f'details {change}',
                 change_id=str(change.change),
                 host=change.host,
                 query_params=[
                     'CURRENT_COMMIT',
                     'CURRENT_FILES',
                     'CURRENT_REVISION',
                 ],
                 max_attempts=2,
                 timeout=30,
                 test_data=self.m.json.test_api.output(
                     {
                         '_number': change.change,
                         'current_revision': 'HASH',
                         'revisions': {
                             'HASH': {
                                 '_number': 1,
                                 'commit': {'message': ''},
                                 'files': {
                                     'filename': {},
                                 },
                             },
                         },
                         'project': 'project',
                         'status': test_status,
                     }
                 ),
             )
             details: dict[str, Any] = step.json.output or {}

         except self.m.step.StepFailure:
             details: dict[str, Any] = {}

         self._details[change.name] = details
         if details:
             change.project = details['project']
             current_revision = details['revisions'][details['current_revision']]
             change.patchset = current_revision['_number']
             change.status = details['status']
             change.commit = details['current_revision']

         return self._details[change.name]
	# 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.
	"""Find dependencies on pending CLs."""

	import base64
	import json
	import re
	from typing import Any, Generator, Literal, Sequence

	import attrs
	from PB.recipe_modules.pigweed.cq_deps.properties import InputProperties
	from recipe_engine import recipe_api

	_PATCHES_FILE = 'patches.json'


	@attrs.define
	class Change:
	"""Details of a matching change.

	This is meant to match the GerritChange message defined at
	https://source.chromium.org/chromium/infra/infra/+/HEAD:go/src/go.chromium.org/luci/buildbucket/proto/common.proto?q=GerritChange
	A few fields have been added for use by this module.
	"""

	gerrit_name: str
	change: int = attrs.field(converter=int)
	project: str \| None = None
	patchset: str \| None = None
	status: str \| None = None
	commit: str \| None = None

	@property
	def host(self) -> str:
	return f'{self.gerrit_name}-review.googlesource.com'

	@property
	def link(self) -> str:
	return f'https://{self.host}/{self.change}'

	@property
	def name(self) -> str:
	return f'{self.gerrit_name}:{self.change}'

	@property
	def remote(self) -> str:
	return f'https://{self.gerrit_name}.googlesource.com/{self.project}'

	@property
	def gerrit_url(self) -> str:
	return f'https://{self.gerrit_name}-review.googlesource.com/c/{self.change}'

	def __str__(self) -> str:
	return self.name


	@attrs.define
	class CqDepsResult:
	"""Result of resolve(), containing references that were found or not."""

	resolved: tuple[Change]
	unresolved: tuple[Change]


	ChangeState = Literal['NEW', 'MERGED', 'ABANDONED']


	class CqDepsApi(recipe_api.RecipeApi):
	"""Find dependencies on pending CLs."""

	ChangeState = ChangeState
	Change = Change
	Result = CqDepsResult

	def __init__(self, props: InputProperties, args, *kwargs):
	super().__init__(args, *kwargs)
	self._details = {}
	self._patches_json_enabled: bool = not props.patches_json_disabled
	self._topics_enabled: bool = props.topics_enabled

	def _lookup_cl(
	self, gerrit_name: str, commit: str
	) -> list[dict[str, Any]] \| None:
	query_results = self.m.gerrit.change_query(
	f'number {commit}',
	f'commit:{commit}',
	host=f'{gerrit_name}-review.googlesource.com',
	max_attempts=2,
	timeout=30,
	# The strange value for _number is to ensure different hashes result
	# in different CL numbers by default. This way we don't accidentally
	# create a cycle by specifying a fixed default value.
	test_data=self.m.json.test_api.output(
	[
	{
	'_number': sum(ord(x) for x in commit),
	}
	]
	),
	).json.output
	if query_results:
	return query_results[0]
	return None

	def resolve(
	self,
	gerrit_name: str,
	number: int \| str,
	topic: str \| None = None,
	) -> CqDepsResult:
	"""Recursively resolve dependencies of a change.

	Resolve dependencies of a change. Depending on the module properties,
	this might be by parsing a 'patches.json' file in the change or by
	looking up other changes in the same topic.

	The two systems are evaluated independently. Topics of patches found via
	'patches.json' are not evaluated, and 'patches.json' files on other
	changes in the same topic are not evaluated.

	Args:
	gerrit_name: Name of Gerrit host of initial change.
	number: Gerrit number or git commit hash of initial change.
	topic: Gerrit topic containing this the change, if any.

	Returns resolved Change objects (excluding the given change) and
	unresolved Change objects (which are not completely populated).
	"""

	with self.m.step.nest('resolve CL deps', status='last') as pres:
	with self.m.context(infra_steps=True):
	if isinstance(number, str):
	cl = self._lookup_cl(gerrit_name, number)
	if not cl:
	return CqDepsResult((), ())
	number: int = cl['_number']

	seen: set[str] = set()
	deps: dict[str, Change] = {}
	unresolved: list[str] = []
	initial_change = Change(gerrit_name, number)
	to_process: list[str] = [initial_change]

	# Want all initial details calls to come from this step level so
	# their step names are predictable. Further calls can be made
	# anywhere and will hit the cache.
	_ = self._change_details(initial_change)

	if self._patches_json_enabled:
	while to_process:
	current: Change = to_process.pop()
	deps[current.name] = current
	seen.add(current.name)

	for dep in self._resolve(current):
	if dep.name not in deps:
	details = self._change_details(dep)
	if not details:
	unresolved.append(dep)
	continue
	if details['status'] != 'NEW':
	continue

	if dep.name not in seen:
	to_process.append(dep)
	deps[dep.name] = dep

	if not topic:
	self.m.step.empty('no topic')

	else:
	if self._topics_enabled:
	assert '"' not in topic, f'invalid topic: {topic!r}'
	changes = self.m.gerrit.change_query(
	f'topic {topic}',
	f'topic:"{topic}" status:new',
	host=f'{gerrit_name}-review.googlesource.com',
	max_attempts=2,
	timeout=30,
	# Always include the current change as well as the
	# "next" one. We would only get here if there was a
	# topic set, so always assume there's another change
	# in the topic in testing.
	test_data=self.m.json.test_api.output(
	[
	{
	'_number': number,
	},
	{
	'_number': number + 1,
	},
	]
	),
	).json.output

	for change in changes:
	dep = Change(gerrit_name, change['_number'])
	deps[dep.name] = dep

	else:
	self.m.step.empty('topic set but topics not enabled')

	if initial_change.name in deps:
	del deps[initial_change.name]

	for dep in deps.values():
	pres.links[dep.name] = dep.link

	# Make sure the last step passes because _change_details might
	# fail for an individual CL in ways we don't care about.
	with self.m.step.nest('pass'):
	pass

	return CqDepsResult(deps.values(), unresolved)

	def _parse_deps(self, change: Change) -> Generator[Change, None, None]:
	"""Parse dependencies from patches.json in a change.

	For a change with the following patches.json file,

	[
	{"gerrit_name": "pigweed", "number": 1001},
	{"gerrit_name": "pigweed", "number": 1002},
	{"gerrit_name": "other", "number": 1003}
	]

	the return value would look like

	[
	Change("pigweed", 1001),
	Change("pigweed", 1002),
	Change("other", 1003),
	]

	Args:
	change (Change): Change object from which to retrieve dependencies

	Returns a list of direct dependencies of the given change.
	"""

	details = self._change_details(change)
	if not details:
	return [] # pragma: no cover
	host = self.m.gerrit.normalize_host(change.host)
	host = host.replace('-review.', '.')
	current_revision = details['revisions'][details['current_revision']]

	# If there is no _PATCHES_FILE in the CL, don't try to retrieve it.
	if _PATCHES_FILE not in current_revision.get('files', ()):
	return []

	# If _PATCHES_FILE is being deleted, don't try to retrieve it.
	if current_revision['files'][_PATCHES_FILE].get('status', 'M') == 'D':
	return [] # pragma: no cover

	cl = details['_number']
	ps = current_revision['_number']

	raw_data = self.m.gitiles.fetch(
	host,
	details['project'],
	_PATCHES_FILE,
	'refs/changes/{:02}/{}/{}'.format(cl % 100, cl, ps),
	step_name=f'fetch {_PATCHES_FILE}',
	)

	if not raw_data:
	raise self.m.step.InfraFailure(
	f'{_PATCHES_FILE} is in CL but failed to retrieve it'
	)

	deps = json.loads(raw_data)

	for dep in deps:
	yield Change(dep['gerrit_name'], dep['number'])

	def _resolve(self, change: Change) -> list[Change]:
	"""Resolve dependencies of the given change.

	Retrieve and parse the patches.json file of the given change. Does not
	attempt to verify if these dependent changes are accessible or even
	valid.

	Args:
	change (Change): Change to resolve dependencies for.

	Returns a list of changes on which the given change depends.
	"""

	with self.m.step.nest(f'resolve deps for {change}') as pres:
	deps = []
	for dep in self._parse_deps(change):
	deps.append(dep)
	pres.links[dep.name] = dep.link
	if not deps:
	pres.step_summary_text = 'no dependencies'
	return deps

	def _change_details(
	self, change: Change, *, test_status: ChangeState = 'NEW'
	) -> dict[str, Any]:
	if change.name in self._details:
	return self._details[change.name]

	try:
	step = self.m.gerrit.change_details(
	f'details {change}',
	change_id=str(change.change),
	host=change.host,
	query_params=[
	'CURRENT_COMMIT',
	'CURRENT_FILES',
	'CURRENT_REVISION',
	],
	max_attempts=2,
	timeout=30,
	test_data=self.m.json.test_api.output(
	{
	'_number': change.change,
	'current_revision': 'HASH',
	'revisions': {
	'HASH': {
	'_number': 1,
	'commit': {'message': ''},
	'files': {
	'filename': {},
	},
	},
	},
	'project': 'project',
	'status': test_status,
	}
	),
	)
	details: dict[str, Any] = step.json.output or {}

	except self.m.step.StepFailure:
	details: dict[str, Any] = {}

	self._details[change.name] = details
	if details:
	change.project = details['project']
	current_revision = details['revisions'][details['current_revision']]
	change.patchset = current_revision['_number']
	change.status = details['status']
	change.commit = details['current_revision']

	return self._details[change.name]