20632486505

Committed 01 Jan 2026 04:21AM UTC coverage: 43.231% (-37.1%) from 80.281%

Build # 20632486505

Build Type

Pull #22962

github

Committed by

web-flow

Commit Message

Merge 08d5c63b0 into f52ab6675

Pull Request Pull Request #22962: Bump the gha-deps group across 1 directory with 6 updates

Run Details

26122 of 60424 relevant lines covered (43.23%)

0.86 hits per line

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

0.0

/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	×
4
5	import logging	×
6	from dataclasses import dataclass	×
7
8	from pants.backend.docker.goals.package_image import DockerPackageFieldSet	×
9	from pants.backend.k8s.k8s_subsystem import K8sSubsystem	×
10	from pants.backend.k8s.kubectl_subsystem import Kubectl	×
11	from pants.backend.k8s.target_types import (	×
12	K8sBundleContextField,
13	K8sBundleDependenciesField,
14	K8sBundleSourcesField,
15	K8sSourceField,
16	)
17	from pants.core.goals.deploy import DeployFieldSet, DeployProcess	×
18	from pants.core.util_rules.env_vars import environment_vars_subset	×
19	from pants.core.util_rules.external_tool import download_external_tool	×
20	from pants.engine.env_vars import EnvironmentVarsRequest	×
21	from pants.engine.fs import MergeDigests	×
22	from pants.engine.internals.graph import hydrate_sources, resolve_targets	×
23	from pants.engine.internals.native_engine import Digest	×
24	from pants.engine.intrinsics import digest_to_snapshot	×
25	from pants.engine.platform import Platform	×
26	from pants.engine.process import InteractiveProcess, Process, ProcessCacheScope	×
27	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
28	from pants.engine.target import DependenciesRequest, HydrateSourcesRequest, SourcesField	×
29	from pants.engine.unions import UnionRule	×
30	from pants.util.frozendict import FrozenDict	×
31	from pants.util.logging import LogLevel	×
32
33	logger = logging.getLogger(__name__)	×
34
35
36	@dataclass(frozen=True)	×
37	class DeployK8sBundleFieldSet(DeployFieldSet):	×
38	required_fields = (	×
39	K8sBundleSourcesField,
40	K8sBundleContextField,
41	K8sBundleDependenciesField,
42	)
43	sources: K8sBundleSourcesField	×
44	context: K8sBundleContextField	×
45	dependencies: K8sBundleDependenciesField	×
46
47
48	@dataclass(frozen=True)	×
49	class KubectlApply:	×
50	paths: tuple[str, ...]	×
51	input_digest: Digest	×
52	platform: Platform	×
53	env: FrozenDict[str, str] \| None = None	×
54	context: str \| None = None	×
55
56
57	@rule	×
58	async def kubectl_apply_process(	×
59	request: KubectlApply, platform: Platform, kubectl: Kubectl
60	) -> Process:
61	tool_relpath = "__kubectl"	×
62	argv: tuple[str, ...] = (f"{tool_relpath}/kubectl",)	×
63
64	if request.context is not None:	×
65	argv += ("--context", request.context)	×
66
67	argv += ("apply", "-o", "yaml")	×
68
69	for path in request.paths:	×
70	argv += ("-f", path)	×
71
72	kubectl_tool = await download_external_tool(kubectl.get_request(platform))	×
73
74	immutable_input_digests = {	×
75	tool_relpath: kubectl_tool.digest,
76	}
77
78	return Process(	×
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)	×
89	async def run_k8s_deploy(	×
90	field_set: DeployK8sBundleFieldSet,
91	kubectl: Kubectl,
92	k8s_subsystem: K8sSubsystem,
93	platform: Platform,
94	) -> DeployProcess:
95	context = field_set.context.value	×
96	if context is None:	×
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	×
102	if context is not None and context not in k8s_subsystem.available_contexts:	×
103	raise ValueError(	×
104	f"Context `{context}` is not listed in `[{K8sSubsystem.options_scope}].available_contexts`"
105	)
106
107	dependencies = await resolve_targets(	×
108	**implicitly(field_set.sources.to_unparsed_address_inputs())
109	)
110	file_sources = await concurrently(	×
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(	×
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 = [	×
129	tgt for tgt in target_dependencies if DockerPackageFieldSet.is_applicable(tgt)
130	]
131
132	env = await environment_vars_subset(	×
133	**implicitly(
134	{EnvironmentVarsRequest(requested=kubectl.extra_env_vars): EnvironmentVarsRequest}
135	)
136	)
137
138	process = InteractiveProcess.from_process(	×
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	×
152
153	return DeployProcess(	×
154	name=field_set.address.spec,
155	publish_dependencies=tuple(publish_targets),
156	process=process,
157	description=description,
158	)
159
160
161	def rules():	×
162	return [	×
163	*collect_rules(),
164	UnionRule(DeployFieldSet, DeployK8sBundleFieldSet),
165	]

pantsbuild / pants / 20632486505

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