19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

53.66

/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	1✔
5	from dataclasses import dataclass	1✔
6	from pathlib import PurePath	1✔
7
8	from pants.engine.fs import MergeDigests, Snapshot	1✔
9	from pants.engine.internals.graph import hydrate_sources	1✔
10	from pants.engine.intrinsics import digest_to_snapshot	1✔
11	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	1✔
12	from pants.engine.target import HydrateSourcesRequest, SourcesField, Target	1✔
13
14
15	@dataclass(frozen=True)	1✔
16	class SourceFiles:	1✔
17	"""A merged snapshot of the `sources` fields of multiple targets."""
18
19	snapshot: Snapshot
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, ...]
24
25	@property	1✔
26	def files(self) -> tuple[str, ...]:	1✔
UNCOV 27	return self.snapshot.files	×
28
29
30	@dataclass(frozen=True)	1✔
31	class SourceFilesRequest:	1✔
32	sources_fields: tuple[SourcesField, ...]
33	for_sources_types: tuple[type[SourcesField], ...]
34	enable_codegen: bool
35
36	def __init__(	1✔
37	self,
38	sources_fields: Iterable[SourcesField],
39	*,
40	for_sources_types: Iterable[type[SourcesField]] = (SourcesField,),
41	enable_codegen: bool = False,
42	) -> None:
UNCOV 43	object.__setattr__(self, "sources_fields", tuple(sources_fields))	×
UNCOV 44	object.__setattr__(self, "for_sources_types", tuple(for_sources_types))	×
UNCOV 45	object.__setattr__(self, "enable_codegen", enable_codegen)	×
46
47
48	@rule(desc="Get all relevant source files")	1✔
49	async def determine_source_files(request: SourceFilesRequest) -> SourceFiles:	1✔
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)	1✔
76	class ClassifiedSources:	1✔
77	target_type: type[Target]
78	files: Collection[str]
79	name: str \| None = None	1✔
80
81
82	def classify_files_for_sources_and_tests(	1✔
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():	1✔
UNCOV 105	return collect_rules()	×

pantsbuild / pants / 19015773527

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