23177125175

Committed 17 Mar 2026 03:32AM UTC coverage: 52.677% (-40.3%) from 92.932%

Build # 23177125175

Build Type

Pull #23177

github

Committed by

web-flow

Commit Message

Merge 1824dfbf4 into 0b9fdfb0e

Pull Request Pull Request #23177: Bump the gha-deps group across 1 directory with 4 updates

Coverage Stats

31687 of 60153 relevant lines covered (52.68%)

1.05 hits per line

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

71.43

/src/python/pants/backend/terraform/goals/lockfiles.py

# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
import os.path
from dataclasses import dataclass
from pathlib import Path

from pants.backend.terraform.dependencies import TerraformInitRequest, prepare_terraform_invocation
from pants.backend.terraform.target_types import (
    TerraformDependenciesField,
    TerraformLockfileTarget,
    TerraformModuleSourcesField,
    TerraformModuleTarget,
    TerraformRootModuleField,
)
from pants.backend.terraform.tool import TerraformCommand, TerraformProcess, TerraformTool
from pants.backend.terraform.utils import terraform_arg
from pants.core.goals.generate_lockfiles import (
    GenerateLockfile,
    GenerateLockfileResult,
    KnownUserResolveNames,
    KnownUserResolveNamesRequest,
    RequestedUserResolveNames,
    UserGenerateLockfiles,
)
from pants.engine.addresses import Addresses
from pants.engine.fs import PathGlobs
from pants.engine.internals.build_files import resolve_address
from pants.engine.internals.graph import resolve_targets
from pants.engine.internals.native_engine import AddressInput, MergeDigests
from pants.engine.internals.selectors import concurrently
from pants.engine.internals.synthetic_targets import SyntheticAddressMaps, SyntheticTargetsRequest
from pants.engine.internals.target_adaptor import TargetAdaptor
from pants.engine.intrinsics import digest_to_snapshot, execute_process, merge_digests
from pants.engine.process import ProcessExecutionFailure
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import AllTargets
from pants.engine.unions import UnionRule
from pants.option.global_options import KeepSandboxes


class KnownTerraformResolveNamesRequest(KnownUserResolveNamesRequest):
    pass


class RequestedTerraformResolveNames(RequestedUserResolveNames):
    pass


@rule
async def identify_user_resolves_from_terraform_files(
    _: KnownTerraformResolveNamesRequest,
    all_targets: AllTargets,
) -> KnownUserResolveNames:
    """We don't use the TerraformSyntheticLockfileTargetsRequest because those only include
    lockfiles that have been written and not new lockfiles."""
    known_terraform_module_dirs = set()

    for tgt in all_targets:
        if tgt.has_field(TerraformModuleSourcesField):
            known_terraform_module_dirs.add(tgt.address.spec)

    return KnownUserResolveNames(
        names=tuple(known_terraform_module_dirs),
        option_name="[terraform].resolves",
        requested_resolve_names_cls=RequestedTerraformResolveNames,
    )


@dataclass(frozen=True)
class GenerateTerraformLockfile(GenerateLockfile):
    target: TerraformModuleTarget
    platforms: tuple[str, ...]


@rule
async def setup_user_lockfile_requests(
    requested: RequestedTerraformResolveNames,
    terraform: TerraformTool,
) -> UserGenerateLockfiles:
    addrs = await concurrently(
        resolve_address(
            **implicitly(
                {
                    AddressInput.parse(
                        m, description_of_origin="setup Terraform lockfiles"
                    ): AddressInput
                }
            )
        )
        for m in requested
    )

    targets = await resolve_targets(**implicitly(Addresses(addrs)))
    deployment_targets = [t for t in targets if isinstance(t, TerraformModuleTarget)]

    return UserGenerateLockfiles(
        [
            GenerateTerraformLockfile(
                target=tgt,
                platforms=terraform.platforms,
                resolve_name=tgt.residence_dir,
                lockfile_dest=(Path(tgt.residence_dir) / ".terraform.lock.hcl").as_posix(),
                diff=False,
            )
            for tgt in deployment_targets
        ]
    )


@rule
async def generate_lockfile_from_sources(
    lockfile_request: GenerateTerraformLockfile,
    keep_sandboxes: KeepSandboxes,
) -> GenerateLockfileResult:
    """Generate a Terraform lockfile by running `terraform providers lock` on the sources."""
    initialised_terraform = await prepare_terraform_invocation(
        TerraformInitRequest(
            TerraformRootModuleField(
                lockfile_request.target.address.spec, lockfile_request.target.address
            ),
            lockfile_request.target[TerraformDependenciesField],
            initialise_backend=False,
            upgrade=True,
        )
    )

    sources_and_deps = await merge_digests(
        MergeDigests(
            [
                initialised_terraform.terraform_sources.snapshot.digest,
                initialised_terraform.dependencies_files.snapshot.digest,
            ]
        )
    )

    args = (
        "providers",
        "lock",
        *(terraform_arg("-platform", platform) for platform in lockfile_request.platforms),
    )

    provider_lock_description = f"Update terraform lockfile for {lockfile_request.resolve_name}"
    multiplatform_lockfile = await execute_process(
        **implicitly(
            TerraformProcess(
                cmds=(
                    initialised_terraform.init_cmd.to_args(),
                    TerraformCommand(args),
                ),
                input_digest=sources_and_deps,
                output_files=(".terraform.lock.hcl",),
                description=provider_lock_description,
                chdir=initialised_terraform.chdir,
            )
        )
    )
    if multiplatform_lockfile.exit_code != 0:
        raise ProcessExecutionFailure(
            multiplatform_lockfile.exit_code,
            multiplatform_lockfile.stdout,
            multiplatform_lockfile.stderr,
            provider_lock_description,
            keep_sandboxes=keep_sandboxes,
        )

    return GenerateLockfileResult(
        multiplatform_lockfile.output_digest,
        lockfile_request.resolve_name,
        lockfile_request.lockfile_dest,
    )


@dataclass(frozen=True)
class TerraformSyntheticLockfileTargetsRequest(SyntheticTargetsRequest):
    path: str = SyntheticTargetsRequest.REQUEST_TARGETS_PER_DIRECTORY


@rule
async def terraform_lockfile_synthetic_targets(
    request: TerraformSyntheticLockfileTargetsRequest,
) -> SyntheticAddressMaps:
    path = request.path
    lockfile = await digest_to_snapshot(
        **implicitly(PathGlobs([Path(path, ".terraform.lock.hcl").as_posix()]))
    )
    if not lockfile.files:
        return SyntheticAddressMaps(tuple())

    return SyntheticAddressMaps.for_targets_request(
        request,
        [
            (
                os.path.join(path, "BUILD.terraform-lockfiles"),
                (
                    TargetAdaptor(
                        TerraformLockfileTarget.alias,
                        name=".terraform.lock.hcl",
                        source=".terraform.lock.hcl",
                        __description_of_origin__="terraform",
                    ),
                ),
            )
        ],
    )


def rules():
    return (
        *collect_rules(),
        UnionRule(GenerateLockfile, GenerateTerraformLockfile),
        UnionRule(KnownUserResolveNamesRequest, KnownTerraformResolveNamesRequest),
        UnionRule(RequestedUserResolveNames, RequestedTerraformResolveNames),
        UnionRule(SyntheticTargetsRequest, TerraformSyntheticLockfileTargetsRequest),
    )

1	# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	import os.path	2✔
4	from dataclasses import dataclass	2✔
5	from pathlib import Path	2✔
6
7	from pants.backend.terraform.dependencies import TerraformInitRequest, prepare_terraform_invocation	2✔
8	from pants.backend.terraform.target_types import (	2✔
9	TerraformDependenciesField,
10	TerraformLockfileTarget,
11	TerraformModuleSourcesField,
12	TerraformModuleTarget,
13	TerraformRootModuleField,
14	)
15	from pants.backend.terraform.tool import TerraformCommand, TerraformProcess, TerraformTool	2✔
16	from pants.backend.terraform.utils import terraform_arg	2✔
17	from pants.core.goals.generate_lockfiles import (	2✔
18	GenerateLockfile,
19	GenerateLockfileResult,
20	KnownUserResolveNames,
21	KnownUserResolveNamesRequest,
22	RequestedUserResolveNames,
23	UserGenerateLockfiles,
24	)
25	from pants.engine.addresses import Addresses	2✔
26	from pants.engine.fs import PathGlobs	2✔
27	from pants.engine.internals.build_files import resolve_address	2✔
28	from pants.engine.internals.graph import resolve_targets	2✔
29	from pants.engine.internals.native_engine import AddressInput, MergeDigests	2✔
30	from pants.engine.internals.selectors import concurrently	2✔
31	from pants.engine.internals.synthetic_targets import SyntheticAddressMaps, SyntheticTargetsRequest	2✔
32	from pants.engine.internals.target_adaptor import TargetAdaptor	2✔
33	from pants.engine.intrinsics import digest_to_snapshot, 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 AllTargets	2✔
37	from pants.engine.unions import UnionRule	2✔
38	from pants.option.global_options import KeepSandboxes	2✔
39
40
41	class KnownTerraformResolveNamesRequest(KnownUserResolveNamesRequest):	2✔
42	pass	2✔
43
44
45	class RequestedTerraformResolveNames(RequestedUserResolveNames):	2✔
46	pass	2✔
47
48
49	@rule	2✔
50	async def identify_user_resolves_from_terraform_files(	2✔
51	_: KnownTerraformResolveNamesRequest,
52	all_targets: AllTargets,
53	) -> KnownUserResolveNames:
54	"""We don't use the TerraformSyntheticLockfileTargetsRequest because those only include
55	lockfiles that have been written and not new lockfiles."""
56	known_terraform_module_dirs = set()	×
57
58	for tgt in all_targets:	×
59	if tgt.has_field(TerraformModuleSourcesField):	×
60	known_terraform_module_dirs.add(tgt.address.spec)	×
61
62	return KnownUserResolveNames(	×
63	names=tuple(known_terraform_module_dirs),
64	option_name="[terraform].resolves",
65	requested_resolve_names_cls=RequestedTerraformResolveNames,
66	)
67
68
69	@dataclass(frozen=True)	2✔
70	class GenerateTerraformLockfile(GenerateLockfile):	2✔
71	target: TerraformModuleTarget
72	platforms: tuple[str, ...]
73
74
75	@rule	2✔
76	async def setup_user_lockfile_requests(	2✔
77	requested: RequestedTerraformResolveNames,
78	terraform: TerraformTool,
79	) -> UserGenerateLockfiles:
80	addrs = await concurrently(	×
81	resolve_address(
82	**implicitly(
83	{
84	AddressInput.parse(
85	m, description_of_origin="setup Terraform lockfiles"
86	): AddressInput
87	}
88	)
89	)
90	for m in requested
91	)
92
93	targets = await resolve_targets(**implicitly(Addresses(addrs)))	×
94	deployment_targets = [t for t in targets if isinstance(t, TerraformModuleTarget)]	×
95
96	return UserGenerateLockfiles(	×
97	[
98	GenerateTerraformLockfile(
99	target=tgt,
100	platforms=terraform.platforms,
101	resolve_name=tgt.residence_dir,
102	lockfile_dest=(Path(tgt.residence_dir) / ".terraform.lock.hcl").as_posix(),
103	diff=False,
104	)
105	for tgt in deployment_targets
106	]
107	)
108
109
110	@rule	2✔
111	async def generate_lockfile_from_sources(	2✔
112	lockfile_request: GenerateTerraformLockfile,
113	keep_sandboxes: KeepSandboxes,
114	) -> GenerateLockfileResult:
115	"""Generate a Terraform lockfile by running `terraform providers lock` on the sources."""
116	initialised_terraform = await prepare_terraform_invocation(	×
117	TerraformInitRequest(
118	TerraformRootModuleField(
119	lockfile_request.target.address.spec, lockfile_request.target.address
120	),
121	lockfile_request.target[TerraformDependenciesField],
122	initialise_backend=False,
123	upgrade=True,
124	)
125	)
126
127	sources_and_deps = await merge_digests(	×
128	MergeDigests(
129	[
130	initialised_terraform.terraform_sources.snapshot.digest,
131	initialised_terraform.dependencies_files.snapshot.digest,
132	]
133	)
134	)
135
136	args = (	×
137	"providers",
138	"lock",
139	*(terraform_arg("-platform", platform) for platform in lockfile_request.platforms),
140	)
141
142	provider_lock_description = f"Update terraform lockfile for {lockfile_request.resolve_name}"	×
143	multiplatform_lockfile = await execute_process(	×
144	**implicitly(
145	TerraformProcess(
146	cmds=(
147	initialised_terraform.init_cmd.to_args(),
148	TerraformCommand(args),
149	),
150	input_digest=sources_and_deps,
151	output_files=(".terraform.lock.hcl",),
152	description=provider_lock_description,
153	chdir=initialised_terraform.chdir,
154	)
155	)
156	)
157	if multiplatform_lockfile.exit_code != 0:	×
158	raise ProcessExecutionFailure(	×
159	multiplatform_lockfile.exit_code,
160	multiplatform_lockfile.stdout,
161	multiplatform_lockfile.stderr,
162	provider_lock_description,
163	keep_sandboxes=keep_sandboxes,
164	)
165
166	return GenerateLockfileResult(	×
167	multiplatform_lockfile.output_digest,
168	lockfile_request.resolve_name,
169	lockfile_request.lockfile_dest,
170	)
171
172
173	@dataclass(frozen=True)	2✔
174	class TerraformSyntheticLockfileTargetsRequest(SyntheticTargetsRequest):	2✔
175	path: str = SyntheticTargetsRequest.REQUEST_TARGETS_PER_DIRECTORY	2✔
176
177
178	@rule	2✔
179	async def terraform_lockfile_synthetic_targets(	2✔
180	request: TerraformSyntheticLockfileTargetsRequest,
181	) -> SyntheticAddressMaps:
182	path = request.path	2✔
183	lockfile = await digest_to_snapshot(	2✔
184	**implicitly(PathGlobs([Path(path, ".terraform.lock.hcl").as_posix()]))
185	)
186	if not lockfile.files:	2✔
187	return SyntheticAddressMaps(tuple())	2✔
188
189	return SyntheticAddressMaps.for_targets_request(	×
190	request,
191	[
192	(
193	os.path.join(path, "BUILD.terraform-lockfiles"),
194	(
195	TargetAdaptor(
196	TerraformLockfileTarget.alias,
197	name=".terraform.lock.hcl",
198	source=".terraform.lock.hcl",
199	__description_of_origin__="terraform",
200	),
201	),
202	)
203	],
204	)
205
206
207	def rules():	2✔
208	return (	2✔
209	*collect_rules(),
210	UnionRule(GenerateLockfile, GenerateTerraformLockfile),
211	UnionRule(KnownUserResolveNamesRequest, KnownTerraformResolveNamesRequest),
212	UnionRule(RequestedUserResolveNames, RequestedTerraformResolveNames),
213	UnionRule(SyntheticTargetsRequest, TerraformSyntheticLockfileTargetsRequest),
214	)

pantsbuild / pants / 23177125175

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