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

82.22

/src/python/pants/backend/terraform/lint/tffmt/tffmt.py

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

import logging
from dataclasses import dataclass
from typing import cast

from pants.backend.terraform.partition import partition_files_by_directory
from pants.backend.terraform.target_types import TerraformFieldSet
from pants.backend.terraform.tool import TerraformCommand, TerraformProcess
from pants.backend.terraform.tool import rules as tool_rules
from pants.core.goals.fmt import FmtResult, FmtTargetsRequest, Partitions
from pants.core.util_rules import external_tool
from pants.core.util_rules.partitions import Partition
from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files
from pants.engine.process import fallible_to_exec_result_or_raise
from pants.engine.rules import collect_rules, implicitly, rule
from pants.option.option_types import SkipOption
from pants.option.subsystem import Subsystem
from pants.util.strutil import pluralize

logger = logging.getLogger(__name__)


class TfFmtSubsystem(Subsystem):
    options_scope = "terraform-fmt"
    name = "`terraform fmt`"
    help = "Terraform fmt options."

    skip = SkipOption("fmt", "lint")


class TffmtRequest(FmtTargetsRequest):
    field_set_type = TerraformFieldSet
    tool_subsystem = TfFmtSubsystem  # type: ignore[assignment]


@dataclass(frozen=True)
class PartitionMetadata:
    directory: str

    @property
    def description(self) -> str:
        return self.directory


@rule
async def partition_tffmt(
    request: TffmtRequest.PartitionRequest, tffmt: TfFmtSubsystem
) -> Partitions:
    if tffmt.skip:
        return Partitions()

    source_files = await determine_source_files(
        SourceFilesRequest([field_set.sources for field_set in request.field_sets])
    )

    return Partitions(
        Partition(tuple(files), PartitionMetadata(directory))
        for directory, files in partition_files_by_directory(source_files.files).items()
    )


@rule(desc="Format with `terraform fmt`")
async def tffmt_fmt(request: TffmtRequest.Batch, tffmt: TfFmtSubsystem) -> FmtResult:
    directory = cast(PartitionMetadata, request.partition_metadata).directory
    result = await fallible_to_exec_result_or_raise(
        **implicitly(
            TerraformProcess(
                cmds=(
                    TerraformCommand(
                        (
                            "fmt",
                            directory,
                        )
                    ),
                ),
                input_digest=request.snapshot.digest,
                output_files=request.files,
                description=f"Run `terraform fmt` on {pluralize(len(request.files), 'file')}.",
            )
        )
    )

    return await FmtResult.create(request, result)


def rules():
    return [
        *collect_rules(),
        *external_tool.rules(),
        *tool_rules(),
        *TffmtRequest.rules(),
    ]

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	2✔
4
5	import logging	2✔
6	from dataclasses import dataclass	2✔
7	from typing import cast	2✔
8
9	from pants.backend.terraform.partition import partition_files_by_directory	2✔
10	from pants.backend.terraform.target_types import TerraformFieldSet	2✔
11	from pants.backend.terraform.tool import TerraformCommand, TerraformProcess	2✔
12	from pants.backend.terraform.tool import rules as tool_rules	2✔
13	from pants.core.goals.fmt import FmtResult, FmtTargetsRequest, Partitions	2✔
14	from pants.core.util_rules import external_tool	2✔
15	from pants.core.util_rules.partitions import Partition	2✔
16	from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files	2✔
17	from pants.engine.process import fallible_to_exec_result_or_raise	2✔
18	from pants.engine.rules import collect_rules, implicitly, rule	2✔
19	from pants.option.option_types import SkipOption	2✔
20	from pants.option.subsystem import Subsystem	2✔
21	from pants.util.strutil import pluralize	2✔
22
23	logger = logging.getLogger(__name__)	2✔
24
25
26	class TfFmtSubsystem(Subsystem):	2✔
27	options_scope = "terraform-fmt"	2✔
28	name = "`terraform fmt`"	2✔
29	help = "Terraform fmt options."	2✔
30
31	skip = SkipOption("fmt", "lint")	2✔
32
33
34	class TffmtRequest(FmtTargetsRequest):	2✔
35	field_set_type = TerraformFieldSet	2✔
36	tool_subsystem = TfFmtSubsystem # type: ignore[assignment]	2✔
37
38
39	@dataclass(frozen=True)	2✔
40	class PartitionMetadata:	2✔
41	directory: str	2✔
42
43	@property	2✔
44	def description(self) -> str:	2✔
45	return self.directory	2✔
46
47
48	@rule	2✔
49	async def partition_tffmt(	2✔
50	request: TffmtRequest.PartitionRequest, tffmt: TfFmtSubsystem
51	) -> Partitions:
52	if tffmt.skip:	×
53	return Partitions()	×
54
55	source_files = await determine_source_files(	×
56	SourceFilesRequest([field_set.sources for field_set in request.field_sets])
57	)
58
59	return Partitions(	×
60	Partition(tuple(files), PartitionMetadata(directory))
61	for directory, files in partition_files_by_directory(source_files.files).items()
62	)
63
64
65	@rule(desc="Format with `terraform fmt`")	2✔
66	async def tffmt_fmt(request: TffmtRequest.Batch, tffmt: TfFmtSubsystem) -> FmtResult:	2✔
67	directory = cast(PartitionMetadata, request.partition_metadata).directory	×
68	result = await fallible_to_exec_result_or_raise(	×
69	**implicitly(
70	TerraformProcess(
71	cmds=(
72	TerraformCommand(
73	(
74	"fmt",
75	directory,
76	)
77	),
78	),
79	input_digest=request.snapshot.digest,
80	output_files=request.files,
81	description=f"Run `terraform fmt` on {pluralize(len(request.files), 'file')}.",
82	)
83	)
84	)
85
86	return await FmtResult.create(request, result)	×
87
88
89	def rules():	2✔
90	return [	×
91	*collect_rules(),
92	*external_tool.rules(),
93	*tool_rules(),
94	*TffmtRequest.rules(),
95	]

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