20332790708

Committed 18 Dec 2025 09:48AM UTC coverage: 64.992% (-15.3%) from 80.295%

Build # 20332790708

Build Type

Pull #22949

github

Committed by

web-flow

Commit Message

Merge f730a56cd into 407284c67

Pull Request Pull Request #22949: Add experimental uv resolver for Python lockfiles

Run Details

54 of 97 new or added lines in 5 files covered. (55.67%)

8270 existing lines in 295 files now uncovered.

48990 of 75379 relevant lines covered (64.99%)

1.81 hits per line

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

68.45

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

pantsbuild / pants / 20332790708

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