20632486505

Committed 01 Jan 2026 04:21AM UTC coverage: 43.231% (-37.1%) from 80.281%

Build # 20632486505

Build Type

Pull #22962

github

Committed by

web-flow

Commit Message

Merge 08d5c63b0 into f52ab6675

Pull Request Pull Request #22962: Bump the gha-deps group across 1 directory with 6 updates

Run Details

26122 of 60424 relevant lines covered (43.23%)

0.86 hits per line

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

51.09

/src/python/pants/backend/terraform/dependencies.py

# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

from dataclasses import dataclass

from pants.backend.terraform.dependency_inference import (
    TerraformDeploymentInvocationFilesRequest,
    get_terraform_backend_and_vars,
)
from pants.backend.terraform.target_types import (
    LockfileSourceField,
    TerraformBackendConfigField,
    TerraformDependenciesField,
    TerraformDeploymentFieldSet,
    TerraformFieldSet,
    TerraformModuleSourcesField,
    TerraformRootModuleField,
    TerraformVarFileSourceField,
)
from pants.backend.terraform.tool import TerraformCommand, TerraformProcess
from pants.backend.terraform.utils import terraform_arg, terraform_relpath
from pants.core.target_types import FileSourceField, ResourceSourceField
from pants.core.util_rules.source_files import (
    SourceFiles,
    SourceFilesRequest,
    determine_source_files,
)
from pants.engine.internals.build_files import resolve_address
from pants.engine.internals.graph import resolve_target, transitive_targets
from pants.engine.internals.native_engine import Digest, MergeDigests
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import execute_process, merge_digests
from pants.engine.process import ProcessExecutionFailure
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import SourcesField, TransitiveTargetsRequest, WrappedTargetRequest
from pants.option.global_options import KeepSandboxes


@dataclass(frozen=True)
class TerraformDependenciesRequest:
    chdir: str
    backend_config: str | None
    lockfile: bool
    dependencies_files: Digest

    # Not initialising the backend means we won't access remote state. Useful for `validate`
    initialise_backend: bool = False
    upgrade: bool = False

    def to_args(self) -> TerraformCommand:
        args = ["init"]
        if self.backend_config:
            args.append(
                terraform_arg(
                    "-backend-config",
                    terraform_relpath(self.chdir, self.backend_config),
                )
            )

        # If we have a lockfile and aren't regenerating it, don't modify it
        if self.lockfile and not self.upgrade:
            args.append("-lockfile=readonly")

        if self.upgrade:
            args.append("-upgrade")

        args.append(terraform_arg("-backend", str(self.initialise_backend)))

        return TerraformCommand(tuple(args))


@dataclass(frozen=True)
class TerraformDependenciesResponse:
    digest: Digest


@rule
async def get_terraform_providers(
    req: TerraformDependenciesRequest,
    keep_sandboxes: KeepSandboxes,
) -> TerraformDependenciesResponse:
    init_process_description = (
        f"Running `init` on Terraform module at `{req.chdir}` to fetch dependencies"
    )
    fetched_deps = await execute_process(
        **implicitly(
            TerraformProcess(
                cmds=(req.to_args(),),
                input_digest=req.dependencies_files,
                output_files=(".terraform.lock.hcl",),
                output_directories=(".terraform",),
                description=init_process_description,
                chdir=req.chdir,
            )
        )
    )
    if fetched_deps.exit_code != 0:
        raise ProcessExecutionFailure.from_result(
            fetched_deps, init_process_description, keep_sandboxes
        )

    return TerraformDependenciesResponse(fetched_deps.output_digest)


@dataclass(frozen=True)
class TerraformInitRequest:
    root_module: TerraformRootModuleField
    dependencies: TerraformDependenciesField

    # Not initialising the backend means we won't access remote state. Useful for `validate`
    initialise_backend: bool = False
    upgrade: bool = False


@dataclass(frozen=True)
class TerraformInvocationRequirements:
    """The things you need to run a terraform command."""

    terraform_sources: SourceFiles
    dependencies_files: SourceFiles
    init_cmd: TerraformDependenciesRequest
    chdir: str


def terraform_fieldset_to_init_request(
    terraform_fieldset: TerraformDeploymentFieldSet | TerraformFieldSet,
) -> TerraformInitRequest:
    """Create a TerraformInitRequest from both Terraform Modules and Deployments."""
    if isinstance(terraform_fieldset, TerraformDeploymentFieldSet):
        deployment = terraform_fieldset
        return TerraformInitRequest(deployment.root_module, deployment.dependencies)
    elif isinstance(terraform_fieldset, TerraformFieldSet):
        module = terraform_fieldset
        return TerraformInitRequest(
            TerraformRootModuleField(module.address.spec, module.address),
            module.dependencies,
        )
    else:
        raise TypeError(
            f"Invalid type passed to initialise terraform tpye={type(terraform_fieldset)}"
        )


@rule
async def prepare_terraform_invocation(
    request: TerraformInitRequest,
) -> TerraformInvocationRequirements:
    """Prepare a terraform module or deployment to be operated on."""
    this_targets_dependencies = await transitive_targets(
        TransitiveTargetsRequest((request.dependencies.address,)), **implicitly()
    )

    address_input = request.root_module.to_address_input()
    module_address = await resolve_address(**implicitly(address_input))

    chdir = module_address.spec_path  # TODO: spec_path is wrong, that's to the build file
    # if the Terraform module is in the root, chdir will be "". Terraform needs a valid dir to change to
    if not chdir:
        chdir = "."

    # TODO: is this still necessary, or do we pull it in with (transitive) dependencies?
    module = await resolve_target(
        WrappedTargetRequest(
            module_address, description_of_origin=address_input.description_of_origin
        ),
        **implicitly(),
    )

    source_files, dependencies_files = await concurrently(
        determine_source_files(
            SourceFilesRequest([module.target.get(SourcesField)])
        ),  # TODO: get through transitive deps???
        determine_source_files(
            SourceFilesRequest(
                [tgt.get(SourcesField) for tgt in this_targets_dependencies.dependencies],
                for_sources_types=(
                    TerraformModuleSourcesField,
                    TerraformBackendConfigField,
                    TerraformVarFileSourceField,
                    LockfileSourceField,
                    FileSourceField,
                    ResourceSourceField,
                ),
                enable_codegen=True,
            )
        ),
    )
    invocation_files = await get_terraform_backend_and_vars(
        TerraformDeploymentInvocationFilesRequest(
            request.dependencies.address, request.dependencies
        )
    )
    backend_config_tgts = invocation_files.backend_configs
    if len(backend_config_tgts) == 0:
        backend_config = None
    elif len(backend_config_tgts) == 1:
        backend_config_sources = await determine_source_files(
            SourceFilesRequest([backend_config_tgts[0].get(SourcesField)])
        )
        backend_config = backend_config_sources.snapshot.files[0]
    else:
        # We've found multiple backend files, but that's only a problem if we need to initialise the backend.
        # For example, we might be `validate`ing a `terraform_module` that has multiple backend files in the same dir,
        # so we don't need to init the backend.
        # The `terraform_deployment`s will have the references to the correct backends

        if request.initialise_backend:
            backend_config_names = [e.address for e in backend_config_tgts]
            raise ValueError(
                f"Found more than 1 backend config for a Terraform deployment. identified {backend_config_names}"
            )
        else:
            backend_config = None

    source_for_validate = await merge_digests(
        MergeDigests([source_files.snapshot.digest, dependencies_files.snapshot.digest])
    )

    has_lockfile = invocation_files.lockfile is not None

    terraform_init_cmd = TerraformDependenciesRequest(
        chdir,
        backend_config,
        has_lockfile,
        source_for_validate,
        initialise_backend=request.initialise_backend,
        upgrade=request.upgrade,
    )
    return TerraformInvocationRequirements(
        source_files, dependencies_files, terraform_init_cmd, chdir
    )


def rules():
    return collect_rules()

1	# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	2✔
4
5	from dataclasses import dataclass	2✔
6
7	from pants.backend.terraform.dependency_inference import (	2✔
8	TerraformDeploymentInvocationFilesRequest,
9	get_terraform_backend_and_vars,
10	)
11	from pants.backend.terraform.target_types import (	2✔
12	LockfileSourceField,
13	TerraformBackendConfigField,
14	TerraformDependenciesField,
15	TerraformDeploymentFieldSet,
16	TerraformFieldSet,
17	TerraformModuleSourcesField,
18	TerraformRootModuleField,
19	TerraformVarFileSourceField,
20	)
21	from pants.backend.terraform.tool import TerraformCommand, TerraformProcess	2✔
22	from pants.backend.terraform.utils import terraform_arg, terraform_relpath	2✔
23	from pants.core.target_types import FileSourceField, ResourceSourceField	2✔
24	from pants.core.util_rules.source_files import (	2✔
25	SourceFiles,
26	SourceFilesRequest,
27	determine_source_files,
28	)
29	from pants.engine.internals.build_files import resolve_address	2✔
30	from pants.engine.internals.graph import resolve_target, transitive_targets	2✔
31	from pants.engine.internals.native_engine import Digest, MergeDigests	2✔
32	from pants.engine.internals.selectors import concurrently	2✔
33	from pants.engine.intrinsics import execute_process, merge_digests	2✔
34	from pants.engine.process import ProcessExecutionFailure	2✔
35	from pants.engine.rules import collect_rules, implicitly, rule	2✔
36	from pants.engine.target import SourcesField, TransitiveTargetsRequest, WrappedTargetRequest	2✔
37	from pants.option.global_options import KeepSandboxes	2✔
38
39
40	@dataclass(frozen=True)	2✔
41	class TerraformDependenciesRequest:	2✔
42	chdir: str	2✔
43	backend_config: str \| None	2✔
44	lockfile: bool	2✔
45	dependencies_files: Digest	2✔
46
47	# Not initialising the backend means we won't access remote state. Useful for `validate`
48	initialise_backend: bool = False	2✔
49	upgrade: bool = False	2✔
50
51	def to_args(self) -> TerraformCommand:	2✔
52	args = ["init"]	×
53	if self.backend_config:	×
54	args.append(	×
55	terraform_arg(
56	"-backend-config",
57	terraform_relpath(self.chdir, self.backend_config),
58	)
59	)
60
61	# If we have a lockfile and aren't regenerating it, don't modify it
62	if self.lockfile and not self.upgrade:	×
63	args.append("-lockfile=readonly")	×
64
65	if self.upgrade:	×
66	args.append("-upgrade")	×
67
68	args.append(terraform_arg("-backend", str(self.initialise_backend)))	×
69
70	return TerraformCommand(tuple(args))	×
71
72
73	@dataclass(frozen=True)	2✔
74	class TerraformDependenciesResponse:	2✔
75	digest: Digest	2✔
76
77
78	@rule	2✔
79	async def get_terraform_providers(	2✔
80	req: TerraformDependenciesRequest,
81	keep_sandboxes: KeepSandboxes,
82	) -> TerraformDependenciesResponse:
83	init_process_description = (	×
84	f"Running `init` on Terraform module at `{req.chdir}` to fetch dependencies"
85	)
86	fetched_deps = await execute_process(	×
87	**implicitly(
88	TerraformProcess(
89	cmds=(req.to_args(),),
90	input_digest=req.dependencies_files,
91	output_files=(".terraform.lock.hcl",),
92	output_directories=(".terraform",),
93	description=init_process_description,
94	chdir=req.chdir,
95	)
96	)
97	)
98	if fetched_deps.exit_code != 0:	×
99	raise ProcessExecutionFailure.from_result(	×
100	fetched_deps, init_process_description, keep_sandboxes
101	)
102
103	return TerraformDependenciesResponse(fetched_deps.output_digest)	×
104
105
106	@dataclass(frozen=True)	2✔
107	class TerraformInitRequest:	2✔
108	root_module: TerraformRootModuleField	2✔
109	dependencies: TerraformDependenciesField	2✔
110
111	# Not initialising the backend means we won't access remote state. Useful for `validate`
112	initialise_backend: bool = False	2✔
113	upgrade: bool = False	2✔
114
115
116	@dataclass(frozen=True)	2✔
117	class TerraformInvocationRequirements:	2✔
118	"""The things you need to run a terraform command."""
119
120	terraform_sources: SourceFiles	2✔
121	dependencies_files: SourceFiles	2✔
122	init_cmd: TerraformDependenciesRequest	2✔
123	chdir: str	2✔
124
125
126	def terraform_fieldset_to_init_request(	2✔
127	terraform_fieldset: TerraformDeploymentFieldSet \| TerraformFieldSet,
128	) -> TerraformInitRequest:
129	"""Create a TerraformInitRequest from both Terraform Modules and Deployments."""
130	if isinstance(terraform_fieldset, TerraformDeploymentFieldSet):	×
131	deployment = terraform_fieldset	×
132	return TerraformInitRequest(deployment.root_module, deployment.dependencies)	×
133	elif isinstance(terraform_fieldset, TerraformFieldSet):	×
134	module = terraform_fieldset	×
135	return TerraformInitRequest(	×
136	TerraformRootModuleField(module.address.spec, module.address),
137	module.dependencies,
138	)
139	else:
140	raise TypeError(	×
141	f"Invalid type passed to initialise terraform tpye={type(terraform_fieldset)}"
142	)
143
144
145	@rule	2✔
146	async def prepare_terraform_invocation(	2✔
147	request: TerraformInitRequest,
148	) -> TerraformInvocationRequirements:
149	"""Prepare a terraform module or deployment to be operated on."""
150	this_targets_dependencies = await transitive_targets(	×
151	TransitiveTargetsRequest((request.dependencies.address,)), **implicitly()
152	)
153
154	address_input = request.root_module.to_address_input()	×
155	module_address = await resolve_address(**implicitly(address_input))	×
156
157	chdir = module_address.spec_path # TODO: spec_path is wrong, that's to the build file	×
158	# if the Terraform module is in the root, chdir will be "". Terraform needs a valid dir to change to
159	if not chdir:	×
160	chdir = "."	×
161
162	# TODO: is this still necessary, or do we pull it in with (transitive) dependencies?
163	module = await resolve_target(	×
164	WrappedTargetRequest(
165	module_address, description_of_origin=address_input.description_of_origin
166	),
167	**implicitly(),
168	)
169
170	source_files, dependencies_files = await concurrently(	×
171	determine_source_files(
172	SourceFilesRequest([module.target.get(SourcesField)])
173	), # TODO: get through transitive deps???
174	determine_source_files(
175	SourceFilesRequest(
176	[tgt.get(SourcesField) for tgt in this_targets_dependencies.dependencies],
177	for_sources_types=(
178	TerraformModuleSourcesField,
179	TerraformBackendConfigField,
180	TerraformVarFileSourceField,
181	LockfileSourceField,
182	FileSourceField,
183	ResourceSourceField,
184	),
185	enable_codegen=True,
186	)
187	),
188	)
189	invocation_files = await get_terraform_backend_and_vars(	×
190	TerraformDeploymentInvocationFilesRequest(
191	request.dependencies.address, request.dependencies
192	)
193	)
194	backend_config_tgts = invocation_files.backend_configs	×
195	if len(backend_config_tgts) == 0:	×
196	backend_config = None	×
197	elif len(backend_config_tgts) == 1:	×
198	backend_config_sources = await determine_source_files(	×
199	SourceFilesRequest([backend_config_tgts[0].get(SourcesField)])
200	)
201	backend_config = backend_config_sources.snapshot.files[0]	×
202	else:
203	# We've found multiple backend files, but that's only a problem if we need to initialise the backend.
204	# For example, we might be `validate`ing a `terraform_module` that has multiple backend files in the same dir,
205	# so we don't need to init the backend.
206	# The `terraform_deployment`s will have the references to the correct backends
207
208	if request.initialise_backend:	×
209	backend_config_names = [e.address for e in backend_config_tgts]	×
210	raise ValueError(	×
211	f"Found more than 1 backend config for a Terraform deployment. identified {backend_config_names}"
212	)
213	else:
214	backend_config = None	×
215
216	source_for_validate = await merge_digests(	×
217	MergeDigests([source_files.snapshot.digest, dependencies_files.snapshot.digest])
218	)
219
220	has_lockfile = invocation_files.lockfile is not None	×
221
222	terraform_init_cmd = TerraformDependenciesRequest(	×
223	chdir,
224	backend_config,
225	has_lockfile,
226	source_for_validate,
227	initialise_backend=request.initialise_backend,
228	upgrade=request.upgrade,
229	)
230	return TerraformInvocationRequirements(	×
231	source_files, dependencies_files, terraform_init_cmd, chdir
232	)
233
234
235	def rules():	2✔
236	return collect_rules()	×

pantsbuild / pants / 20632486505

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