25441711719

Committed 06 May 2026 02:31PM UTC coverage: 92.915%. Remained the same

Build # 25441711719

Build Type

push

github

Committed by

web-flow

Commit Message

use sha pin (with comment) format for generated actions (#23312)

Per the GitHub Action best practices we recently enabled at #23249, we
should pin each action to a SHA so that the reference is actually
immutable.

This will -- I hope -- knock out a large chunk of the 421 alerts we
currently get from zizmor. The next followup would then be upgrades and
harmonizing the generated and none-generated pins.

Notice: This idea was suggested by Claude while going over pinact output
and I was surprised to see that post processing the yaml wasn't too
gross.

Coverage Stats

92206 of 99237 relevant lines covered (92.91%)

4.04 hits per line

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

98.48

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

# Copyright 2024 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.docker.goals.package_image import DockerPackageFieldSet
from pants.backend.k8s.k8s_subsystem import K8sSubsystem
from pants.backend.k8s.kubectl_subsystem import Kubectl
from pants.backend.k8s.target_types import (
    K8sBundleContextField,
    K8sBundleDependenciesField,
    K8sBundleSourcesField,
    K8sSourceField,
)
from pants.core.goals.deploy import DeployFieldSet, DeployProcess
from pants.core.util_rules.env_vars import environment_vars_subset
from pants.core.util_rules.external_tool import download_external_tool
from pants.engine.env_vars import EnvironmentVarsRequest
from pants.engine.fs import MergeDigests
from pants.engine.internals.graph import hydrate_sources, resolve_targets
from pants.engine.internals.native_engine import Digest
from pants.engine.intrinsics import digest_to_snapshot
from pants.engine.platform import Platform
from pants.engine.process import InteractiveProcess, Process, ProcessCacheScope
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import DependenciesRequest, HydrateSourcesRequest, SourcesField
from pants.engine.unions import UnionRule
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class DeployK8sBundleFieldSet(DeployFieldSet):
    required_fields = (
        K8sBundleSourcesField,
        K8sBundleContextField,
        K8sBundleDependenciesField,
    )
    sources: K8sBundleSourcesField
    context: K8sBundleContextField
    dependencies: K8sBundleDependenciesField


@dataclass(frozen=True)
class KubectlApply:
    paths: tuple[str, ...]
    input_digest: Digest
    platform: Platform
    env: FrozenDict[str, str] | None = None
    context: str | None = None


@rule
async def kubectl_apply_process(
    request: KubectlApply, platform: Platform, kubectl: Kubectl
) -> Process:
    tool_relpath = "__kubectl"
    argv: tuple[str, ...] = (f"{tool_relpath}/kubectl",)

    if request.context is not None:
        argv += ("--context", request.context)

    argv += ("apply", "-o", "yaml")

    for path in request.paths:
        argv += ("-f", path)

    kubectl_tool = await download_external_tool(kubectl.get_request(platform))

    immutable_input_digests = {
        tool_relpath: kubectl_tool.digest,
    }

    return Process(
        argv=argv,
        input_digest=request.input_digest,
        cache_scope=ProcessCacheScope.PER_SESSION,
        description=f"Applying kubernetes config {request.paths}",
        env=request.env,
        immutable_input_digests=immutable_input_digests,
    )


@rule(desc="Run k8s deploy process", level=LogLevel.DEBUG)
async def run_k8s_deploy(
    field_set: DeployK8sBundleFieldSet,
    kubectl: Kubectl,
    k8s_subsystem: K8sSubsystem,
    platform: Platform,
) -> DeployProcess:
    context = field_set.context.value
    if context is None:
        raise ValueError(
            f"Missing `{K8sBundleContextField.alias}` field on target `{field_set.address.spec}`"
        )

    context = context if kubectl.pass_context else None
    if context is not None and context not in k8s_subsystem.available_contexts:
        raise ValueError(
            f"Context `{context}` is not listed in `[{K8sSubsystem.options_scope}].available_contexts`"
        )

    dependencies = await resolve_targets(
        **implicitly(field_set.sources.to_unparsed_address_inputs())
    )
    file_sources = await concurrently(
        hydrate_sources(
            HydrateSourcesRequest(
                t.get(SourcesField),
                for_sources_types=(K8sSourceField,),
                enable_codegen=True,
            ),
            **implicitly(),
        )
        for t in dependencies
    )
    snapshot, target_dependencies = await concurrently(
        digest_to_snapshot(
            **implicitly(MergeDigests((*(sources.snapshot.digest for sources in file_sources),)))
        ),
        resolve_targets(**implicitly(DependenciesRequest(field_set.dependencies))),
    )

    publish_targets = [
        tgt for tgt in target_dependencies if DockerPackageFieldSet.is_applicable(tgt)
    ]

    env = await environment_vars_subset(
        **implicitly(
            {EnvironmentVarsRequest(requested=kubectl.extra_env_vars): EnvironmentVarsRequest}
        )
    )

    process = InteractiveProcess.from_process(
        await kubectl_apply_process(
            KubectlApply(
                snapshot.files,
                platform=platform,
                input_digest=snapshot.digest,
                env=env,
                context=context,
            ),
            **implicitly(),
        )
    )

    description = f"context {context}" if context is not None else None

    return DeployProcess(
        name=field_set.address.spec,
        publish_dependencies=tuple(publish_targets),
        process=process,
        description=description,
    )


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

1	# Copyright 2024 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 logging	1✔
6	from dataclasses import dataclass	1✔
7
8	from pants.backend.docker.goals.package_image import DockerPackageFieldSet	1✔
9	from pants.backend.k8s.k8s_subsystem import K8sSubsystem	1✔
10	from pants.backend.k8s.kubectl_subsystem import Kubectl	1✔
11	from pants.backend.k8s.target_types import (	1✔
12	K8sBundleContextField,
13	K8sBundleDependenciesField,
14	K8sBundleSourcesField,
15	K8sSourceField,
16	)
17	from pants.core.goals.deploy import DeployFieldSet, DeployProcess	1✔
18	from pants.core.util_rules.env_vars import environment_vars_subset	1✔
19	from pants.core.util_rules.external_tool import download_external_tool	1✔
20	from pants.engine.env_vars import EnvironmentVarsRequest	1✔
21	from pants.engine.fs import MergeDigests	1✔
22	from pants.engine.internals.graph import hydrate_sources, resolve_targets	1✔
23	from pants.engine.internals.native_engine import Digest	1✔
24	from pants.engine.intrinsics import digest_to_snapshot	1✔
25	from pants.engine.platform import Platform	1✔
26	from pants.engine.process import InteractiveProcess, Process, ProcessCacheScope	1✔
27	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	1✔
28	from pants.engine.target import DependenciesRequest, HydrateSourcesRequest, SourcesField	1✔
29	from pants.engine.unions import UnionRule	1✔
30	from pants.util.frozendict import FrozenDict	1✔
31	from pants.util.logging import LogLevel	1✔
32
33	logger = logging.getLogger(__name__)	1✔
34
35
36	@dataclass(frozen=True)	1✔
37	class DeployK8sBundleFieldSet(DeployFieldSet):	1✔
38	required_fields = (	1✔
39	K8sBundleSourcesField,
40	K8sBundleContextField,
41	K8sBundleDependenciesField,
42	)
43	sources: K8sBundleSourcesField	1✔
44	context: K8sBundleContextField	1✔
45	dependencies: K8sBundleDependenciesField	1✔
46
47
48	@dataclass(frozen=True)	1✔
49	class KubectlApply:	1✔
50	paths: tuple[str, ...]	1✔
51	input_digest: Digest	1✔
52	platform: Platform	1✔
53	env: FrozenDict[str, str] \| None = None	1✔
54	context: str \| None = None	1✔
55
56
57	@rule	1✔
58	async def kubectl_apply_process(	1✔
59	request: KubectlApply, platform: Platform, kubectl: Kubectl
60	) -> Process:
61	tool_relpath = "__kubectl"	1✔
62	argv: tuple[str, ...] = (f"{tool_relpath}/kubectl",)	1✔
63
64	if request.context is not None:	1✔
65	argv += ("--context", request.context)	1✔
66
67	argv += ("apply", "-o", "yaml")	1✔
68
69	for path in request.paths:	1✔
70	argv += ("-f", path)	1✔
71
72	kubectl_tool = await download_external_tool(kubectl.get_request(platform))	1✔
73
74	immutable_input_digests = {	1✔
75	tool_relpath: kubectl_tool.digest,
76	}
77
78	return Process(	1✔
79	argv=argv,
80	input_digest=request.input_digest,
81	cache_scope=ProcessCacheScope.PER_SESSION,
82	description=f"Applying kubernetes config {request.paths}",
83	env=request.env,
84	immutable_input_digests=immutable_input_digests,
85	)
86
87
88	@rule(desc="Run k8s deploy process", level=LogLevel.DEBUG)	1✔
89	async def run_k8s_deploy(	1✔
90	field_set: DeployK8sBundleFieldSet,
91	kubectl: Kubectl,
92	k8s_subsystem: K8sSubsystem,
93	platform: Platform,
94	) -> DeployProcess:
95	context = field_set.context.value	1✔
96	if context is None:	1✔
97	raise ValueError(	×
98	f"Missing `{K8sBundleContextField.alias}` field on target `{field_set.address.spec}`"
99	)
100
101	context = context if kubectl.pass_context else None	1✔
102	if context is not None and context not in k8s_subsystem.available_contexts:	1✔
103	raise ValueError(	1✔
104	f"Context `{context}` is not listed in `[{K8sSubsystem.options_scope}].available_contexts`"
105	)
106
107	dependencies = await resolve_targets(	1✔
108	**implicitly(field_set.sources.to_unparsed_address_inputs())
109	)
110	file_sources = await concurrently(	1✔
111	hydrate_sources(
112	HydrateSourcesRequest(
113	t.get(SourcesField),
114	for_sources_types=(K8sSourceField,),
115	enable_codegen=True,
116	),
117	**implicitly(),
118	)
119	for t in dependencies
120	)
121	snapshot, target_dependencies = await concurrently(	1✔
122	digest_to_snapshot(
123	*implicitly(MergeDigests(((sources.snapshot.digest for sources in file_sources),)))
124	),
125	resolve_targets(**implicitly(DependenciesRequest(field_set.dependencies))),
126	)
127
128	publish_targets = [	1✔
129	tgt for tgt in target_dependencies if DockerPackageFieldSet.is_applicable(tgt)
130	]
131
132	env = await environment_vars_subset(	1✔
133	**implicitly(
134	{EnvironmentVarsRequest(requested=kubectl.extra_env_vars): EnvironmentVarsRequest}
135	)
136	)
137
138	process = InteractiveProcess.from_process(	1✔
139	await kubectl_apply_process(
140	KubectlApply(
141	snapshot.files,
142	platform=platform,
143	input_digest=snapshot.digest,
144	env=env,
145	context=context,
146	),
147	**implicitly(),
148	)
149	)
150
151	description = f"context {context}" if context is not None else None	1✔
152
153	return DeployProcess(	1✔
154	name=field_set.address.spec,
155	publish_dependencies=tuple(publish_targets),
156	process=process,
157	description=description,
158	)
159
160
161	def rules():	1✔
162	return [	1✔
163	*collect_rules(),
164	UnionRule(DeployFieldSet, DeployK8sBundleFieldSet),
165	]

pantsbuild / pants / 25441711719

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