18517631058

Committed 15 Oct 2025 04:18AM UTC coverage: 69.207% (-11.1%) from 80.267%

Build # 18517631058

Build Type

Pull #22745

github

Committed by

web-flow

Commit Message

Merge 642a76ca1 into 99919310e

Pull Request Pull Request #22745: [windows] Add windows support in the stdio crate.

Run Details

53815 of 77759 relevant lines covered (69.21%)

2.42 hits per line

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

68.64

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

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

from __future__ import annotations

import logging
from collections import defaultdict
from collections.abc import Callable, Coroutine, Iterable
from dataclasses import dataclass
from typing import Any, ClassVar, Protocol, TypeVar, cast, final

from pants.base.specs import Specs
from pants.core.environments.rules import _warn_on_non_local_environments
from pants.core.goals.multi_tool_goal_helper import (
    BatchSizeOption,
    OnlyOption,
    SkippableSubsystem,
    determine_specified_tool_ids,
)
from pants.core.util_rules.partitions import PartitionElementT, PartitionerType, PartitionMetadataT
from pants.core.util_rules.partitions import Partitions as Partitions  # re-export
from pants.core.util_rules.partitions import (
    _BatchBase,
    _PartitionFieldSetsRequestBase,
    _PartitionFilesRequestBase,
)
from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType
from pants.engine.environment import EnvironmentName
from pants.engine.fs import EMPTY_DIGEST, Digest
from pants.engine.goal import Goal, GoalSubsystem
from pants.engine.internals.graph import filter_targets
from pants.engine.internals.specs_rules import resolve_specs_paths
from pants.engine.process import FallibleProcessResult
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import FieldSet
from pants.engine.unions import UnionMembership, UnionRule, distinct_union_type_per_subclass, union
from pants.option.option_types import BoolOption
from pants.util.docutil import bin_name, doc_url
from pants.util.logging import LogLevel
from pants.util.meta import classproperty
from pants.util.strutil import Simplifier, softwrap

logger = logging.getLogger(__name__)

_T = TypeVar("_T")
_FieldSetT = TypeVar("_FieldSetT", bound=FieldSet)


@dataclass(frozen=True)
class LintResult(EngineAwareReturnType):
    exit_code: int
    stdout: str
    stderr: str
    linter_name: str
    partition_description: str | None = None
    report: Digest = EMPTY_DIGEST
    _render_message: bool = True

    @classmethod
    def create(
        cls,
        request: AbstractLintRequest.Batch,
        process_result: FallibleProcessResult,
        *,
        output_simplifier: Simplifier = Simplifier(),
        report: Digest = EMPTY_DIGEST,
    ) -> LintResult:
        return cls(
            exit_code=process_result.exit_code,
            stdout=output_simplifier.simplify(process_result.stdout),
            stderr=output_simplifier.simplify(process_result.stderr),
            linter_name=request.tool_name,
            partition_description=request.partition_metadata.description,
            report=report,
        )

    def metadata(self) -> dict[str, Any]:
        return {"partition": self.partition_description}

    def level(self) -> LogLevel | None:
        if not self._render_message:
            return LogLevel.TRACE
        if self.exit_code != 0:
            return LogLevel.ERROR
        return LogLevel.INFO

    def message(self) -> str | None:
        message = self.linter_name
        message += (
            " succeeded." if self.exit_code == 0 else f" failed (exit code {self.exit_code})."
        )
        if self.partition_description:
            message += f"\nPartition: {self.partition_description}"
        if self.stdout:
            message += f"\n{self.stdout}"
        if self.stderr:
            message += f"\n{self.stderr}"
        if self.partition_description or self.stdout or self.stderr:
            message += "\n\n"

        return message

    def cacheable(self) -> bool:
        """Is marked uncacheable to ensure that it always renders."""
        return False


@union
class AbstractLintRequest:
    """Base class for plugin types wanting to be run as part of `lint`.

    Plugins should define a new type which subclasses either `LintTargetsRequest` (to lint targets)
    or `LintFilesRequest` (to lint arbitrary files), and set the appropriate class variables.
    E.g.
        class DryCleaningRequest(LintTargetsRequest):
            name = DryCleaningSubsystem.options_scope
            field_set_type = DryCleaningFieldSet

    Then, define 2 `@rule`s:
        1. A rule which takes an instance of your request type's `PartitionRequest` class property,
            and returns a `Partitions` instance.
            E.g.
                @rule
                async def partition(
                    request: DryCleaningRequest.PartitionRequest[DryCleaningFieldSet]
                    # or `request: DryCleaningRequest.PartitionRequest` if file linter
                    subsystem: DryCleaningSubsystem,
                ) -> Partitions[DryCleaningFieldSet, Any]:
                    if subsystem.skip:
                        return Partitions()

                    # One possible implementation
                    return Partitions.single_partition(request.field_sets)

        2. A rule which takes an instance of your request type's `Batch` class property, and
            returns a `LintResult instance.
            E.g.
                @rule
                async def dry_clean(
                    request: DryCleaningRequest.Batch,
                ) -> LintResult:
                    ...

    Lastly, register the rules which tell Pants about your plugin.
    E.g.
        def rules():
            return [
                *collect_rules(),
                *DryCleaningRequest.rules()
            ]
    """

    tool_subsystem: ClassVar[type[SkippableSubsystem]]
    partitioner_type: ClassVar[PartitionerType] = PartitionerType.CUSTOM

    is_formatter: ClassVar[bool] = False
    is_fixer: ClassVar[bool] = False

    @final
    @classproperty
    def _requires_snapshot(cls) -> bool:
        return cls.is_formatter or cls.is_fixer

    @classproperty
    def tool_name(cls) -> str:
        """The user-facing "name" of the tool."""
        return cls.tool_subsystem.options_scope

    @classproperty
    def tool_id(cls) -> str:
        """The "id" of the tool, used in tool selection (Eg --only=<id>)."""
        return cls.tool_subsystem.options_scope

    @distinct_union_type_per_subclass(in_scope_types=[EnvironmentName])
    class Batch(_BatchBase[PartitionElementT, PartitionMetadataT]):
        pass

    @final
    @classmethod
    def rules(cls) -> Iterable:
        yield from cls._get_rules()

    @classmethod
    def _get_rules(cls) -> Iterable:
        yield UnionRule(AbstractLintRequest, cls)
        yield UnionRule(AbstractLintRequest.Batch, cls.Batch)


class LintTargetsRequest(AbstractLintRequest):
    """The entry point for linters that operate on targets."""

    field_set_type: ClassVar[type[FieldSet]]

    @distinct_union_type_per_subclass(in_scope_types=[EnvironmentName])
    class PartitionRequest(_PartitionFieldSetsRequestBase[_FieldSetT]):
        pass

    @classmethod
    def _get_rules(cls) -> Iterable:
        yield from cls.partitioner_type.default_rules(cls, by_file=False)
        yield from super()._get_rules()
        yield UnionRule(LintTargetsRequest.PartitionRequest, cls.PartitionRequest)


class LintFilesRequest(AbstractLintRequest, EngineAwareParameter):
    """The entry point for linters that do not use targets."""

    @distinct_union_type_per_subclass(in_scope_types=[EnvironmentName])
    class PartitionRequest(_PartitionFilesRequestBase):
        pass

    @classmethod
    def _get_rules(cls) -> Iterable:
        if cls.partitioner_type is not PartitionerType.CUSTOM:
            raise ValueError(
                "Pants does not provide default partitioners for `LintFilesRequest`."
                + " You will need to provide your own partitioner rule."
            )

        yield from super()._get_rules()
        yield UnionRule(LintFilesRequest.PartitionRequest, cls.PartitionRequest)


# If a user wants linter reports to show up in dist/ they must ensure that the reports
# are written under this directory. E.g.,
# ./pants --flake8-args="--output-file=reports/report.txt" lint <target>
REPORT_DIR = "reports"


class LintSubsystem(GoalSubsystem):
    name = "lint"
    help = softwrap(
        f"""
        Run linters/formatters/fixers in check mode.

        This goal runs tools that check code quality/styling etc, without changing that code. This
        includes running formatters and fixers, but instead of writing changes back to the
        workspace, Pants treats any changes they would make as a linting failure.

        See also:

        - [The `fmt` goal]({doc_url("reference/goals/fix")} will save the the result of formatters
          (code-editing tools that make only "syntactic" changes) back to the workspace.

        - [The `fmt` goal]({doc_url("reference/goals/fix")} will save the the result of fixers
          (code-editing tools that may make "semantic" changes too) back to the workspace.

        - Documentation about linters for various ecosystems, such as:
          [Python]({doc_url("docs/python/overview/linters-and-formatters")}), [Go]({doc_url("docs/go")}),
          [JVM]({doc_url("jvm/java-and-scala#lint-and-format")}), [Shell]({doc_url("docs/shell")}),
          [Docker]({doc_url("docs/docker#linting-dockerfiles-with-hadolint")}).

        """
    )

    @classmethod
    def activated(cls, union_membership: UnionMembership) -> bool:
        return AbstractLintRequest in union_membership

    only = OnlyOption("linter", "flake8", "shellcheck")
    skip_formatters = BoolOption(
        default=False,
        help=softwrap(
            f"""
            If true, skip running all formatters in check-only mode.

            FYI: when running `{bin_name()} fmt lint ::`, there should be diminishing performance
            benefit to using this flag. Pants attempts to reuse the results from `fmt` when running
            `lint` where possible.
            """
        ),
    )
    skip_fixers = BoolOption(
        default=False,
        help=softwrap(
            f"""
            If true, skip running all fixers in check-only mode.

            FYI: when running `{bin_name()} fix lint ::`, there should be diminishing performance
            benefit to using this flag. Pants attempts to reuse the results from `fix` when running
            `lint` where possible.
            """
        ),
    )
    batch_size = BatchSizeOption(uppercase="Linter", lowercase="linter")


class Lint(Goal):
    subsystem_cls = LintSubsystem
    environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY


_CoreRequestType = TypeVar("_CoreRequestType", bound=AbstractLintRequest)
_TargetPartitioner = TypeVar("_TargetPartitioner", bound=LintTargetsRequest.PartitionRequest)
_FilePartitioner = TypeVar("_FilePartitioner", bound=LintFilesRequest.PartitionRequest)


class _MultiToolGoalSubsystem(Protocol):
    name: str
    only: OnlyOption


async def get_partitions_by_request_type(
    core_request_types: Iterable[type[_CoreRequestType]],
    target_partitioners: Iterable[type[_TargetPartitioner]],
    file_partitioners: Iterable[type[_FilePartitioner]],
    subsystem: _MultiToolGoalSubsystem,
    specs: Specs,
    # NB: Because the rule parser code will collect rule calls from caller's scope, these allow the
    # caller to customize the specific rule.
    make_targets_partition_request: Callable[[_TargetPartitioner], Coroutine[Any, Any, Partitions]],
    make_files_partition_request: Callable[[_FilePartitioner], Coroutine[Any, Any, Partitions]],
) -> dict[type[_CoreRequestType], list[Partitions]]:
    specified_ids = determine_specified_tool_ids(
        subsystem.name,
        subsystem.only,
        core_request_types,
    )

    filtered_core_request_types = [
        request_type for request_type in core_request_types if request_type.tool_id in specified_ids
    ]
    if not filtered_core_request_types:
        return {}

    core_partition_request_types = {
        getattr(request_type, "PartitionRequest") for request_type in filtered_core_request_types
    }
    target_partitioners = [
        target_partitioner
        for target_partitioner in target_partitioners
        if target_partitioner in core_partition_request_types
    ]
    file_partitioners = [
        file_partitioner
        for file_partitioner in file_partitioners
        if file_partitioner in core_partition_request_types
    ]

    _get_targets = filter_targets(
        **implicitly({(specs if target_partitioners else Specs.empty()): Specs})
    )
    _get_specs_paths = resolve_specs_paths(specs if file_partitioners else Specs.empty())

    targets, specs_paths = await concurrently(_get_targets, _get_specs_paths)

    await _warn_on_non_local_environments(targets, f"the {subsystem.name} goal")

    def partition_request_get(
        request_type: type[AbstractLintRequest],
    ) -> Coroutine[Any, Any, Partitions]:
        partition_request_type: type = getattr(request_type, "PartitionRequest")
        if partition_request_type in target_partitioners:
            partition_targets_type = cast(LintTargetsRequest, request_type)
            field_set_type = partition_targets_type.field_set_type
            field_sets = tuple(
                field_set_type.create(target)
                for target in targets
                if field_set_type.is_applicable(target)
            )
            return make_targets_partition_request(
                partition_targets_type.PartitionRequest(field_sets)  # type: ignore[arg-type]
            )
        else:
            assert partition_request_type in file_partitioners
            partition_files_type = cast(LintFilesRequest, request_type)
            return make_files_partition_request(
                partition_files_type.PartitionRequest(specs_paths.files)  # type: ignore[arg-type]
            )

    all_partitions = await concurrently(
        partition_request_get(request_type) for request_type in filtered_core_request_types
    )
    partitions_by_request_type = defaultdict(list)
    for request_type, partition in zip(filtered_core_request_types, all_partitions):
        partitions_by_request_type[request_type].append(partition)

    return partitions_by_request_type


@rule(polymorphic=True)
async def partition_targets(req: LintTargetsRequest.PartitionRequest) -> Partitions:
    raise NotImplementedError()


@rule(polymorphic=True)
async def partition_files(req: LintFilesRequest.PartitionRequest) -> Partitions:
    raise NotImplementedError()


@rule(polymorphic=True)
async def lint_batch(batch: AbstractLintRequest.Batch) -> LintResult:
    raise NotImplementedError()


def rules():
    return collect_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	7✔
5
6	import logging	7✔
7	from collections import defaultdict	7✔
8	from collections.abc import Callable, Coroutine, Iterable	7✔
9	from dataclasses import dataclass	7✔
10	from typing import Any, ClassVar, Protocol, TypeVar, cast, final	7✔
11
12	from pants.base.specs import Specs	7✔
13	from pants.core.environments.rules import _warn_on_non_local_environments	7✔
14	from pants.core.goals.multi_tool_goal_helper import (	7✔
15	BatchSizeOption,
16	OnlyOption,
17	SkippableSubsystem,
18	determine_specified_tool_ids,
19	)
20	from pants.core.util_rules.partitions import PartitionElementT, PartitionerType, PartitionMetadataT	7✔
21	from pants.core.util_rules.partitions import Partitions as Partitions # re-export	7✔
22	from pants.core.util_rules.partitions import (	7✔
23	_BatchBase,
24	_PartitionFieldSetsRequestBase,
25	_PartitionFilesRequestBase,
26	)
27	from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType	7✔
28	from pants.engine.environment import EnvironmentName	7✔
29	from pants.engine.fs import EMPTY_DIGEST, Digest	7✔
30	from pants.engine.goal import Goal, GoalSubsystem	7✔
31	from pants.engine.internals.graph import filter_targets	7✔
32	from pants.engine.internals.specs_rules import resolve_specs_paths	7✔
33	from pants.engine.process import FallibleProcessResult	7✔
34	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	7✔
35	from pants.engine.target import FieldSet	7✔
36	from pants.engine.unions import UnionMembership, UnionRule, distinct_union_type_per_subclass, union	7✔
37	from pants.option.option_types import BoolOption	7✔
38	from pants.util.docutil import bin_name, doc_url	7✔
39	from pants.util.logging import LogLevel	7✔
40	from pants.util.meta import classproperty	7✔
41	from pants.util.strutil import Simplifier, softwrap	7✔
42
43	logger = logging.getLogger(__name__)	7✔
44
45	_T = TypeVar("_T")	7✔
46	_FieldSetT = TypeVar("_FieldSetT", bound=FieldSet)	7✔
47
48
49	@dataclass(frozen=True)	7✔
50	class LintResult(EngineAwareReturnType):	7✔
51	exit_code: int	7✔
52	stdout: str	7✔
53	stderr: str	7✔
54	linter_name: str	7✔
55	partition_description: str \| None = None	7✔
56	report: Digest = EMPTY_DIGEST	7✔
57	_render_message: bool = True	7✔
58
59	@classmethod	7✔
60	def create(	7✔
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
77	def metadata(self) -> dict[str, Any]:	7✔
78	return {"partition": self.partition_description}	×
79
80	def level(self) -> LogLevel \| None:	7✔
81	if not self._render_message:	×
82	return LogLevel.TRACE	×
83	if self.exit_code != 0:	×
84	return LogLevel.ERROR	×
85	return LogLevel.INFO	×
86
87	def message(self) -> str \| None:	7✔
88	message = self.linter_name	×
89	message += (	×
90	" succeeded." if self.exit_code == 0 else f" failed (exit code {self.exit_code})."
91	)
92	if self.partition_description:	×
93	message += f"\nPartition: {self.partition_description}"	×
94	if self.stdout:	×
95	message += f"\n{self.stdout}"	×
96	if self.stderr:	×
97	message += f"\n{self.stderr}"	×
98	if self.partition_description or self.stdout or self.stderr:	×
99	message += "\n\n"	×
100
101	return message	×
102
103	def cacheable(self) -> bool:	7✔
104	"""Is marked uncacheable to ensure that it always renders."""
105	return False	×
106
107
108	@union	7✔
109	class AbstractLintRequest:	7✔
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
153	tool_subsystem: ClassVar[type[SkippableSubsystem]]	7✔
154	partitioner_type: ClassVar[PartitionerType] = PartitionerType.CUSTOM	7✔
155
156	is_formatter: ClassVar[bool] = False	7✔
157	is_fixer: ClassVar[bool] = False	7✔
158
159	@final	7✔
160	@classproperty	7✔
161	def _requires_snapshot(cls) -> bool:	7✔
162	return cls.is_formatter or cls.is_fixer	×
163
164	@classproperty	7✔
165	def tool_name(cls) -> str:	7✔
166	"""The user-facing "name" of the tool."""
167	return cls.tool_subsystem.options_scope	×
168
169	@classproperty	7✔
170	def tool_id(cls) -> str:	7✔
171	"""The "id" of the tool, used in tool selection (Eg --only=<id>)."""
172	return cls.tool_subsystem.options_scope	×
173
174	@distinct_union_type_per_subclass(in_scope_types=[EnvironmentName])	7✔
175	class Batch(_BatchBase[PartitionElementT, PartitionMetadataT]):	7✔
176	pass	7✔
177
178	@final	7✔
179	@classmethod	7✔
180	def rules(cls) -> Iterable:	7✔
181	yield from cls._get_rules()	7✔
182
183	@classmethod	7✔
184	def _get_rules(cls) -> Iterable:	7✔
185	yield UnionRule(AbstractLintRequest, cls)	7✔
186	yield UnionRule(AbstractLintRequest.Batch, cls.Batch)	7✔
187
188
189	class LintTargetsRequest(AbstractLintRequest):	7✔
190	"""The entry point for linters that operate on targets."""
191
192	field_set_type: ClassVar[type[FieldSet]]	7✔
193
194	@distinct_union_type_per_subclass(in_scope_types=[EnvironmentName])	7✔
195	class PartitionRequest(_PartitionFieldSetsRequestBase[_FieldSetT]):	7✔
196	pass	7✔
197
198	@classmethod	7✔
199	def _get_rules(cls) -> Iterable:	7✔
200	yield from cls.partitioner_type.default_rules(cls, by_file=False)	7✔
201	yield from super()._get_rules()	7✔
202	yield UnionRule(LintTargetsRequest.PartitionRequest, cls.PartitionRequest)	7✔
203
204
205	class LintFilesRequest(AbstractLintRequest, EngineAwareParameter):	7✔
206	"""The entry point for linters that do not use targets."""
207
208	@distinct_union_type_per_subclass(in_scope_types=[EnvironmentName])	7✔
209	class PartitionRequest(_PartitionFilesRequestBase):	7✔
210	pass	7✔
211
212	@classmethod	7✔
213	def _get_rules(cls) -> Iterable:	7✔
214	if cls.partitioner_type is not PartitionerType.CUSTOM:	6✔
215	raise ValueError(	×
216	"Pants does not provide default partitioners for `LintFilesRequest`."
217	+ " You will need to provide your own partitioner rule."
218	)
219
220	yield from super()._get_rules()	6✔
221	yield UnionRule(LintFilesRequest.PartitionRequest, cls.PartitionRequest)	6✔
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>
227	REPORT_DIR = "reports"	7✔
228
229
230	class LintSubsystem(GoalSubsystem):	7✔
231	name = "lint"	7✔
232	help = softwrap(	7✔
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
256	@classmethod	7✔
257	def activated(cls, union_membership: UnionMembership) -> bool:	7✔
258	return AbstractLintRequest in union_membership	×
259
260	only = OnlyOption("linter", "flake8", "shellcheck")	7✔
261	skip_formatters = BoolOption(	7✔
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	)
273	skip_fixers = BoolOption(	7✔
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	)
285	batch_size = BatchSizeOption(uppercase="Linter", lowercase="linter")	7✔
286
287
288	class Lint(Goal):	7✔
289	subsystem_cls = LintSubsystem	7✔
290	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	7✔
291
292
293	_CoreRequestType = TypeVar("_CoreRequestType", bound=AbstractLintRequest)	7✔
294	_TargetPartitioner = TypeVar("_TargetPartitioner", bound=LintTargetsRequest.PartitionRequest)	7✔
295	_FilePartitioner = TypeVar("_FilePartitioner", bound=LintFilesRequest.PartitionRequest)	7✔
296
297
298	class _MultiToolGoalSubsystem(Protocol):	7✔
299	name: str	7✔
300	only: OnlyOption	7✔
301
302
303	async def get_partitions_by_request_type(	7✔
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]]:
314	specified_ids = determine_specified_tool_ids(	×
315	subsystem.name,
316	subsystem.only,
317	core_request_types,
318	)
319
320	filtered_core_request_types = [	×
321	request_type for request_type in core_request_types if request_type.tool_id in specified_ids
322	]
323	if not filtered_core_request_types:	×
324	return {}	×
325
326	core_partition_request_types = {	×
327	getattr(request_type, "PartitionRequest") for request_type in filtered_core_request_types
328	}
329	target_partitioners = [	×
330	target_partitioner
331	for target_partitioner in target_partitioners
332	if target_partitioner in core_partition_request_types
333	]
334	file_partitioners = [	×
335	file_partitioner
336	for file_partitioner in file_partitioners
337	if file_partitioner in core_partition_request_types
338	]
339
340	_get_targets = filter_targets(	×
341	**implicitly({(specs if target_partitioners else Specs.empty()): Specs})
342	)
343	_get_specs_paths = resolve_specs_paths(specs if file_partitioners else Specs.empty())	×
344
345	targets, specs_paths = await concurrently(_get_targets, _get_specs_paths)	×
346
347	await _warn_on_non_local_environments(targets, f"the {subsystem.name} goal")	×
348
349	def partition_request_get(	×
350	request_type: type[AbstractLintRequest],
351	) -> Coroutine[Any, Any, Partitions]:
352	partition_request_type: type = getattr(request_type, "PartitionRequest")	×
353	if partition_request_type in target_partitioners:	×
354	partition_targets_type = cast(LintTargetsRequest, request_type)	×
355	field_set_type = partition_targets_type.field_set_type	×
356	field_sets = tuple(	×
357	field_set_type.create(target)
358	for target in targets
359	if field_set_type.is_applicable(target)
360	)
361	return make_targets_partition_request(	×
362	partition_targets_type.PartitionRequest(field_sets) # type: ignore[arg-type]
363	)
364	else:
365	assert partition_request_type in file_partitioners	×
366	partition_files_type = cast(LintFilesRequest, request_type)	×
367	return make_files_partition_request(	×
368	partition_files_type.PartitionRequest(specs_paths.files) # type: ignore[arg-type]
369	)
370
371	all_partitions = await concurrently(	×
372	partition_request_get(request_type) for request_type in filtered_core_request_types
373	)
374	partitions_by_request_type = defaultdict(list)	×
375	for request_type, partition in zip(filtered_core_request_types, all_partitions):	×
376	partitions_by_request_type[request_type].append(partition)	×
377
378	return partitions_by_request_type	×
379
380
381	@rule(polymorphic=True)	7✔
382	async def partition_targets(req: LintTargetsRequest.PartitionRequest) -> Partitions:	7✔
383	raise NotImplementedError()	×
384
385
386	@rule(polymorphic=True)	7✔
387	async def partition_files(req: LintFilesRequest.PartitionRequest) -> Partitions:	7✔
388	raise NotImplementedError()	×
389
390
391	@rule(polymorphic=True)	7✔
392	async def lint_batch(batch: AbstractLintRequest.Batch) -> LintResult:	7✔
393	raise NotImplementedError()	×
394
395
396	def rules():	7✔
397	return collect_rules()	3✔

pantsbuild / pants / 18517631058

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