22507851448

Committed 27 Feb 2026 11:28PM UTC coverage: 92.928% (-0.007%) from 92.935%

Build # 22507851448

Build Type

push

github

Committed by

web-flow

Commit Message

silence new HdrHistogram induced deprecation warning on Python 3.14 (#23144)

No more:
```
/usr/lib/python3.14/ctypes/_endian.py:33: DeprecationWarning: Due to '_pack_', the
'ExternalHeader' Structure will use memory layout compatible with MSVC (Windows). If this is intended, set _layout_ to 'ms'. The
 implicit default is deprecated and slated to become an error in Python 3.19.
  super().__setattr__(attrname, value)
/usr/lib/python3.14/ctypes/_endian.py:33: DeprecationWarning: Due to '_pack_', the 'PayloadHeader' Structure will use memory
layout compatible with MSVC (Windows). If this is intended, set _layout_ to 'ms'. The implicit default is deprecated and slated
to become an error in Python 3.19.
  super().__setattr__(attrname, value)
```

Run Details

0 of 1 new or added line in 1 file covered. (0.0%)

71 existing lines in 12 files now uncovered.

90912 of 97831 relevant lines covered (92.93%)

4.06 hits per line

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

97.18

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

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

from __future__ import annotations

from abc import ABCMeta, abstractmethod
from collections.abc import Iterable, Sequence
from pathlib import Path
from textwrap import dedent

import pytest

from pants.core.goals.check import (
    Check,
    CheckRequest,
    CheckResult,
    CheckResults,
    CheckSubsystem,
    check_goal,
)
from pants.core.util_rules.distdir import DistDir
from pants.engine.addresses import Address
from pants.engine.environment import EnvironmentName
from pants.engine.fs import EMPTY_DIGEST, EMPTY_FILE_DIGEST, Workspace
from pants.engine.internals.session import RunId
from pants.engine.platform import Platform
from pants.engine.process import (
    FallibleProcessResult,
    ProcessExecutionEnvironment,
    ProcessResultMetadata,
)
from pants.engine.target import FieldSet, MultipleSourcesField, Target, Targets
from pants.engine.unions import UnionMembership, UnionRule
from pants.testutil.option_util import create_subsystem
from pants.testutil.rule_runner import RuleRunner, mock_console, run_rule_with_mocks
from pants.util.logging import LogLevel
from pants.util.meta import classproperty
from pants.util.strutil import Simplifier


class MockMultipleSourcesField(MultipleSourcesField):
    pass


class MockTarget(Target):
    alias = "mock_target"
    core_fields = (MockMultipleSourcesField,)


class MockCheckFieldSet(FieldSet):
    required_fields = (MultipleSourcesField,)


class MockCheckRequest(CheckRequest, metaclass=ABCMeta):
    field_set_type = MockCheckFieldSet

    @staticmethod
    @abstractmethod
    def exit_code(_: Iterable[Address]) -> int:
        pass

    @property
    def check_results(self) -> CheckResults:
        addresses = [config.address for config in self.field_sets]
        return CheckResults(
            [CheckResult(self.exit_code(addresses), "", "")],
            checker_name=self.tool_name,
        )


class SuccessfulRequest(MockCheckRequest):
    tool_name = "Successful Checker"

    @classproperty
    def tool_id(cls) -> str:
        return "successfulchecker"

    @staticmethod
    def exit_code(_: Iterable[Address]) -> int:
        return 0


class FailingRequest(MockCheckRequest):
    tool_name = "Failing Checker"

    @classproperty
    def tool_id(cls) -> str:
        return "failingchecker"

    @staticmethod
    def exit_code(_: Iterable[Address]) -> int:
        return 1


class ConditionallySucceedsRequest(MockCheckRequest):
    tool_name = "Conditionally Succeeds Checker"

    @classproperty
    def tool_id(cls) -> str:
        return "conditionallysucceedschecker"

    @staticmethod
    def exit_code(addresses: Iterable[Address]) -> int:
        if any(address.target_name == "bad" for address in addresses):
            return 127
        return 0


class SkippedRequest(MockCheckRequest):
    tool_name = "Skipped Checker"

    @classproperty
    def tool_id(cls) -> str:
        return "skippedchecker"

    @staticmethod
    def exit_code(_) -> int:
        return 0

    @property
    def check_results(self) -> CheckResults:
        return CheckResults([], checker_name=self.tool_name)


class InvalidField(MultipleSourcesField):
    pass


class InvalidFieldSet(MockCheckFieldSet):
    required_fields = (InvalidField,)


class InvalidRequest(MockCheckRequest):
    field_set_type = InvalidFieldSet
    tool_name = "Invalid Checker"

    @classproperty
    def tool_id(cls) -> str:
        return "invalidchecker"

    @staticmethod
    def exit_code(_: Iterable[Address]) -> int:
        return -1


def make_target(address: Address | None = None) -> Target:
    if address is None:
        address = Address("", target_name="tests")
    return MockTarget({}, address)


def mock_check(__implicitly: tuple) -> CheckResults:
    req, typ = next(iter(__implicitly[0].items()))
    assert typ == CheckRequest
    return req.check_results  # type: ignore


def run_typecheck_rule(
    *,
    request_types: Sequence[type[CheckRequest]],
    targets: list[Target],
    only: list[str] | None = None,
) -> tuple[int, str]:
    union_membership = UnionMembership.from_rules(UnionRule(CheckRequest, t) for t in request_types)
    check_subsystem = create_subsystem(CheckSubsystem, only=only or [])
    rule_runner = RuleRunner(bootstrap_args=["-lwarn"])
    with mock_console(rule_runner.options_bootstrapper) as (console, stdio_reader):
        result: Check = run_rule_with_mocks(
            check_goal,
            rule_args=[
                console,
                Workspace(rule_runner.scheduler, _enforce_effects=False),
                Targets(targets),
                DistDir(relpath=Path("dist")),
                union_membership,
                check_subsystem,
                RunId(0),
            ],
            mock_calls={
                "pants.core.environments.rules.resolve_environment_name": lambda a: EnvironmentName(
                    a.raw_value
                ),
                "pants.core.goals.check.check": mock_check,
            },
            union_membership=union_membership,
            # We don't want temporary warnings to interfere with our expected output.
            show_warnings=False,
        )
        assert not stdio_reader.get_stdout()
        return result.exit_code, stdio_reader.get_stderr()


def test_invalid_target_noops() -> None:
    exit_code, stderr = run_typecheck_rule(request_types=[InvalidRequest], targets=[make_target()])
    assert exit_code == 0
    assert stderr == ""


def test_summary() -> None:
    good_address = Address("", target_name="good")
    bad_address = Address("", target_name="bad")
    targets = [make_target(good_address), make_target(bad_address)]
    requests = [
        ConditionallySucceedsRequest,
        FailingRequest,
        SkippedRequest,
        SuccessfulRequest,
    ]

    exit_code, stderr = run_typecheck_rule(request_types=requests, targets=targets)
    assert exit_code == FailingRequest.exit_code([bad_address])
    assert stderr == dedent(
        """\

        ✕ Conditionally Succeeds Checker failed.
        ✕ Failing Checker failed.
        ✓ Successful Checker succeeded.
        """
    )

    exit_code, stderr = run_typecheck_rule(
        request_types=requests,
        targets=targets,
        only=[FailingRequest.tool_id, SuccessfulRequest.tool_id],
    )
    assert stderr == dedent(
        """\

        ✕ Failing Checker failed.
        ✓ Successful Checker succeeded.
        """
    )


def test_streaming_output_skip() -> None:
    results = CheckResults([], checker_name="typechecker")
    assert results.level() == LogLevel.DEBUG
    assert results.message() == "typechecker skipped."


def test_streaming_output_success() -> None:
    results = CheckResults([CheckResult(0, "stdout", "stderr")], checker_name="typechecker")
    assert results.level() == LogLevel.INFO
    assert results.message() == dedent(
        """\
        typechecker succeeded.
        stdout
        stderr

        """
    )


def test_streaming_output_failure() -> None:
    results = CheckResults([CheckResult(18, "stdout", "stderr")], checker_name="typechecker")
    assert results.level() == LogLevel.ERROR
    assert results.message() == dedent(
        """\
        typechecker failed (exit code 18).
        stdout
        stderr

        """
    )


def test_streaming_output_partitions() -> None:
    results = CheckResults(
        [
            CheckResult(21, "", "", partition_description="ghc8.1"),
            CheckResult(0, "stdout", "stderr", partition_description="ghc9.2"),
        ],
        checker_name="typechecker",
    )
    assert results.level() == LogLevel.ERROR
    assert results.message() == dedent(
        """\
        typechecker failed (exit code 21).
        Partition #1 - ghc8.1:

        Partition #2 - ghc9.2:
        stdout
        stderr

        """
    )


def test_from_fallible_process_result_output_prepping() -> None:
    # Check that this calls the simplifier.
    class DistinctiveException(Exception):
        pass

    class SubSimplifier(Simplifier):
        def simplify(self, v: bytes | str) -> str:
            raise DistinctiveException()

    with pytest.raises(DistinctiveException):
        CheckResult.from_fallible_process_result(
            FallibleProcessResult(
                exit_code=0,
                stdout=b"",
                stdout_digest=EMPTY_FILE_DIGEST,
                stderr=b"",
                stderr_digest=EMPTY_FILE_DIGEST,
                output_digest=EMPTY_DIGEST,
                metadata=ProcessResultMetadata(
                    0,
                    ProcessExecutionEnvironment(
                        environment_name=None,
                        platform=Platform.create_for_localhost().value,
                        docker_image=None,
                        remote_execution=False,
                        remote_execution_extra_platform_properties=[],
                        execute_in_workspace=False,
                        keep_sandboxes="never",
                    ),
                    "ran_locally",
                    0,
                ),
            ),
            output_simplifier=SubSimplifier(),
        )

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	1✔
5
6	from abc import ABCMeta, abstractmethod	1✔
7	from collections.abc import Iterable, Sequence	1✔
8	from pathlib import Path	1✔
9	from textwrap import dedent	1✔
10
11	import pytest	1✔
12
13	from pants.core.goals.check import (	1✔
14	Check,
15	CheckRequest,
16	CheckResult,
17	CheckResults,
18	CheckSubsystem,
19	check_goal,
20	)
21	from pants.core.util_rules.distdir import DistDir	1✔
22	from pants.engine.addresses import Address	1✔
23	from pants.engine.environment import EnvironmentName	1✔
24	from pants.engine.fs import EMPTY_DIGEST, EMPTY_FILE_DIGEST, Workspace	1✔
25	from pants.engine.internals.session import RunId	1✔
26	from pants.engine.platform import Platform	1✔
27	from pants.engine.process import (	1✔
28	FallibleProcessResult,
29	ProcessExecutionEnvironment,
30	ProcessResultMetadata,
31	)
32	from pants.engine.target import FieldSet, MultipleSourcesField, Target, Targets	1✔
33	from pants.engine.unions import UnionMembership, UnionRule	1✔
34	from pants.testutil.option_util import create_subsystem	1✔
35	from pants.testutil.rule_runner import RuleRunner, mock_console, run_rule_with_mocks	1✔
36	from pants.util.logging import LogLevel	1✔
37	from pants.util.meta import classproperty	1✔
38	from pants.util.strutil import Simplifier	1✔
39
40
41	class MockMultipleSourcesField(MultipleSourcesField):	1✔
42	pass	1✔
43
44
45	class MockTarget(Target):	1✔
46	alias = "mock_target"	1✔
47	core_fields = (MockMultipleSourcesField,)	1✔
48
49
50	class MockCheckFieldSet(FieldSet):	1✔
51	required_fields = (MultipleSourcesField,)	1✔
52
53
54	class MockCheckRequest(CheckRequest, metaclass=ABCMeta):	1✔
55	field_set_type = MockCheckFieldSet	1✔
56
57	@staticmethod	1✔
58	@abstractmethod	1✔
59	def exit_code(_: Iterable[Address]) -> int:	1✔
UNCOV 60	pass	×
61
62	@property	1✔
63	def check_results(self) -> CheckResults:	1✔
64	addresses = [config.address for config in self.field_sets]	1✔
65	return CheckResults(	1✔
66	[CheckResult(self.exit_code(addresses), "", "")],
67	checker_name=self.tool_name,
68	)
69
70
71	class SuccessfulRequest(MockCheckRequest):	1✔
72	tool_name = "Successful Checker"	1✔
73
74	@classproperty	1✔
75	def tool_id(cls) -> str:	1✔
76	return "successfulchecker"	1✔
77
78	@staticmethod	1✔
79	def exit_code(_: Iterable[Address]) -> int:	1✔
80	return 0	1✔
81
82
83	class FailingRequest(MockCheckRequest):	1✔
84	tool_name = "Failing Checker"	1✔
85
86	@classproperty	1✔
87	def tool_id(cls) -> str:	1✔
88	return "failingchecker"	1✔
89
90	@staticmethod	1✔
91	def exit_code(_: Iterable[Address]) -> int:	1✔
92	return 1	1✔
93
94
95	class ConditionallySucceedsRequest(MockCheckRequest):	1✔
96	tool_name = "Conditionally Succeeds Checker"	1✔
97
98	@classproperty	1✔
99	def tool_id(cls) -> str:	1✔
100	return "conditionallysucceedschecker"	1✔
101
102	@staticmethod	1✔
103	def exit_code(addresses: Iterable[Address]) -> int:	1✔
104	if any(address.target_name == "bad" for address in addresses):	1✔
105	return 127	1✔
UNCOV 106	return 0	×
107
108
109	class SkippedRequest(MockCheckRequest):	1✔
110	tool_name = "Skipped Checker"	1✔
111
112	@classproperty	1✔
113	def tool_id(cls) -> str:	1✔
114	return "skippedchecker"	1✔
115
116	@staticmethod	1✔
117	def exit_code(_) -> int:	1✔
UNCOV 118	return 0	×
119
120	@property	1✔
121	def check_results(self) -> CheckResults:	1✔
122	return CheckResults([], checker_name=self.tool_name)	1✔
123
124
125	class InvalidField(MultipleSourcesField):	1✔
126	pass	1✔
127
128
129	class InvalidFieldSet(MockCheckFieldSet):	1✔
130	required_fields = (InvalidField,)	1✔
131
132
133	class InvalidRequest(MockCheckRequest):	1✔
134	field_set_type = InvalidFieldSet	1✔
135	tool_name = "Invalid Checker"	1✔
136
137	@classproperty	1✔
138	def tool_id(cls) -> str:	1✔
139	return "invalidchecker"	1✔
140
141	@staticmethod	1✔
142	def exit_code(_: Iterable[Address]) -> int:	1✔
UNCOV 143	return -1	×
144
145
146	def make_target(address: Address \| None = None) -> Target:	1✔
147	if address is None:	1✔
148	address = Address("", target_name="tests")	1✔
149	return MockTarget({}, address)	1✔
150
151
152	def mock_check(__implicitly: tuple) -> CheckResults:	1✔
153	req, typ = next(iter(__implicitly[0].items()))	1✔
154	assert typ == CheckRequest	1✔
155	return req.check_results # type: ignore	1✔
156
157
158	def run_typecheck_rule(	1✔
159	*,
160	request_types: Sequence[type[CheckRequest]],
161	targets: list[Target],
162	only: list[str] \| None = None,
163	) -> tuple[int, str]:
164	union_membership = UnionMembership.from_rules(UnionRule(CheckRequest, t) for t in request_types)	1✔
165	check_subsystem = create_subsystem(CheckSubsystem, only=only or [])	1✔
166	rule_runner = RuleRunner(bootstrap_args=["-lwarn"])	1✔
167	with mock_console(rule_runner.options_bootstrapper) as (console, stdio_reader):	1✔
168	result: Check = run_rule_with_mocks(	1✔
169	check_goal,
170	rule_args=[
171	console,
172	Workspace(rule_runner.scheduler, _enforce_effects=False),
173	Targets(targets),
174	DistDir(relpath=Path("dist")),
175	union_membership,
176	check_subsystem,
177	RunId(0),
178	],
179	mock_calls={
180	"pants.core.environments.rules.resolve_environment_name": lambda a: EnvironmentName(
181	a.raw_value
182	),
183	"pants.core.goals.check.check": mock_check,
184	},
185	union_membership=union_membership,
186	# We don't want temporary warnings to interfere with our expected output.
187	show_warnings=False,
188	)
189	assert not stdio_reader.get_stdout()	1✔
190	return result.exit_code, stdio_reader.get_stderr()	1✔
191
192
193	def test_invalid_target_noops() -> None:	1✔
194	exit_code, stderr = run_typecheck_rule(request_types=[InvalidRequest], targets=[make_target()])	1✔
195	assert exit_code == 0	1✔
196	assert stderr == ""	1✔
197
198
199	def test_summary() -> None:	1✔
200	good_address = Address("", target_name="good")	1✔
201	bad_address = Address("", target_name="bad")	1✔
202	targets = [make_target(good_address), make_target(bad_address)]	1✔
203	requests = [	1✔
204	ConditionallySucceedsRequest,
205	FailingRequest,
206	SkippedRequest,
207	SuccessfulRequest,
208	]
209
210	exit_code, stderr = run_typecheck_rule(request_types=requests, targets=targets)	1✔
211	assert exit_code == FailingRequest.exit_code([bad_address])	1✔
212	assert stderr == dedent(	1✔
213	"""\
214
215	✕ Conditionally Succeeds Checker failed.
216	✕ Failing Checker failed.
217	✓ Successful Checker succeeded.
218	"""
219	)
220
221	exit_code, stderr = run_typecheck_rule(	1✔
222	request_types=requests,
223	targets=targets,
224	only=[FailingRequest.tool_id, SuccessfulRequest.tool_id],
225	)
226	assert stderr == dedent(	1✔
227	"""\
228
229	✕ Failing Checker failed.
230	✓ Successful Checker succeeded.
231	"""
232	)
233
234
235	def test_streaming_output_skip() -> None:	1✔
236	results = CheckResults([], checker_name="typechecker")	1✔
237	assert results.level() == LogLevel.DEBUG	1✔
238	assert results.message() == "typechecker skipped."	1✔
239
240
241	def test_streaming_output_success() -> None:	1✔
242	results = CheckResults([CheckResult(0, "stdout", "stderr")], checker_name="typechecker")	1✔
243	assert results.level() == LogLevel.INFO	1✔
244	assert results.message() == dedent(	1✔
245	"""\
246	typechecker succeeded.
247	stdout
248	stderr
249
250	"""
251	)
252
253
254	def test_streaming_output_failure() -> None:	1✔
255	results = CheckResults([CheckResult(18, "stdout", "stderr")], checker_name="typechecker")	1✔
256	assert results.level() == LogLevel.ERROR	1✔
257	assert results.message() == dedent(	1✔
258	"""\
259	typechecker failed (exit code 18).
260	stdout
261	stderr
262
263	"""
264	)
265
266
267	def test_streaming_output_partitions() -> None:	1✔
268	results = CheckResults(	1✔
269	[
270	CheckResult(21, "", "", partition_description="ghc8.1"),
271	CheckResult(0, "stdout", "stderr", partition_description="ghc9.2"),
272	],
273	checker_name="typechecker",
274	)
275	assert results.level() == LogLevel.ERROR	1✔
276	assert results.message() == dedent(	1✔
277	"""\
278	typechecker failed (exit code 21).
279	Partition #1 - ghc8.1:
280
281	Partition #2 - ghc9.2:
282	stdout
283	stderr
284
285	"""
286	)
287
288
289	def test_from_fallible_process_result_output_prepping() -> None:	1✔
290	# Check that this calls the simplifier.
291	class DistinctiveException(Exception):	1✔
292	pass	1✔
293
294	class SubSimplifier(Simplifier):	1✔
295	def simplify(self, v: bytes \| str) -> str:	1✔
296	raise DistinctiveException()	1✔
297
298	with pytest.raises(DistinctiveException):	1✔
299	CheckResult.from_fallible_process_result(	1✔
300	FallibleProcessResult(
301	exit_code=0,
302	stdout=b"",
303	stdout_digest=EMPTY_FILE_DIGEST,
304	stderr=b"",
305	stderr_digest=EMPTY_FILE_DIGEST,
306	output_digest=EMPTY_DIGEST,
307	metadata=ProcessResultMetadata(
308	0,
309	ProcessExecutionEnvironment(
310	environment_name=None,
311	platform=Platform.create_for_localhost().value,
312	docker_image=None,
313	remote_execution=False,
314	remote_execution_extra_platform_properties=[],
315	execute_in_workspace=False,
316	keep_sandboxes="never",
317	),
318	"ran_locally",
319	0,
320	),
321	),
322	output_simplifier=SubSimplifier(),
323	)

pantsbuild / pants / 22507851448

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