18252174847

Committed 05 Oct 2025 01:36AM UTC coverage: 43.382% (-36.9%) from 80.261%

Build # 18252174847

Build Type

push

github

Committed by

web-flow

Commit Message

run tests on mac arm (#22717)

Just doing the minimal to pull forward the x86_64 pattern.

ref #20993

Run Details

25776 of 59416 relevant lines covered (43.38%)

1.3 hits per line

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

0.0

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

# Copyright 2023 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 pants.backend.terraform.dependencies import TerraformInitRequest, prepare_terraform_invocation
from pants.backend.terraform.dependency_inference import (
    TerraformDeploymentInvocationFilesRequest,
    get_terraform_backend_and_vars,
)
from pants.backend.terraform.target_types import TerraformDeploymentFieldSet
from pants.backend.terraform.tool import (
    TerraformCommand,
    TerraformProcess,
    TerraformTool,
    setup_terraform_process,
)
from pants.backend.terraform.utils import terraform_arg, terraform_relpath
from pants.core.goals.deploy import DeployFieldSet, DeployProcess, DeploySubsystem
from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.internals.native_engine import MergeDigests
from pants.engine.intrinsics import merge_digests
from pants.engine.process import InteractiveProcess
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import SourcesField
from pants.engine.unions import UnionRule
from pants.option.global_options import KeepSandboxes
from pants.util.logging import LogLevel

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class DeployTerraformFieldSet(TerraformDeploymentFieldSet, DeployFieldSet):
    pass


@dataclass(frozen=True)
class TerraformDeploymentRequest(EngineAwareParameter):
    field_set: TerraformDeploymentFieldSet


@rule
async def prepare_terraform_deployment(
    request: TerraformDeploymentRequest,
    terraform_subsystem: TerraformTool,
    deploy_subsystem: DeploySubsystem,
    keep_sandboxes: KeepSandboxes,
) -> InteractiveProcess:
    deployment = await prepare_terraform_invocation(
        TerraformInitRequest(
            request.field_set.root_module,
            request.field_set.dependencies,
            initialise_backend=True,
        )
    )

    terraform_command = "plan" if deploy_subsystem.dry_run else "apply"
    args = [terraform_command]

    invocation_files = await get_terraform_backend_and_vars(
        TerraformDeploymentInvocationFilesRequest(
            request.field_set.dependencies.address, request.field_set.dependencies
        )
    )
    var_files = await determine_source_files(
        SourceFilesRequest(e.get(SourcesField) for e in invocation_files.vars_files)
    )
    for var_file in var_files.files:
        args.append(terraform_arg("-var-file", terraform_relpath(deployment.chdir, var_file)))

    with_vars = await merge_digests(
        MergeDigests(
            [
                var_files.snapshot.digest,
                deployment.terraform_sources.snapshot.digest,
                deployment.dependencies_files.snapshot.digest,
            ]
        )
    )

    if terraform_subsystem.args:
        args.extend(terraform_subsystem.args)

    process = await setup_terraform_process(
        TerraformProcess(
            cmds=(
                deployment.init_cmd.to_args(),
                TerraformCommand(tuple(args)),
            ),
            input_digest=with_vars,
            description=f"Terraform {terraform_command}",
            chdir=deployment.chdir,
        ),
        **implicitly(),
    )
    return InteractiveProcess.from_process(process, keep_sandboxes=keep_sandboxes)


@rule(desc="Run Terraform deploy process", level=LogLevel.DEBUG)
async def run_terraform_deploy(field_set: DeployTerraformFieldSet) -> DeployProcess:
    interactive_process = await prepare_terraform_deployment(
        TerraformDeploymentRequest(field_set=field_set), **implicitly()
    )

    return DeployProcess(
        name=field_set.address.spec,
        process=interactive_process,
    )


def rules():
    return [*collect_rules(), UnionRule(DeployFieldSet, DeployTerraformFieldSet)]

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	×
4
5	import logging	×
6	from dataclasses import dataclass	×
7
8	from pants.backend.terraform.dependencies import TerraformInitRequest, prepare_terraform_invocation	×
9	from pants.backend.terraform.dependency_inference import (	×
10	TerraformDeploymentInvocationFilesRequest,
11	get_terraform_backend_and_vars,
12	)
13	from pants.backend.terraform.target_types import TerraformDeploymentFieldSet	×
14	from pants.backend.terraform.tool import (	×
15	TerraformCommand,
16	TerraformProcess,
17	TerraformTool,
18	setup_terraform_process,
19	)
20	from pants.backend.terraform.utils import terraform_arg, terraform_relpath	×
21	from pants.core.goals.deploy import DeployFieldSet, DeployProcess, DeploySubsystem	×
22	from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files	×
23	from pants.engine.engine_aware import EngineAwareParameter	×
24	from pants.engine.internals.native_engine import MergeDigests	×
25	from pants.engine.intrinsics import merge_digests	×
26	from pants.engine.process import InteractiveProcess	×
27	from pants.engine.rules import collect_rules, implicitly, rule	×
28	from pants.engine.target import SourcesField	×
29	from pants.engine.unions import UnionRule	×
30	from pants.option.global_options import KeepSandboxes	×
31	from pants.util.logging import LogLevel	×
32
33	logger = logging.getLogger(__name__)	×
34
35
36	@dataclass(frozen=True)	×
37	class DeployTerraformFieldSet(TerraformDeploymentFieldSet, DeployFieldSet):	×
38	pass	×
39
40
41	@dataclass(frozen=True)	×
42	class TerraformDeploymentRequest(EngineAwareParameter):	×
43	field_set: TerraformDeploymentFieldSet	×
44
45
46	@rule	×
47	async def prepare_terraform_deployment(	×
48	request: TerraformDeploymentRequest,
49	terraform_subsystem: TerraformTool,
50	deploy_subsystem: DeploySubsystem,
51	keep_sandboxes: KeepSandboxes,
52	) -> InteractiveProcess:
53	deployment = await prepare_terraform_invocation(	×
54	TerraformInitRequest(
55	request.field_set.root_module,
56	request.field_set.dependencies,
57	initialise_backend=True,
58	)
59	)
60
61	terraform_command = "plan" if deploy_subsystem.dry_run else "apply"	×
62	args = [terraform_command]	×
63
64	invocation_files = await get_terraform_backend_and_vars(	×
65	TerraformDeploymentInvocationFilesRequest(
66	request.field_set.dependencies.address, request.field_set.dependencies
67	)
68	)
69	var_files = await determine_source_files(	×
70	SourceFilesRequest(e.get(SourcesField) for e in invocation_files.vars_files)
71	)
72	for var_file in var_files.files:	×
73	args.append(terraform_arg("-var-file", terraform_relpath(deployment.chdir, var_file)))	×
74
75	with_vars = await merge_digests(	×
76	MergeDigests(
77	[
78	var_files.snapshot.digest,
79	deployment.terraform_sources.snapshot.digest,
80	deployment.dependencies_files.snapshot.digest,
81	]
82	)
83	)
84
85	if terraform_subsystem.args:	×
86	args.extend(terraform_subsystem.args)	×
87
88	process = await setup_terraform_process(	×
89	TerraformProcess(
90	cmds=(
91	deployment.init_cmd.to_args(),
92	TerraformCommand(tuple(args)),
93	),
94	input_digest=with_vars,
95	description=f"Terraform {terraform_command}",
96	chdir=deployment.chdir,
97	),
98	**implicitly(),
99	)
100	return InteractiveProcess.from_process(process, keep_sandboxes=keep_sandboxes)	×
101
102
103	@rule(desc="Run Terraform deploy process", level=LogLevel.DEBUG)	×
104	async def run_terraform_deploy(field_set: DeployTerraformFieldSet) -> DeployProcess:	×
105	interactive_process = await prepare_terraform_deployment(	×
106	TerraformDeploymentRequest(field_set=field_set), **implicitly()
107	)
108
109	return DeployProcess(	×
110	name=field_set.address.spec,
111	process=interactive_process,
112	)
113
114
115	def rules():	×
116	return [*collect_rules(), UnionRule(DeployFieldSet, DeployTerraformFieldSet)]	×

pantsbuild / pants / 18252174847

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