22740642519

Committed 05 Mar 2026 11:00PM UTC coverage: 52.677% (-40.3%) from 92.931%

Build # 22740642519

Build Type

Pull #23157

github

Committed by

web-flow

Commit Message

Merge 2aa18e6d4 into f0030f5e7

Pull Request Pull Request #23157: [pants ng] Partition source files by config.

Run Details

31678 of 60136 relevant lines covered (52.68%)

0.53 hits per line

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

64.38

/src/python/pants/backend/helm/util_rules/post_renderer.py

# Copyright 2022 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 itertools import chain
from typing import Any

from pants.backend.docker.goals.package_image import DockerPackageFieldSet
from pants.backend.docker.subsystems import dockerfile_parser
from pants.backend.docker.subsystems.docker_options import DockerOptions
from pants.backend.docker.target_types import (
    DockerImageTags,
    DockerImageTagsRequest,
    get_docker_image_tags,
)
from pants.backend.docker.util_rules import (
    docker_binary,
    docker_build_args,
    docker_build_context,
    docker_build_env,
    dockerfile,
)
from pants.backend.docker.util_rules.docker_build_context import (
    DockerBuildContext,
    DockerBuildContextRequest,
    create_docker_build_context,
)
from pants.backend.helm.dependency_inference.deployment import (
    FirstPartyHelmDeploymentMappingRequest,
    first_party_helm_deployment_mapping,
)
from pants.backend.helm.subsystems import post_renderer
from pants.backend.helm.subsystems.post_renderer import SetupHelmPostRenderer
from pants.backend.helm.target_types import HelmDeploymentFieldSet
from pants.engine.addresses import Address, Addresses
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.internals.graph import resolve_target, resolve_targets
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import WrappedTargetRequest
from pants.engine.unions import UnionMembership
from pants.util.logging import LogLevel
from pants.util.strutil import bullet_list, softwrap

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class HelmDeploymentPostRendererRequest(EngineAwareParameter):
    field_set: HelmDeploymentFieldSet

    def debug_hint(self) -> str | None:
        return self.field_set.address.spec

    def metadata(self) -> dict[str, Any] | None:
        return {"address": self.field_set.address.spec}


async def _obtain_custom_image_tags(
    address: Address, union_membership: UnionMembership
) -> DockerImageTags:
    wrapped_target = await resolve_target(
        WrappedTargetRequest(address, description_of_origin="<infallible>"), **implicitly()
    )

    image_tags_requests = union_membership.get(DockerImageTagsRequest)
    found_image_tags = await concurrently(
        get_docker_image_tags(
            **implicitly({image_tags_request_cls(wrapped_target.target): DockerImageTagsRequest})
        )
        for image_tags_request_cls in image_tags_requests
        if image_tags_request_cls.is_applicable(wrapped_target.target)
    )

    return DockerImageTags(chain.from_iterable(found_image_tags))


@rule(desc="Prepare Helm deployment post-renderer", level=LogLevel.DEBUG)
async def prepare_post_renderer_for_helm_deployment(
    request: HelmDeploymentPostRendererRequest,
    union_membership: UnionMembership,
    docker_options: DockerOptions,
) -> SetupHelmPostRenderer:
    mapping = await first_party_helm_deployment_mapping(
        FirstPartyHelmDeploymentMappingRequest(request.field_set), **implicitly()
    )

    docker_addresses = [addr for _, addr in mapping.indexed_docker_addresses.values()]

    logger.debug(
        softwrap(
            f"""
            Resolving Docker image references for targets:

            {bullet_list([addr.spec for addr in docker_addresses])}
            """
        )
    )
    docker_contexts = await concurrently(
        create_docker_build_context(
            DockerBuildContextRequest(
                address=addr,
                build_upstream_images=False,
            ),
            **implicitly(),
        )
        for addr in docker_addresses
    )

    docker_targets = await resolve_targets(**implicitly(Addresses(docker_addresses)))
    field_sets = [DockerPackageFieldSet.create(tgt) for tgt in docker_targets]

    async def resolve_docker_image_ref(address: Address, context: DockerBuildContext) -> str | None:
        docker_field_sets = [fs for fs in field_sets if fs.address == address]
        if not docker_field_sets:
            return None

        additional_image_tags = await _obtain_custom_image_tags(address, union_membership)

        docker_field_set = docker_field_sets[0]
        image_refs = docker_field_set.image_refs(
            default_repository=docker_options.default_repository,
            registries=docker_options.registries(),
            interpolation_context=context.interpolation_context,
            additional_tags=tuple(additional_image_tags),
        )

        # Choose first non-latest image reference found, or fallback to 'latest'.
        found_ref: str | None = None
        fallback_ref: str | None = None
        for registry in image_refs:
            for tag in registry.tags:
                ref = tag.full_name
                if ref.endswith(":latest"):
                    fallback_ref = ref
                else:
                    found_ref = ref
                    break

        resolved_ref = found_ref or fallback_ref
        if resolved_ref:
            logger.debug(f"Resolved Docker image ref '{resolved_ref}' for address {address}.")
        else:
            logger.warning(f"Could not resolve a valid image ref for Docker target {address}.")

        return resolved_ref

    docker_addr_ref_mapping = {
        addr: await resolve_docker_image_ref(addr, ctx)
        for addr, ctx in zip(docker_addresses, docker_contexts)
    }

    def find_replacement(value: tuple[str, Address]) -> str | None:
        _, addr = value
        return docker_addr_ref_mapping.get(addr)

    replacements = mapping.indexed_docker_addresses.transform_values(find_replacement)

    return SetupHelmPostRenderer(
        replacements, description_of_origin=f"the `helm_deployment` {request.field_set.address}"
    )


def rules():
    return [
        *collect_rules(),
        *docker_binary.rules(),
        *docker_build_args.rules(),
        *docker_build_context.rules(),
        *docker_build_env.rules(),
        *dockerfile.rules(),
        *dockerfile_parser.rules(),
        *post_renderer.rules(),
    ]

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	1✔
5
6	import logging	1✔
7	from dataclasses import dataclass	1✔
8	from itertools import chain	1✔
9	from typing import Any	1✔
10
11	from pants.backend.docker.goals.package_image import DockerPackageFieldSet	1✔
12	from pants.backend.docker.subsystems import dockerfile_parser	1✔
13	from pants.backend.docker.subsystems.docker_options import DockerOptions	1✔
14	from pants.backend.docker.target_types import (	1✔
15	DockerImageTags,
16	DockerImageTagsRequest,
17	get_docker_image_tags,
18	)
19	from pants.backend.docker.util_rules import (	1✔
20	docker_binary,
21	docker_build_args,
22	docker_build_context,
23	docker_build_env,
24	dockerfile,
25	)
26	from pants.backend.docker.util_rules.docker_build_context import (	1✔
27	DockerBuildContext,
28	DockerBuildContextRequest,
29	create_docker_build_context,
30	)
31	from pants.backend.helm.dependency_inference.deployment import (	1✔
32	FirstPartyHelmDeploymentMappingRequest,
33	first_party_helm_deployment_mapping,
34	)
35	from pants.backend.helm.subsystems import post_renderer	1✔
36	from pants.backend.helm.subsystems.post_renderer import SetupHelmPostRenderer	1✔
37	from pants.backend.helm.target_types import HelmDeploymentFieldSet	1✔
38	from pants.engine.addresses import Address, Addresses	1✔
39	from pants.engine.engine_aware import EngineAwareParameter	1✔
40	from pants.engine.internals.graph import resolve_target, resolve_targets	1✔
41	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	1✔
42	from pants.engine.target import WrappedTargetRequest	1✔
43	from pants.engine.unions import UnionMembership	1✔
44	from pants.util.logging import LogLevel	1✔
45	from pants.util.strutil import bullet_list, softwrap	1✔
46
47	logger = logging.getLogger(__name__)	1✔
48
49
50	@dataclass(frozen=True)	1✔
51	class HelmDeploymentPostRendererRequest(EngineAwareParameter):	1✔
52	field_set: HelmDeploymentFieldSet	1✔
53
54	def debug_hint(self) -> str \| None:	1✔
55	return self.field_set.address.spec	1✔
56
57	def metadata(self) -> dict[str, Any] \| None:	1✔
58	return {"address": self.field_set.address.spec}	1✔
59
60
61	async def _obtain_custom_image_tags(	1✔
62	address: Address, union_membership: UnionMembership
63	) -> DockerImageTags:
64	wrapped_target = await resolve_target(	×
65	WrappedTargetRequest(address, description_of_origin="<infallible>"), **implicitly()
66	)
67
68	image_tags_requests = union_membership.get(DockerImageTagsRequest)	×
69	found_image_tags = await concurrently(	×
70	get_docker_image_tags(
71	**implicitly({image_tags_request_cls(wrapped_target.target): DockerImageTagsRequest})
72	)
73	for image_tags_request_cls in image_tags_requests
74	if image_tags_request_cls.is_applicable(wrapped_target.target)
75	)
76
77	return DockerImageTags(chain.from_iterable(found_image_tags))	×
78
79
80	@rule(desc="Prepare Helm deployment post-renderer", level=LogLevel.DEBUG)	1✔
81	async def prepare_post_renderer_for_helm_deployment(	1✔
82	request: HelmDeploymentPostRendererRequest,
83	union_membership: UnionMembership,
84	docker_options: DockerOptions,
85	) -> SetupHelmPostRenderer:
86	mapping = await first_party_helm_deployment_mapping(	1✔
87	FirstPartyHelmDeploymentMappingRequest(request.field_set), **implicitly()
88	)
89
90	docker_addresses = [addr for _, addr in mapping.indexed_docker_addresses.values()]	1✔
91
92	logger.debug(	1✔
93	softwrap(
94	f"""
95	Resolving Docker image references for targets:
96
97	{bullet_list([addr.spec for addr in docker_addresses])}
98	"""
99	)
100	)
101	docker_contexts = await concurrently(	1✔
102	create_docker_build_context(
103	DockerBuildContextRequest(
104	address=addr,
105	build_upstream_images=False,
106	),
107	**implicitly(),
108	)
109	for addr in docker_addresses
110	)
111
112	docker_targets = await resolve_targets(**implicitly(Addresses(docker_addresses)))	1✔
113	field_sets = [DockerPackageFieldSet.create(tgt) for tgt in docker_targets]	1✔
114
115	async def resolve_docker_image_ref(address: Address, context: DockerBuildContext) -> str \| None:	1✔
116	docker_field_sets = [fs for fs in field_sets if fs.address == address]	×
117	if not docker_field_sets:	×
118	return None	×
119
120	additional_image_tags = await _obtain_custom_image_tags(address, union_membership)	×
121
122	docker_field_set = docker_field_sets[0]	×
123	image_refs = docker_field_set.image_refs(	×
124	default_repository=docker_options.default_repository,
125	registries=docker_options.registries(),
126	interpolation_context=context.interpolation_context,
127	additional_tags=tuple(additional_image_tags),
128	)
129
130	# Choose first non-latest image reference found, or fallback to 'latest'.
131	found_ref: str \| None = None	×
132	fallback_ref: str \| None = None	×
133	for registry in image_refs:	×
134	for tag in registry.tags:	×
135	ref = tag.full_name	×
136	if ref.endswith(":latest"):	×
137	fallback_ref = ref	×
138	else:
139	found_ref = ref	×
140	break	×
141
142	resolved_ref = found_ref or fallback_ref	×
143	if resolved_ref:	×
144	logger.debug(f"Resolved Docker image ref '{resolved_ref}' for address {address}.")	×
145	else:
146	logger.warning(f"Could not resolve a valid image ref for Docker target {address}.")	×
147
148	return resolved_ref	×
149
150	docker_addr_ref_mapping = {	1✔
151	addr: await resolve_docker_image_ref(addr, ctx)
152	for addr, ctx in zip(docker_addresses, docker_contexts)
153	}
154
155	def find_replacement(value: tuple[str, Address]) -> str \| None:	1✔
156	_, addr = value	×
157	return docker_addr_ref_mapping.get(addr)	×
158
159	replacements = mapping.indexed_docker_addresses.transform_values(find_replacement)	1✔
160
161	return SetupHelmPostRenderer(	1✔
162	replacements, description_of_origin=f"the `helm_deployment` {request.field_set.address}"
163	)
164
165
166	def rules():	1✔
167	return [	1✔
168	*collect_rules(),
169	*docker_binary.rules(),
170	*docker_build_args.rules(),
171	*docker_build_context.rules(),
172	*docker_build_env.rules(),
173	*dockerfile.rules(),
174	*dockerfile_parser.rules(),
175	*post_renderer.rules(),
176	]

pantsbuild / pants / 22740642519

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