18847018991

Committed 27 Oct 2025 03:45PM UTC coverage: 92.254% (+12.0%) from 80.282%

Build # 18847018991

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge f1312fa87 into 06e216752

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

39 of 40 new or added lines in 11 files covered. (97.5%)

382 existing lines in 22 files now uncovered.

89230 of 96722 relevant lines covered (92.25%)

3.72 hits per line

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

74.74

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

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

import textwrap
from collections.abc import Sequence

import pytest

from pants.backend.terraform import dependencies, dependency_inference, tool
from pants.backend.terraform.goals import check
from pants.backend.terraform.goals.check import TerraformCheckRequest
from pants.backend.terraform.target_types import (
    TerraformDeploymentTarget,
    TerraformFieldSet,
    TerraformModuleTarget,
)
from pants.core.goals.check import CheckResult, CheckResults
from pants.core.util_rules import external_tool, source_files
from pants.core.util_rules.source_files import SourceFiles, SourceFilesRequest
from pants.engine.addresses import Address
from pants.engine.fs import CreateDigest, Digest, DigestContents, FileContent
from pants.engine.target import Target
from pants.testutil.rule_runner import QueryRule, RuleRunner


@pytest.fixture()
def rule_runner() -> RuleRunner:
    return RuleRunner(
        target_types=[TerraformModuleTarget, TerraformDeploymentTarget],
        rules=[
            *external_tool.rules(),
            *check.rules(),
            *tool.rules(),
            *source_files.rules(),
            *dependencies.rules(),
            *dependency_inference.rules(),
            QueryRule(CheckResults, (TerraformCheckRequest,)),
            QueryRule(SourceFiles, (SourceFilesRequest,)),
        ],
    )


GOOD_SOURCE = FileContent(
    "good.tf",
    textwrap.dedent(
        """
        locals {
          good_base = "xyzzy"
          foo       = "${local.good_base}"
        }
        """
    ).encode("utf-8"),
)

# The invalid part is the use of interpolation solely for a variable.
BAD_SOURCE = FileContent(
    "bad.tf",
    textwrap.dedent(
        """
        locals {
          bad_base = "xyzzy"
          bar = "${bad_base}"
        }
        """
    ).encode("utf-8"),
)

# This resource uses the null_resource provider. Terraform will need to run `init` to init the provider
SOURCE_WITH_PROVIDER = FileContent(
    "provided.tf",
    textwrap.dedent(
        """
        resource "null_resource" "dep" {}
        """
    ).encode("utf-8"),
)


def make_target(
    rule_runner: RuleRunner, source_files: list[FileContent], *, target_name="target"
) -> Target:
    files = {
        "BUILD": f"terraform_module(name='{target_name}mod')\nterraform_deployment(name='{target_name}', root_module=':{target_name}mod')\n",
    }
    files.update({source_file.path: source_file.content.decode() for source_file in source_files})
    rule_runner.write_files(files)
    return rule_runner.get_target(Address("", target_name=f"{target_name}mod"))


def run_terraform_validate(
    rule_runner: RuleRunner,
    targets: list[Target],
    *,
    args: list[str] | None = None,
) -> Sequence[CheckResult]:
    rule_runner.set_options(args or ())
    field_sets = [TerraformFieldSet.create(tgt) for tgt in targets]
    check_results = rule_runner.request(CheckResults, [TerraformCheckRequest(field_sets)])
    return check_results.results


def get_content(rule_runner: RuleRunner, digest: Digest) -> DigestContents:
    return rule_runner.request(DigestContents, [digest])


def get_digest(rule_runner: RuleRunner, source_files: list[FileContent]) -> Digest:
    return rule_runner.request(Digest, [CreateDigest(source_files)])


def test_passing_source(rule_runner: RuleRunner) -> None:
    target = make_target(rule_runner, [GOOD_SOURCE])
    check_results = run_terraform_validate(rule_runner, [target])
    assert len(check_results) == 1
    assert check_results[0].exit_code == 0
    assert check_results[0].stderr == ""


def test_failing_source(rule_runner: RuleRunner) -> None:
    target = make_target(rule_runner, [BAD_SOURCE])
    check_results = run_terraform_validate(rule_runner, [target])
    assert len(check_results) == 1
    assert check_results[0].exit_code == 1
    assert "bad.tf" in check_results[0].stderr


def test_mixed_sources(rule_runner: RuleRunner) -> None:
    target = make_target(rule_runner, [GOOD_SOURCE, BAD_SOURCE])
    check_results = run_terraform_validate(rule_runner, [target])
    assert len(check_results) == 1
    assert check_results[0].exit_code == 1
    assert "bad.tf" in check_results[0].stderr
    assert "good.tf" not in check_results[0].stderr


def test_multiple_targets(rule_runner: RuleRunner) -> None:
    source_files = [
        FileContent(
            "BUILD",
            textwrap.dedent(
                """\
                terraform_module(name="good", sources=["good.tf"])
                terraform_deployment(name="tgt_good", root_module=":good")
                terraform_module(name="bad", sources=["bad.tf"])
                terraform_deployment(name="tgt_bad", root_module=":bad")
            """
            ).encode("utf-8"),
        ),
        BAD_SOURCE,
        GOOD_SOURCE,
    ]

    rule_runner.write_files(
        {source_file.path: source_file.content.decode() for source_file in source_files}
    )

    targets = [
        rule_runner.get_target(Address("", target_name="good")),
        rule_runner.get_target(Address("", target_name="bad")),
    ]

    check_results = run_terraform_validate(rule_runner, targets)
    assert len(check_results) == 2
    for check_result in check_results:
        assert check_result.partition_description
        if "bad" in check_result.partition_description:
            assert check_result.exit_code == 1
        elif "good" in check_result.partition_description:
            assert check_result.exit_code == 0
        else:
            raise AssertionError(f"Did not find expected target in check result {check_result}")


def test_skip(rule_runner: RuleRunner) -> None:
    target = make_target(rule_runner, [BAD_SOURCE])
    lint_results = run_terraform_validate(rule_runner, [target], args=["--terraform-validate-skip"])
    assert not lint_results


def test_with_dependency(rule_runner: RuleRunner) -> None:
    """Sources with a provider need to have `terraform init` run before to initialise the provider.

    Without `init`, `terraform validate` fails. It is therefore sufficient to just test that the
    process ran successfully
    """
    targets = [make_target(rule_runner, [SOURCE_WITH_PROVIDER])]
    check_results = run_terraform_validate(rule_runner, targets)
    assert check_results[0].exit_code == 0


def test_in_folder(rule_runner: RuleRunner) -> None:
    """Test that we can `check` terraform files not in the root folder."""
    target_name = "in_folder"
    files = {
        "folder/BUILD": textwrap.dedent(
            f"""\
            terraform_deployment(name='{target_name}', root_module=':mod0')
            terraform_module(name='mod0')
            """
        ),
        "folder/provided.tf": textwrap.dedent(
            """
            resource "null_resource" "dep" {}
            resource "random_pet" "random" {}
            """
        ),
    }
    rule_runner.write_files(files)
    target = rule_runner.get_target(Address("folder", target_name="mod0"))

    check_results = run_terraform_validate(rule_runner, [target])
    assert check_results[0].exit_code == 0


def test_conflicting_provider_versions(rule_runner: RuleRunner) -> None:
    """Test that 2 separate terraform_modules can request conflicting providers.

    I think this test is really only necessary because we don't really separate the files we pass.
    If a large target glob is used (`::`), we get all the sources
    """
    target_name = "in_folder"
    versions = ["3.2.1", "3.2.2"]

    def make_terraform_module(version: str) -> dict[str, str]:
        return {
            f"folder{version}/BUILD": textwrap.dedent(
                f"""\
                terraform_deployment(name='{target_name}', root_module=':mod')
                terraform_module(name='mod')
            """
            ),
            f"folder{version}/provided.tf": textwrap.dedent(
                """
            terraform {
              required_providers {
                null = {
                  source = "hashicorp/null"
                  version = "%s"
                }
              }
            }
            resource "null_resource" "res" {}
            """
                % version
            ),
        }

    files = {}
    for version in versions:
        files.update(make_terraform_module(version))

    rule_runner.write_files(files)
    targets = [
        rule_runner.get_target(Address(folder, target_name="mod"))
        for folder in (f"folder{version}" for version in versions)
    ]

    check_results = run_terraform_validate(rule_runner, targets)
    assert len(check_results) == len(versions)
    assert all(check_result.exit_code == 0 for check_result in check_results)

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	1✔
4
5	import textwrap	1✔
6	from collections.abc import Sequence	1✔
7
8	import pytest	1✔
9
10	from pants.backend.terraform import dependencies, dependency_inference, tool	1✔
11	from pants.backend.terraform.goals import check	1✔
12	from pants.backend.terraform.goals.check import TerraformCheckRequest	1✔
13	from pants.backend.terraform.target_types import (	1✔
14	TerraformDeploymentTarget,
15	TerraformFieldSet,
16	TerraformModuleTarget,
17	)
18	from pants.core.goals.check import CheckResult, CheckResults	1✔
19	from pants.core.util_rules import external_tool, source_files	1✔
20	from pants.core.util_rules.source_files import SourceFiles, SourceFilesRequest	1✔
21	from pants.engine.addresses import Address	1✔
22	from pants.engine.fs import CreateDigest, Digest, DigestContents, FileContent	1✔
23	from pants.engine.target import Target	1✔
24	from pants.testutil.rule_runner import QueryRule, RuleRunner	1✔
25
26
27	@pytest.fixture()	1✔
28	def rule_runner() -> RuleRunner:	1✔
29	return RuleRunner(	1✔
30	target_types=[TerraformModuleTarget, TerraformDeploymentTarget],
31	rules=[
32	*external_tool.rules(),
33	*check.rules(),
34	*tool.rules(),
35	*source_files.rules(),
36	*dependencies.rules(),
37	*dependency_inference.rules(),
38	QueryRule(CheckResults, (TerraformCheckRequest,)),
39	QueryRule(SourceFiles, (SourceFilesRequest,)),
40	],
41	)
42
43
44	GOOD_SOURCE = FileContent(	1✔
45	"good.tf",
46	textwrap.dedent(
47	"""
48	locals {
49	good_base = "xyzzy"
50	foo = "${local.good_base}"
51	}
52	"""
53	).encode("utf-8"),
54	)
55
56	# The invalid part is the use of interpolation solely for a variable.
57	BAD_SOURCE = FileContent(	1✔
58	"bad.tf",
59	textwrap.dedent(
60	"""
61	locals {
62	bad_base = "xyzzy"
63	bar = "${bad_base}"
64	}
65	"""
66	).encode("utf-8"),
67	)
68
69	# This resource uses the null_resource provider. Terraform will need to run `init` to init the provider
70	SOURCE_WITH_PROVIDER = FileContent(	1✔
71	"provided.tf",
72	textwrap.dedent(
73	"""
74	resource "null_resource" "dep" {}
75	"""
76	).encode("utf-8"),
77	)
78
79
80	def make_target(	1✔
81	rule_runner: RuleRunner, source_files: list[FileContent], *, target_name="target"
82	) -> Target:
83	files = {	1✔
84	"BUILD": f"terraform_module(name='{target_name}mod')\nterraform_deployment(name='{target_name}', root_module=':{target_name}mod')\n",
85	}
86	files.update({source_file.path: source_file.content.decode() for source_file in source_files})	1✔
87	rule_runner.write_files(files)	1✔
88	return rule_runner.get_target(Address("", target_name=f"{target_name}mod"))	1✔
89
90
91	def run_terraform_validate(	1✔
92	rule_runner: RuleRunner,
93	targets: list[Target],
94	*,
95	args: list[str] \| None = None,
96	) -> Sequence[CheckResult]:
97	rule_runner.set_options(args or ())	1✔
98	field_sets = [TerraformFieldSet.create(tgt) for tgt in targets]	1✔
99	check_results = rule_runner.request(CheckResults, [TerraformCheckRequest(field_sets)])	1✔
100	return check_results.results	1✔
101
102
103	def get_content(rule_runner: RuleRunner, digest: Digest) -> DigestContents:	1✔
104	return rule_runner.request(DigestContents, [digest])	×
105
106
107	def get_digest(rule_runner: RuleRunner, source_files: list[FileContent]) -> Digest:	1✔
108	return rule_runner.request(Digest, [CreateDigest(source_files)])	×
109
110
111	def test_passing_source(rule_runner: RuleRunner) -> None:	1✔
112	target = make_target(rule_runner, [GOOD_SOURCE])	1✔
113	check_results = run_terraform_validate(rule_runner, [target])	1✔
UNCOV 114	assert len(check_results) == 1	×
UNCOV 115	assert check_results[0].exit_code == 0	×
UNCOV 116	assert check_results[0].stderr == ""	×
117
118
119	def test_failing_source(rule_runner: RuleRunner) -> None:	1✔
120	target = make_target(rule_runner, [BAD_SOURCE])	1✔
121	check_results = run_terraform_validate(rule_runner, [target])	1✔
UNCOV 122	assert len(check_results) == 1	×
UNCOV 123	assert check_results[0].exit_code == 1	×
UNCOV 124	assert "bad.tf" in check_results[0].stderr	×
125
126
127	def test_mixed_sources(rule_runner: RuleRunner) -> None:	1✔
128	target = make_target(rule_runner, [GOOD_SOURCE, BAD_SOURCE])	1✔
129	check_results = run_terraform_validate(rule_runner, [target])	1✔
UNCOV 130	assert len(check_results) == 1	×
UNCOV 131	assert check_results[0].exit_code == 1	×
UNCOV 132	assert "bad.tf" in check_results[0].stderr	×
UNCOV 133	assert "good.tf" not in check_results[0].stderr	×
134
135
136	def test_multiple_targets(rule_runner: RuleRunner) -> None:	1✔
137	source_files = [	1✔
138	FileContent(
139	"BUILD",
140	textwrap.dedent(
141	"""\
142	terraform_module(name="good", sources=["good.tf"])
143	terraform_deployment(name="tgt_good", root_module=":good")
144	terraform_module(name="bad", sources=["bad.tf"])
145	terraform_deployment(name="tgt_bad", root_module=":bad")
146	"""
147	).encode("utf-8"),
148	),
149	BAD_SOURCE,
150	GOOD_SOURCE,
151	]
152
153	rule_runner.write_files(	1✔
154	{source_file.path: source_file.content.decode() for source_file in source_files}
155	)
156
157	targets = [	1✔
158	rule_runner.get_target(Address("", target_name="good")),
159	rule_runner.get_target(Address("", target_name="bad")),
160	]
161
162	check_results = run_terraform_validate(rule_runner, targets)	1✔
UNCOV 163	assert len(check_results) == 2	×
UNCOV 164	for check_result in check_results:	×
UNCOV 165	assert check_result.partition_description	×
UNCOV 166	if "bad" in check_result.partition_description:	×
UNCOV 167	assert check_result.exit_code == 1	×
UNCOV 168	elif "good" in check_result.partition_description:	×
UNCOV 169	assert check_result.exit_code == 0	×
170	else:
171	raise AssertionError(f"Did not find expected target in check result {check_result}")	×
172
173
174	def test_skip(rule_runner: RuleRunner) -> None:	1✔
175	target = make_target(rule_runner, [BAD_SOURCE])	1✔
176	lint_results = run_terraform_validate(rule_runner, [target], args=["--terraform-validate-skip"])	1✔
177	assert not lint_results	1✔
178
179
180	def test_with_dependency(rule_runner: RuleRunner) -> None:	1✔
181	"""Sources with a provider need to have `terraform init` run before to initialise the provider.
182
183	Without `init`, `terraform validate` fails. It is therefore sufficient to just test that the
184	process ran successfully
185	"""
186	targets = [make_target(rule_runner, [SOURCE_WITH_PROVIDER])]	1✔
187	check_results = run_terraform_validate(rule_runner, targets)	1✔
UNCOV 188	assert check_results[0].exit_code == 0	×
189
190
191	def test_in_folder(rule_runner: RuleRunner) -> None:	1✔
192	"""Test that we can `check` terraform files not in the root folder."""
193	target_name = "in_folder"	1✔
194	files = {	1✔
195	"folder/BUILD": textwrap.dedent(
196	f"""\
197	terraform_deployment(name='{target_name}', root_module=':mod0')
198	terraform_module(name='mod0')
199	"""
200	),
201	"folder/provided.tf": textwrap.dedent(
202	"""
203	resource "null_resource" "dep" {}
204	resource "random_pet" "random" {}
205	"""
206	),
207	}
208	rule_runner.write_files(files)	1✔
209	target = rule_runner.get_target(Address("folder", target_name="mod0"))	1✔
210
211	check_results = run_terraform_validate(rule_runner, [target])	1✔
UNCOV 212	assert check_results[0].exit_code == 0	×
213
214
215	def test_conflicting_provider_versions(rule_runner: RuleRunner) -> None:	1✔
216	"""Test that 2 separate terraform_modules can request conflicting providers.
217
218	I think this test is really only necessary because we don't really separate the files we pass.
219	If a large target glob is used (`::`), we get all the sources
220	"""
221	target_name = "in_folder"	1✔
222	versions = ["3.2.1", "3.2.2"]	1✔
223
224	def make_terraform_module(version: str) -> dict[str, str]:	1✔
225	return {	1✔
226	f"folder{version}/BUILD": textwrap.dedent(
227	f"""\
228	terraform_deployment(name='{target_name}', root_module=':mod')
229	terraform_module(name='mod')
230	"""
231	),
232	f"folder{version}/provided.tf": textwrap.dedent(
233	"""
234	terraform {
235	required_providers {
236	null = {
237	source = "hashicorp/null"
238	version = "%s"
239	}
240	}
241	}
242	resource "null_resource" "res" {}
243	"""
244	% version
245	),
246	}
247
248	files = {}	1✔
249	for version in versions:	1✔
250	files.update(make_terraform_module(version))	1✔
251
252	rule_runner.write_files(files)	1✔
253	targets = [	1✔
254	rule_runner.get_target(Address(folder, target_name="mod"))
255	for folder in (f"folder{version}" for version in versions)
256	]
257
258	check_results = run_terraform_validate(rule_runner, targets)	1✔
UNCOV 259	assert len(check_results) == len(versions)	×
UNCOV 260	assert all(check_result.exit_code == 0 for check_result in check_results)	×

pantsbuild / pants / 18847018991

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