21552830208

Committed 31 Jan 2026 11:40PM UTC coverage: 80.277% (-0.05%) from 80.324%

Build # 21552830208

Build Type

Pull #23062

github

Committed by

web-flow

Commit Message

Merge 808a9786c into 2c4dcf9cf

Pull Request Pull Request #23062: Remove support for Get

Run Details

18 of 25 new or added lines in 4 files covered. (72.0%)

17119 existing lines in 541 files now uncovered.

78278 of 97510 relevant lines covered (80.28%)

3.36 hits per line

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

70.83

/src/python/pants/core/util_rules/source_files.py

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

from collections.abc import Collection, Iterable
from dataclasses import dataclass
from pathlib import PurePath

from pants.engine.fs import MergeDigests, Snapshot
from pants.engine.internals.graph import hydrate_sources
from pants.engine.intrinsics import digest_to_snapshot
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import HydrateSourcesRequest, SourcesField, Target


@dataclass(frozen=True)
class SourceFiles:
    """A merged snapshot of the `sources` fields of multiple targets."""

    snapshot: Snapshot

    # The subset of files in snapshot that are not intended to have an associated source root.
    # That is, the sources of files() targets.
    unrooted_files: tuple[str, ...]

    @property
    def files(self) -> tuple[str, ...]:
        return self.snapshot.files


@dataclass(frozen=True)
class SourceFilesRequest:
    sources_fields: tuple[SourcesField, ...]
    for_sources_types: tuple[type[SourcesField], ...]
    enable_codegen: bool

    def __init__(
        self,
        sources_fields: Iterable[SourcesField],
        *,
        for_sources_types: Iterable[type[SourcesField]] = (SourcesField,),
        enable_codegen: bool = False,
    ) -> None:
        object.__setattr__(self, "sources_fields", tuple(sources_fields))
        object.__setattr__(self, "for_sources_types", tuple(for_sources_types))
        object.__setattr__(self, "enable_codegen", enable_codegen)


@rule(desc="Get all relevant source files")
async def determine_source_files(request: SourceFilesRequest) -> SourceFiles:
    """Merge all `SourceBaseField`s into one Snapshot."""
    unrooted_files: set[str] = set()
    all_hydrated_sources = await concurrently(
        hydrate_sources(
            HydrateSourcesRequest(
                sources_field,
                for_sources_types=request.for_sources_types,
                enable_codegen=request.enable_codegen,
            ),
            **implicitly(),
        )
        for sources_field in request.sources_fields
    )

    for hydrated_sources, sources_field in zip(all_hydrated_sources, request.sources_fields):
        if not sources_field.uses_source_roots:
            unrooted_files.update(hydrated_sources.snapshot.files)

    digests_to_merge = tuple(
        hydrated_sources.snapshot.digest for hydrated_sources in all_hydrated_sources
    )
    result = await digest_to_snapshot(**implicitly(MergeDigests(digests_to_merge)))
    return SourceFiles(result, tuple(sorted(unrooted_files)))


@dataclass(frozen=True)
class ClassifiedSources:
    target_type: type[Target]
    files: Collection[str]
    name: str | None = None


def classify_files_for_sources_and_tests(
    paths: Iterable[str],
    test_file_glob: tuple[str, ...],
    sources_generator: type[Target],
    tests_generator: type[Target],
) -> Iterable[ClassifiedSources]:
    """Classify files when running the tailor goal logic.

    This function is to be re-used by any tailor related code that needs to separate sources
    collected for the target generators to own sources code (`language-name_sources` targets) and
    tests code (`language-name_tests` targets).
    """
    sources_files = set(paths)
    test_files = {
        path for path in paths if any(PurePath(path).match(glob) for glob in test_file_glob)
    }
    if sources_files:
        yield ClassifiedSources(sources_generator, files=sources_files - test_files)
    if test_files:
        yield ClassifiedSources(tests_generator, test_files, "tests")


def rules():
    return collect_rules()

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from collections.abc import Collection, Iterable	12✔
5	from dataclasses import dataclass	12✔
6	from pathlib import PurePath	12✔
7
8	from pants.engine.fs import MergeDigests, Snapshot	12✔
9	from pants.engine.internals.graph import hydrate_sources	12✔
10	from pants.engine.intrinsics import digest_to_snapshot	12✔
11	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	12✔
12	from pants.engine.target import HydrateSourcesRequest, SourcesField, Target	12✔
13
14
15	@dataclass(frozen=True)	12✔
16	class SourceFiles:	12✔
17	"""A merged snapshot of the `sources` fields of multiple targets."""
18
19	snapshot: Snapshot	12✔
20
21	# The subset of files in snapshot that are not intended to have an associated source root.
22	# That is, the sources of files() targets.
23	unrooted_files: tuple[str, ...]	12✔
24
25	@property	12✔
26	def files(self) -> tuple[str, ...]:	12✔
UNCOV 27	return self.snapshot.files	2✔
28
29
30	@dataclass(frozen=True)	12✔
31	class SourceFilesRequest:	12✔
32	sources_fields: tuple[SourcesField, ...]	12✔
33	for_sources_types: tuple[type[SourcesField], ...]	12✔
34	enable_codegen: bool	12✔
35
36	def __init__(	12✔
37	self,
38	sources_fields: Iterable[SourcesField],
39	*,
40	for_sources_types: Iterable[type[SourcesField]] = (SourcesField,),
41	enable_codegen: bool = False,
42	) -> None:
43	object.__setattr__(self, "sources_fields", tuple(sources_fields))	11✔
44	object.__setattr__(self, "for_sources_types", tuple(for_sources_types))	11✔
45	object.__setattr__(self, "enable_codegen", enable_codegen)	11✔
46
47
48	@rule(desc="Get all relevant source files")	12✔
49	async def determine_source_files(request: SourceFilesRequest) -> SourceFiles:	12✔
50	"""Merge all `SourceBaseField`s into one Snapshot."""
51	unrooted_files: set[str] = set()	×
52	all_hydrated_sources = await concurrently(	×
53	hydrate_sources(
54	HydrateSourcesRequest(
55	sources_field,
56	for_sources_types=request.for_sources_types,
57	enable_codegen=request.enable_codegen,
58	),
59	**implicitly(),
60	)
61	for sources_field in request.sources_fields
62	)
63
64	for hydrated_sources, sources_field in zip(all_hydrated_sources, request.sources_fields):	×
65	if not sources_field.uses_source_roots:	×
66	unrooted_files.update(hydrated_sources.snapshot.files)	×
67
68	digests_to_merge = tuple(	×
69	hydrated_sources.snapshot.digest for hydrated_sources in all_hydrated_sources
70	)
71	result = await digest_to_snapshot(**implicitly(MergeDigests(digests_to_merge)))	×
72	return SourceFiles(result, tuple(sorted(unrooted_files)))	×
73
74
75	@dataclass(frozen=True)	12✔
76	class ClassifiedSources:	12✔
77	target_type: type[Target]	12✔
78	files: Collection[str]	12✔
79	name: str \| None = None	12✔
80
81
82	def classify_files_for_sources_and_tests(	12✔
83	paths: Iterable[str],
84	test_file_glob: tuple[str, ...],
85	sources_generator: type[Target],
86	tests_generator: type[Target],
87	) -> Iterable[ClassifiedSources]:
88	"""Classify files when running the tailor goal logic.
89
90	This function is to be re-used by any tailor related code that needs to separate sources
91	collected for the target generators to own sources code (`language-name_sources` targets) and
92	tests code (`language-name_tests` targets).
93	"""
94	sources_files = set(paths)	×
95	test_files = {	×
96	path for path in paths if any(PurePath(path).match(glob) for glob in test_file_glob)
97	}
98	if sources_files:	×
99	yield ClassifiedSources(sources_generator, files=sources_files - test_files)	×
100	if test_files:	×
101	yield ClassifiedSources(tests_generator, test_files, "tests")	×
102
103
104	def rules():	12✔
105	return collect_rules()	12✔

pantsbuild / pants / 21552830208

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