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/terraform/dependency_inference.py

1	# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
UNCOV 3	from __future__ import annotations	×
4
UNCOV 5	from collections.abc import Iterable, Sequence	×
UNCOV 6	from dataclasses import dataclass	×
UNCOV 7	from pathlib import PurePath	×
8
UNCOV 9	from pants.backend.python.subsystems.python_tool_base import PythonToolRequirementsBase	×
UNCOV 10	from pants.backend.python.target_types import EntryPoint	×
UNCOV 11	from pants.backend.python.util_rules.pex import VenvPex, VenvPexProcess, create_venv_pex	×
UNCOV 12	from pants.backend.python.util_rules.pex import rules as pex_rules	×
UNCOV 13	from pants.backend.python.util_rules.pex import setup_venv_pex_process	×
UNCOV 14	from pants.backend.terraform.target_types import (	×
15	TerraformBackendTarget,
16	TerraformDependenciesField,
17	TerraformDeploymentFieldSet,
18	TerraformLockfileTarget,
19	TerraformModuleSourcesField,
20	TerraformVarFileTarget,
21	)
UNCOV 22	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	×
UNCOV 23	from pants.base.specs import DirGlobSpec, DirLiteralSpec, RawSpecs	×
UNCOV 24	from pants.core.target_types import LockfileTarget	×
UNCOV 25	from pants.engine.addresses import Addresses	×
UNCOV 26	from pants.engine.fs import CreateDigest, Digest, FileContent	×
UNCOV 27	from pants.engine.internals.build_files import resolve_address	×
UNCOV 28	from pants.engine.internals.graph import (	×
29	determine_explicitly_provided_dependencies,
30	hydrate_sources,
31	resolve_targets,
32	)
UNCOV 33	from pants.engine.internals.native_engine import Address, AddressInput	×
UNCOV 34	from pants.engine.internals.selectors import concurrently	×
UNCOV 35	from pants.engine.intrinsics import create_digest	×
UNCOV 36	from pants.engine.process import Process, execute_process_or_raise	×
UNCOV 37	from pants.engine.rules import collect_rules, implicitly, rule	×
UNCOV 38	from pants.engine.target import (	×
39	DependenciesRequest,
40	FieldSet,
41	HydrateSourcesRequest,
42	InferDependenciesRequest,
43	InferredDependencies,
44	Target,
45	)
UNCOV 46	from pants.engine.unions import UnionRule	×
UNCOV 47	from pants.util.dirutil import group_by_dir	×
UNCOV 48	from pants.util.logging import LogLevel	×
UNCOV 49	from pants.util.ordered_set import OrderedSet	×
UNCOV 50	from pants.util.resources import read_resource	×
UNCOV 51	from pants.util.strutil import bullet_list, softwrap	×
52
53
UNCOV 54	class TerraformHcl2Parser(PythonToolRequirementsBase):	×
UNCOV 55	options_scope = "terraform-hcl2-parser"	×
UNCOV 56	help_short = "Used to parse Terraform modules to infer their dependencies."	×
57
58	# versions 4.3.2+ have parsing issues; bump once resolved
UNCOV 59	default_requirements = ["python-hcl2>=3.0.5,<=4.3.0"]	×
60
UNCOV 61	register_interpreter_constraints = True	×
62
UNCOV 63	default_lockfile_resource = ("pants.backend.terraform", "hcl2.lock")	×
64
65
UNCOV 66	@dataclass(frozen=True)	×
UNCOV 67	class ParserSetup:	×
UNCOV 68	pex: VenvPex	×
69
70
UNCOV 71	@rule	×
UNCOV 72	async def setup_parser(hcl2_parser: TerraformHcl2Parser) -> ParserSetup:	×
73	parser_script_content = read_resource("pants.backend.terraform", "hcl2_parser.py")	×
74	if not parser_script_content:	×
75	raise ValueError("Unable to find source to hcl2_parser.py wrapper script.")	×
76
77	parser_content = FileContent(	×
78	path="__pants_tf_parser.py", content=parser_script_content, is_executable=True
79	)
80	parser_digest = await create_digest(CreateDigest([parser_content]))	×
81
82	parser_pex = await create_venv_pex(	×
83	**implicitly(
84	hcl2_parser.to_pex_request(
85	main=EntryPoint(PurePath(parser_content.path).stem), sources=parser_digest
86	)
87	)
88	)
89	return ParserSetup(parser_pex)	×
90
91
UNCOV 92	@dataclass(frozen=True)	×
UNCOV 93	class ParseTerraformModuleSources:	×
UNCOV 94	sources_digest: Digest	×
UNCOV 95	paths: tuple[str, ...]	×
96
97
UNCOV 98	@rule	×
UNCOV 99	async def setup_process_for_parse_terraform_module_sources(	×
100	request: ParseTerraformModuleSources, parser: ParserSetup
101	) -> Process:
102	dir_paths = ", ".join(sorted(group_by_dir(request.paths).keys()))	×
103
104	process = await setup_venv_pex_process(	×
105	VenvPexProcess(
106	parser.pex,
107	argv=request.paths,
108	input_digest=request.sources_digest,
109	description=f"Parse Terraform module sources: {dir_paths}",
110	level=LogLevel.DEBUG,
111	),
112	**implicitly(),
113	)
114	return process	×
115
116
UNCOV 117	@dataclass(frozen=True)	×
UNCOV 118	class TerraformModuleDependenciesInferenceFieldSet(FieldSet):	×
UNCOV 119	required_fields = (TerraformModuleSourcesField, TerraformDependenciesField)	×
120
UNCOV 121	sources: TerraformModuleSourcesField	×
UNCOV 122	dependencies: TerraformDependenciesField	×
123
124
UNCOV 125	class InferTerraformModuleDependenciesRequest(InferDependenciesRequest):	×
UNCOV 126	infer_from = TerraformModuleDependenciesInferenceFieldSet	×
127
128
UNCOV 129	@dataclass(frozen=True)	×
UNCOV 130	class TerraformDeploymentDependenciesInferenceFieldSet(TerraformDeploymentFieldSet):	×
UNCOV 131	pass	×
132
133
UNCOV 134	class InferTerraformDeploymentDependenciesRequest(InferDependenciesRequest):	×
UNCOV 135	infer_from = TerraformDeploymentDependenciesInferenceFieldSet	×
136
137
UNCOV 138	def find_targets_of_type(tgts, of_type) -> tuple:	×
139	if tgts:	×
140	return tuple(e for e in tgts if isinstance(e, of_type))	×
141	else:
142	return ()	×
143
144
UNCOV 145	@dataclass(frozen=True)	×
UNCOV 146	class TerraformDeploymentInvocationFilesRequest:	×
147	"""TODO: is there a way to convert between FS? We could convert the inference FS to the deployment FS itself"""
148
UNCOV 149	address: Address	×
UNCOV 150	dependencies: TerraformDependenciesField	×
151
152
UNCOV 153	@dataclass(frozen=True)	×
UNCOV 154	class TerraformDeploymentInvocationFiles:	×
155	"""The files passed in to the invocation of `terraform`"""
156
UNCOV 157	backend_configs: tuple[TerraformBackendTarget, ...]	×
UNCOV 158	vars_files: tuple[TerraformVarFileTarget, ...]	×
UNCOV 159	lockfile: LockfileTarget \| None	×
160
161
UNCOV 162	@rule	×
UNCOV 163	async def get_terraform_backend_and_vars(	×
164	field_set: TerraformDeploymentInvocationFilesRequest,
165	) -> TerraformDeploymentInvocationFiles:
166	this_address = field_set.address	×
167
168	explicit_deps = await determine_explicitly_provided_dependencies(	×
169	**implicitly(DependenciesRequest(field_set.dependencies))
170	)
171	tgts_in_dir, explicit_deps_tgt = await concurrently(	×
172	resolve_targets(
173	**implicitly(
174	RawSpecs(
175	description_of_origin="terraform infer deployment dependencies",
176	dir_literals=(DirLiteralSpec(this_address.spec_path),),
177	)
178	)
179	),
180	resolve_targets(**implicitly(Addresses(explicit_deps.includes))),
181	)
182	return identify_terraform_backend_and_vars(explicit_deps_tgt, tgts_in_dir)	×
183
184
UNCOV 185	class InvalidLockfileException(Exception):	×
UNCOV 186	@classmethod	×
UNCOV 187	def too_many_lockfiles(	×
188	cls, lockfiles: Iterable[TerraformLockfileTarget]
189	) -> InvalidLockfileException:
190	addresses = sorted(tgt.address.spec for tgt in lockfiles)	×
191	return cls(	×
192	softwrap(
193	f"""\
194	A Terraform deployment has {len(addresses)} lockfiles supplied:
195	{bullet_list(addresses)}
196	Terraform requires at most 1 lockfile; it must be called `.terraform.lock.hcl`;
197	and it must be in the same directory as the root module.
198
199	Pants generates targets for Terraform lockfiles automatically.
200	If you manually added `{TerraformLockfileTarget.alias}` targets, removing them should resolve this error.
201	If you have not, please report this as a bug.
202	"""
203	)
204	)
205
206
UNCOV 207	def identify_terraform_backend_and_vars(	×
208	explicit_deps: Sequence[Target], tgts_in_dir: Sequence[Target]
209	) -> TerraformDeploymentInvocationFiles:
210	has_explicit_backend = find_targets_of_type(explicit_deps, TerraformBackendTarget)	×
211	if not has_explicit_backend:	×
212	# Note: Terraform does not support multiple backends, but dep inference isn't the place to enforce that
213	backend_targets = find_targets_of_type(tgts_in_dir, TerraformBackendTarget)	×
214	else:
215	backend_targets = has_explicit_backend	×
216
217	has_explicit_var = find_targets_of_type(explicit_deps, TerraformVarFileTarget)	×
218	if not has_explicit_var:	×
219	vars_targets = find_targets_of_type(tgts_in_dir, TerraformVarFileTarget)	×
220	else:
221	vars_targets = has_explicit_var	×
222
223	lockfiles = find_targets_of_type(tgts_in_dir, TerraformLockfileTarget)	×
224	if len(lockfiles) == 1:	×
225	lockfile = lockfiles[0]	×
226	elif len(lockfiles) > 1:	×
227	# Unlikely, since we generate them based on a constant filename.
228	# Indicates manual specification of targets
229	raise InvalidLockfileException.too_many_lockfiles(lockfiles)	×
230	else:
231	lockfile = None	×
232
233	return TerraformDeploymentInvocationFiles(backend_targets, vars_targets, lockfile)	×
234
235
UNCOV 236	async def _infer_dependencies_from_sources(	×
237	request: InferTerraformModuleDependenciesRequest,
238	) -> list[Address]:
239	"""Parse the source code for references to other modules."""
240	hydrated_sources = await hydrate_sources(	×
241	HydrateSourcesRequest(request.field_set.sources), **implicitly()
242	)
243	paths = OrderedSet(	×
244	filename for filename in hydrated_sources.snapshot.files if filename.endswith(".tf")
245	)
246	result = await execute_process_or_raise(	×
247	**implicitly(
248	ParseTerraformModuleSources(
249	sources_digest=hydrated_sources.snapshot.digest,
250	paths=tuple(paths),
251	)
252	)
253	)
254	candidate_spec_paths = [line for line in result.stdout.decode("utf-8").split("\n") if line]	×
255	# For each path, see if there is a `terraform_module` target at the specified spec_path.
256	candidate_targets = await resolve_targets(	×
257	**implicitly(
258	RawSpecs(
259	dir_globs=tuple(DirGlobSpec(path) for path in candidate_spec_paths),
260	unmatched_glob_behavior=GlobMatchErrorBehavior.ignore,
261	description_of_origin="the `terraform_module` dependency inference rule",
262	)
263	)
264	)
265	# TODO: Need to either implement the standard ambiguous dependency logic or ban >1 terraform_module
266	# per directory.
267	terraform_module_addresses = [	×
268	tgt.address for tgt in candidate_targets if tgt.has_field(TerraformModuleSourcesField)
269	]
270	return terraform_module_addresses	×
271
272
UNCOV 273	async def _infer_lockfile(request: InferTerraformModuleDependenciesRequest) -> list[Address]:	×
274	"""Pull in the lockfile for a Terraform module.
275
276	This is necessary for `terraform validate`.
277	"""
278	invocation_files = await get_terraform_backend_and_vars(	×
279	TerraformDeploymentInvocationFilesRequest(
280	request.field_set.address, request.field_set.dependencies
281	)
282	)
283	if invocation_files.lockfile:	×
284	return [invocation_files.lockfile.address]	×
285	else:
286	return []	×
287
288
UNCOV 289	@rule	×
UNCOV 290	async def infer_terraform_module_dependencies(	×
291	request: InferTerraformModuleDependenciesRequest,
292	) -> InferredDependencies:
293	terraform_module_addresses = await _infer_dependencies_from_sources(request)	×
294	lockfile_address = await _infer_lockfile(request)	×
295
296	return InferredDependencies([terraform_module_addresses, lockfile_address])	×
297
298
UNCOV 299	@rule	×
UNCOV 300	async def infer_terraform_deployment_dependencies(	×
301	request: InferTerraformDeploymentDependenciesRequest,
302	) -> InferredDependencies:
303	root_module_address_input = request.field_set.root_module.to_address_input()	×
304	root_module = await resolve_address(**implicitly({root_module_address_input: AddressInput}))	×
305	deps = [root_module]	×
306
307	invocation_files = await get_terraform_backend_and_vars(	×
308	TerraformDeploymentInvocationFilesRequest(
309	request.field_set.address, request.field_set.dependencies
310	)
311	)
312	deps.extend(e.address for e in invocation_files.backend_configs)	×
313	deps.extend(e.address for e in invocation_files.vars_files)	×
314	# lockfile is attached to the module itself
315
316	return InferredDependencies(deps)	×
317
318
UNCOV 319	def rules():	×
UNCOV 320	return [	×
321	*collect_rules(),
322	*pex_rules(),
323	UnionRule(InferDependenciesRequest, InferTerraformModuleDependenciesRequest),
324	UnionRule(InferDependenciesRequest, InferTerraformDeploymentDependenciesRequest),
325	]

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