23068140808

Committed 13 Mar 2026 07:56PM UTC coverage: 52.677% (-40.3%) from 92.932%

Build # 23068140808

Build Type

Pull #23170

github

Committed by

web-flow

Commit Message

Merge e8ca01cfa into f07276df6

Pull Request Pull Request #23170: Debug reapi test cache misses

Run Details

31687 of 60153 relevant lines covered (52.68%)

1.05 hits per line

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

96.15

/src/python/pants/backend/python/lint/flake8/rules.py

# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

from collections import defaultdict

from pants.backend.python.lint.flake8.subsystem import (
    Flake8,
    Flake8FieldSet,
    Flake8FirstPartyPlugins,
)
from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.util_rules import pex
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.pex import VenvPexProcess, create_venv_pex
from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior
from pants.core.goals.lint import REPORT_DIR, LintResult, LintTargetsRequest, Partitions
from pants.core.util_rules.config_files import find_config_file
from pants.core.util_rules.partitions import Partition
from pants.core.util_rules.source_files import (
    SourceFiles,
    SourceFilesRequest,
    determine_source_files,
)
from pants.engine.fs import CreateDigest, Directory, MergeDigests, PathGlobs, RemovePrefix
from pants.engine.intrinsics import (
    create_digest,
    execute_process,
    merge_digests,
    path_globs_to_digest,
    remove_prefix,
)
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.util.logging import LogLevel
from pants.util.strutil import pluralize


class Flake8Request(LintTargetsRequest):
    field_set_type = Flake8FieldSet
    tool_subsystem = Flake8  # type: ignore[assignment]


def generate_argv(source_files: SourceFiles, flake8: Flake8) -> tuple[str, ...]:
    args: list[str] = []
    if flake8.config:
        args.append(f"--config={flake8.config}")
    args.append("--jobs={pants_concurrency}")
    args.extend(flake8.args)
    args.extend(source_files.files)
    return tuple(args)


@rule
async def partition_flake8(
    request: Flake8Request.PartitionRequest[Flake8FieldSet],
    flake8: Flake8,
    python_setup: PythonSetup,
    first_party_plugins: Flake8FirstPartyPlugins,
) -> Partitions[Flake8FieldSet, InterpreterConstraints]:
    if flake8.skip:
        return Partitions()

    results: dict[InterpreterConstraints, list[Flake8FieldSet]] = defaultdict(list)
    for fs in request.field_sets:
        constraints = InterpreterConstraints.create_from_compatibility_fields(
            [
                (fs.interpreter_constraints, fs.resolve),
                *first_party_plugins.interpreter_constraints_and_resolve_fields,
            ],
            python_setup,
        )
        results[constraints].append(fs)

    return Partitions(
        Partition(tuple(field_sets), interpreter_constraints)
        for interpreter_constraints, field_sets in results.items()
    )


@rule(desc="Lint with Flake8", level=LogLevel.DEBUG)
async def run_flake8(
    request: Flake8Request.Batch[Flake8FieldSet, InterpreterConstraints],
    flake8: Flake8,
    first_party_plugins: Flake8FirstPartyPlugins,
) -> LintResult:
    interpreter_constraints = request.partition_metadata
    flake8_pex_get = create_venv_pex(
        **implicitly(
            flake8.to_pex_request(
                interpreter_constraints=interpreter_constraints,
                extra_requirements=first_party_plugins.requirement_strings,
            )
        )
    )
    config_files_get = find_config_file(flake8.config_request)
    source_files_get = determine_source_files(
        SourceFilesRequest(field_set.source for field_set in request.elements)
    )
    extra_files_get = path_globs_to_digest(
        PathGlobs(
            globs=flake8.extra_files,
            glob_match_error_behavior=GlobMatchErrorBehavior.error,
            description_of_origin="the option [flake8].extra_files",
        )
    )
    # Ensure that the empty report dir exists.
    report_directory_digest_get = create_digest(CreateDigest([Directory(REPORT_DIR)]))
    flake8_pex, config_files, report_directory, source_files, extra_files = await concurrently(
        flake8_pex_get,
        config_files_get,
        report_directory_digest_get,
        source_files_get,
        extra_files_get,
    )

    input_digest = await merge_digests(
        MergeDigests(
            (
                source_files.snapshot.digest,
                first_party_plugins.sources_digest,
                config_files.snapshot.digest,
                extra_files,
                report_directory,
            )
        )
    )
    result = await execute_process(
        **implicitly(
            VenvPexProcess(
                flake8_pex,
                argv=generate_argv(source_files, flake8),
                input_digest=input_digest,
                output_directories=(REPORT_DIR,),
                extra_env={"PEX_EXTRA_SYS_PATH": first_party_plugins.PREFIX},
                concurrency_available=len(request.elements),
                description=f"Run Flake8 on {pluralize(len(request.elements), 'file')}.",
                level=LogLevel.DEBUG,
            )
        )
    )
    report = await remove_prefix(RemovePrefix(result.output_digest, REPORT_DIR))
    return LintResult.create(request, result, report=report)


def rules():
    return (
        *collect_rules(),
        *Flake8Request.rules(),
        *pex.rules(),
    )

1	# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	2✔
5
6	from collections import defaultdict	2✔
7
8	from pants.backend.python.lint.flake8.subsystem import (	2✔
9	Flake8,
10	Flake8FieldSet,
11	Flake8FirstPartyPlugins,
12	)
13	from pants.backend.python.subsystems.setup import PythonSetup	2✔
14	from pants.backend.python.util_rules import pex	2✔
15	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	2✔
16	from pants.backend.python.util_rules.pex import VenvPexProcess, create_venv_pex	2✔
17	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	2✔
18	from pants.core.goals.lint import REPORT_DIR, LintResult, LintTargetsRequest, Partitions	2✔
19	from pants.core.util_rules.config_files import find_config_file	2✔
20	from pants.core.util_rules.partitions import Partition	2✔
21	from pants.core.util_rules.source_files import (	2✔
22	SourceFiles,
23	SourceFilesRequest,
24	determine_source_files,
25	)
26	from pants.engine.fs import CreateDigest, Directory, MergeDigests, PathGlobs, RemovePrefix	2✔
27	from pants.engine.intrinsics import (	2✔
28	create_digest,
29	execute_process,
30	merge_digests,
31	path_globs_to_digest,
32	remove_prefix,
33	)
34	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	2✔
35	from pants.util.logging import LogLevel	2✔
36	from pants.util.strutil import pluralize	2✔
37
38
39	class Flake8Request(LintTargetsRequest):	2✔
40	field_set_type = Flake8FieldSet	2✔
41	tool_subsystem = Flake8 # type: ignore[assignment]	2✔
42
43
44	def generate_argv(source_files: SourceFiles, flake8: Flake8) -> tuple[str, ...]:	2✔
45	args: list[str] = []	2✔
46	if flake8.config:	2✔
47	args.append(f"--config={flake8.config}")	×
48	args.append("--jobs={pants_concurrency}")	2✔
49	args.extend(flake8.args)	2✔
50	args.extend(source_files.files)	2✔
51	return tuple(args)	2✔
52
53
54	@rule	2✔
55	async def partition_flake8(	2✔
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:	2✔
62	return Partitions()	×
63
64	results: dict[InterpreterConstraints, list[Flake8FieldSet]] = defaultdict(list)	2✔
65	for fs in request.field_sets:	2✔
66	constraints = InterpreterConstraints.create_from_compatibility_fields(	2✔
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)	2✔
74
75	return Partitions(	2✔
76	Partition(tuple(field_sets), interpreter_constraints)
77	for interpreter_constraints, field_sets in results.items()
78	)
79
80
81	@rule(desc="Lint with Flake8", level=LogLevel.DEBUG)	2✔
82	async def run_flake8(	2✔
83	request: Flake8Request.Batch[Flake8FieldSet, InterpreterConstraints],
84	flake8: Flake8,
85	first_party_plugins: Flake8FirstPartyPlugins,
86	) -> LintResult:
87	interpreter_constraints = request.partition_metadata	2✔
88	flake8_pex_get = create_venv_pex(	2✔
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)	2✔
97	source_files_get = determine_source_files(	2✔
98	SourceFilesRequest(field_set.source for field_set in request.elements)
99	)
100	extra_files_get = path_globs_to_digest(	2✔
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)]))	2✔
109	flake8_pex, config_files, report_directory, source_files, extra_files = await concurrently(	2✔
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(	2✔
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(	2✔
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))	2✔
143	return LintResult.create(request, result, report=report)	2✔
144
145
146	def rules():	2✔
147	return (	2✔
148	*collect_rules(),
149	*Flake8Request.rules(),
150	*pex.rules(),
151	)

pantsbuild / pants / 23068140808

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