22285099215

Committed 22 Feb 2026 08:52PM UTC coverage: 75.854% (-17.1%) from 92.936%

Build # 22285099215

Build Type

Pull #23121

github

Committed by

web-flow

Commit Message

Merge c7299df9c into ba8359840

Pull Request Pull Request #23121: fix issue with optional fields in dependency validator

Run Details

28 of 29 new or added lines in 2 files covered. (96.55%)

11174 existing lines in 400 files now uncovered.

53694 of 70786 relevant lines covered (75.85%)

1.88 hits per line

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

79.14

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

pantsbuild / pants / 22285099215

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