18791134616

Committed 24 Oct 2025 08:18PM UTC coverage: 75.519% (-4.8%) from 80.282%

Build # 18791134616

Build Type

Pull #22794

github

Committed by

web-flow

Commit Message

Merge 098c595a0 into 7971a20bf

Pull Request Pull Request #22794: Use self-hosted MacOS Intel runner

Run Details

65803 of 87134 relevant lines covered (75.52%)

3.07 hits per line

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

0.0

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

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

import logging
import os
from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import PurePath

from pants.backend.helm.goals.tailor import _SNAPSHOT_FOLDER_NAME, _TESTS_FOLDER_NAME
from pants.backend.helm.target_types import AllHelmChartTargets, HelmUnitTestDependenciesField
from pants.core.target_types import AllAssetTargetsByPath
from pants.engine.addresses import Address
from pants.engine.internals.graph import determine_explicitly_provided_dependencies
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import (
    DependenciesRequest,
    FieldSet,
    InferDependenciesRequest,
    InferredDependencies,
    Target,
)
from pants.engine.unions import UnionRule
from pants.util.ordered_set import OrderedSet
from pants.util.strutil import bullet_list, pluralize

logger = logging.getLogger(__name__)


class InvalidUnitTestTestFolder(Exception):
    def __init__(self, address: Address, found_folder: str) -> None:
        super().__init__(
            f"`helm_unittest_test` target at {address.spec_path} is at the wrong folder, "
            f"it should be inside a `tests` folder under the Helm chart root sources, it was found at: {found_folder}"
        )


class AmbiguousHelmUnitTestChart(Exception):
    def __init__(self, *, target_addr: str, putative_addresses: Sequence[str]) -> None:
        super().__init__(
            f"The actual Helm chart for the target at '{target_addr}' is ambiguous and can not be inferred. "
            f"Found {pluralize(len(putative_addresses), 'candidate')}:\n{bullet_list(putative_addresses)}"
        )


@dataclass(frozen=True)
class HelmUnitTestChartDependencyInferenceFieldSet(FieldSet):
    required_fields = (HelmUnitTestDependenciesField,)

    dependencies: HelmUnitTestDependenciesField


class InferHelmUnitTestChartDependencyRequest(InferDependenciesRequest):
    infer_from = HelmUnitTestChartDependencyInferenceFieldSet


@rule
async def infer_chart_dependency_into_unittests(
    request: InferHelmUnitTestChartDependencyRequest,
    all_helm_charts: AllHelmChartTargets,
    all_asset_targets: AllAssetTargetsByPath,
) -> InferredDependencies:
    unittest_target_addr: Address = request.field_set.address

    putative_chart_path, unittest_target_dir = os.path.split(unittest_target_addr.spec_path)
    if unittest_target_dir != _TESTS_FOLDER_NAME:
        raise InvalidUnitTestTestFolder(unittest_target_addr, unittest_target_addr.spec_path)

    explicitly_provided_deps = await determine_explicitly_provided_dependencies(
        **implicitly(DependenciesRequest(request.field_set.dependencies))
    )

    def is_snapshot_resource(path: PurePath) -> bool:
        if not path.parent:
            return False
        if path.parent.name != _SNAPSHOT_FOLDER_NAME:
            return False
        if not path.parent.parent:
            return False
        return str(path.parent.parent) == unittest_target_addr.spec_path

    candidate_snapshot_resources = {
        tgt.address
        for path, targets in all_asset_targets.resources.items()
        if is_snapshot_resource(path)
        for tgt in targets
    }
    found_snapshots = set()

    for candidate_snapshot in candidate_snapshot_resources:
        explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(
            (candidate_snapshot,),
            unittest_target_addr,
            import_reference="snapshot",
            context=f"The target {candidate_snapshot} is nested under {unittest_target_addr}",
        )

        maybe_disambiguated = explicitly_provided_deps.disambiguated((candidate_snapshot,))
        if maybe_disambiguated:
            found_snapshots.add(maybe_disambiguated)

    def is_parent_chart(target: Target) -> bool:
        chart_folder = target.address.spec_path
        return chart_folder == putative_chart_path

    candidate_charts: OrderedSet[Address] = OrderedSet(
        [tgt.address for tgt in all_helm_charts if is_parent_chart(tgt)]
    )
    found_charts = set()

    for candidate_chart in candidate_charts:
        explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(
            (candidate_chart,),
            unittest_target_addr,
            import_reference="chart",
            context=f"The target {unittest_target_addr} is nested under the chart {candidate_chart}",
        )
        maybe_disambiguated = explicitly_provided_deps.disambiguated((candidate_chart,))
        if maybe_disambiguated:
            found_charts.add(maybe_disambiguated)

    if len(found_charts) > 1:
        raise AmbiguousHelmUnitTestChart(
            target_addr=unittest_target_addr.spec,
            putative_addresses=[addr.spec for addr in found_charts],
        )

    if len(found_charts) == 1:
        found_dep = list(found_charts)[0]
        logger.debug(
            f"Found Helm chart at '{found_dep.spec}' for unittest at: {unittest_target_addr.spec}"
        )

    dependencies: OrderedSet[Address] = OrderedSet()
    dependencies.update(found_snapshots)
    dependencies.update(found_charts)

    return InferredDependencies(dependencies)


def rules():
    return [
        *collect_rules(),
        UnionRule(InferDependenciesRequest, InferHelmUnitTestChartDependencyRequest),
    ]

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	import logging	×
5	import os	×
6	from collections.abc import Sequence	×
7	from dataclasses import dataclass	×
8	from pathlib import PurePath	×
9
10	from pants.backend.helm.goals.tailor import _SNAPSHOT_FOLDER_NAME, _TESTS_FOLDER_NAME	×
11	from pants.backend.helm.target_types import AllHelmChartTargets, HelmUnitTestDependenciesField	×
12	from pants.core.target_types import AllAssetTargetsByPath	×
13	from pants.engine.addresses import Address	×
14	from pants.engine.internals.graph import determine_explicitly_provided_dependencies	×
15	from pants.engine.rules import collect_rules, implicitly, rule	×
16	from pants.engine.target import (	×
17	DependenciesRequest,
18	FieldSet,
19	InferDependenciesRequest,
20	InferredDependencies,
21	Target,
22	)
23	from pants.engine.unions import UnionRule	×
24	from pants.util.ordered_set import OrderedSet	×
25	from pants.util.strutil import bullet_list, pluralize	×
26
27	logger = logging.getLogger(__name__)	×
28
29
30	class InvalidUnitTestTestFolder(Exception):	×
31	def __init__(self, address: Address, found_folder: str) -> None:	×
32	super().__init__(	×
33	f"`helm_unittest_test` target at {address.spec_path} is at the wrong folder, "
34	f"it should be inside a `tests` folder under the Helm chart root sources, it was found at: {found_folder}"
35	)
36
37
38	class AmbiguousHelmUnitTestChart(Exception):	×
39	def __init__(self, *, target_addr: str, putative_addresses: Sequence[str]) -> None:	×
40	super().__init__(	×
41	f"The actual Helm chart for the target at '{target_addr}' is ambiguous and can not be inferred. "
42	f"Found {pluralize(len(putative_addresses), 'candidate')}:\n{bullet_list(putative_addresses)}"
43	)
44
45
46	@dataclass(frozen=True)	×
47	class HelmUnitTestChartDependencyInferenceFieldSet(FieldSet):	×
48	required_fields = (HelmUnitTestDependenciesField,)	×
49
50	dependencies: HelmUnitTestDependenciesField	×
51
52
53	class InferHelmUnitTestChartDependencyRequest(InferDependenciesRequest):	×
54	infer_from = HelmUnitTestChartDependencyInferenceFieldSet	×
55
56
57	@rule	×
58	async def infer_chart_dependency_into_unittests(	×
59	request: InferHelmUnitTestChartDependencyRequest,
60	all_helm_charts: AllHelmChartTargets,
61	all_asset_targets: AllAssetTargetsByPath,
62	) -> InferredDependencies:
63	unittest_target_addr: Address = request.field_set.address	×
64
65	putative_chart_path, unittest_target_dir = os.path.split(unittest_target_addr.spec_path)	×
66	if unittest_target_dir != _TESTS_FOLDER_NAME:	×
67	raise InvalidUnitTestTestFolder(unittest_target_addr, unittest_target_addr.spec_path)	×
68
69	explicitly_provided_deps = await determine_explicitly_provided_dependencies(	×
70	**implicitly(DependenciesRequest(request.field_set.dependencies))
71	)
72
73	def is_snapshot_resource(path: PurePath) -> bool:	×
74	if not path.parent:	×
75	return False	×
76	if path.parent.name != _SNAPSHOT_FOLDER_NAME:	×
77	return False	×
78	if not path.parent.parent:	×
79	return False	×
80	return str(path.parent.parent) == unittest_target_addr.spec_path	×
81
82	candidate_snapshot_resources = {	×
83	tgt.address
84	for path, targets in all_asset_targets.resources.items()
85	if is_snapshot_resource(path)
86	for tgt in targets
87	}
88	found_snapshots = set()	×
89
90	for candidate_snapshot in candidate_snapshot_resources:	×
91	explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(	×
92	(candidate_snapshot,),
93	unittest_target_addr,
94	import_reference="snapshot",
95	context=f"The target {candidate_snapshot} is nested under {unittest_target_addr}",
96	)
97
98	maybe_disambiguated = explicitly_provided_deps.disambiguated((candidate_snapshot,))	×
99	if maybe_disambiguated:	×
100	found_snapshots.add(maybe_disambiguated)	×
101
102	def is_parent_chart(target: Target) -> bool:	×
103	chart_folder = target.address.spec_path	×
104	return chart_folder == putative_chart_path	×
105
106	candidate_charts: OrderedSet[Address] = OrderedSet(	×
107	[tgt.address for tgt in all_helm_charts if is_parent_chart(tgt)]
108	)
109	found_charts = set()	×
110
111	for candidate_chart in candidate_charts:	×
112	explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(	×
113	(candidate_chart,),
114	unittest_target_addr,
115	import_reference="chart",
116	context=f"The target {unittest_target_addr} is nested under the chart {candidate_chart}",
117	)
118	maybe_disambiguated = explicitly_provided_deps.disambiguated((candidate_chart,))	×
119	if maybe_disambiguated:	×
120	found_charts.add(maybe_disambiguated)	×
121
122	if len(found_charts) > 1:	×
123	raise AmbiguousHelmUnitTestChart(	×
124	target_addr=unittest_target_addr.spec,
125	putative_addresses=[addr.spec for addr in found_charts],
126	)
127
128	if len(found_charts) == 1:	×
129	found_dep = list(found_charts)[0]	×
130	logger.debug(	×
131	f"Found Helm chart at '{found_dep.spec}' for unittest at: {unittest_target_addr.spec}"
132	)
133
134	dependencies: OrderedSet[Address] = OrderedSet()	×
135	dependencies.update(found_snapshots)	×
136	dependencies.update(found_charts)	×
137
138	return InferredDependencies(dependencies)	×
139
140
141	def rules():	×
142	return [	×
143	*collect_rules(),
144	UnionRule(InferDependenciesRequest, InferHelmUnitTestChartDependencyRequest),
145	]

pantsbuild / pants / 18791134616

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