18225878979

Committed 03 Oct 2025 02:59PM UTC coverage: 80.269% (-0.006%) from 80.275%

Build # 18225878979

Build Type

Pull #22717

github

Committed by

web-flow

Commit Message

Merge b17e03226 into f7b1ba442

Pull Request Pull Request #22717: wip: run tests on mac arm

Run Details

77232 of 96217 relevant lines covered (80.27%)

3.63 hits per line

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

74.87

/src/python/pants/backend/adhoc/code_quality_tool.py

# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from collections.abc import Iterable
from dataclasses import dataclass
from typing import ClassVar

from pants.core.goals.fix import Fix, FixFilesRequest, FixResult
from pants.core.goals.fmt import Fmt, FmtFilesRequest, FmtResult
from pants.core.goals.lint import Lint, LintFilesRequest, LintResult
from pants.core.util_rules.adhoc_process_support import (
    ToolRunner,
    ToolRunnerRequest,
    create_tool_runner,
    prepare_env_vars,
)
from pants.core.util_rules.adhoc_process_support import rules as adhoc_process_support_rules
from pants.core.util_rules.partitions import Partitions
from pants.engine.addresses import Addresses
from pants.engine.fs import PathGlobs
from pants.engine.goal import Goal
from pants.engine.internals.build_files import resolve_address
from pants.engine.internals.graph import resolve_targets
from pants.engine.internals.native_engine import (
    AddressInput,
    FilespecMatcher,
    MergeDigests,
    Snapshot,
)
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import digest_to_snapshot, execute_process, merge_digests
from pants.engine.process import FallibleProcessResult, Process
from pants.engine.rules import Rule, collect_rules, implicitly, rule
from pants.engine.target import (
    COMMON_TARGET_FIELDS,
    SpecialCasedDependencies,
    StringField,
    StringSequenceField,
    Target,
)
from pants.option.option_types import SkipOption
from pants.option.subsystem import Subsystem
from pants.util.frozendict import FrozenDict
from pants.util.strutil import Simplifier, help_text


class CodeQualityToolFileGlobIncludeField(StringSequenceField):
    alias: ClassVar[str] = "file_glob_include"
    required = True
    help = help_text(
        """
        Globs that identify files that can be processed by this tool

        A file matching any of the supplied globs is eligible for processing.
        Example: ["**/*.py"]
        """
    )


class CodeQualityToolFileGlobExcludeField(StringSequenceField):
    alias: ClassVar[str] = "file_glob_exclude"
    required = False
    default = ()
    help = help_text(
        """
        Globs matching files that should not be processed by this tool
        """
    )


class CodeQualityToolRunnableField(StringField):
    alias: ClassVar[str] = "runnable"
    required = True
    help = help_text(
        lambda: f"""
        Address to a target that can be invoked by the `run` goal (and does not set
        `run_in_sandbox_behavior=NOT_SUPPORTED`). This will be executed along with any arguments
        specified by `{CodeQualityToolArgumentsField.alias}`, in a sandbox with that target's transitive
        dependencies, along with the transitive dependencies specified by
        `{CodeQualityToolExecutionDependenciesField.alias}`.
        """
    )


class CodeQualityToolArgumentsField(StringSequenceField):
    alias: ClassVar[str] = "args"
    default = ()
    help = help_text(
        lambda: f"""
        Extra arguments to pass into the `{CodeQualityToolRunnableField.alias}` field
        before the list of source files
        """
    )


class CodeQualityToolExecutionDependenciesField(SpecialCasedDependencies):
    alias: ClassVar[str] = "execution_dependencies"
    required = False
    default = None

    help = help_text(
        """
        Additional dependencies that need to be available when running the tool.
        Typically used to point to config files.
        """
    )


class CodeQualityToolRunnableDependenciesField(SpecialCasedDependencies):
    alias: ClassVar[str] = "runnable_dependencies"
    required = False
    default = None

    help = help_text(
        lambda: f"""
        The runnable dependencies for this command.

        Dependencies specified here are those required to exist on the `PATH` to make the command
        complete successfully (interpreters specified in a `#!` command, etc). Note that these
        dependencies will be made available on the `PATH` with the name of the target.

        See also `{CodeQualityToolExecutionDependenciesField.alias}`.
        """
    )


class CodeQualityToolTarget(Target):
    alias: ClassVar[str] = "code_quality_tool"
    core_fields = (
        *COMMON_TARGET_FIELDS,
        CodeQualityToolRunnableField,
        CodeQualityToolArgumentsField,
        CodeQualityToolExecutionDependenciesField,
        CodeQualityToolRunnableDependenciesField,
        CodeQualityToolFileGlobIncludeField,
        CodeQualityToolFileGlobExcludeField,
    )

    help = help_text(
        lambda: f"""
        Configure a runnable to use as a linter, fixer or formatter

        Example BUILD file:

            {CodeQualityToolTarget.alias}(
                {CodeQualityToolRunnableField.alias}=":flake8_req",
                {CodeQualityToolExecutionDependenciesField.alias}=[":config_file"],
                {CodeQualityToolFileGlobIncludeField.alias}=["**/*.py"],
            )
        """
    )


@dataclass(frozen=True)
class CodeQualityToolAddressString:
    address: str


@dataclass(frozen=True)
class CodeQualityTool:
    runnable_address_str: str
    args: tuple[str, ...]
    execution_dependencies: tuple[str, ...]
    runnable_dependencies: tuple[str, ...]
    file_glob_include: tuple[str, ...]
    file_glob_exclude: tuple[str, ...]
    target: Target


@rule
async def find_code_quality_tool(request: CodeQualityToolAddressString) -> CodeQualityTool:
    tool_address = await resolve_address(
        **implicitly(
            {
                AddressInput.parse(
                    request.address, description_of_origin="code quality tool target"
                ): AddressInput
            }
        )
    )

    addresses = Addresses((tool_address,))
    addresses.expect_single()

    tool_targets = await resolve_targets(**implicitly({addresses: Addresses}))
    target = tool_targets[0]
    runnable_address_str = target[CodeQualityToolRunnableField].value
    if not runnable_address_str:
        raise Exception(f"Must supply a value for `runnable` for {request.address}.")

    return CodeQualityTool(
        runnable_address_str=runnable_address_str,
        execution_dependencies=target[CodeQualityToolExecutionDependenciesField].value or (),
        runnable_dependencies=target[CodeQualityToolRunnableDependenciesField].value or (),
        args=target[CodeQualityToolArgumentsField].value or (),
        file_glob_include=target[CodeQualityToolFileGlobIncludeField].value or (),
        file_glob_exclude=target[CodeQualityToolFileGlobExcludeField].value or (),
        target=target,
    )


@dataclass(frozen=True)
class CodeQualityToolBatch:
    runner: ToolRunner
    sources_snapshot: Snapshot
    output_files: tuple[str, ...]


@rule
async def process_files(batch: CodeQualityToolBatch) -> FallibleProcessResult:
    runner = batch.runner

    input_digest = await merge_digests(MergeDigests((runner.digest, batch.sources_snapshot.digest)))

    env_vars = await prepare_env_vars(
        runner.extra_env,
        (),
        extra_paths=runner.extra_paths,
        description_of_origin="code quality tool",
    )

    result = await execute_process(
        **implicitly(
            Process(
                argv=tuple(runner.args + batch.sources_snapshot.files),
                description="Running code quality tool",
                input_digest=input_digest,
                append_only_caches=runner.append_only_caches,
                immutable_input_digests=FrozenDict.frozen(runner.immutable_input_digests),
                env=env_vars,
                output_files=batch.output_files,
            )
        )
    )
    return result


@rule
async def runner_request_for_code_quality_tool(
    cqt: CodeQualityTool,
) -> ToolRunnerRequest:
    return ToolRunnerRequest(
        runnable_address_str=cqt.runnable_address_str,
        args=cqt.args,
        execution_dependencies=cqt.execution_dependencies,
        runnable_dependencies=cqt.runnable_dependencies,
        target=cqt.target,
    )


_name_to_supported_goal = {g.name: g for g in (Lint, Fmt, Fix)}


class CodeQualityToolUnsupportedGoalError(Exception):
    """Raised when a rule builder is instantiated for an unrecognized or unsupported goal."""


@dataclass
class CodeQualityToolRuleBuilder:
    goal: str
    target: str
    name: str
    scope: str

    @property
    def goal_type(self) -> type[Goal]:
        return _name_to_supported_goal[self.goal]

    def __post_init__(self):
        if self.goal not in _name_to_supported_goal:
            raise CodeQualityToolUnsupportedGoalError(
                f"""goal must be one of {sorted(_name_to_supported_goal)}"""
            )

    def rules(self) -> Iterable[Rule]:
        if self.goal_type is Fmt:
            return self._build_fmt_rules()
        elif self.goal_type is Fix:
            return self._build_fix_rules()
        elif self.goal_type is Lint:
            return self._build_lint_rules()
        else:
            raise ValueError(f"Unsupported goal for code quality tool: {self.goal}")

    def _build_lint_rules(self) -> Iterable[Rule]:
        class CodeQualityToolInstance(Subsystem):
            options_scope = self.scope
            name = self.name
            help = f"{self.goal.capitalize()} with {self.name}. Tool defined in {self.target}"

            skip = SkipOption("lint")

        class CodeQualityProcessingRequest(LintFilesRequest):
            tool_subsystem = CodeQualityToolInstance

        @rule(canonical_name_suffix=self.scope)
        async def partition_inputs(
            request: CodeQualityProcessingRequest.PartitionRequest,
            subsystem: CodeQualityToolInstance,
        ) -> Partitions:
            if subsystem.skip:
                return Partitions()

            cqt = await find_code_quality_tool(CodeQualityToolAddressString(address=self.target))

            matching_filepaths = FilespecMatcher(
                includes=cqt.file_glob_include,
                excludes=cqt.file_glob_exclude,
            ).matches(request.files)

            return Partitions.single_partition(sorted(matching_filepaths))

        @rule(canonical_name_suffix=self.scope)
        async def run_code_quality(request: CodeQualityProcessingRequest.Batch) -> LintResult:
            sources_snapshot, code_quality_tool_runner = await concurrently(
                digest_to_snapshot(**implicitly(PathGlobs(request.elements))),
                create_tool_runner(**implicitly(CodeQualityToolAddressString(address=self.target))),
            )

            proc_result = await process_files(
                CodeQualityToolBatch(
                    runner=code_quality_tool_runner,
                    sources_snapshot=sources_snapshot,
                    output_files=(),
                ),
            )

            return LintResult.create(request, process_result=proc_result)

        namespace = dict(locals())

        return [
            *collect_rules(namespace),
            *CodeQualityProcessingRequest.rules(),
        ]

    def _build_fmt_rules(self) -> Iterable[Rule]:
        class CodeQualityToolInstance(Subsystem):
            options_scope = self.scope
            name = self.name
            help = f"{self.goal.capitalize()} with {self.name}. Tool defined in {self.target}"

            skip = SkipOption("lint", "fmt")

        class CodeQualityProcessingRequest(FmtFilesRequest):
            tool_subsystem = CodeQualityToolInstance

        @rule(canonical_name_suffix=self.scope)
        async def partition_inputs(
            request: CodeQualityProcessingRequest.PartitionRequest,
            subsystem: CodeQualityToolInstance,
        ) -> Partitions:
            if subsystem.skip:
                return Partitions()

            cqt = await find_code_quality_tool(CodeQualityToolAddressString(address=self.target))

            matching_filepaths = FilespecMatcher(
                includes=cqt.file_glob_include,
                excludes=cqt.file_glob_exclude,
            ).matches(request.files)

            return Partitions.single_partition(sorted(matching_filepaths))

        @rule(canonical_name_suffix=self.scope)
        async def run_code_quality(request: CodeQualityProcessingRequest.Batch) -> FmtResult:
            sources_snapshot = request.snapshot

            cqt = await find_code_quality_tool(CodeQualityToolAddressString(address=self.target))
            code_quality_tool_runner_request = await runner_request_for_code_quality_tool(cqt)
            code_quality_tool_runner = await create_tool_runner(code_quality_tool_runner_request)

            proc_result = await process_files(
                CodeQualityToolBatch(
                    runner=code_quality_tool_runner,
                    sources_snapshot=sources_snapshot,
                    output_files=request.files,
                ),
            )

            output = await digest_to_snapshot(proc_result.output_digest)

            return FmtResult(
                input=request.snapshot,
                output=output,
                stdout=Simplifier().simplify(proc_result.stdout),
                stderr=Simplifier().simplify(proc_result.stderr),
                tool_name=request.tool_name,
            )

        namespace = dict(locals())

        return [
            *collect_rules(namespace),
            *CodeQualityProcessingRequest.rules(),
        ]

    def _build_fix_rules(self) -> Iterable[Rule]:
        class CodeQualityToolInstance(Subsystem):
            options_scope = self.scope
            name = self.name
            help = f"{self.goal.capitalize()} with {self.name}. Tool defined in {self.target}"

            skip = SkipOption("lint", "fmt", "fix")

        class CodeQualityProcessingRequest(FixFilesRequest):
            tool_subsystem = CodeQualityToolInstance

        @rule(canonical_name_suffix=self.scope)
        async def partition_inputs(
            request: CodeQualityProcessingRequest.PartitionRequest,
            subsystem: CodeQualityToolInstance,
        ) -> Partitions:
            if subsystem.skip:
                return Partitions()

            cqt = await find_code_quality_tool(CodeQualityToolAddressString(address=self.target))

            matching_filepaths = FilespecMatcher(
                includes=cqt.file_glob_include,
                excludes=cqt.file_glob_exclude,
            ).matches(request.files)

            return Partitions.single_partition(sorted(matching_filepaths))

        @rule(canonical_name_suffix=self.scope)
        async def run_code_quality(request: CodeQualityProcessingRequest.Batch) -> FixResult:
            sources_snapshot = request.snapshot

            cqt = await find_code_quality_tool(CodeQualityToolAddressString(address=self.target))
            code_quality_tool_runner_request = await runner_request_for_code_quality_tool(cqt)
            code_quality_tool_runner = await create_tool_runner(code_quality_tool_runner_request)

            proc_result = await process_files(
                CodeQualityToolBatch(
                    runner=code_quality_tool_runner,
                    sources_snapshot=sources_snapshot,
                    output_files=request.files,
                ),
            )

            output = await digest_to_snapshot(proc_result.output_digest)

            return FixResult(
                input=request.snapshot,
                output=output,
                stdout=Simplifier().simplify(proc_result.stdout),
                stderr=Simplifier().simplify(proc_result.stderr),
                tool_name=request.tool_name,
            )

        namespace = dict(locals())

        return [
            *collect_rules(namespace),
            *CodeQualityProcessingRequest.rules(),
        ]


def base_rules():
    return [
        *collect_rules(),
        *adhoc_process_support_rules(),
    ]

1	# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from collections.abc import Iterable	1✔
4	from dataclasses import dataclass	1✔
5	from typing import ClassVar	1✔
6
7	from pants.core.goals.fix import Fix, FixFilesRequest, FixResult	1✔
8	from pants.core.goals.fmt import Fmt, FmtFilesRequest, FmtResult	1✔
9	from pants.core.goals.lint import Lint, LintFilesRequest, LintResult	1✔
10	from pants.core.util_rules.adhoc_process_support import (	1✔
11	ToolRunner,
12	ToolRunnerRequest,
13	create_tool_runner,
14	prepare_env_vars,
15	)
16	from pants.core.util_rules.adhoc_process_support import rules as adhoc_process_support_rules	1✔
17	from pants.core.util_rules.partitions import Partitions	1✔
18	from pants.engine.addresses import Addresses	1✔
19	from pants.engine.fs import PathGlobs	1✔
20	from pants.engine.goal import Goal	1✔
21	from pants.engine.internals.build_files import resolve_address	1✔
22	from pants.engine.internals.graph import resolve_targets	1✔
23	from pants.engine.internals.native_engine import (	1✔
24	AddressInput,
25	FilespecMatcher,
26	MergeDigests,
27	Snapshot,
28	)
29	from pants.engine.internals.selectors import concurrently	1✔
30	from pants.engine.intrinsics import digest_to_snapshot, execute_process, merge_digests	1✔
31	from pants.engine.process import FallibleProcessResult, Process	1✔
32	from pants.engine.rules import Rule, collect_rules, implicitly, rule	1✔
33	from pants.engine.target import (	1✔
34	COMMON_TARGET_FIELDS,
35	SpecialCasedDependencies,
36	StringField,
37	StringSequenceField,
38	Target,
39	)
40	from pants.option.option_types import SkipOption	1✔
41	from pants.option.subsystem import Subsystem	1✔
42	from pants.util.frozendict import FrozenDict	1✔
43	from pants.util.strutil import Simplifier, help_text	1✔
44
45
46	class CodeQualityToolFileGlobIncludeField(StringSequenceField):	1✔
47	alias: ClassVar[str] = "file_glob_include"	1✔
48	required = True	1✔
49	help = help_text(	1✔
50	"""
51	Globs that identify files that can be processed by this tool
52
53	A file matching any of the supplied globs is eligible for processing.
54	Example: ["*/.py"]
55	"""
56	)
57
58
59	class CodeQualityToolFileGlobExcludeField(StringSequenceField):	1✔
60	alias: ClassVar[str] = "file_glob_exclude"	1✔
61	required = False	1✔
62	default = ()	1✔
63	help = help_text(	1✔
64	"""
65	Globs matching files that should not be processed by this tool
66	"""
67	)
68
69
70	class CodeQualityToolRunnableField(StringField):	1✔
71	alias: ClassVar[str] = "runnable"	1✔
72	required = True	1✔
73	help = help_text(	1✔
74	lambda: f"""
75	Address to a target that can be invoked by the `run` goal (and does not set
76	`run_in_sandbox_behavior=NOT_SUPPORTED`). This will be executed along with any arguments
77	specified by `{CodeQualityToolArgumentsField.alias}`, in a sandbox with that target's transitive
78	dependencies, along with the transitive dependencies specified by
79	`{CodeQualityToolExecutionDependenciesField.alias}`.
80	"""
81	)
82
83
84	class CodeQualityToolArgumentsField(StringSequenceField):	1✔
85	alias: ClassVar[str] = "args"	1✔
86	default = ()	1✔
87	help = help_text(	1✔
88	lambda: f"""
89	Extra arguments to pass into the `{CodeQualityToolRunnableField.alias}` field
90	before the list of source files
91	"""
92	)
93
94
95	class CodeQualityToolExecutionDependenciesField(SpecialCasedDependencies):	1✔
96	alias: ClassVar[str] = "execution_dependencies"	1✔
97	required = False	1✔
98	default = None	1✔
99
100	help = help_text(	1✔
101	"""
102	Additional dependencies that need to be available when running the tool.
103	Typically used to point to config files.
104	"""
105	)
106
107
108	class CodeQualityToolRunnableDependenciesField(SpecialCasedDependencies):	1✔
109	alias: ClassVar[str] = "runnable_dependencies"	1✔
110	required = False	1✔
111	default = None	1✔
112
113	help = help_text(	1✔
114	lambda: f"""
115	The runnable dependencies for this command.
116
117	Dependencies specified here are those required to exist on the `PATH` to make the command
118	complete successfully (interpreters specified in a `#!` command, etc). Note that these
119	dependencies will be made available on the `PATH` with the name of the target.
120
121	See also `{CodeQualityToolExecutionDependenciesField.alias}`.
122	"""
123	)
124
125
126	class CodeQualityToolTarget(Target):	1✔
127	alias: ClassVar[str] = "code_quality_tool"	1✔
128	core_fields = (	1✔
129	*COMMON_TARGET_FIELDS,
130	CodeQualityToolRunnableField,
131	CodeQualityToolArgumentsField,
132	CodeQualityToolExecutionDependenciesField,
133	CodeQualityToolRunnableDependenciesField,
134	CodeQualityToolFileGlobIncludeField,
135	CodeQualityToolFileGlobExcludeField,
136	)
137
138	help = help_text(	1✔
139	lambda: f"""
140	Configure a runnable to use as a linter, fixer or formatter
141
142	Example BUILD file:
143
144	{CodeQualityToolTarget.alias}(
145	{CodeQualityToolRunnableField.alias}=":flake8_req",
146	{CodeQualityToolExecutionDependenciesField.alias}=[":config_file"],
147	{CodeQualityToolFileGlobIncludeField.alias}=["*/.py"],
148	)
149	"""
150	)
151
152
153	@dataclass(frozen=True)	1✔
154	class CodeQualityToolAddressString:	1✔
155	address: str	1✔
156
157
158	@dataclass(frozen=True)	1✔
159	class CodeQualityTool:	1✔
160	runnable_address_str: str	1✔
161	args: tuple[str, ...]	1✔
162	execution_dependencies: tuple[str, ...]	1✔
163	runnable_dependencies: tuple[str, ...]	1✔
164	file_glob_include: tuple[str, ...]	1✔
165	file_glob_exclude: tuple[str, ...]	1✔
166	target: Target	1✔
167
168
169	@rule	1✔
170	async def find_code_quality_tool(request: CodeQualityToolAddressString) -> CodeQualityTool:	1✔
171	tool_address = await resolve_address(	×
172	**implicitly(
173	{
174	AddressInput.parse(
175	request.address, description_of_origin="code quality tool target"
176	): AddressInput
177	}
178	)
179	)
180
181	addresses = Addresses((tool_address,))	×
182	addresses.expect_single()	×
183
184	tool_targets = await resolve_targets(**implicitly({addresses: Addresses}))	×
185	target = tool_targets[0]	×
186	runnable_address_str = target[CodeQualityToolRunnableField].value	×
187	if not runnable_address_str:	×
188	raise Exception(f"Must supply a value for `runnable` for {request.address}.")	×
189
190	return CodeQualityTool(	×
191	runnable_address_str=runnable_address_str,
192	execution_dependencies=target[CodeQualityToolExecutionDependenciesField].value or (),
193	runnable_dependencies=target[CodeQualityToolRunnableDependenciesField].value or (),
194	args=target[CodeQualityToolArgumentsField].value or (),
195	file_glob_include=target[CodeQualityToolFileGlobIncludeField].value or (),
196	file_glob_exclude=target[CodeQualityToolFileGlobExcludeField].value or (),
197	target=target,
198	)
199
200
201	@dataclass(frozen=True)	1✔
202	class CodeQualityToolBatch:	1✔
203	runner: ToolRunner	1✔
204	sources_snapshot: Snapshot	1✔
205	output_files: tuple[str, ...]	1✔
206
207
208	@rule	1✔
209	async def process_files(batch: CodeQualityToolBatch) -> FallibleProcessResult:	1✔
210	runner = batch.runner	×
211
212	input_digest = await merge_digests(MergeDigests((runner.digest, batch.sources_snapshot.digest)))	×
213
214	env_vars = await prepare_env_vars(	×
215	runner.extra_env,
216	(),
217	extra_paths=runner.extra_paths,
218	description_of_origin="code quality tool",
219	)
220
221	result = await execute_process(	×
222	**implicitly(
223	Process(
224	argv=tuple(runner.args + batch.sources_snapshot.files),
225	description="Running code quality tool",
226	input_digest=input_digest,
227	append_only_caches=runner.append_only_caches,
228	immutable_input_digests=FrozenDict.frozen(runner.immutable_input_digests),
229	env=env_vars,
230	output_files=batch.output_files,
231	)
232	)
233	)
234	return result	×
235
236
237	@rule	1✔
238	async def runner_request_for_code_quality_tool(	1✔
239	cqt: CodeQualityTool,
240	) -> ToolRunnerRequest:
241	return ToolRunnerRequest(	×
242	runnable_address_str=cqt.runnable_address_str,
243	args=cqt.args,
244	execution_dependencies=cqt.execution_dependencies,
245	runnable_dependencies=cqt.runnable_dependencies,
246	target=cqt.target,
247	)
248
249
250	_name_to_supported_goal = {g.name: g for g in (Lint, Fmt, Fix)}	1✔
251
252
253	class CodeQualityToolUnsupportedGoalError(Exception):	1✔
254	"""Raised when a rule builder is instantiated for an unrecognized or unsupported goal."""
255
256
257	@dataclass	1✔
258	class CodeQualityToolRuleBuilder:	1✔
259	goal: str	1✔
260	target: str	1✔
261	name: str	1✔
262	scope: str	1✔
263
264	@property	1✔
265	def goal_type(self) -> type[Goal]:	1✔
266	return _name_to_supported_goal[self.goal]	1✔
267
268	def __post_init__(self):	1✔
269	if self.goal not in _name_to_supported_goal:	1✔
270	raise CodeQualityToolUnsupportedGoalError(	1✔
271	f"""goal must be one of {sorted(_name_to_supported_goal)}"""
272	)
273
274	def rules(self) -> Iterable[Rule]:	1✔
275	if self.goal_type is Fmt:	1✔
276	return self._build_fmt_rules()	1✔
277	elif self.goal_type is Fix:	1✔
278	return self._build_fix_rules()	1✔
279	elif self.goal_type is Lint:	1✔
280	return self._build_lint_rules()	1✔
281	else:
282	raise ValueError(f"Unsupported goal for code quality tool: {self.goal}")	×
283
284	def _build_lint_rules(self) -> Iterable[Rule]:	1✔
285	class CodeQualityToolInstance(Subsystem):	1✔
286	options_scope = self.scope	1✔
287	name = self.name	1✔
288	help = f"{self.goal.capitalize()} with {self.name}. Tool defined in {self.target}"	1✔
289
290	skip = SkipOption("lint")	1✔
291
292	class CodeQualityProcessingRequest(LintFilesRequest):	1✔
293	tool_subsystem = CodeQualityToolInstance	1✔
294
295	@rule(canonical_name_suffix=self.scope)	1✔
296	async def partition_inputs(	1✔
297	request: CodeQualityProcessingRequest.PartitionRequest,
298	subsystem: CodeQualityToolInstance,
299	) -> Partitions:
300	if subsystem.skip:	×
301	return Partitions()	×
302
303	cqt = await find_code_quality_tool(CodeQualityToolAddressString(address=self.target))	×
304
305	matching_filepaths = FilespecMatcher(	×
306	includes=cqt.file_glob_include,
307	excludes=cqt.file_glob_exclude,
308	).matches(request.files)
309
310	return Partitions.single_partition(sorted(matching_filepaths))	×
311
312	@rule(canonical_name_suffix=self.scope)	1✔
313	async def run_code_quality(request: CodeQualityProcessingRequest.Batch) -> LintResult:	1✔
314	sources_snapshot, code_quality_tool_runner = await concurrently(	×
315	digest_to_snapshot(**implicitly(PathGlobs(request.elements))),
316	create_tool_runner(**implicitly(CodeQualityToolAddressString(address=self.target))),
317	)
318
319	proc_result = await process_files(	×
320	CodeQualityToolBatch(
321	runner=code_quality_tool_runner,
322	sources_snapshot=sources_snapshot,
323	output_files=(),
324	),
325	)
326
327	return LintResult.create(request, process_result=proc_result)	×
328
329	namespace = dict(locals())	1✔
330
331	return [	1✔
332	*collect_rules(namespace),
333	*CodeQualityProcessingRequest.rules(),
334	]
335
336	def _build_fmt_rules(self) -> Iterable[Rule]:	1✔
337	class CodeQualityToolInstance(Subsystem):	1✔
338	options_scope = self.scope	1✔
339	name = self.name	1✔
340	help = f"{self.goal.capitalize()} with {self.name}. Tool defined in {self.target}"	1✔
341
342	skip = SkipOption("lint", "fmt")	1✔
343
344	class CodeQualityProcessingRequest(FmtFilesRequest):	1✔
345	tool_subsystem = CodeQualityToolInstance	1✔
346
347	@rule(canonical_name_suffix=self.scope)	1✔
348	async def partition_inputs(	1✔
349	request: CodeQualityProcessingRequest.PartitionRequest,
350	subsystem: CodeQualityToolInstance,
351	) -> Partitions:
352	if subsystem.skip:	×
353	return Partitions()	×
354
355	cqt = await find_code_quality_tool(CodeQualityToolAddressString(address=self.target))	×
356
357	matching_filepaths = FilespecMatcher(	×
358	includes=cqt.file_glob_include,
359	excludes=cqt.file_glob_exclude,
360	).matches(request.files)
361
362	return Partitions.single_partition(sorted(matching_filepaths))	×
363
364	@rule(canonical_name_suffix=self.scope)	1✔
365	async def run_code_quality(request: CodeQualityProcessingRequest.Batch) -> FmtResult:	1✔
366	sources_snapshot = request.snapshot	×
367
368	cqt = await find_code_quality_tool(CodeQualityToolAddressString(address=self.target))	×
369	code_quality_tool_runner_request = await runner_request_for_code_quality_tool(cqt)	×
370	code_quality_tool_runner = await create_tool_runner(code_quality_tool_runner_request)	×
371
372	proc_result = await process_files(	×
373	CodeQualityToolBatch(
374	runner=code_quality_tool_runner,
375	sources_snapshot=sources_snapshot,
376	output_files=request.files,
377	),
378	)
379
380	output = await digest_to_snapshot(proc_result.output_digest)	×
381
382	return FmtResult(	×
383	input=request.snapshot,
384	output=output,
385	stdout=Simplifier().simplify(proc_result.stdout),
386	stderr=Simplifier().simplify(proc_result.stderr),
387	tool_name=request.tool_name,
388	)
389
390	namespace = dict(locals())	1✔
391
392	return [	1✔
393	*collect_rules(namespace),
394	*CodeQualityProcessingRequest.rules(),
395	]
396
397	def _build_fix_rules(self) -> Iterable[Rule]:	1✔
398	class CodeQualityToolInstance(Subsystem):	1✔
399	options_scope = self.scope	1✔
400	name = self.name	1✔
401	help = f"{self.goal.capitalize()} with {self.name}. Tool defined in {self.target}"	1✔
402
403	skip = SkipOption("lint", "fmt", "fix")	1✔
404
405	class CodeQualityProcessingRequest(FixFilesRequest):	1✔
406	tool_subsystem = CodeQualityToolInstance	1✔
407
408	@rule(canonical_name_suffix=self.scope)	1✔
409	async def partition_inputs(	1✔
410	request: CodeQualityProcessingRequest.PartitionRequest,
411	subsystem: CodeQualityToolInstance,
412	) -> Partitions:
413	if subsystem.skip:	×
414	return Partitions()	×
415
416	cqt = await find_code_quality_tool(CodeQualityToolAddressString(address=self.target))	×
417
418	matching_filepaths = FilespecMatcher(	×
419	includes=cqt.file_glob_include,
420	excludes=cqt.file_glob_exclude,
421	).matches(request.files)
422
423	return Partitions.single_partition(sorted(matching_filepaths))	×
424
425	@rule(canonical_name_suffix=self.scope)	1✔
426	async def run_code_quality(request: CodeQualityProcessingRequest.Batch) -> FixResult:	1✔
427	sources_snapshot = request.snapshot	×
428
429	cqt = await find_code_quality_tool(CodeQualityToolAddressString(address=self.target))	×
430	code_quality_tool_runner_request = await runner_request_for_code_quality_tool(cqt)	×
431	code_quality_tool_runner = await create_tool_runner(code_quality_tool_runner_request)	×
432
433	proc_result = await process_files(	×
434	CodeQualityToolBatch(
435	runner=code_quality_tool_runner,
436	sources_snapshot=sources_snapshot,
437	output_files=request.files,
438	),
439	)
440
441	output = await digest_to_snapshot(proc_result.output_digest)	×
442
443	return FixResult(	×
444	input=request.snapshot,
445	output=output,
446	stdout=Simplifier().simplify(proc_result.stdout),
447	stderr=Simplifier().simplify(proc_result.stderr),
448	tool_name=request.tool_name,
449	)
450
451	namespace = dict(locals())	1✔
452
453	return [	1✔
454	*collect_rules(namespace),
455	*CodeQualityProcessingRequest.rules(),
456	]
457
458
459	def base_rules():	1✔
460	return [	1✔
461	*collect_rules(),
462	*adhoc_process_support_rules(),
463	]

pantsbuild / pants / 18225878979

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