19381742489

Committed 15 Nov 2025 12:52AM UTC coverage: 49.706% (-30.6%) from 80.29%

Build # 19381742489

Build Type

Pull #22890

github

Committed by

web-flow

Commit Message

Merge d961abf79 into 42e1ebd41

Pull Request Pull Request #22890: Updated all python subsystem constraints to 3.14

Run Details

4 of 5 new or added lines in 5 files covered. (80.0%)

14659 existing lines in 485 files now uncovered.

31583 of 63540 relevant lines covered (49.71%)

0.79 hits per line

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

53.96

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

# Copyright 2020 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 Iterable
from dataclasses import dataclass
from typing import Any, ClassVar, Generic, TypeVar, cast

from pants.core.environments.rules import EnvironmentNameRequest, resolve_environment_name
from pants.core.goals.lint import REPORT_DIR as REPORT_DIR  # noqa: F401
from pants.core.goals.multi_tool_goal_helper import (
    OnlyOption,
    determine_specified_tool_ids,
    write_reports,
)
from pants.core.util_rules.distdir import DistDir
from pants.engine.collection import Collection
from pants.engine.console import Console
from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType
from pants.engine.environment import EnvironmentName
from pants.engine.fs import EMPTY_DIGEST, Digest, Workspace
from pants.engine.goal import Goal, GoalSubsystem
from pants.engine.internals.selectors import concurrently
from pants.engine.process import FallibleProcessResult
from pants.engine.rules import QueryRule, collect_rules, goal_rule, implicitly, rule
from pants.engine.target import FieldSet, FilteredTargets
from pants.engine.unions import UnionMembership, union
from pants.util.logging import LogLevel
from pants.util.memo import memoized_property
from pants.util.meta import classproperty
from pants.util.strutil import Simplifier

logger = logging.getLogger(__name__)

_FS = TypeVar("_FS", bound=FieldSet)


@dataclass(frozen=True)
class CheckResult:
    exit_code: int
    stdout: str
    stderr: str
    partition_description: str | None = None
    report: Digest = EMPTY_DIGEST

    @staticmethod
    def from_fallible_process_result(
        process_result: FallibleProcessResult,
        *,
        partition_description: str | None = None,
        output_simplifier: Simplifier = Simplifier(),
        report: Digest = EMPTY_DIGEST,
    ) -> CheckResult:
        return CheckResult(
            exit_code=process_result.exit_code,
            stdout=output_simplifier.simplify(process_result.stdout),
            stderr=output_simplifier.simplify(process_result.stderr),
            partition_description=partition_description,
            report=report,
        )

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


@dataclass(frozen=True)
class CheckResults(EngineAwareReturnType):
    """Zero or more CheckResult objects for a single type checker.

    Typically, type checkers will return one result. If they no-oped, they will return zero results.
    However, some type checkers may need to partition their input and thus may need to return
    multiple results.
    """

    results: tuple[CheckResult, ...]
    checker_name: str

    def __init__(self, results: Iterable[CheckResult], *, checker_name: str) -> None:
        object.__setattr__(self, "results", tuple(results))
        object.__setattr__(self, "checker_name", checker_name)

    @property
    def skipped(self) -> bool:
        return bool(self.results) is False

    @memoized_property
    def exit_code(self) -> int:
        return next((result.exit_code for result in self.results if result.exit_code != 0), 0)

    def level(self) -> LogLevel | None:
        if self.skipped:
            return LogLevel.DEBUG
        return LogLevel.ERROR if self.exit_code != 0 else LogLevel.INFO

    def message(self) -> str | None:
        if self.skipped:
            return f"{self.checker_name} skipped."
        message = self.checker_name
        message += (
            " succeeded." if self.exit_code == 0 else f" failed (exit code {self.exit_code})."
        )

        def msg_for_result(result: CheckResult) -> str:
            msg = ""
            if result.stdout:
                msg += f"\n{result.stdout}"
            if result.stderr:
                msg += f"\n{result.stderr}"
            if msg:
                msg = f"{msg.rstrip()}\n\n"
            return msg

        if len(self.results) == 1:
            results_msg = msg_for_result(self.results[0])
        else:
            results_msg = "\n"
            for i, result in enumerate(self.results):
                msg = f"Partition #{i + 1}"
                msg += (
                    f" - {result.partition_description}:" if result.partition_description else ":"
                )
                msg += msg_for_result(result) or "\n\n"
                results_msg += msg
        message += results_msg
        return message

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


@dataclass(frozen=True)
@union(in_scope_types=[EnvironmentName])
class CheckRequest(Generic[_FS], EngineAwareParameter):
    """A union for targets that should be checked.

    Subclass and install a member of this type to provide a checker.
    """

    field_set_type: ClassVar[type[_FS]]
    tool_name: ClassVar[str]

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

    field_sets: Collection[_FS]

    def __init__(self, field_sets: Iterable[_FS]) -> None:
        object.__setattr__(self, "field_sets", Collection[_FS](field_sets))

    def debug_hint(self) -> str:
        return self.tool_name

    def metadata(self) -> dict[str, Any]:
        return {"addresses": [fs.address.spec for fs in self.field_sets]}


@rule(polymorphic=True)
async def check(
    req: CheckRequest,
    environment_name: EnvironmentName,
) -> CheckResults:
    raise NotImplementedError()


class CheckSubsystem(GoalSubsystem):
    name = "check"
    help = "Run type checking or the lightest variant of compilation available for a language."

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

    only = OnlyOption("checker", "mypy", "javac")


class Check(Goal):
    subsystem_cls = CheckSubsystem
    environment_behavior = Goal.EnvironmentBehavior.USES_ENVIRONMENTS


@goal_rule
async def check_goal(
    console: Console,
    workspace: Workspace,
    targets: FilteredTargets,
    dist_dir: DistDir,
    union_membership: UnionMembership,
    check_subsystem: CheckSubsystem,
) -> Check:
    request_types = cast("Iterable[type[CheckRequest]]", union_membership[CheckRequest])
    specified_ids = determine_specified_tool_ids("check", check_subsystem.only, request_types)

    requests = tuple(
        request_type(
            request_type.field_set_type.create(target)
            for target in targets
            if (
                request_type.tool_id in specified_ids
                and request_type.field_set_type.is_applicable(target)
            )
        )
        for request_type in request_types
    )

    request_to_field_set = [
        (request, field_set) for request in requests for field_set in request.field_sets
    ]

    environment_names = await concurrently(
        resolve_environment_name(EnvironmentNameRequest.from_field_set(field_set), **implicitly())
        for (_, field_set) in request_to_field_set
    )

    request_to_env_name = {
        (request, env_name)
        for (request, _), env_name in zip(request_to_field_set, environment_names)
    }

    # Run each check request in each valid environment (potentially multiple runs per tool)
    all_results = await concurrently(
        check(**implicitly({request: CheckRequest, env_name: EnvironmentName}))
        for (request, env_name) in request_to_env_name
    )

    results_by_tool: dict[str, list[CheckResult]] = defaultdict(list)
    for results in all_results:
        results_by_tool[results.checker_name].extend(results.results)

    write_reports(
        results_by_tool,
        workspace,
        dist_dir,
        goal_name=CheckSubsystem.name,
    )

    exit_code = 0
    if all_results:
        console.print_stderr("")
    for results in sorted(all_results, key=lambda results: results.checker_name):
        if results.skipped:
            continue
        elif results.exit_code == 0:
            sigil = console.sigil_succeeded()
            status = "succeeded"
        else:
            sigil = console.sigil_failed()
            status = "failed"
            exit_code = results.exit_code
        console.print_stderr(f"{sigil} {results.checker_name} {status}.")

    return Check(exit_code)


def rules():
    return [
        *collect_rules(),
        # NB: Would be unused otherwise.
        QueryRule(CheckSubsystem, []),
    ]

1	# Copyright 2020 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	import logging	2✔
7	from collections import defaultdict	2✔
8	from collections.abc import Iterable	2✔
9	from dataclasses import dataclass	2✔
10	from typing import Any, ClassVar, Generic, TypeVar, cast	2✔
11
12	from pants.core.environments.rules import EnvironmentNameRequest, resolve_environment_name	2✔
13	from pants.core.goals.lint import REPORT_DIR as REPORT_DIR # noqa: F401	2✔
14	from pants.core.goals.multi_tool_goal_helper import (	2✔
15	OnlyOption,
16	determine_specified_tool_ids,
17	write_reports,
18	)
19	from pants.core.util_rules.distdir import DistDir	2✔
20	from pants.engine.collection import Collection	2✔
21	from pants.engine.console import Console	2✔
22	from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType	2✔
23	from pants.engine.environment import EnvironmentName	2✔
24	from pants.engine.fs import EMPTY_DIGEST, Digest, Workspace	2✔
25	from pants.engine.goal import Goal, GoalSubsystem	2✔
26	from pants.engine.internals.selectors import concurrently	2✔
27	from pants.engine.process import FallibleProcessResult	2✔
28	from pants.engine.rules import QueryRule, collect_rules, goal_rule, implicitly, rule	2✔
29	from pants.engine.target import FieldSet, FilteredTargets	2✔
30	from pants.engine.unions import UnionMembership, union	2✔
31	from pants.util.logging import LogLevel	2✔
32	from pants.util.memo import memoized_property	2✔
33	from pants.util.meta import classproperty	2✔
34	from pants.util.strutil import Simplifier	2✔
35
36	logger = logging.getLogger(__name__)	2✔
37
38	_FS = TypeVar("_FS", bound=FieldSet)	2✔
39
40
41	@dataclass(frozen=True)	2✔
42	class CheckResult:	2✔
43	exit_code: int	2✔
44	stdout: str	2✔
45	stderr: str	2✔
46	partition_description: str \| None = None	2✔
47	report: Digest = EMPTY_DIGEST	2✔
48
49	@staticmethod	2✔
50	def from_fallible_process_result(	2✔
51	process_result: FallibleProcessResult,
52	*,
53	partition_description: str \| None = None,
54	output_simplifier: Simplifier = Simplifier(),
55	report: Digest = EMPTY_DIGEST,
56	) -> CheckResult:
UNCOV 57	return CheckResult(	×
58	exit_code=process_result.exit_code,
59	stdout=output_simplifier.simplify(process_result.stdout),
60	stderr=output_simplifier.simplify(process_result.stderr),
61	partition_description=partition_description,
62	report=report,
63	)
64
65	def metadata(self) -> dict[str, Any]:	2✔
66	return {"partition": self.partition_description}	×
67
68
69	@dataclass(frozen=True)	2✔
70	class CheckResults(EngineAwareReturnType):	2✔
71	"""Zero or more CheckResult objects for a single type checker.
72
73	Typically, type checkers will return one result. If they no-oped, they will return zero results.
74	However, some type checkers may need to partition their input and thus may need to return
75	multiple results.
76	"""
77
78	results: tuple[CheckResult, ...]	2✔
79	checker_name: str	2✔
80
81	def __init__(self, results: Iterable[CheckResult], *, checker_name: str) -> None:	2✔
UNCOV 82	object.__setattr__(self, "results", tuple(results))	×
UNCOV 83	object.__setattr__(self, "checker_name", checker_name)	×
84
85	@property	2✔
86	def skipped(self) -> bool:	2✔
UNCOV 87	return bool(self.results) is False	×
88
89	@memoized_property	2✔
90	def exit_code(self) -> int:	2✔
UNCOV 91	return next((result.exit_code for result in self.results if result.exit_code != 0), 0)	×
92
93	def level(self) -> LogLevel \| None:	2✔
UNCOV 94	if self.skipped:	×
UNCOV 95	return LogLevel.DEBUG	×
UNCOV 96	return LogLevel.ERROR if self.exit_code != 0 else LogLevel.INFO	×
97
98	def message(self) -> str \| None:	2✔
UNCOV 99	if self.skipped:	×
UNCOV 100	return f"{self.checker_name} skipped."	×
UNCOV 101	message = self.checker_name	×
UNCOV 102	message += (	×
103	" succeeded." if self.exit_code == 0 else f" failed (exit code {self.exit_code})."
104	)
105
UNCOV 106	def msg_for_result(result: CheckResult) -> str:	×
UNCOV 107	msg = ""	×
UNCOV 108	if result.stdout:	×
UNCOV 109	msg += f"\n{result.stdout}"	×
UNCOV 110	if result.stderr:	×
UNCOV 111	msg += f"\n{result.stderr}"	×
UNCOV 112	if msg:	×
UNCOV 113	msg = f"{msg.rstrip()}\n\n"	×
UNCOV 114	return msg	×
115
UNCOV 116	if len(self.results) == 1:	×
UNCOV 117	results_msg = msg_for_result(self.results[0])	×
118	else:
UNCOV 119	results_msg = "\n"	×
UNCOV 120	for i, result in enumerate(self.results):	×
UNCOV 121	msg = f"Partition #{i + 1}"	×
UNCOV 122	msg += (	×
123	f" - {result.partition_description}:" if result.partition_description else ":"
124	)
UNCOV 125	msg += msg_for_result(result) or "\n\n"	×
UNCOV 126	results_msg += msg	×
UNCOV 127	message += results_msg	×
UNCOV 128	return message	×
129
130	def cacheable(self) -> bool:	2✔
131	"""Is marked uncacheable to ensure that it always renders."""
132	return False	×
133
134
135	@dataclass(frozen=True)	2✔
136	@union(in_scope_types=[EnvironmentName])	2✔
137	class CheckRequest(Generic[_FS], EngineAwareParameter):	2✔
138	"""A union for targets that should be checked.
139
140	Subclass and install a member of this type to provide a checker.
141	"""
142
143	field_set_type: ClassVar[type[_FS]]	2✔
144	tool_name: ClassVar[str]	2✔
145
146	@classproperty	2✔
147	def tool_id(cls) -> str:	2✔
148	"""The "id" of the tool, used in tool selection (Eg --only=<id>)."""
149	return cls.tool_name	×
150
151	field_sets: Collection[_FS]	2✔
152
153	def __init__(self, field_sets: Iterable[_FS]) -> None:	2✔
154	object.__setattr__(self, "field_sets", Collection[_FS](field_sets))	2✔
155
156	def debug_hint(self) -> str:	2✔
157	return self.tool_name	×
158
159	def metadata(self) -> dict[str, Any]:	2✔
160	return {"addresses": [fs.address.spec for fs in self.field_sets]}	×
161
162
163	@rule(polymorphic=True)	2✔
164	async def check(	2✔
165	req: CheckRequest,
166	environment_name: EnvironmentName,
167	) -> CheckResults:
168	raise NotImplementedError()	×
169
170
171	class CheckSubsystem(GoalSubsystem):	2✔
172	name = "check"	2✔
173	help = "Run type checking or the lightest variant of compilation available for a language."	2✔
174
175	@classmethod	2✔
176	def activated(cls, union_membership: UnionMembership) -> bool:	2✔
177	return CheckRequest in union_membership	×
178
179	only = OnlyOption("checker", "mypy", "javac")	2✔
180
181
182	class Check(Goal):	2✔
183	subsystem_cls = CheckSubsystem	2✔
184	environment_behavior = Goal.EnvironmentBehavior.USES_ENVIRONMENTS	2✔
185
186
187	@goal_rule	2✔
188	async def check_goal(	2✔
189	console: Console,
190	workspace: Workspace,
191	targets: FilteredTargets,
192	dist_dir: DistDir,
193	union_membership: UnionMembership,
194	check_subsystem: CheckSubsystem,
195	) -> Check:
UNCOV 196	request_types = cast("Iterable[type[CheckRequest]]", union_membership[CheckRequest])	×
UNCOV 197	specified_ids = determine_specified_tool_ids("check", check_subsystem.only, request_types)	×
198
UNCOV 199	requests = tuple(	×
200	request_type(
201	request_type.field_set_type.create(target)
202	for target in targets
203	if (
204	request_type.tool_id in specified_ids
205	and request_type.field_set_type.is_applicable(target)
206	)
207	)
208	for request_type in request_types
209	)
210
UNCOV 211	request_to_field_set = [	×
212	(request, field_set) for request in requests for field_set in request.field_sets
213	]
214
UNCOV 215	environment_names = await concurrently(	×
216	resolve_environment_name(EnvironmentNameRequest.from_field_set(field_set), **implicitly())
217	for (_, field_set) in request_to_field_set
218	)
219
UNCOV 220	request_to_env_name = {	×
221	(request, env_name)
222	for (request, _), env_name in zip(request_to_field_set, environment_names)
223	}
224
225	# Run each check request in each valid environment (potentially multiple runs per tool)
UNCOV 226	all_results = await concurrently(	×
227	check(**implicitly({request: CheckRequest, env_name: EnvironmentName}))
228	for (request, env_name) in request_to_env_name
229	)
230
UNCOV 231	results_by_tool: dict[str, list[CheckResult]] = defaultdict(list)	×
UNCOV 232	for results in all_results:	×
UNCOV 233	results_by_tool[results.checker_name].extend(results.results)	×
234
UNCOV 235	write_reports(	×
236	results_by_tool,
237	workspace,
238	dist_dir,
239	goal_name=CheckSubsystem.name,
240	)
241
UNCOV 242	exit_code = 0	×
UNCOV 243	if all_results:	×
UNCOV 244	console.print_stderr("")	×
UNCOV 245	for results in sorted(all_results, key=lambda results: results.checker_name):	×
UNCOV 246	if results.skipped:	×
UNCOV 247	continue	×
UNCOV 248	elif results.exit_code == 0:	×
UNCOV 249	sigil = console.sigil_succeeded()	×
UNCOV 250	status = "succeeded"	×
251	else:
UNCOV 252	sigil = console.sigil_failed()	×
UNCOV 253	status = "failed"	×
UNCOV 254	exit_code = results.exit_code	×
UNCOV 255	console.print_stderr(f"{sigil} {results.checker_name} {status}.")	×
256
UNCOV 257	return Check(exit_code)	×
258
259
260	def rules():	2✔
UNCOV 261	return [	×
262	*collect_rules(),
263	# NB: Would be unused otherwise.
264	QueryRule(CheckSubsystem, []),
265	]

pantsbuild / pants / 19381742489

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