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

90.7

/src/python/pants/backend/helm/dependency_inference/chart.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
import os
from collections.abc import Iterable
from dataclasses import dataclass

from pants.backend.helm.resolve import artifacts
from pants.backend.helm.resolve.artifacts import ThirdPartyHelmArtifactMapping
from pants.backend.helm.resolve.remotes import HelmRemotes
from pants.backend.helm.subsystems.helm import HelmSubsystem
from pants.backend.helm.target_types import (
    AllHelmChartTargets,
    HelmChartDependenciesField,
    HelmChartMetaSourceField,
    HelmChartTarget,
)
from pants.backend.helm.target_types import rules as helm_target_types_rules
from pants.backend.helm.util_rules import chart_metadata
from pants.backend.helm.util_rules.chart_metadata import (
    HelmChartDependency,
    parse_chart_metadata_from_field,
)
from pants.engine.addresses import Address
from pants.engine.internals.graph import determine_explicitly_provided_dependencies
from pants.engine.internals.selectors import concurrently
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import (
    DependenciesRequest,
    FieldSet,
    InferDependenciesRequest,
    InferredDependencies,
)
from pants.engine.unions import UnionRule
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.ordered_set import OrderedSet
from pants.util.strutil import bullet_list, pluralize

logger = logging.getLogger(__name__)


class DuplicateHelmChartNamesFound(Exception):
    def __init__(self, duplicates: Iterable[tuple[str, Address]]) -> None:
        super().__init__(
            f"Found more than one `{HelmChartTarget.alias}` target using the same chart name:\n\n"
            f"{bullet_list([f'{addr} -> {name}' for name, addr in duplicates])}"
        )


class UnknownHelmChartDependency(Exception):
    def __init__(self, address: Address, dependency: HelmChartDependency) -> None:
        super().__init__(
            f"Can not find any declared artifact for dependency '{dependency.name}' "
            f"declared at `Chart.yaml` in Helm chart at address: {address}"
        )


class FirstPartyHelmChartMapping(FrozenDict[str, Address]):
    pass


@rule
async def first_party_helm_chart_mapping(
    all_helm_chart_tgts: AllHelmChartTargets,
) -> FirstPartyHelmChartMapping:
    charts_metadata = await concurrently(
        parse_chart_metadata_from_field(tgt[HelmChartMetaSourceField])
        for tgt in all_helm_chart_tgts
    )

    name_addr_mapping: dict[str, Address] = {}
    duplicate_chart_names: OrderedSet[tuple[str, Address]] = OrderedSet()

    for meta, tgt in zip(charts_metadata, all_helm_chart_tgts):
        if meta.name in name_addr_mapping:
            duplicate_chart_names.add((meta.name, name_addr_mapping[meta.name]))
            duplicate_chart_names.add((meta.name, tgt.address))
            continue

        name_addr_mapping[meta.name] = tgt.address

    if duplicate_chart_names:
        raise DuplicateHelmChartNamesFound(duplicate_chart_names)

    return FirstPartyHelmChartMapping(name_addr_mapping)


@dataclass(frozen=True)
class HelmChartDependenciesInferenceFieldSet(FieldSet):
    required_fields = (HelmChartMetaSourceField, HelmChartDependenciesField)

    source: HelmChartMetaSourceField
    dependencies: HelmChartDependenciesField


class InferHelmChartDependenciesRequest(InferDependenciesRequest):
    infer_from = HelmChartDependenciesInferenceFieldSet


def resolve_dependency_url(remotes: HelmRemotes, dependency: HelmChartDependency) -> str | None:
    if not dependency.repository:
        registry = remotes.default_registry
        if registry:
            return os.path.join(registry.address, dependency.name)
        return None
    else:
        return os.path.join(dependency.repository, dependency.name)


@rule(desc="Inferring Helm chart dependencies", level=LogLevel.DEBUG)
async def infer_chart_dependencies_via_metadata(
    request: InferHelmChartDependenciesRequest,
    first_party_mapping: FirstPartyHelmChartMapping,
    third_party_mapping: ThirdPartyHelmArtifactMapping,
    subsystem: HelmSubsystem,
) -> InferredDependencies:
    address = request.field_set.address

    # Parse Chart.yaml for explicitly set dependencies.
    explicitly_provided_deps, metadata = await concurrently(
        determine_explicitly_provided_dependencies(
            **implicitly(DependenciesRequest(request.field_set.dependencies))
        ),
        parse_chart_metadata_from_field(request.field_set.source),
    )

    remotes = subsystem.remotes()

    # Associate dependencies in Chart.yaml with addresses.
    dependencies: OrderedSet[Address] = OrderedSet()
    for chart_dep in metadata.dependencies:
        candidate_addrs = []

        first_party_dep = first_party_mapping.get(chart_dep.name)
        if first_party_dep:
            candidate_addrs.append(first_party_dep)

        dependency_url = resolve_dependency_url(remotes, chart_dep)
        third_party_dep = third_party_mapping.get(dependency_url) if dependency_url else None
        if third_party_dep:
            candidate_addrs.append(third_party_dep)

        if not candidate_addrs:
            raise UnknownHelmChartDependency(address, chart_dep)

        matches = frozenset(candidate_addrs).difference(explicitly_provided_deps.includes)

        explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(
            matches,
            address,
            context=f"The Helm chart {address} declares `{chart_dep.name}` as dependency",
            import_reference="helm dependency",
        )

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

    logger.debug(
        f"Inferred {pluralize(len(dependencies), 'dependency')} for target at address: {address}"
    )
    return InferredDependencies(dependencies)


def rules():
    return [
        *collect_rules(),
        *artifacts.rules(),
        *helm_target_types_rules(),
        *chart_metadata.rules(),
        UnionRule(InferDependenciesRequest, InferHelmChartDependenciesRequest),
    ]

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	8✔
5
6	import logging	8✔
7	import os	8✔
8	from collections.abc import Iterable	8✔
9	from dataclasses import dataclass	8✔
10
11	from pants.backend.helm.resolve import artifacts	8✔
12	from pants.backend.helm.resolve.artifacts import ThirdPartyHelmArtifactMapping	8✔
13	from pants.backend.helm.resolve.remotes import HelmRemotes	8✔
14	from pants.backend.helm.subsystems.helm import HelmSubsystem	8✔
15	from pants.backend.helm.target_types import (	8✔
16	AllHelmChartTargets,
17	HelmChartDependenciesField,
18	HelmChartMetaSourceField,
19	HelmChartTarget,
20	)
21	from pants.backend.helm.target_types import rules as helm_target_types_rules	8✔
22	from pants.backend.helm.util_rules import chart_metadata	8✔
23	from pants.backend.helm.util_rules.chart_metadata import (	8✔
24	HelmChartDependency,
25	parse_chart_metadata_from_field,
26	)
27	from pants.engine.addresses import Address	8✔
28	from pants.engine.internals.graph import determine_explicitly_provided_dependencies	8✔
29	from pants.engine.internals.selectors import concurrently	8✔
30	from pants.engine.rules import collect_rules, implicitly, rule	8✔
31	from pants.engine.target import (	8✔
32	DependenciesRequest,
33	FieldSet,
34	InferDependenciesRequest,
35	InferredDependencies,
36	)
37	from pants.engine.unions import UnionRule	8✔
38	from pants.util.frozendict import FrozenDict	8✔
39	from pants.util.logging import LogLevel	8✔
40	from pants.util.ordered_set import OrderedSet	8✔
41	from pants.util.strutil import bullet_list, pluralize	8✔
42
43	logger = logging.getLogger(__name__)	8✔
44
45
46	class DuplicateHelmChartNamesFound(Exception):	8✔
47	def __init__(self, duplicates: Iterable[tuple[str, Address]]) -> None:	8✔
48	super().__init__(	×
49	f"Found more than one `{HelmChartTarget.alias}` target using the same chart name:\n\n"
50	f"{bullet_list([f'{addr} -> {name}' for name, addr in duplicates])}"
51	)
52
53
54	class UnknownHelmChartDependency(Exception):	8✔
55	def __init__(self, address: Address, dependency: HelmChartDependency) -> None:	8✔
56	super().__init__(	×
57	f"Can not find any declared artifact for dependency '{dependency.name}' "
58	f"declared at `Chart.yaml` in Helm chart at address: {address}"
59	)
60
61
62	class FirstPartyHelmChartMapping(FrozenDict[str, Address]):	8✔
63	pass	8✔
64
65
66	@rule	8✔
67	async def first_party_helm_chart_mapping(	8✔
68	all_helm_chart_tgts: AllHelmChartTargets,
69	) -> FirstPartyHelmChartMapping:
70	charts_metadata = await concurrently(	7✔
71	parse_chart_metadata_from_field(tgt[HelmChartMetaSourceField])
72	for tgt in all_helm_chart_tgts
73	)
74
75	name_addr_mapping: dict[str, Address] = {}	7✔
76	duplicate_chart_names: OrderedSet[tuple[str, Address]] = OrderedSet()	7✔
77
78	for meta, tgt in zip(charts_metadata, all_helm_chart_tgts):	7✔
79	if meta.name in name_addr_mapping:	7✔
80	duplicate_chart_names.add((meta.name, name_addr_mapping[meta.name]))	×
81	duplicate_chart_names.add((meta.name, tgt.address))	×
82	continue	×
83
84	name_addr_mapping[meta.name] = tgt.address	7✔
85
86	if duplicate_chart_names:	7✔
87	raise DuplicateHelmChartNamesFound(duplicate_chart_names)	×
88
89	return FirstPartyHelmChartMapping(name_addr_mapping)	7✔
90
91
92	@dataclass(frozen=True)	8✔
93	class HelmChartDependenciesInferenceFieldSet(FieldSet):	8✔
94	required_fields = (HelmChartMetaSourceField, HelmChartDependenciesField)	8✔
95
96	source: HelmChartMetaSourceField	8✔
97	dependencies: HelmChartDependenciesField	8✔
98
99
100	class InferHelmChartDependenciesRequest(InferDependenciesRequest):	8✔
101	infer_from = HelmChartDependenciesInferenceFieldSet	8✔
102
103
104	def resolve_dependency_url(remotes: HelmRemotes, dependency: HelmChartDependency) -> str \| None:	8✔
105	if not dependency.repository:	1✔
106	registry = remotes.default_registry	1✔
107	if registry:	1✔
108	return os.path.join(registry.address, dependency.name)	×
109	return None	1✔
110	else:
111	return os.path.join(dependency.repository, dependency.name)	1✔
112
113
114	@rule(desc="Inferring Helm chart dependencies", level=LogLevel.DEBUG)	8✔
115	async def infer_chart_dependencies_via_metadata(	8✔
116	request: InferHelmChartDependenciesRequest,
117	first_party_mapping: FirstPartyHelmChartMapping,
118	third_party_mapping: ThirdPartyHelmArtifactMapping,
119	subsystem: HelmSubsystem,
120	) -> InferredDependencies:
121	address = request.field_set.address	7✔
122
123	# Parse Chart.yaml for explicitly set dependencies.
124	explicitly_provided_deps, metadata = await concurrently(	7✔
125	determine_explicitly_provided_dependencies(
126	**implicitly(DependenciesRequest(request.field_set.dependencies))
127	),
128	parse_chart_metadata_from_field(request.field_set.source),
129	)
130
131	remotes = subsystem.remotes()	7✔
132
133	# Associate dependencies in Chart.yaml with addresses.
134	dependencies: OrderedSet[Address] = OrderedSet()	7✔
135	for chart_dep in metadata.dependencies:	7✔
136	candidate_addrs = []	1✔
137
138	first_party_dep = first_party_mapping.get(chart_dep.name)	1✔
139	if first_party_dep:	1✔
140	candidate_addrs.append(first_party_dep)	1✔
141
142	dependency_url = resolve_dependency_url(remotes, chart_dep)	1✔
143	third_party_dep = third_party_mapping.get(dependency_url) if dependency_url else None	1✔
144	if third_party_dep:	1✔
145	candidate_addrs.append(third_party_dep)	1✔
146
147	if not candidate_addrs:	1✔
148	raise UnknownHelmChartDependency(address, chart_dep)	×
149
150	matches = frozenset(candidate_addrs).difference(explicitly_provided_deps.includes)	1✔
151
152	explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(	1✔
153	matches,
154	address,
155	context=f"The Helm chart {address} declares `{chart_dep.name}` as dependency",
156	import_reference="helm dependency",
157	)
158
159	maybe_disambiguated = explicitly_provided_deps.disambiguated(matches)	1✔
160	if maybe_disambiguated:	1✔
161	dependencies.add(maybe_disambiguated)	1✔
162
163	logger.debug(	7✔
164	f"Inferred {pluralize(len(dependencies), 'dependency')} for target at address: {address}"
165	)
166	return InferredDependencies(dependencies)	7✔
167
168
169	def rules():	8✔
170	return [	7✔
171	*collect_rules(),
172	*artifacts.rules(),
173	*helm_target_types_rules(),
174	*chart_metadata.rules(),
175	UnionRule(InferDependenciesRequest, InferHelmChartDependenciesRequest),
176	]

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