19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/python/pants/core/goals/lint.py

1	# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
UNCOV 4	from __future__ import annotations	×
5
UNCOV 6	import logging	×
UNCOV 7	from collections import defaultdict	×
UNCOV 8	from collections.abc import Callable, Coroutine, Iterable	×
UNCOV 9	from dataclasses import dataclass	×
UNCOV 10	from typing import Any, ClassVar, Protocol, TypeVar, cast, final	×
11
UNCOV 12	from pants.base.specs import Specs	×
UNCOV 13	from pants.core.environments.rules import _warn_on_non_local_environments	×
UNCOV 14	from pants.core.goals.multi_tool_goal_helper import (	×
15	BatchSizeOption,
16	OnlyOption,
17	SkippableSubsystem,
18	determine_specified_tool_ids,
19	)
UNCOV 20	from pants.core.util_rules.partitions import PartitionElementT, PartitionerType, PartitionMetadataT	×
UNCOV 21	from pants.core.util_rules.partitions import Partitions as Partitions # re-export	×
UNCOV 22	from pants.core.util_rules.partitions import (	×
23	_BatchBase,
24	_PartitionFieldSetsRequestBase,
25	_PartitionFilesRequestBase,
26	)
UNCOV 27	from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType	×
UNCOV 28	from pants.engine.environment import EnvironmentName	×
UNCOV 29	from pants.engine.fs import EMPTY_DIGEST, Digest	×
UNCOV 30	from pants.engine.goal import Goal, GoalSubsystem	×
UNCOV 31	from pants.engine.internals.graph import filter_targets	×
UNCOV 32	from pants.engine.internals.specs_rules import resolve_specs_paths	×
UNCOV 33	from pants.engine.process import FallibleProcessResult	×
UNCOV 34	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
UNCOV 35	from pants.engine.target import FieldSet	×
UNCOV 36	from pants.engine.unions import UnionMembership, UnionRule, distinct_union_type_per_subclass, union	×
UNCOV 37	from pants.option.option_types import BoolOption	×
UNCOV 38	from pants.util.docutil import bin_name, doc_url	×
UNCOV 39	from pants.util.logging import LogLevel	×
UNCOV 40	from pants.util.meta import classproperty	×
UNCOV 41	from pants.util.strutil import Simplifier, softwrap	×
42
UNCOV 43	logger = logging.getLogger(__name__)	×
44
UNCOV 45	_T = TypeVar("_T")	×
UNCOV 46	_FieldSetT = TypeVar("_FieldSetT", bound=FieldSet)	×
47
48
UNCOV 49	@dataclass(frozen=True)	×
UNCOV 50	class LintResult(EngineAwareReturnType):	×
UNCOV 51	exit_code: int	×
UNCOV 52	stdout: str	×
UNCOV 53	stderr: str	×
UNCOV 54	linter_name: str	×
UNCOV 55	partition_description: str \| None = None	×
UNCOV 56	report: Digest = EMPTY_DIGEST	×
UNCOV 57	_render_message: bool = True	×
58
UNCOV 59	@classmethod	×
UNCOV 60	def create(	×
61	cls,
62	request: AbstractLintRequest.Batch,
63	process_result: FallibleProcessResult,
64	*,
65	output_simplifier: Simplifier = Simplifier(),
66	report: Digest = EMPTY_DIGEST,
67	) -> LintResult:
68	return cls(	×
69	exit_code=process_result.exit_code,
70	stdout=output_simplifier.simplify(process_result.stdout),
71	stderr=output_simplifier.simplify(process_result.stderr),
72	linter_name=request.tool_name,
73	partition_description=request.partition_metadata.description,
74	report=report,
75	)
76
UNCOV 77	def metadata(self) -> dict[str, Any]:	×
78	return {"partition": self.partition_description}	×
79
UNCOV 80	def level(self) -> LogLevel \| None:	×
UNCOV 81	if not self._render_message:	×
82	return LogLevel.TRACE	×
UNCOV 83	if self.exit_code != 0:	×
UNCOV 84	return LogLevel.ERROR	×
UNCOV 85	return LogLevel.INFO	×
86
UNCOV 87	def message(self) -> str \| None:	×
UNCOV 88	message = self.linter_name	×
UNCOV 89	message += (	×
90	" succeeded." if self.exit_code == 0 else f" failed (exit code {self.exit_code})."
91	)
UNCOV 92	if self.partition_description:	×
UNCOV 93	message += f"\nPartition: {self.partition_description}"	×
UNCOV 94	if self.stdout:	×
UNCOV 95	message += f"\n{self.stdout}"	×
UNCOV 96	if self.stderr:	×
UNCOV 97	message += f"\n{self.stderr}"	×
UNCOV 98	if self.partition_description or self.stdout or self.stderr:	×
UNCOV 99	message += "\n\n"	×
100
UNCOV 101	return message	×
102
UNCOV 103	def cacheable(self) -> bool:	×
104	"""Is marked uncacheable to ensure that it always renders."""
105	return False	×
106
107
UNCOV 108	@union	×
UNCOV 109	class AbstractLintRequest:	×
110	"""Base class for plugin types wanting to be run as part of `lint`.
111
112	Plugins should define a new type which subclasses either `LintTargetsRequest` (to lint targets)
113	or `LintFilesRequest` (to lint arbitrary files), and set the appropriate class variables.
114	E.g.
115	class DryCleaningRequest(LintTargetsRequest):
116	name = DryCleaningSubsystem.options_scope
117	field_set_type = DryCleaningFieldSet
118
119	Then, define 2 `@rule`s:
120	1. A rule which takes an instance of your request type's `PartitionRequest` class property,
121	and returns a `Partitions` instance.
122	E.g.
123	@rule
124	async def partition(
125	request: DryCleaningRequest.PartitionRequest[DryCleaningFieldSet]
126	# or `request: DryCleaningRequest.PartitionRequest` if file linter
127	subsystem: DryCleaningSubsystem,
128	) -> Partitions[DryCleaningFieldSet, Any]:
129	if subsystem.skip:
130	return Partitions()
131
132	# One possible implementation
133	return Partitions.single_partition(request.field_sets)
134
135	2. A rule which takes an instance of your request type's `Batch` class property, and
136	returns a `LintResult instance.
137	E.g.
138	@rule
139	async def dry_clean(
140	request: DryCleaningRequest.Batch,
141	) -> LintResult:
142	...
143
144	Lastly, register the rules which tell Pants about your plugin.
145	E.g.
146	def rules():
147	return [
148	*collect_rules(),
149	*DryCleaningRequest.rules()
150	]
151	"""
152
UNCOV 153	tool_subsystem: ClassVar[type[SkippableSubsystem]]	×
UNCOV 154	partitioner_type: ClassVar[PartitionerType] = PartitionerType.CUSTOM	×
155
UNCOV 156	is_formatter: ClassVar[bool] = False	×
UNCOV 157	is_fixer: ClassVar[bool] = False	×
158
UNCOV 159	@final	×
UNCOV 160	@classproperty	×
UNCOV 161	def _requires_snapshot(cls) -> bool:	×
UNCOV 162	return cls.is_formatter or cls.is_fixer	×
163
UNCOV 164	@classproperty	×
UNCOV 165	def tool_name(cls) -> str:	×
166	"""The user-facing "name" of the tool."""
167	return cls.tool_subsystem.options_scope	×
168
UNCOV 169	@classproperty	×
UNCOV 170	def tool_id(cls) -> str:	×
171	"""The "id" of the tool, used in tool selection (Eg --only=<id>)."""
172	return cls.tool_subsystem.options_scope	×
173
UNCOV 174	@distinct_union_type_per_subclass(in_scope_types=[EnvironmentName])	×
UNCOV 175	class Batch(_BatchBase[PartitionElementT, PartitionMetadataT]):	×
UNCOV 176	pass	×
177
UNCOV 178	@final	×
UNCOV 179	@classmethod	×
UNCOV 180	def rules(cls) -> Iterable:	×
UNCOV 181	yield from cls._get_rules()	×
182
UNCOV 183	@classmethod	×
UNCOV 184	def _get_rules(cls) -> Iterable:	×
UNCOV 185	yield UnionRule(AbstractLintRequest, cls)	×
UNCOV 186	yield UnionRule(AbstractLintRequest.Batch, cls.Batch)	×
187
188
UNCOV 189	class LintTargetsRequest(AbstractLintRequest):	×
190	"""The entry point for linters that operate on targets."""
191
UNCOV 192	field_set_type: ClassVar[type[FieldSet]]	×
193
UNCOV 194	@distinct_union_type_per_subclass(in_scope_types=[EnvironmentName])	×
UNCOV 195	class PartitionRequest(_PartitionFieldSetsRequestBase[_FieldSetT]):	×
UNCOV 196	pass	×
197
UNCOV 198	@classmethod	×
UNCOV 199	def _get_rules(cls) -> Iterable:	×
UNCOV 200	yield from cls.partitioner_type.default_rules(cls, by_file=False)	×
UNCOV 201	yield from super()._get_rules()	×
UNCOV 202	yield UnionRule(LintTargetsRequest.PartitionRequest, cls.PartitionRequest)	×
203
204
UNCOV 205	class LintFilesRequest(AbstractLintRequest, EngineAwareParameter):	×
206	"""The entry point for linters that do not use targets."""
207
UNCOV 208	@distinct_union_type_per_subclass(in_scope_types=[EnvironmentName])	×
UNCOV 209	class PartitionRequest(_PartitionFilesRequestBase):	×
UNCOV 210	pass	×
211
UNCOV 212	@classmethod	×
UNCOV 213	def _get_rules(cls) -> Iterable:	×
UNCOV 214	if cls.partitioner_type is not PartitionerType.CUSTOM:	×
215	raise ValueError(	×
216	"Pants does not provide default partitioners for `LintFilesRequest`."
217	+ " You will need to provide your own partitioner rule."
218	)
219
UNCOV 220	yield from super()._get_rules()	×
UNCOV 221	yield UnionRule(LintFilesRequest.PartitionRequest, cls.PartitionRequest)	×
222
223
224	# If a user wants linter reports to show up in dist/ they must ensure that the reports
225	# are written under this directory. E.g.,
226	# ./pants --flake8-args="--output-file=reports/report.txt" lint <target>
UNCOV 227	REPORT_DIR = "reports"	×
228
229
UNCOV 230	class LintSubsystem(GoalSubsystem):	×
UNCOV 231	name = "lint"	×
UNCOV 232	help = softwrap(	×
233	f"""
234	Run linters/formatters/fixers in check mode.
235
236	This goal runs tools that check code quality/styling etc, without changing that code. This
237	includes running formatters and fixers, but instead of writing changes back to the
238	workspace, Pants treats any changes they would make as a linting failure.
239
240	See also:
241
242	- [The `fmt` goal]({doc_url("reference/goals/fix")} will save the the result of formatters
243	(code-editing tools that make only "syntactic" changes) back to the workspace.
244
245	- [The `fmt` goal]({doc_url("reference/goals/fix")} will save the the result of fixers
246	(code-editing tools that may make "semantic" changes too) back to the workspace.
247
248	- Documentation about linters for various ecosystems, such as:
249	[Python]({doc_url("docs/python/overview/linters-and-formatters")}), [Go]({doc_url("docs/go")}),
250	[JVM]({doc_url("jvm/java-and-scala#lint-and-format")}), [Shell]({doc_url("docs/shell")}),
251	[Docker]({doc_url("docs/docker#linting-dockerfiles-with-hadolint")}).
252
253	"""
254	)
255
UNCOV 256	@classmethod	×
UNCOV 257	def activated(cls, union_membership: UnionMembership) -> bool:	×
258	return AbstractLintRequest in union_membership	×
259
UNCOV 260	only = OnlyOption("linter", "flake8", "shellcheck")	×
UNCOV 261	skip_formatters = BoolOption(	×
262	default=False,
263	help=softwrap(
264	f"""
265	If true, skip running all formatters in check-only mode.
266
267	FYI: when running `{bin_name()} fmt lint ::`, there should be diminishing performance
268	benefit to using this flag. Pants attempts to reuse the results from `fmt` when running
269	`lint` where possible.
270	"""
271	),
272	)
UNCOV 273	skip_fixers = BoolOption(	×
274	default=False,
275	help=softwrap(
276	f"""
277	If true, skip running all fixers in check-only mode.
278
279	FYI: when running `{bin_name()} fix lint ::`, there should be diminishing performance
280	benefit to using this flag. Pants attempts to reuse the results from `fix` when running
281	`lint` where possible.
282	"""
283	),
284	)
UNCOV 285	batch_size = BatchSizeOption(uppercase="Linter", lowercase="linter")	×
286
287
UNCOV 288	class Lint(Goal):	×
UNCOV 289	subsystem_cls = LintSubsystem	×
UNCOV 290	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	×
291
292
UNCOV 293	_CoreRequestType = TypeVar("_CoreRequestType", bound=AbstractLintRequest)	×
UNCOV 294	_TargetPartitioner = TypeVar("_TargetPartitioner", bound=LintTargetsRequest.PartitionRequest)	×
UNCOV 295	_FilePartitioner = TypeVar("_FilePartitioner", bound=LintFilesRequest.PartitionRequest)	×
296
297
UNCOV 298	class _MultiToolGoalSubsystem(Protocol):	×
UNCOV 299	name: str	×
UNCOV 300	only: OnlyOption	×
301
302
UNCOV 303	async def get_partitions_by_request_type(	×
304	core_request_types: Iterable[type[_CoreRequestType]],
305	target_partitioners: Iterable[type[_TargetPartitioner]],
306	file_partitioners: Iterable[type[_FilePartitioner]],
307	subsystem: _MultiToolGoalSubsystem,
308	specs: Specs,
309	# NB: Because the rule parser code will collect rule calls from caller's scope, these allow the
310	# caller to customize the specific rule.
311	make_targets_partition_request: Callable[[_TargetPartitioner], Coroutine[Any, Any, Partitions]],
312	make_files_partition_request: Callable[[_FilePartitioner], Coroutine[Any, Any, Partitions]],
313	) -> dict[type[_CoreRequestType], list[Partitions]]:
UNCOV 314	specified_ids = determine_specified_tool_ids(	×
315	subsystem.name,
316	subsystem.only, # type: ignore[arg-type]
317	core_request_types,
318	)
319
UNCOV 320	filtered_core_request_types = [	×
321	request_type for request_type in core_request_types if request_type.tool_id in specified_ids
322	]
UNCOV 323	if not filtered_core_request_types:	×
324	return {}	×
325
UNCOV 326	core_partition_request_types = {	×
327	getattr(request_type, "PartitionRequest") for request_type in filtered_core_request_types
328	}
UNCOV 329	target_partitioners = [	×
330	target_partitioner
331	for target_partitioner in target_partitioners
332	if target_partitioner in core_partition_request_types
333	]
UNCOV 334	file_partitioners = [	×
335	file_partitioner
336	for file_partitioner in file_partitioners
337	if file_partitioner in core_partition_request_types
338	]
339
UNCOV 340	_get_targets = filter_targets(	×
341	**implicitly({(specs if target_partitioners else Specs.empty()): Specs})
342	)
UNCOV 343	_get_specs_paths = resolve_specs_paths(specs if file_partitioners else Specs.empty())	×
344
UNCOV 345	targets, specs_paths = await concurrently(_get_targets, _get_specs_paths)	×
346
UNCOV 347	await _warn_on_non_local_environments(targets, f"the {subsystem.name} goal")	×
348
UNCOV 349	def partition_request_get(	×
350	request_type: type[AbstractLintRequest],
351	) -> Coroutine[Any, Any, Partitions]:
UNCOV 352	partition_request_type: type = getattr(request_type, "PartitionRequest")	×
UNCOV 353	if partition_request_type in target_partitioners:	×
UNCOV 354	partition_targets_type = cast(LintTargetsRequest, request_type)	×
UNCOV 355	field_set_type = partition_targets_type.field_set_type	×
UNCOV 356	field_sets = tuple(	×
357	field_set_type.create(target)
358	for target in targets
359	if field_set_type.is_applicable(target)
360	)
UNCOV 361	return make_targets_partition_request(	×
362	partition_targets_type.PartitionRequest(field_sets) # type: ignore[arg-type]
363	)
364	else:
UNCOV 365	assert partition_request_type in file_partitioners	×
UNCOV 366	partition_files_type = cast(LintFilesRequest, request_type)	×
UNCOV 367	return make_files_partition_request(	×
368	partition_files_type.PartitionRequest(specs_paths.files) # type: ignore[arg-type]
369	)
370
UNCOV 371	all_partitions = await concurrently(	×
372	partition_request_get(request_type) for request_type in filtered_core_request_types
373	)
UNCOV 374	partitions_by_request_type = defaultdict(list)	×
UNCOV 375	for request_type, partition in zip(filtered_core_request_types, all_partitions):	×
UNCOV 376	partitions_by_request_type[request_type].append(partition)	×
377
UNCOV 378	return partitions_by_request_type	×
379
380
UNCOV 381	@rule(polymorphic=True)	×
UNCOV 382	async def partition_targets(req: LintTargetsRequest.PartitionRequest) -> Partitions:	×
383	raise NotImplementedError()	×
384
385
UNCOV 386	@rule(polymorphic=True)	×
UNCOV 387	async def partition_files(req: LintFilesRequest.PartitionRequest) -> Partitions:	×
388	raise NotImplementedError()	×
389
390
UNCOV 391	@rule(polymorphic=True)	×
UNCOV 392	async def lint_batch(batch: AbstractLintRequest.Batch) -> LintResult:	×
393	raise NotImplementedError()	×
394
395
UNCOV 396	def rules():	×
UNCOV 397	return collect_rules()	×

pantsbuild / pants / 19015773527

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous