18517631058

Committed 15 Oct 2025 04:18AM UTC coverage: 69.207% (-11.1%) from 80.267%

Build # 18517631058

Build Type

Pull #22745

github

Committed by

web-flow

Commit Message

Merge 642a76ca1 into 99919310e

Pull Request Pull Request #22745: [windows] Add windows support in the stdio crate.

Run Details

53815 of 77759 relevant lines covered (69.21%)

2.42 hits per line

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

56.63

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

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

from __future__ import annotations

import os
from dataclasses import dataclass

from pants.backend.helm.target_types import (
    HelmChartFieldSet,
    HelmChartMetaSourceField,
    HelmChartSourcesField,
)
from pants.core.target_types import FileSourceField, ResourceSourceField
from pants.core.util_rules import source_files
from pants.core.util_rules.source_files import (
    SourceFiles,
    SourceFilesRequest,
    determine_source_files,
)
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.fs import DigestSubset, MergeDigests, PathGlobs, Snapshot
from pants.engine.internals.graph import hydrate_sources, resolve_targets
from pants.engine.internals.native_engine import RemovePrefix
from pants.engine.intrinsics import digest_subset_to_digest, digest_to_snapshot
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import DependenciesRequest, HydrateSourcesRequest, SourcesField, Target


@dataclass(frozen=True)
class HelmChartRootRequest(EngineAwareParameter):
    source: HelmChartMetaSourceField

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


@dataclass(frozen=True)
class HelmChartRoot:
    path: str


@rule(desc="Detect Helm chart source root")
async def find_chart_source_root(request: HelmChartRootRequest) -> HelmChartRoot:
    source = await hydrate_sources(
        HydrateSourcesRequest(
            request.source, for_sources_types=[HelmChartMetaSourceField], enable_codegen=True
        ),
        **implicitly(),
    )
    assert len(source.snapshot.files) == 1

    return HelmChartRoot(os.path.dirname(source.snapshot.files[0]))


@dataclass(frozen=True)
class HelmChartSourceFilesRequest(EngineAwareParameter):
    field_set: HelmChartFieldSet
    include_resources: bool
    include_files: bool
    include_metadata: bool

    @classmethod
    def create(
        cls,
        target: Target,
        *,
        include_resources: bool = True,
        include_files: bool = False,
        include_metadata: bool = True,
    ) -> HelmChartSourceFilesRequest:
        return cls.for_field_set(
            HelmChartFieldSet.create(target),
            include_resources=include_resources,
            include_files=include_files,
            include_metadata=include_metadata,
        )

    @classmethod
    def for_field_set(
        cls,
        field_set: HelmChartFieldSet,
        *,
        include_resources: bool = True,
        include_files: bool = False,
        include_metadata: bool = True,
    ) -> HelmChartSourceFilesRequest:
        return cls(
            field_set=field_set,
            include_resources=include_resources,
            include_files=include_files,
            include_metadata=include_metadata,
        )

    @property
    def sources_fields(self) -> tuple[SourcesField, ...]:
        fields: list[SourcesField] = [self.field_set.sources]
        if self.include_metadata:
            fields.append(self.field_set.chart)
        return tuple(fields)

    @property
    def valid_sources_types(self) -> tuple[type[SourcesField], ...]:
        types: list[type[SourcesField]] = [HelmChartSourcesField]
        if self.include_metadata:
            types.append(HelmChartMetaSourceField)
        if self.include_resources:
            types.append(ResourceSourceField)
        if self.include_files:
            types.append(FileSourceField)
        return tuple(types)

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


@dataclass(frozen=True)
class HelmChartSourceFiles:
    snapshot: Snapshot
    unrooted_files: tuple[str, ...]


async def _strip_chart_source_root(
    source_files: SourceFiles, chart_root: HelmChartRoot
) -> Snapshot:
    if not source_files.snapshot.files:
        return source_files.snapshot

    if source_files.unrooted_files:
        rooted_files = set(source_files.snapshot.files) - set(source_files.unrooted_files)
        rooted_files_snapshot = await digest_to_snapshot(
            **implicitly(DigestSubset(source_files.snapshot.digest, PathGlobs(rooted_files)))
        )
    else:
        rooted_files_snapshot = source_files.snapshot

    resulting_snapshot = await digest_to_snapshot(
        **implicitly(RemovePrefix(rooted_files_snapshot.digest, chart_root.path))
    )
    if source_files.unrooted_files:
        # Add unrooted files back in
        unrooted_digest = await digest_subset_to_digest(
            DigestSubset(source_files.snapshot.digest, PathGlobs(source_files.unrooted_files)),
        )
        resulting_snapshot = await digest_to_snapshot(
            **implicitly(MergeDigests([resulting_snapshot.digest, unrooted_digest]))
        )

    return resulting_snapshot


@rule
async def get_helm_source_files(request: HelmChartSourceFilesRequest) -> HelmChartSourceFiles:
    chart_root, dependencies = await concurrently(
        find_chart_source_root(HelmChartRootRequest(request.field_set.chart)),
        resolve_targets(**implicitly(DependenciesRequest(request.field_set.dependencies))),
    )

    source_files, original_sources = await concurrently(
        determine_source_files(
            SourceFilesRequest(
                sources_fields=[
                    *request.sources_fields,
                    *(
                        tgt.get(SourcesField)
                        for tgt in dependencies
                        if not HelmChartFieldSet.is_applicable(tgt)
                    ),
                ],
                for_sources_types=request.valid_sources_types,
                enable_codegen=True,
            )
        ),
        determine_source_files(
            SourceFilesRequest([request.field_set.sources], enable_codegen=False)
        ),
    )

    stripped_source_files = await _strip_chart_source_root(source_files, chart_root)
    stripped_original_sources = await _strip_chart_source_root(original_sources, chart_root)

    all_files_snapshot = await digest_to_snapshot(
        **implicitly(MergeDigests([stripped_source_files.digest, stripped_original_sources.digest]))
    )
    return HelmChartSourceFiles(
        snapshot=all_files_snapshot,
        unrooted_files=(*source_files.unrooted_files, *original_sources.unrooted_files),
    )


def rules():
    return [*collect_rules(), *source_files.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	7✔
5
6	import os	7✔
7	from dataclasses import dataclass	7✔
8
9	from pants.backend.helm.target_types import (	7✔
10	HelmChartFieldSet,
11	HelmChartMetaSourceField,
12	HelmChartSourcesField,
13	)
14	from pants.core.target_types import FileSourceField, ResourceSourceField	7✔
15	from pants.core.util_rules import source_files	7✔
16	from pants.core.util_rules.source_files import (	7✔
17	SourceFiles,
18	SourceFilesRequest,
19	determine_source_files,
20	)
21	from pants.engine.engine_aware import EngineAwareParameter	7✔
22	from pants.engine.fs import DigestSubset, MergeDigests, PathGlobs, Snapshot	7✔
23	from pants.engine.internals.graph import hydrate_sources, resolve_targets	7✔
24	from pants.engine.internals.native_engine import RemovePrefix	7✔
25	from pants.engine.intrinsics import digest_subset_to_digest, digest_to_snapshot	7✔
26	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	7✔
27	from pants.engine.target import DependenciesRequest, HydrateSourcesRequest, SourcesField, Target	7✔
28
29
30	@dataclass(frozen=True)	7✔
31	class HelmChartRootRequest(EngineAwareParameter):	7✔
32	source: HelmChartMetaSourceField	7✔
33
34	def debug_hint(self) -> str \| None:	7✔
35	return self.source.address.spec	×
36
37
38	@dataclass(frozen=True)	7✔
39	class HelmChartRoot:	7✔
40	path: str	7✔
41
42
43	@rule(desc="Detect Helm chart source root")	7✔
44	async def find_chart_source_root(request: HelmChartRootRequest) -> HelmChartRoot:	7✔
45	source = await hydrate_sources(	×
46	HydrateSourcesRequest(
47	request.source, for_sources_types=[HelmChartMetaSourceField], enable_codegen=True
48	),
49	**implicitly(),
50	)
51	assert len(source.snapshot.files) == 1	×
52
53	return HelmChartRoot(os.path.dirname(source.snapshot.files[0]))	×
54
55
56	@dataclass(frozen=True)	7✔
57	class HelmChartSourceFilesRequest(EngineAwareParameter):	7✔
58	field_set: HelmChartFieldSet	7✔
59	include_resources: bool	7✔
60	include_files: bool	7✔
61	include_metadata: bool	7✔
62
63	@classmethod	7✔
64	def create(	7✔
65	cls,
66	target: Target,
67	*,
68	include_resources: bool = True,
69	include_files: bool = False,
70	include_metadata: bool = True,
71	) -> HelmChartSourceFilesRequest:
72	return cls.for_field_set(	×
73	HelmChartFieldSet.create(target),
74	include_resources=include_resources,
75	include_files=include_files,
76	include_metadata=include_metadata,
77	)
78
79	@classmethod	7✔
80	def for_field_set(	7✔
81	cls,
82	field_set: HelmChartFieldSet,
83	*,
84	include_resources: bool = True,
85	include_files: bool = False,
86	include_metadata: bool = True,
87	) -> HelmChartSourceFilesRequest:
88	return cls(	×
89	field_set=field_set,
90	include_resources=include_resources,
91	include_files=include_files,
92	include_metadata=include_metadata,
93	)
94
95	@property	7✔
96	def sources_fields(self) -> tuple[SourcesField, ...]:	7✔
97	fields: list[SourcesField] = [self.field_set.sources]	×
98	if self.include_metadata:	×
99	fields.append(self.field_set.chart)	×
100	return tuple(fields)	×
101
102	@property	7✔
103	def valid_sources_types(self) -> tuple[type[SourcesField], ...]:	7✔
104	types: list[type[SourcesField]] = [HelmChartSourcesField]	×
105	if self.include_metadata:	×
106	types.append(HelmChartMetaSourceField)	×
107	if self.include_resources:	×
108	types.append(ResourceSourceField)	×
109	if self.include_files:	×
110	types.append(FileSourceField)	×
111	return tuple(types)	×
112
113	def debug_hint(self) -> str \| None:	7✔
114	return self.field_set.address.spec	×
115
116
117	@dataclass(frozen=True)	7✔
118	class HelmChartSourceFiles:	7✔
119	snapshot: Snapshot	7✔
120	unrooted_files: tuple[str, ...]	7✔
121
122
123	async def _strip_chart_source_root(	7✔
124	source_files: SourceFiles, chart_root: HelmChartRoot
125	) -> Snapshot:
126	if not source_files.snapshot.files:	×
127	return source_files.snapshot	×
128
129	if source_files.unrooted_files:	×
130	rooted_files = set(source_files.snapshot.files) - set(source_files.unrooted_files)	×
131	rooted_files_snapshot = await digest_to_snapshot(	×
132	**implicitly(DigestSubset(source_files.snapshot.digest, PathGlobs(rooted_files)))
133	)
134	else:
135	rooted_files_snapshot = source_files.snapshot	×
136
137	resulting_snapshot = await digest_to_snapshot(	×
138	**implicitly(RemovePrefix(rooted_files_snapshot.digest, chart_root.path))
139	)
140	if source_files.unrooted_files:	×
141	# Add unrooted files back in
142	unrooted_digest = await digest_subset_to_digest(	×
143	DigestSubset(source_files.snapshot.digest, PathGlobs(source_files.unrooted_files)),
144	)
145	resulting_snapshot = await digest_to_snapshot(	×
146	**implicitly(MergeDigests([resulting_snapshot.digest, unrooted_digest]))
147	)
148
149	return resulting_snapshot	×
150
151
152	@rule	7✔
153	async def get_helm_source_files(request: HelmChartSourceFilesRequest) -> HelmChartSourceFiles:	7✔
154	chart_root, dependencies = await concurrently(	×
155	find_chart_source_root(HelmChartRootRequest(request.field_set.chart)),
156	resolve_targets(**implicitly(DependenciesRequest(request.field_set.dependencies))),
157	)
158
159	source_files, original_sources = await concurrently(	×
160	determine_source_files(
161	SourceFilesRequest(
162	sources_fields=[
163	*request.sources_fields,
164	*(
165	tgt.get(SourcesField)
166	for tgt in dependencies
167	if not HelmChartFieldSet.is_applicable(tgt)
168	),
169	],
170	for_sources_types=request.valid_sources_types,
171	enable_codegen=True,
172	)
173	),
174	determine_source_files(
175	SourceFilesRequest([request.field_set.sources], enable_codegen=False)
176	),
177	)
178
179	stripped_source_files = await _strip_chart_source_root(source_files, chart_root)	×
180	stripped_original_sources = await _strip_chart_source_root(original_sources, chart_root)	×
181
182	all_files_snapshot = await digest_to_snapshot(	×
183	**implicitly(MergeDigests([stripped_source_files.digest, stripped_original_sources.digest]))
184	)
185	return HelmChartSourceFiles(	×
186	snapshot=all_files_snapshot,
187	unrooted_files=(source_files.unrooted_files, original_sources.unrooted_files),
188	)
189
190
191	def rules():	7✔
192	return [collect_rules(), source_files.rules()]	5✔

pantsbuild / pants / 18517631058

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