26342152999

Committed 23 May 2026 07:59PM UTC coverage: 91.165% (-1.6%) from 92.792%

Build # 26342152999

Build Type

push

github

Committed by

web-flow

Commit Message

Run Linux ARM CI on Depot runners (#23363)

RunsOn is deprecating their v2 stack, and rather than migrate
to v3 we should use the resources kindly donated by Depot.

GitHub also now has Linux ARM runners, should we need them.

Coverage Stats

87305 of 95766 relevant lines covered (91.16%)

3.87 hits per line

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

94.84

/src/python/pants/backend/helm/dependency_inference/deployment.py

# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import fnmatch
import logging
import re
from dataclasses import dataclass, field
from pathlib import PurePath
from typing import Any

from pants.backend.docker.target_types import AllDockerImageTargets
from pants.backend.docker.target_types import rules as docker_target_types_rules
from pants.backend.docker.utils import image_ref_regexp
from pants.backend.helm.dependency_inference.subsystem import (
    HelmInferSubsystem,
    UnownedDependencyError,
    UnownedHelmDependencyUsage,
)
from pants.backend.helm.subsystems import k8s_parser
from pants.backend.helm.subsystems.k8s_parser import ParseKubeManifestRequest, parse_kube_manifest
from pants.backend.helm.target_types import HelmDeploymentFieldSet
from pants.backend.helm.target_types import rules as helm_target_types_rules
from pants.backend.helm.util_rules import renderer
from pants.backend.helm.util_rules.renderer import (
    HelmDeploymentCmd,
    HelmDeploymentRequest,
    run_renderer,
)
from pants.backend.helm.utils.yaml import FrozenYamlIndex, MutableYamlIndex
from pants.build_graph.address import MaybeAddress
from pants.engine.addresses import Address
from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType
from pants.engine.fs import FileEntry
from pants.engine.internals.build_files import maybe_resolve_address, resolve_address
from pants.engine.internals.graph import determine_explicitly_provided_dependencies
from pants.engine.internals.native_engine import AddressInput, AddressParseException
from pants.engine.intrinsics import get_digest_entries
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import DependenciesRequest, InferDependenciesRequest, InferredDependencies
from pants.engine.unions import UnionRule
from pants.util.logging import LogLevel
from pants.util.ordered_set import FrozenOrderedSet, OrderedSet
from pants.util.strutil import pluralize, softwrap

logger = logging.getLogger(__name__)


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

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


@dataclass(frozen=True)
class HelmDeploymentReport(EngineAwareReturnType):
    address: Address
    image_refs: FrozenYamlIndex[str]

    @property
    def all_image_refs(self) -> FrozenOrderedSet[str]:
        return FrozenOrderedSet(self.image_refs.values())

    def level(self) -> LogLevel | None:
        return LogLevel.DEBUG

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


@rule(desc="Analyse Helm deployment", level=LogLevel.DEBUG)
async def analyse_deployment(request: AnalyseHelmDeploymentRequest) -> HelmDeploymentReport:
    rendered_deployment = await run_renderer(
        **implicitly(
            HelmDeploymentRequest(
                cmd=HelmDeploymentCmd.RENDER,
                field_set=request.field_set,
                description=f"Rendering Helm deployment {request.field_set.address}",
            )
        )
    )

    rendered_entries = await get_digest_entries(rendered_deployment.snapshot.digest)
    parsed_manifests = await concurrently(
        parse_kube_manifest(ParseKubeManifestRequest(file=entry), **implicitly())
        for entry in rendered_entries
        if isinstance(entry, FileEntry)
    )

    # Build YAML index of Docker image refs for future processing during dependency inference or post-rendering.
    image_refs_index: MutableYamlIndex[str] = MutableYamlIndex()
    for manifest in parsed_manifests:
        for entry in manifest.found_image_refs:
            image_refs_index.insert(
                file_path=PurePath(manifest.filename),
                document_index=entry.document_index,
                yaml_path=entry.path,
                item=entry.unparsed_image_ref,
            )

    return HelmDeploymentReport(
        address=request.field_set.address, image_refs=image_refs_index.frozen()
    )


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

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


@dataclass(frozen=True)
class FirstPartyHelmDeploymentMapping:
    """A mapping between `helm_deployment` target addresses and tuples made up of a Docker image
    reference and a `docker_image` target address.

    The tuples of Docker image references and addresses are stored in a YAML index so we can track
    the locations in which the Docker image refs appear in the deployment files.
    """

    address: Address
    indexed_docker_addresses: FrozenYamlIndex[tuple[str, Address]]


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


@rule
async def _first_party_helm_deployment_mapping(
    request: _FirstPartyHelmDeploymentMappingRequest,
    docker_targets: AllDockerImageTargets,
    helm_infer: HelmInferSubsystem,
) -> FirstPartyHelmDeploymentMapping:
    deployment_report = await analyse_deployment(AnalyseHelmDeploymentRequest(request.field_set))

    def image_ref_to_address_input(image_ref: str) -> tuple[str, AddressInput] | None:
        try:
            return image_ref, AddressInput.parse(
                image_ref,
                description_of_origin=f"the helm_deployment at {request.field_set.address}",
                relative_to=request.field_set.address.spec_path,
            )
        except AddressParseException:
            return None

    indexed_address_inputs = deployment_report.image_refs.transform_values(
        image_ref_to_address_input
    )
    maybe_addresses = await concurrently(
        maybe_resolve_address(ai) for _, ai in indexed_address_inputs.values()
    )

    docker_target_addresses = {tgt.address for tgt in docker_targets}
    maybe_addresses_by_ref = {
        ref: maybe_addr
        for ((ref, _), maybe_addr) in zip(indexed_address_inputs.values(), maybe_addresses)
    }

    resolver = ImageReferenceResolver(helm_infer, maybe_addresses_by_ref, docker_target_addresses)

    indexed_docker_addresses = indexed_address_inputs.transform_values(
        lambda image_ref_ai: resolver.image_ref_to_actual_address(image_ref_ai[0])
    )

    resolver.report_errors()
    return FirstPartyHelmDeploymentMapping(
        address=request.field_set.address,
        indexed_docker_addresses=indexed_docker_addresses,
    )


@rule
async def first_party_helm_deployment_mapping(
    request: FirstPartyHelmDeploymentMappingRequest,
    helm_infer: HelmInferSubsystem,
) -> FirstPartyHelmDeploymentMapping:
    if not helm_infer.deployment_dependencies:
        return FirstPartyHelmDeploymentMapping(
            request.field_set.address,
            FrozenYamlIndex.empty(),
        )
    # Use a small proxy rule to make sure we don't calculate AllDockerImageTargets
    # if `[helm-infer].deployment_dependencies` is set to true.
    return await _first_party_helm_deployment_mapping(
        _FirstPartyHelmDeploymentMappingRequest(field_set=request.field_set), **implicitly()
    )


@dataclass
class ImageReferenceResolver:
    """Attempt to resolve images to their references.

    Errors are stored internally and are surfaced by a call to `report_errors`, so we can report all
    problems and avoid only raising 1 per run.
    """

    helm_infer: HelmInferSubsystem
    maybe_addresses_by_ref: dict[str, MaybeAddress]
    docker_target_addresses: set[Address]

    errors: list[str] = field(default_factory=list)

    def image_ref_to_actual_address(self, image_ref: str) -> tuple[str, Address] | None:
        """Attempt to resolve an image reference to its Pants Address or the docker image."""
        maybe_addr = self.maybe_addresses_by_ref.get(image_ref)
        if not maybe_addr:
            return None
        if not isinstance(maybe_addr.val, Address):
            # obviously intended to be a Pants target
            if (
                image_ref.startswith("//")
                or image_ref.startswith("./")
                or image_ref.startswith(":")
            ):
                message = f"`{image_ref}` was supplied but the docker_image target at `{maybe_addr.val}` does not exist."
                self._handle_missing_docker_image(message)
                return None
            # explicit 3rd party
            elif self._image_ref_is_known_external(image_ref):
                return None
            else:
                message = f"""\
                `{image_ref}` was supplied, but Pants cannot determine
                whether this should be a target's address or a 3rd-party dependency.
                If this should be an external image, add `{image_ref}` to `[{HelmInferSubsystem.options_scope}].external_docker_images`
                If this should be a target address, use an absolute path instead (possibly `//{image_ref}`).
                """
                self._handle_missing_docker_image(message)
                return None

        if maybe_addr.val not in self.docker_target_addresses:
            message = f"The address `{image_ref}` was supplied, but the target at `{maybe_addr.val}` is not a docker_image target."
            self._handle_missing_docker_image(message)
            return None
        return image_ref, maybe_addr.val

    def _image_ref_is_known_external(self, image_ref: str) -> bool:
        parsed = re.match(image_ref_regexp, image_ref.strip("\"'"))
        if not parsed:
            return False
        if parsed.group("registry"):
            image_name = parsed.group("registry") + "/" + parsed.group("repository")
        else:
            image_name = parsed.group("repository")

        return any(
            (fnmatch.fnmatch(image_name, pattern) or fnmatch.fnmatch(image_ref, pattern))
            for pattern in self.helm_infer.external_docker_images
        )

    def _handle_missing_docker_image(self, message):
        self.errors.append(message)

    def report_errors(self):
        """Raise all gathered errors.

        Invoke this method after all resolving is done.
        """
        if not self.errors:
            return

        message = "\n".join(
            [
                "Error resolving Docker image dependency of a Helm chart.",
                *self.errors,
                f"The behavior for unowned imports can also be set with the `[{HelmInferSubsystem.options_scope}].unowned_dependency_behavior`",
            ]
        )
        if self.helm_infer.unowned_dependency_behavior == UnownedHelmDependencyUsage.RaiseError:
            raise UnownedDependencyError(message)
        elif self.helm_infer.unowned_dependency_behavior == UnownedHelmDependencyUsage.LogWarning:
            logging.warning(message)
        else:
            return


class InferHelmDeploymentDependenciesRequest(InferDependenciesRequest):
    infer_from = HelmDeploymentFieldSet


@rule(desc="Find the dependencies needed by a Helm deployment")
async def inject_deployment_dependencies(
    request: InferHelmDeploymentDependenciesRequest,
    infer_subsystem: HelmInferSubsystem,
) -> InferredDependencies:
    get_address = resolve_address(
        **implicitly({request.field_set.chart.to_address_input(): AddressInput})
    )
    get_explicit_deps = determine_explicitly_provided_dependencies(
        **implicitly(DependenciesRequest(request.field_set.dependencies))
    )

    chart_address, explicitly_provided_deps, mapping = await concurrently(
        get_address,
        get_explicit_deps,
        first_party_helm_deployment_mapping(
            FirstPartyHelmDeploymentMappingRequest(request.field_set), **implicitly()
        ),
    )

    dependencies: OrderedSet[Address] = OrderedSet()
    dependencies.add(chart_address)

    for imager_ref, candidate_address in mapping.indexed_docker_addresses.values():
        matches = frozenset([candidate_address]).difference(explicitly_provided_deps.includes)
        explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(
            matches,
            request.field_set.address,
            context=softwrap(
                f"""
                The Helm deployment {request.field_set.address} declares
                {imager_ref} as Docker image reference
                """
            ),
            import_reference="manifest",
        )

        maybe_disambiguated = explicitly_provided_deps.disambiguated(matches)
        if maybe_disambiguated:
            dependencies.add(maybe_disambiguated)

    logging.debug(
        f"Found {pluralize(len(dependencies), 'dependency')} for target {request.field_set.address}"
    )
    return InferredDependencies(dependencies)


def rules():
    return [
        *collect_rules(),
        *renderer.rules(),
        *k8s_parser.rules(),
        *helm_target_types_rules(),
        *docker_target_types_rules(),
        UnionRule(InferDependenciesRequest, InferHelmDeploymentDependenciesRequest),
    ]

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	5✔
5
6	import fnmatch	5✔
7	import logging	5✔
8	import re	5✔
9	from dataclasses import dataclass, field	5✔
10	from pathlib import PurePath	5✔
11	from typing import Any	5✔
12
13	from pants.backend.docker.target_types import AllDockerImageTargets	5✔
14	from pants.backend.docker.target_types import rules as docker_target_types_rules	5✔
15	from pants.backend.docker.utils import image_ref_regexp	5✔
16	from pants.backend.helm.dependency_inference.subsystem import (	5✔
17	HelmInferSubsystem,
18	UnownedDependencyError,
19	UnownedHelmDependencyUsage,
20	)
21	from pants.backend.helm.subsystems import k8s_parser	5✔
22	from pants.backend.helm.subsystems.k8s_parser import ParseKubeManifestRequest, parse_kube_manifest	5✔
23	from pants.backend.helm.target_types import HelmDeploymentFieldSet	5✔
24	from pants.backend.helm.target_types import rules as helm_target_types_rules	5✔
25	from pants.backend.helm.util_rules import renderer	5✔
26	from pants.backend.helm.util_rules.renderer import (	5✔
27	HelmDeploymentCmd,
28	HelmDeploymentRequest,
29	run_renderer,
30	)
31	from pants.backend.helm.utils.yaml import FrozenYamlIndex, MutableYamlIndex	5✔
32	from pants.build_graph.address import MaybeAddress	5✔
33	from pants.engine.addresses import Address	5✔
34	from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType	5✔
35	from pants.engine.fs import FileEntry	5✔
36	from pants.engine.internals.build_files import maybe_resolve_address, resolve_address	5✔
37	from pants.engine.internals.graph import determine_explicitly_provided_dependencies	5✔
38	from pants.engine.internals.native_engine import AddressInput, AddressParseException	5✔
39	from pants.engine.intrinsics import get_digest_entries	5✔
40	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	5✔
41	from pants.engine.target import DependenciesRequest, InferDependenciesRequest, InferredDependencies	5✔
42	from pants.engine.unions import UnionRule	5✔
43	from pants.util.logging import LogLevel	5✔
44	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	5✔
45	from pants.util.strutil import pluralize, softwrap	5✔
46
47	logger = logging.getLogger(__name__)	5✔
48
49
50	@dataclass(frozen=True)	5✔
51	class AnalyseHelmDeploymentRequest(EngineAwareParameter):	5✔
52	field_set: HelmDeploymentFieldSet	5✔
53
54	def debug_hint(self) -> str \| None:	5✔
55	return self.field_set.address.spec	1✔
56
57
58	@dataclass(frozen=True)	5✔
59	class HelmDeploymentReport(EngineAwareReturnType):	5✔
60	address: Address	5✔
61	image_refs: FrozenYamlIndex[str]	5✔
62
63	@property	5✔
64	def all_image_refs(self) -> FrozenOrderedSet[str]:	5✔
65	return FrozenOrderedSet(self.image_refs.values())	1✔
66
67	def level(self) -> LogLevel \| None:	5✔
68	return LogLevel.DEBUG	5✔
69
70	def metadata(self) -> dict[str, Any] \| None:	5✔
71	return {"address": self.address, "image_refs": self.image_refs}	5✔
72
73
74	@rule(desc="Analyse Helm deployment", level=LogLevel.DEBUG)	5✔
75	async def analyse_deployment(request: AnalyseHelmDeploymentRequest) -> HelmDeploymentReport:	5✔
76	rendered_deployment = await run_renderer(	5✔
77	**implicitly(
78	HelmDeploymentRequest(
79	cmd=HelmDeploymentCmd.RENDER,
80	field_set=request.field_set,
81	description=f"Rendering Helm deployment {request.field_set.address}",
82	)
83	)
84	)
85
86	rendered_entries = await get_digest_entries(rendered_deployment.snapshot.digest)	5✔
87	parsed_manifests = await concurrently(	5✔
88	parse_kube_manifest(ParseKubeManifestRequest(file=entry), **implicitly())
89	for entry in rendered_entries
90	if isinstance(entry, FileEntry)
91	)
92
93	# Build YAML index of Docker image refs for future processing during dependency inference or post-rendering.
94	image_refs_index: MutableYamlIndex[str] = MutableYamlIndex()	5✔
95	for manifest in parsed_manifests:	5✔
96	for entry in manifest.found_image_refs:	5✔
97	image_refs_index.insert(	3✔
98	file_path=PurePath(manifest.filename),
99	document_index=entry.document_index,
100	yaml_path=entry.path,
101	item=entry.unparsed_image_ref,
102	)
103
104	return HelmDeploymentReport(	5✔
105	address=request.field_set.address, image_refs=image_refs_index.frozen()
106	)
107
108
109	@dataclass(frozen=True)	5✔
110	class FirstPartyHelmDeploymentMappingRequest(EngineAwareParameter):	5✔
111	field_set: HelmDeploymentFieldSet	5✔
112
113	def debug_hint(self) -> str \| None:	5✔
114	return self.field_set.address.spec	×
115
116
117	@dataclass(frozen=True)	5✔
118	class FirstPartyHelmDeploymentMapping:	5✔
119	"""A mapping between `helm_deployment` target addresses and tuples made up of a Docker image
120	reference and a `docker_image` target address.
121
122	The tuples of Docker image references and addresses are stored in a YAML index so we can track
123	the locations in which the Docker image refs appear in the deployment files.
124	"""
125
126	address: Address	5✔
127	indexed_docker_addresses: FrozenYamlIndex[tuple[str, Address]]	5✔
128
129
130	@dataclass(frozen=True)	5✔
131	class _FirstPartyHelmDeploymentMappingRequest(EngineAwareParameter):	5✔
132	field_set: HelmDeploymentFieldSet	5✔
133
134
135	@rule	5✔
136	async def _first_party_helm_deployment_mapping(	5✔
137	request: _FirstPartyHelmDeploymentMappingRequest,
138	docker_targets: AllDockerImageTargets,
139	helm_infer: HelmInferSubsystem,
140	) -> FirstPartyHelmDeploymentMapping:
141	deployment_report = await analyse_deployment(AnalyseHelmDeploymentRequest(request.field_set))	5✔
142
143	def image_ref_to_address_input(image_ref: str) -> tuple[str, AddressInput] \| None:	5✔
144	try:	3✔
145	return image_ref, AddressInput.parse(	3✔
146	image_ref,
147	description_of_origin=f"the helm_deployment at {request.field_set.address}",
148	relative_to=request.field_set.address.spec_path,
149	)
150	except AddressParseException:	×
151	return None	×
152
153	indexed_address_inputs = deployment_report.image_refs.transform_values(	5✔
154	image_ref_to_address_input
155	)
156	maybe_addresses = await concurrently(	5✔
157	maybe_resolve_address(ai) for _, ai in indexed_address_inputs.values()
158	)
159
160	docker_target_addresses = {tgt.address for tgt in docker_targets}	5✔
161	maybe_addresses_by_ref = {	5✔
162	ref: maybe_addr
163	for ((ref, _), maybe_addr) in zip(indexed_address_inputs.values(), maybe_addresses)
164	}
165
166	resolver = ImageReferenceResolver(helm_infer, maybe_addresses_by_ref, docker_target_addresses)	5✔
167
168	indexed_docker_addresses = indexed_address_inputs.transform_values(	5✔
169	lambda image_ref_ai: resolver.image_ref_to_actual_address(image_ref_ai[0])
170	)
171
172	resolver.report_errors()	5✔
173	return FirstPartyHelmDeploymentMapping(	5✔
174	address=request.field_set.address,
175	indexed_docker_addresses=indexed_docker_addresses,
176	)
177
178
179	@rule	5✔
180	async def first_party_helm_deployment_mapping(	5✔
181	request: FirstPartyHelmDeploymentMappingRequest,
182	helm_infer: HelmInferSubsystem,
183	) -> FirstPartyHelmDeploymentMapping:
184	if not helm_infer.deployment_dependencies:	5✔
185	return FirstPartyHelmDeploymentMapping(	1✔
186	request.field_set.address,
187	FrozenYamlIndex.empty(),
188	)
189	# Use a small proxy rule to make sure we don't calculate AllDockerImageTargets
190	# if `[helm-infer].deployment_dependencies` is set to true.
191	return await _first_party_helm_deployment_mapping(	5✔
192	_FirstPartyHelmDeploymentMappingRequest(field_set=request.field_set), **implicitly()
193	)
194
195
196	@dataclass	5✔
197	class ImageReferenceResolver:	5✔
198	"""Attempt to resolve images to their references.
199
200	Errors are stored internally and are surfaced by a call to `report_errors`, so we can report all
201	problems and avoid only raising 1 per run.
202	"""
203
204	helm_infer: HelmInferSubsystem	5✔
205	maybe_addresses_by_ref: dict[str, MaybeAddress]	5✔
206	docker_target_addresses: set[Address]	5✔
207
208	errors: list[str] = field(default_factory=list)	5✔
209
210	def image_ref_to_actual_address(self, image_ref: str) -> tuple[str, Address] \| None:	5✔
211	"""Attempt to resolve an image reference to its Pants Address or the docker image."""
212	maybe_addr = self.maybe_addresses_by_ref.get(image_ref)	3✔
213	if not maybe_addr:	3✔
214	return None	×
215	if not isinstance(maybe_addr.val, Address):	3✔
216	# obviously intended to be a Pants target
217	if (	3✔
218	image_ref.startswith("//")
219	or image_ref.startswith("./")
220	or image_ref.startswith(":")
221	):
222	message = f"`{image_ref}` was supplied but the docker_image target at `{maybe_addr.val}` does not exist."	1✔
223	self._handle_missing_docker_image(message)	1✔
224	return None	1✔
225	# explicit 3rd party
226	elif self._image_ref_is_known_external(image_ref):	3✔
227	return None	3✔
228	else:
229	message = f"""\	1✔
230	`{image_ref}` was supplied, but Pants cannot determine
231	whether this should be a target's address or a 3rd-party dependency.
232	If this should be an external image, add `{image_ref}` to `[{HelmInferSubsystem.options_scope}].external_docker_images`
233	If this should be a target address, use an absolute path instead (possibly `//{image_ref}`).
234	"""
235	self._handle_missing_docker_image(message)	1✔
236	return None	1✔
237
238	if maybe_addr.val not in self.docker_target_addresses:	1✔
239	message = f"The address `{image_ref}` was supplied, but the target at `{maybe_addr.val}` is not a docker_image target."	1✔
240	self._handle_missing_docker_image(message)	1✔
241	return None	1✔
242	return image_ref, maybe_addr.val	1✔
243
244	def _image_ref_is_known_external(self, image_ref: str) -> bool:	5✔
245	parsed = re.match(image_ref_regexp, image_ref.strip("\"'"))	3✔
246	if not parsed:	3✔
247	return False	×
248	if parsed.group("registry"):	3✔
249	image_name = parsed.group("registry") + "/" + parsed.group("repository")	2✔
250	else:
251	image_name = parsed.group("repository")	3✔
252
253	return any(	3✔
254	(fnmatch.fnmatch(image_name, pattern) or fnmatch.fnmatch(image_ref, pattern))
255	for pattern in self.helm_infer.external_docker_images
256	)
257
258	def _handle_missing_docker_image(self, message):	5✔
259	self.errors.append(message)	1✔
260
261	def report_errors(self):	5✔
262	"""Raise all gathered errors.
263
264	Invoke this method after all resolving is done.
265	"""
266	if not self.errors:	5✔
267	return	5✔
268
269	message = "\n".join(	1✔
270	[
271	"Error resolving Docker image dependency of a Helm chart.",
272	*self.errors,
273	f"The behavior for unowned imports can also be set with the `[{HelmInferSubsystem.options_scope}].unowned_dependency_behavior`",
274	]
275	)
276	if self.helm_infer.unowned_dependency_behavior == UnownedHelmDependencyUsage.RaiseError:	1✔
277	raise UnownedDependencyError(message)	1✔
278	elif self.helm_infer.unowned_dependency_behavior == UnownedHelmDependencyUsage.LogWarning:	×
279	logging.warning(message)	×
280	else:
281	return	×
282
283
284	class InferHelmDeploymentDependenciesRequest(InferDependenciesRequest):	5✔
285	infer_from = HelmDeploymentFieldSet	5✔
286
287
288	@rule(desc="Find the dependencies needed by a Helm deployment")	5✔
289	async def inject_deployment_dependencies(	5✔
290	request: InferHelmDeploymentDependenciesRequest,
291	infer_subsystem: HelmInferSubsystem,
292	) -> InferredDependencies:
293	get_address = resolve_address(	1✔
294	**implicitly({request.field_set.chart.to_address_input(): AddressInput})
295	)
296	get_explicit_deps = determine_explicitly_provided_dependencies(	1✔
297	**implicitly(DependenciesRequest(request.field_set.dependencies))
298	)
299
300	chart_address, explicitly_provided_deps, mapping = await concurrently(	1✔
301	get_address,
302	get_explicit_deps,
303	first_party_helm_deployment_mapping(
304	FirstPartyHelmDeploymentMappingRequest(request.field_set), **implicitly()
305	),
306	)
307
308	dependencies: OrderedSet[Address] = OrderedSet()	1✔
309	dependencies.add(chart_address)	1✔
310
311	for imager_ref, candidate_address in mapping.indexed_docker_addresses.values():	1✔
312	matches = frozenset([candidate_address]).difference(explicitly_provided_deps.includes)	1✔
313	explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(	1✔
314	matches,
315	request.field_set.address,
316	context=softwrap(
317	f"""
318	The Helm deployment {request.field_set.address} declares
319	{imager_ref} as Docker image reference
320	"""
321	),
322	import_reference="manifest",
323	)
324
325	maybe_disambiguated = explicitly_provided_deps.disambiguated(matches)	1✔
326	if maybe_disambiguated:	1✔
327	dependencies.add(maybe_disambiguated)	1✔
328
329	logging.debug(	1✔
330	f"Found {pluralize(len(dependencies), 'dependency')} for target {request.field_set.address}"
331	)
332	return InferredDependencies(dependencies)	1✔
333
334
335	def rules():	5✔
336	return [	5✔
337	*collect_rules(),
338	*renderer.rules(),
339	*k8s_parser.rules(),
340	*helm_target_types_rules(),
341	*docker_target_types_rules(),
342	UnionRule(InferDependenciesRequest, InferHelmDeploymentDependenciesRequest),
343	]

pantsbuild / pants / 26342152999

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