18252174847

Committed 05 Oct 2025 01:36AM UTC coverage: 43.382% (-36.9%) from 80.261%

Build # 18252174847

Build Type

push

github

Committed by

web-flow

Commit Message

run tests on mac arm (#22717)

Just doing the minimal to pull forward the x86_64 pattern.

ref #20993

Run Details

25776 of 59416 relevant lines covered (43.38%)

1.3 hits per line

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

68.75

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

pantsbuild / pants / 18252174847

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