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/adhoc/run_system_binary.py

1	# Copyright 2023 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 logging	×
UNCOV 7	import re	×
UNCOV 8	from collections.abc import Mapping	×
UNCOV 9	from dataclasses import dataclass	×
UNCOV 10	from io import StringIO	×
11
UNCOV 12	from pants.backend.adhoc.target_types import (	×
13	SystemBinaryExtraSearchPathsField,
14	SystemBinaryFingerprintArgsField,
15	SystemBinaryFingerprintDependenciesField,
16	SystemBinaryFingerprintPattern,
17	SystemBinaryLogFingerprintingErrorsField,
18	SystemBinaryNameField,
19	)
UNCOV 20	from pants.build_graph.address import Address	×
UNCOV 21	from pants.core.goals.run import RunFieldSet, RunInSandboxBehavior, RunRequest	×
UNCOV 22	from pants.core.util_rules.adhoc_process_support import (	×
23	ResolveExecutionDependenciesRequest,
24	resolve_execution_environment,
25	)
UNCOV 26	from pants.core.util_rules.system_binaries import (	×
27	BinaryPath,
28	BinaryPathRequest,
29	SearchPath,
30	SystemBinariesSubsystem,
31	find_binary,
32	)
UNCOV 33	from pants.engine.internals.native_engine import EMPTY_DIGEST, Digest	×
UNCOV 34	from pants.engine.internals.selectors import concurrently	×
UNCOV 35	from pants.engine.intrinsics import execute_process	×
UNCOV 36	from pants.engine.process import FallibleProcessResult, Process	×
UNCOV 37	from pants.engine.rules import collect_rules, implicitly, rule	×
UNCOV 38	from pants.util.logging import LogLevel	×
39
UNCOV 40	logger = logging.getLogger(__name__)	×
41
42
UNCOV 43	@dataclass(frozen=True)	×
UNCOV 44	class SystemBinaryFieldSet(RunFieldSet):	×
UNCOV 45	run_in_sandbox_behavior = RunInSandboxBehavior.RUN_REQUEST_HERMETIC	×
46
UNCOV 47	required_fields = (	×
48	SystemBinaryNameField,
49	SystemBinaryExtraSearchPathsField,
50	SystemBinaryFingerprintPattern,
51	SystemBinaryFingerprintArgsField,
52	SystemBinaryFingerprintDependenciesField,
53	SystemBinaryLogFingerprintingErrorsField,
54	)
55
UNCOV 56	name: SystemBinaryNameField	×
UNCOV 57	extra_search_paths: SystemBinaryExtraSearchPathsField	×
UNCOV 58	fingerprint_pattern: SystemBinaryFingerprintPattern	×
UNCOV 59	fingerprint_argv: SystemBinaryFingerprintArgsField	×
UNCOV 60	fingerprint_dependencies: SystemBinaryFingerprintDependenciesField	×
UNCOV 61	log_fingerprinting_errors: SystemBinaryLogFingerprintingErrorsField	×
62
63
UNCOV 64	async def _find_binary(	×
65	address: Address,
66	binary_name: str,
67	search_path: SearchPath,
68	fingerprint_pattern: str \| None,
69	fingerprint_args: tuple[str, ...] \| None,
70	fingerprint_dependencies: tuple[str, ...] \| None,
71	log_fingerprinting_errors: bool,
72	) -> BinaryPath:
73	binaries = await find_binary(	×
74	BinaryPathRequest(
75	binary_name=binary_name,
76	search_path=search_path,
77	),
78	**implicitly(),
79	)
80
81	fingerprint_args = fingerprint_args or ()	×
82
83	deps = await resolve_execution_environment(	×
84	ResolveExecutionDependenciesRequest(address, (), fingerprint_dependencies), **implicitly()
85	)
86	rds = deps.runnable_dependencies	×
87	env: dict[str, str] = {}	×
88	append_only_caches: Mapping[str, str] = {}	×
89	immutable_input_digests: Mapping[str, Digest] = {}	×
90	if rds:	×
91	env = {"PATH": rds.path_component}	×
92	env.update(**(rds.extra_env or {}))	×
93	append_only_caches = rds.append_only_caches	×
94	immutable_input_digests = rds.immutable_input_digests	×
95
96	tests: tuple[FallibleProcessResult, ...] = await concurrently(	×
97	execute_process(
98	Process(
99	description=f"Testing candidate for `{binary_name}` at `{path.path}`",
100	argv=(path.path,) + fingerprint_args,
101	input_digest=deps.digest,
102	env=env,
103	append_only_caches=append_only_caches,
104	immutable_input_digests=immutable_input_digests,
105	),
106	**implicitly(),
107	)
108	for path in binaries.paths
109	)
110
111	for test, binary in zip(tests, binaries.paths):	×
112	if test.exit_code != 0:	×
113	if log_fingerprinting_errors:	×
114	logger.warning(	×
115	f"Error occurred while fingerprinting candidate binary `{binary.path}` "
116	f"for binary `{binary_name}` (exit code {test.exit_code}) (use the `{SystemBinaryLogFingerprintingErrorsField.alias}` field to control this warning):\n\n"
117	f"stdout:\n{test.stdout.decode(errors='ignore')}\n"
118	f"stderr:\n{test.stderr.decode(errors='ignore')}"
119	)
120
121	# Skip this binary since fingerprinting failed.
122	continue	×
123
124	if fingerprint_pattern:	×
125	fingerprint = test.stdout.decode().strip()	×
126	match = re.match(fingerprint_pattern, fingerprint)	×
127	if not match:	×
128	continue	×
129
130	return binary	×
131
132	message = StringIO()	×
133	message.write(f"Could not find a binary with name `{binary_name}`")	×
134	if fingerprint_pattern:	×
135	message.write(	×
136	f" with output matching `{fingerprint_pattern}` when run with arguments `{' '.join(fingerprint_args or ())}`"
137	)
138
139	message.write(". The following paths were searched:\n")	×
140	for sp in search_path:	×
141	message.write(f"- {sp}\n")	×
142
143	failed_tests = [	×
144	(test, binary) for test, binary in zip(tests, binaries.paths) if test.exit_code != 0
145	]
146	if failed_tests:	×
147	message.write(	×
148	"\n\nThe following binaries were skipped because each binary returned an error when invoked:"
149	)
150	for failed_test, failed_binary in failed_tests:	×
151	message.write(f"\n\n- {failed_binary.path} (exit code {failed_test.exit_code})\n")	×
152	message.write(f" stdout:\n{failed_test.stdout.decode(errors='ignore')}\n")	×
153	message.write(f" stderr:\n{failed_test.stderr.decode(errors='ignore')}\n")	×
154
155	raise ValueError(message.getvalue())	×
156
157
UNCOV 158	@rule(level=LogLevel.DEBUG)	×
UNCOV 159	async def create_system_binary_run_request(	×
160	field_set: SystemBinaryFieldSet,
161	system_binaries: SystemBinariesSubsystem.EnvironmentAware,
162	) -> RunRequest:
163	assert field_set.name.value is not None	×
164	extra_search_paths = field_set.extra_search_paths.value or ()	×
165
166	search_path = SearchPath((extra_search_paths, system_binaries.system_binary_paths))	×
167
168	path = await _find_binary(	×
169	field_set.address,
170	field_set.name.value,
171	search_path,
172	field_set.fingerprint_pattern.value,
173	field_set.fingerprint_argv.value,
174	field_set.fingerprint_dependencies.value,
175	field_set.log_fingerprinting_errors.value,
176	)
177
178	return RunRequest(	×
179	digest=EMPTY_DIGEST,
180	args=[path.path],
181	)
182
183
UNCOV 184	def rules():	×
UNCOV 185	return [	×
186	*collect_rules(),
187	*SystemBinaryFieldSet.rules(),
188	]

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