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

56.12

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

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