• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

pantsbuild / pants / 19572556425

21 Nov 2025 01:50PM UTC coverage: 80.295% (+0.007%) from 80.288%
19572556425

Pull #22906

github

web-flow
Merge c59ca89b1 into 70dc9fe34
Pull Request #22906: Update Coursier default version to v2.1.24

3 of 3 new or added lines in 2 files covered. (100.0%)

294 existing lines in 12 files now uncovered.

78385 of 97621 relevant lines covered (80.3%)

3.36 hits per line

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

59.85
/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).
3
from __future__ import annotations
6✔
4

5
from collections.abc import Iterable, Sequence
6✔
6
from dataclasses import dataclass
6✔
7
from pathlib import PurePath
6✔
8

9
from pants.backend.python.subsystems.python_tool_base import PythonToolRequirementsBase
6✔
10
from pants.backend.python.target_types import EntryPoint
6✔
11
from pants.backend.python.util_rules.pex import VenvPex, VenvPexProcess, create_venv_pex
6✔
12
from pants.backend.python.util_rules.pex import rules as pex_rules
6✔
13
from pants.backend.python.util_rules.pex import setup_venv_pex_process
6✔
14
from pants.backend.terraform.target_types import (
6✔
15
    TerraformBackendTarget,
16
    TerraformDependenciesField,
17
    TerraformDeploymentFieldSet,
18
    TerraformLockfileTarget,
19
    TerraformModuleSourcesField,
20
    TerraformVarFileTarget,
21
)
22
from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior
6✔
23
from pants.base.specs import DirGlobSpec, DirLiteralSpec, RawSpecs
6✔
24
from pants.core.target_types import LockfileTarget
6✔
25
from pants.engine.addresses import Addresses
6✔
26
from pants.engine.fs import CreateDigest, Digest, FileContent
6✔
27
from pants.engine.internals.build_files import resolve_address
6✔
28
from pants.engine.internals.graph import (
6✔
29
    determine_explicitly_provided_dependencies,
30
    hydrate_sources,
31
    resolve_targets,
32
)
33
from pants.engine.internals.native_engine import Address, AddressInput
6✔
34
from pants.engine.internals.selectors import concurrently
6✔
35
from pants.engine.intrinsics import create_digest
6✔
36
from pants.engine.process import Process, execute_process_or_raise
6✔
37
from pants.engine.rules import collect_rules, implicitly, rule
6✔
38
from pants.engine.target import (
6✔
39
    DependenciesRequest,
40
    FieldSet,
41
    HydrateSourcesRequest,
42
    InferDependenciesRequest,
43
    InferredDependencies,
44
    Target,
45
)
46
from pants.engine.unions import UnionRule
6✔
47
from pants.util.dirutil import group_by_dir
6✔
48
from pants.util.logging import LogLevel
6✔
49
from pants.util.ordered_set import OrderedSet
6✔
50
from pants.util.resources import read_resource
6✔
51
from pants.util.strutil import bullet_list, softwrap
6✔
52

53

54
class TerraformHcl2Parser(PythonToolRequirementsBase):
6✔
55
    options_scope = "terraform-hcl2-parser"
6✔
56
    help_short = "Used to parse Terraform modules to infer their dependencies."
6✔
57

58
    # versions 4.3.2+ have parsing issues; bump once resolved
59
    default_requirements = ["python-hcl2>=3.0.5,<=4.3.0"]
6✔
60

61
    register_interpreter_constraints = True
6✔
62

63
    default_lockfile_resource = ("pants.backend.terraform", "hcl2.lock")
6✔
64

65

66
@dataclass(frozen=True)
6✔
67
class ParserSetup:
6✔
68
    pex: VenvPex
6✔
69

70

71
@rule
6✔
72
async def setup_parser(hcl2_parser: TerraformHcl2Parser) -> ParserSetup:
6✔
UNCOV
73
    parser_script_content = read_resource("pants.backend.terraform", "hcl2_parser.py")
×
UNCOV
74
    if not parser_script_content:
×
UNCOV
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
    )
UNCOV
80
    parser_digest = await create_digest(CreateDigest([parser_content]))
×
81

UNCOV
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
    )
UNCOV
89
    return ParserSetup(parser_pex)
×
90

91

92
@dataclass(frozen=True)
6✔
93
class ParseTerraformModuleSources:
6✔
94
    sources_digest: Digest
6✔
95
    paths: tuple[str, ...]
6✔
96

97

98
@rule
6✔
99
async def setup_process_for_parse_terraform_module_sources(
6✔
100
    request: ParseTerraformModuleSources, parser: ParserSetup
101
) -> Process:
UNCOV
102
    dir_paths = ", ".join(sorted(group_by_dir(request.paths).keys()))
×
103

UNCOV
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
    )
UNCOV
114
    return process
×
115

116

117
@dataclass(frozen=True)
6✔
118
class TerraformModuleDependenciesInferenceFieldSet(FieldSet):
6✔
119
    required_fields = (TerraformModuleSourcesField, TerraformDependenciesField)
6✔
120

121
    sources: TerraformModuleSourcesField
6✔
122
    dependencies: TerraformDependenciesField
6✔
123

124

125
class InferTerraformModuleDependenciesRequest(InferDependenciesRequest):
6✔
126
    infer_from = TerraformModuleDependenciesInferenceFieldSet
6✔
127

128

129
@dataclass(frozen=True)
6✔
130
class TerraformDeploymentDependenciesInferenceFieldSet(TerraformDeploymentFieldSet):
6✔
131
    pass
6✔
132

133

134
class InferTerraformDeploymentDependenciesRequest(InferDependenciesRequest):
6✔
135
    infer_from = TerraformDeploymentDependenciesInferenceFieldSet
6✔
136

137

138
def find_targets_of_type(tgts, of_type) -> tuple:
6✔
UNCOV
139
    if tgts:
×
UNCOV
140
        return tuple(e for e in tgts if isinstance(e, of_type))
×
141
    else:
UNCOV
142
        return ()
×
143

144

145
@dataclass(frozen=True)
6✔
146
class TerraformDeploymentInvocationFilesRequest:
6✔
147
    """TODO: is there a way to convert between FS? We could convert the inference FS to the deployment FS itself"""
148

149
    address: Address
6✔
150
    dependencies: TerraformDependenciesField
6✔
151

152

153
@dataclass(frozen=True)
6✔
154
class TerraformDeploymentInvocationFiles:
6✔
155
    """The files passed in to the invocation of `terraform`"""
156

157
    backend_configs: tuple[TerraformBackendTarget, ...]
6✔
158
    vars_files: tuple[TerraformVarFileTarget, ...]
6✔
159
    lockfile: LockfileTarget | None
6✔
160

161

162
@rule
6✔
163
async def get_terraform_backend_and_vars(
6✔
164
    field_set: TerraformDeploymentInvocationFilesRequest,
165
) -> TerraformDeploymentInvocationFiles:
UNCOV
166
    this_address = field_set.address
×
167

UNCOV
168
    explicit_deps = await determine_explicitly_provided_dependencies(
×
169
        **implicitly(DependenciesRequest(field_set.dependencies))
170
    )
UNCOV
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
    )
UNCOV
182
    return identify_terraform_backend_and_vars(explicit_deps_tgt, tgts_in_dir)
×
183

184

185
class InvalidLockfileException(Exception):
6✔
186
    @classmethod
6✔
187
    def too_many_lockfiles(
6✔
188
        cls, lockfiles: Iterable[TerraformLockfileTarget]
189
    ) -> InvalidLockfileException:
UNCOV
190
        addresses = sorted(tgt.address.spec for tgt in lockfiles)
×
UNCOV
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

207
def identify_terraform_backend_and_vars(
6✔
208
    explicit_deps: Sequence[Target], tgts_in_dir: Sequence[Target]
209
) -> TerraformDeploymentInvocationFiles:
UNCOV
210
    has_explicit_backend = find_targets_of_type(explicit_deps, TerraformBackendTarget)
×
UNCOV
211
    if not has_explicit_backend:
×
212
        # Note: Terraform does not support multiple backends, but dep inference isn't the place to enforce that
UNCOV
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)
×
UNCOV
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)
×
UNCOV
224
    if len(lockfiles) == 1:
×
225
        lockfile = lockfiles[0]
×
UNCOV
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:
UNCOV
231
        lockfile = None
×
232

233
    return TerraformDeploymentInvocationFiles(backend_targets, vars_targets, lockfile)
×
234

235

236
async def _infer_dependencies_from_sources(
6✔
237
    request: InferTerraformModuleDependenciesRequest,
238
) -> list[Address]:
239
    """Parse the source code for references to other modules."""
UNCOV
240
    hydrated_sources = await hydrate_sources(
×
241
        HydrateSourcesRequest(request.field_set.sources), **implicitly()
242
    )
UNCOV
243
    paths = OrderedSet(
×
244
        filename for filename in hydrated_sources.snapshot.files if filename.endswith(".tf")
245
    )
UNCOV
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
    )
UNCOV
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.
UNCOV
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.
UNCOV
267
    terraform_module_addresses = [
×
268
        tgt.address for tgt in candidate_targets if tgt.has_field(TerraformModuleSourcesField)
269
    ]
UNCOV
270
    return terraform_module_addresses
×
271

272

273
async def _infer_lockfile(request: InferTerraformModuleDependenciesRequest) -> list[Address]:
6✔
274
    """Pull in the lockfile for a Terraform module.
275

276
    This is necessary for `terraform validate`.
277
    """
UNCOV
278
    invocation_files = await get_terraform_backend_and_vars(
×
279
        TerraformDeploymentInvocationFilesRequest(
280
            request.field_set.address, request.field_set.dependencies
281
        )
282
    )
UNCOV
283
    if invocation_files.lockfile:
×
UNCOV
284
        return [invocation_files.lockfile.address]
×
285
    else:
UNCOV
286
        return []
×
287

288

289
@rule
6✔
290
async def infer_terraform_module_dependencies(
6✔
291
    request: InferTerraformModuleDependenciesRequest,
292
) -> InferredDependencies:
UNCOV
293
    terraform_module_addresses = await _infer_dependencies_from_sources(request)
×
UNCOV
294
    lockfile_address = await _infer_lockfile(request)
×
295

UNCOV
296
    return InferredDependencies([*terraform_module_addresses, *lockfile_address])
×
297

298

299
@rule
6✔
300
async def infer_terraform_deployment_dependencies(
6✔
301
    request: InferTerraformDeploymentDependenciesRequest,
302
) -> InferredDependencies:
UNCOV
303
    root_module_address_input = request.field_set.root_module.to_address_input()
×
UNCOV
304
    root_module = await resolve_address(**implicitly({root_module_address_input: AddressInput}))
×
UNCOV
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
    )
UNCOV
312
    deps.extend(e.address for e in invocation_files.backend_configs)
×
UNCOV
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

319
def rules():
6✔
320
    return [
6✔
321
        *collect_rules(),
322
        *pex_rules(),
323
        UnionRule(InferDependenciesRequest, InferTerraformModuleDependenciesRequest),
324
        UnionRule(InferDependenciesRequest, InferTerraformDeploymentDependenciesRequest),
325
    ]
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc