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/backend/python/lint/flake8/rules.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	from collections import defaultdict	×
7
UNCOV 8	from pants.backend.python.lint.flake8.subsystem import (	×
9	Flake8,
10	Flake8FieldSet,
11	Flake8FirstPartyPlugins,
12	)
UNCOV 13	from pants.backend.python.subsystems.setup import PythonSetup	×
UNCOV 14	from pants.backend.python.util_rules import pex	×
UNCOV 15	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	×
UNCOV 16	from pants.backend.python.util_rules.pex import VenvPexProcess, create_venv_pex	×
UNCOV 17	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	×
UNCOV 18	from pants.core.goals.lint import REPORT_DIR, LintResult, LintTargetsRequest, Partitions	×
UNCOV 19	from pants.core.util_rules.config_files import find_config_file	×
UNCOV 20	from pants.core.util_rules.partitions import Partition	×
UNCOV 21	from pants.core.util_rules.source_files import (	×
22	SourceFiles,
23	SourceFilesRequest,
24	determine_source_files,
25	)
UNCOV 26	from pants.engine.fs import CreateDigest, Directory, MergeDigests, PathGlobs, RemovePrefix	×
UNCOV 27	from pants.engine.intrinsics import (	×
28	create_digest,
29	execute_process,
30	merge_digests,
31	path_globs_to_digest,
32	remove_prefix,
33	)
UNCOV 34	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
UNCOV 35	from pants.util.logging import LogLevel	×
UNCOV 36	from pants.util.strutil import pluralize	×
37
38
UNCOV 39	class Flake8Request(LintTargetsRequest):	×
UNCOV 40	field_set_type = Flake8FieldSet	×
UNCOV 41	tool_subsystem = Flake8 # type: ignore[assignment]	×
42
43
UNCOV 44	def generate_argv(source_files: SourceFiles, flake8: Flake8) -> tuple[str, ...]:	×
45	args: list[str] = []	×
46	if flake8.config:	×
47	args.append(f"--config={flake8.config}")	×
48	args.append("--jobs={pants_concurrency}")	×
49	args.extend(flake8.args)	×
50	args.extend(source_files.files)	×
51	return tuple(args)	×
52
53
UNCOV 54	@rule	×
UNCOV 55	async def partition_flake8(	×
56	request: Flake8Request.PartitionRequest[Flake8FieldSet],
57	flake8: Flake8,
58	python_setup: PythonSetup,
59	first_party_plugins: Flake8FirstPartyPlugins,
60	) -> Partitions[Flake8FieldSet, InterpreterConstraints]:
61	if flake8.skip:	×
62	return Partitions()	×
63
64	results: dict[InterpreterConstraints, list[Flake8FieldSet]] = defaultdict(list)	×
65	for fs in request.field_sets:	×
66	constraints = InterpreterConstraints.create_from_compatibility_fields(	×
67	[
68	(fs.interpreter_constraints, fs.resolve),
69	*first_party_plugins.interpreter_constraints_and_resolve_fields,
70	],
71	python_setup,
72	)
73	results[constraints].append(fs)	×
74
75	return Partitions(	×
76	Partition(tuple(field_sets), interpreter_constraints)
77	for interpreter_constraints, field_sets in results.items()
78	)
79
80
UNCOV 81	@rule(desc="Lint with Flake8", level=LogLevel.DEBUG)	×
UNCOV 82	async def run_flake8(	×
83	request: Flake8Request.Batch[Flake8FieldSet, InterpreterConstraints],
84	flake8: Flake8,
85	first_party_plugins: Flake8FirstPartyPlugins,
86	) -> LintResult:
87	interpreter_constraints = request.partition_metadata	×
88	flake8_pex_get = create_venv_pex(	×
89	**implicitly(
90	flake8.to_pex_request(
91	interpreter_constraints=interpreter_constraints,
92	extra_requirements=first_party_plugins.requirement_strings,
93	)
94	)
95	)
96	config_files_get = find_config_file(flake8.config_request)	×
97	source_files_get = determine_source_files(	×
98	SourceFilesRequest(field_set.source for field_set in request.elements)
99	)
100	extra_files_get = path_globs_to_digest(	×
101	PathGlobs(
102	globs=flake8.extra_files,
103	glob_match_error_behavior=GlobMatchErrorBehavior.error,
104	description_of_origin="the option [flake8].extra_files",
105	)
106	)
107	# Ensure that the empty report dir exists.
108	report_directory_digest_get = create_digest(CreateDigest([Directory(REPORT_DIR)]))	×
109	flake8_pex, config_files, report_directory, source_files, extra_files = await concurrently(	×
110	flake8_pex_get,
111	config_files_get,
112	report_directory_digest_get,
113	source_files_get,
114	extra_files_get,
115	)
116
117	input_digest = await merge_digests(	×
118	MergeDigests(
119	(
120	source_files.snapshot.digest,
121	first_party_plugins.sources_digest,
122	config_files.snapshot.digest,
123	extra_files,
124	report_directory,
125	)
126	)
127	)
128	result = await execute_process(	×
129	**implicitly(
130	VenvPexProcess(
131	flake8_pex,
132	argv=generate_argv(source_files, flake8),
133	input_digest=input_digest,
134	output_directories=(REPORT_DIR,),
135	extra_env={"PEX_EXTRA_SYS_PATH": first_party_plugins.PREFIX},
136	concurrency_available=len(request.elements),
137	description=f"Run Flake8 on {pluralize(len(request.elements), 'file')}.",
138	level=LogLevel.DEBUG,
139	)
140	)
141	)
142	report = await remove_prefix(RemovePrefix(result.output_digest, REPORT_DIR))	×
143	return LintResult.create(request, result, report=report)	×
144
145
UNCOV 146	def rules():	×
UNCOV 147	return (	×
148	*collect_rules(),
149	*Flake8Request.rules(),
150	*pex.rules(),
151	)

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