19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

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

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

0.0

/src/python/pants/backend/shell/goals/test.py

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
UNCOV 4	from __future__ import annotations	×
5
UNCOV 6	import dataclasses	×
UNCOV 7	from typing import Any	×
8
UNCOV 9	from pants.backend.shell.subsystems.shell_test_subsys import ShellTestSubsystem	×
UNCOV 10	from pants.backend.shell.target_types import (	×
11	ShellCommandCacheScopeField,
12	ShellCommandCommandField,
13	ShellCommandTestDependenciesField,
14	SkipShellCommandTestsField,
15	)
UNCOV 16	from pants.backend.shell.util_rules import shell_command	×
UNCOV 17	from pants.backend.shell.util_rules.shell_command import (	×
18	ShellCommandProcessFromTargetRequest,
19	prepare_process_request_from_target,
20	)
UNCOV 21	from pants.core.environments.target_types import EnvironmentField	×
UNCOV 22	from pants.core.goals.test import (	×
23	TestDebugRequest,
24	TestExtraEnv,
25	TestFieldSet,
26	TestRequest,
27	TestResult,
28	TestSubsystem,
29	)
UNCOV 30	from pants.core.util_rules.adhoc_process_support import (	×
31	AdhocProcessRequest,
32	FallibleAdhocProcessResult,
33	prepare_adhoc_process,
34	)
UNCOV 35	from pants.core.util_rules.adhoc_process_support import rules as adhoc_process_support_rules	×
UNCOV 36	from pants.core.util_rules.adhoc_process_support import run_prepared_adhoc_process	×
UNCOV 37	from pants.engine.fs import EMPTY_DIGEST, Snapshot	×
UNCOV 38	from pants.engine.internals.graph import resolve_target	×
UNCOV 39	from pants.engine.intrinsics import digest_to_snapshot	×
UNCOV 40	from pants.engine.process import InteractiveProcess, ProcessCacheScope	×
UNCOV 41	from pants.engine.rules import collect_rules, implicitly, rule	×
UNCOV 42	from pants.engine.target import Target, WrappedTargetRequest	×
UNCOV 43	from pants.util.frozendict import FrozenDict	×
UNCOV 44	from pants.util.logging import LogLevel	×
45
46
UNCOV 47	@dataclasses.dataclass(frozen=True)	×
UNCOV 48	class TestShellCommandFieldSet(TestFieldSet):	×
UNCOV 49	required_fields = (	×
50	ShellCommandCommandField,
51	ShellCommandTestDependenciesField,
52	)
53
UNCOV 54	environment: EnvironmentField	×
UNCOV 55	cache_scope: ShellCommandCacheScopeField	×
56
UNCOV 57	@classmethod	×
UNCOV 58	def opt_out(cls, tgt: Target) -> bool:	×
59	return tgt.get(SkipShellCommandTestsField).value	×
60
61
UNCOV 62	class ShellTestRequest(TestRequest):	×
UNCOV 63	tool_subsystem = ShellTestSubsystem # type: ignore[assignment]	×
UNCOV 64	field_set_type = TestShellCommandFieldSet	×
UNCOV 65	supports_debug = True	×
66
67
UNCOV 68	@rule(desc="Test with shell command", level=LogLevel.DEBUG)	×
UNCOV 69	async def test_shell_command(	×
70	batch: ShellTestRequest.Batch[TestShellCommandFieldSet, Any],
71	test_subsystem: TestSubsystem,
72	test_extra_env: TestExtraEnv,
73	) -> TestResult:
74	field_set = batch.single_element	×
75	wrapped_tgt = await resolve_target(	×
76	WrappedTargetRequest(field_set.address, description_of_origin="<infallible>"),
77	**implicitly(),
78	)
79
80	shell_process = await prepare_process_request_from_target(	×
81	ShellCommandProcessFromTargetRequest(wrapped_tgt.target), **implicitly()
82	)
83
84	shell_process = dataclasses.replace(	×
85	shell_process,
86	env_vars=FrozenDict(
87	{
88	**test_extra_env.env,
89	**shell_process.env_vars,
90	}
91	),
92	)
93
94	if field_set.cache_scope.value is None and test_subsystem.force:	×
95	shell_process = dataclasses.replace(	×
96	shell_process,
97	cache_scope=ProcessCacheScope.PER_SESSION,
98	)
99
100	results: list[FallibleAdhocProcessResult] = []	×
101	for _ in range(test_subsystem.attempts_default):	×
102	result = await run_prepared_adhoc_process(	×
103	**implicitly({shell_process: AdhocProcessRequest})
104	) # noqa: PNT30: retry loop
105	results.append(result)	×
106	if result.process_result.exit_code == 0:	×
107	break	×
108
109	extra_output: Snapshot \| None = None	×
110	if results[-1].adjusted_digest != EMPTY_DIGEST:	×
111	extra_output = await digest_to_snapshot(results[-1].adjusted_digest)	×
112
113	return TestResult.from_fallible_process_result(	×
114	process_results=tuple(r.process_result for r in results),
115	address=field_set.address,
116	output_setting=test_subsystem.output,
117	extra_output=extra_output,
118	log_extra_output=extra_output is not None,
119	)
120
121
UNCOV 122	@rule(desc="Test with shell command (interactively)", level=LogLevel.DEBUG)	×
UNCOV 123	async def test_shell_command_interactively(	×
124	batch: ShellTestRequest.Batch[TestShellCommandFieldSet, Any],
125	) -> TestDebugRequest:
126	field_set = batch.single_element	×
127	wrapped_tgt = await resolve_target(	×
128	WrappedTargetRequest(field_set.address, description_of_origin="<infallible>"),
129	**implicitly(),
130	)
131
132	prepared_request = await prepare_adhoc_process(	×
133	**implicitly(ShellCommandProcessFromTargetRequest(wrapped_tgt.target))
134	)
135
136	# This is probably not strictly necessary given the use of `InteractiveProcess` but good to be correct in any event.
137	shell_process = dataclasses.replace(	×
138	prepared_request.process, cache_scope=ProcessCacheScope.PER_SESSION
139	)
140
141	return TestDebugRequest(	×
142	InteractiveProcess.from_process(
143	shell_process, forward_signals_to_process=False, restartable=True
144	)
145	)
146
147
UNCOV 148	def rules():	×
UNCOV 149	return (	×
150	*collect_rules(),
151	*shell_command.rules(),
152	*ShellTestRequest.rules(),
153	*adhoc_process_support_rules(),
154	)

pantsbuild / pants / 19015773527

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