19250292619

Committed 11 Nov 2025 12:09AM UTC coverage: 77.865% (-2.4%) from 80.298%

Build # 19250292619

Build Type

push

github

Committed by

web-flow

Commit Message

flag non-runnable targets used with `code_quality_tool` (#22875)

Run Details

2 of 5 new or added lines in 2 files covered. (40.0%)

1487 existing lines in 72 files now uncovered.

71448 of 91759 relevant lines covered (77.86%)

3.22 hits per line

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

61.76

/src/python/pants/backend/python/util_rules/python_sources.py

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

from __future__ import annotations

from collections.abc import Iterable
from dataclasses import dataclass

from pants.backend.python.target_types import PythonSourceField
from pants.backend.python.util_rules import ancestor_files
from pants.backend.python.util_rules.ancestor_files import AncestorFilesRequest, find_ancestor_files
from pants.core.target_types import FileSourceField, ResourceSourceField
from pants.core.util_rules import source_files, stripped_source_files
from pants.core.util_rules.source_files import (
    SourceFiles,
    SourceFilesRequest,
    determine_source_files,
)
from pants.core.util_rules.stripped_source_files import StrippedSourceFiles, strip_source_roots
from pants.engine.fs import EMPTY_SNAPSHOT, MergeDigests
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
from pants.engine.unions import UnionMembership
from pants.source.source_root import SourceRootRequest, get_source_root
from pants.util.logging import LogLevel


@dataclass(frozen=True)
class PythonSourceFiles:
    """Sources that can be introspected by Python, relative to a set of source roots.

    Specifically, this will filter out to only have Python, and, optionally, resource and
    file targets; and will add any missing `__init__.py` files to ensure that modules are
    recognized correctly.

    Use-cases that introspect Python source code (e.g., the `test, `lint`, `fmt` goals) can
    request this type to get relevant sources that are still relative to their source roots.
    That way the paths they report are the unstripped ones the user is familiar with.

    The sources can also be imported and used by Python (e.g., for the `test` goal), but only
    if sys.path is modified to include the source roots.
    """

    source_files: SourceFiles
    source_roots: tuple[str, ...]  # Source roots for the specified source files.

    @classmethod
    def empty(cls) -> PythonSourceFiles:
        return cls(SourceFiles(EMPTY_SNAPSHOT, tuple()), tuple())


@dataclass(frozen=True)
class StrippedPythonSourceFiles:
    """A PythonSourceFiles that has had its source roots stripped."""

    stripped_source_files: StrippedSourceFiles


@dataclass(unsafe_hash=True)
class PythonSourceFilesRequest:
    targets: tuple[Target, ...]
    include_resources: bool
    # NB: Setting this to True may cause Digest merge collisions, in the presence of
    #  relocated_files targets that relocate two different files to the same destination.
    #  Set this to True only if you know this cannot logically happen (e.g., if it did
    #  happen the underlying user operation would make no sense anyway).
    #  For example, if the user relocates different config files to the same location
    #  in different tests, there is no problem as long as we sandbox each test's sources
    #  separately. The user may hit this issue if they run tests in batches, but in that
    #  case they couldn't make that work even without Pants.
    include_files: bool

    def __init__(
        self,
        targets: Iterable[Target],
        *,
        include_resources: bool = True,
        include_files: bool = False,
    ) -> None:
        object.__setattr__(self, "targets", tuple(targets))
        object.__setattr__(self, "include_resources", include_resources)
        object.__setattr__(self, "include_files", include_files)

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


@rule(level=LogLevel.DEBUG)
async def prepare_python_sources(
    request: PythonSourceFilesRequest, union_membership: UnionMembership
) -> PythonSourceFiles:
    sources = await determine_source_files(
        SourceFilesRequest(
            (tgt.get(SourcesField) for tgt in request.targets),
            for_sources_types=request.valid_sources_types,
            enable_codegen=True,
        )
    )

    missing_init_files = await find_ancestor_files(
        AncestorFilesRequest(
            input_files=sources.snapshot.files, requested=("__init__.py", "__init__.pyi")
        )
    )
    init_injected = await digest_to_snapshot(
        **implicitly(MergeDigests((sources.snapshot.digest, missing_init_files.snapshot.digest)))
    )

    # Codegen is able to generate code in any arbitrary location, unlike sources normally being
    # rooted under the target definition. To determine source roots for these generated files, we
    # cannot use the normal `SourceRootRequest.for_target()` and we instead must determine
    # a source root for every individual generated file. So, we re-resolve the codegen sources here.
    python_and_resources_targets = []
    codegen_targets = []
    for tgt in request.targets:
        if tgt.has_field(PythonSourceField) or tgt.has_field(ResourceSourceField):
            python_and_resources_targets.append(tgt)
        elif tgt.get(SourcesField).can_generate(PythonSourceField, union_membership) or tgt.get(
            SourcesField
        ).can_generate(ResourceSourceField, union_membership):
            codegen_targets.append(tgt)
    codegen_sources = await concurrently(
        hydrate_sources(
            HydrateSourcesRequest(
                tgt.get(SourcesField),
                for_sources_types=request.valid_sources_types,
                enable_codegen=True,
            ),
            **implicitly(),
        )
        for tgt in codegen_targets
    )
    source_root_requests = [
        *(SourceRootRequest.for_target(tgt) for tgt in python_and_resources_targets),
        *(
            SourceRootRequest.for_file(f)
            for sources in codegen_sources
            for f in sources.snapshot.files
        ),
    ]

    source_root_objs = await concurrently(get_source_root(req) for req in source_root_requests)
    source_root_paths = {source_root_obj.path for source_root_obj in source_root_objs}
    return PythonSourceFiles(
        SourceFiles(init_injected, sources.unrooted_files), tuple(sorted(source_root_paths))
    )


@rule(level=LogLevel.DEBUG)
async def strip_python_sources(python_sources: PythonSourceFiles) -> StrippedPythonSourceFiles:
    stripped = await strip_source_roots(python_sources.source_files)
    return StrippedPythonSourceFiles(stripped)


def rules():
    return [
        *collect_rules(),
        *ancestor_files.rules(),
        *source_files.rules(),
        *stripped_source_files.rules(),
    ]

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	11✔
5
6	from collections.abc import Iterable	11✔
7	from dataclasses import dataclass	11✔
8
9	from pants.backend.python.target_types import PythonSourceField	11✔
10	from pants.backend.python.util_rules import ancestor_files	11✔
11	from pants.backend.python.util_rules.ancestor_files import AncestorFilesRequest, find_ancestor_files	11✔
12	from pants.core.target_types import FileSourceField, ResourceSourceField	11✔
13	from pants.core.util_rules import source_files, stripped_source_files	11✔
14	from pants.core.util_rules.source_files import (	11✔
15	SourceFiles,
16	SourceFilesRequest,
17	determine_source_files,
18	)
19	from pants.core.util_rules.stripped_source_files import StrippedSourceFiles, strip_source_roots	11✔
20	from pants.engine.fs import EMPTY_SNAPSHOT, MergeDigests	11✔
21	from pants.engine.internals.graph import hydrate_sources	11✔
22	from pants.engine.intrinsics import digest_to_snapshot	11✔
23	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	11✔
24	from pants.engine.target import HydrateSourcesRequest, SourcesField, Target	11✔
25	from pants.engine.unions import UnionMembership	11✔
26	from pants.source.source_root import SourceRootRequest, get_source_root	11✔
27	from pants.util.logging import LogLevel	11✔
28
29
30	@dataclass(frozen=True)	11✔
31	class PythonSourceFiles:	11✔
32	"""Sources that can be introspected by Python, relative to a set of source roots.
33
34	Specifically, this will filter out to only have Python, and, optionally, resource and
35	file targets; and will add any missing `__init__.py` files to ensure that modules are
36	recognized correctly.
37
38	Use-cases that introspect Python source code (e.g., the `test, `lint`, `fmt` goals) can
39	request this type to get relevant sources that are still relative to their source roots.
40	That way the paths they report are the unstripped ones the user is familiar with.
41
42	The sources can also be imported and used by Python (e.g., for the `test` goal), but only
43	if sys.path is modified to include the source roots.
44	"""
45
46	source_files: SourceFiles	11✔
47	source_roots: tuple[str, ...] # Source roots for the specified source files.	11✔
48
49	@classmethod	11✔
50	def empty(cls) -> PythonSourceFiles:	11✔
51	return cls(SourceFiles(EMPTY_SNAPSHOT, tuple()), tuple())	11✔
52
53
54	@dataclass(frozen=True)	11✔
55	class StrippedPythonSourceFiles:	11✔
56	"""A PythonSourceFiles that has had its source roots stripped."""
57
58	stripped_source_files: StrippedSourceFiles	11✔
59
60
61	@dataclass(unsafe_hash=True)	11✔
62	class PythonSourceFilesRequest:	11✔
63	targets: tuple[Target, ...]	11✔
64	include_resources: bool	11✔
65	# NB: Setting this to True may cause Digest merge collisions, in the presence of
66	# relocated_files targets that relocate two different files to the same destination.
67	# Set this to True only if you know this cannot logically happen (e.g., if it did
68	# happen the underlying user operation would make no sense anyway).
69	# For example, if the user relocates different config files to the same location
70	# in different tests, there is no problem as long as we sandbox each test's sources
71	# separately. The user may hit this issue if they run tests in batches, but in that
72	# case they couldn't make that work even without Pants.
73	include_files: bool	11✔
74
75	def __init__(	11✔
76	self,
77	targets: Iterable[Target],
78	*,
79	include_resources: bool = True,
80	include_files: bool = False,
81	) -> None:
UNCOV 82	object.__setattr__(self, "targets", tuple(targets))	×
UNCOV 83	object.__setattr__(self, "include_resources", include_resources)	×
UNCOV 84	object.__setattr__(self, "include_files", include_files)	×
85
86	@property	11✔
87	def valid_sources_types(self) -> tuple[type[SourcesField], ...]:	11✔
88	types: list[type[SourcesField]] = [PythonSourceField]	×
89	if self.include_resources:	×
90	types.append(ResourceSourceField)	×
91	if self.include_files:	×
92	types.append(FileSourceField)	×
93	return tuple(types)	×
94
95
96	@rule(level=LogLevel.DEBUG)	11✔
97	async def prepare_python_sources(	11✔
98	request: PythonSourceFilesRequest, union_membership: UnionMembership
99	) -> PythonSourceFiles:
100	sources = await determine_source_files(	×
101	SourceFilesRequest(
102	(tgt.get(SourcesField) for tgt in request.targets),
103	for_sources_types=request.valid_sources_types,
104	enable_codegen=True,
105	)
106	)
107
108	missing_init_files = await find_ancestor_files(	×
109	AncestorFilesRequest(
110	input_files=sources.snapshot.files, requested=("__init__.py", "__init__.pyi")
111	)
112	)
113	init_injected = await digest_to_snapshot(	×
114	**implicitly(MergeDigests((sources.snapshot.digest, missing_init_files.snapshot.digest)))
115	)
116
117	# Codegen is able to generate code in any arbitrary location, unlike sources normally being
118	# rooted under the target definition. To determine source roots for these generated files, we
119	# cannot use the normal `SourceRootRequest.for_target()` and we instead must determine
120	# a source root for every individual generated file. So, we re-resolve the codegen sources here.
121	python_and_resources_targets = []	×
122	codegen_targets = []	×
123	for tgt in request.targets:	×
124	if tgt.has_field(PythonSourceField) or tgt.has_field(ResourceSourceField):	×
125	python_and_resources_targets.append(tgt)	×
126	elif tgt.get(SourcesField).can_generate(PythonSourceField, union_membership) or tgt.get(	×
127	SourcesField
128	).can_generate(ResourceSourceField, union_membership):
129	codegen_targets.append(tgt)	×
130	codegen_sources = await concurrently(	×
131	hydrate_sources(
132	HydrateSourcesRequest(
133	tgt.get(SourcesField),
134	for_sources_types=request.valid_sources_types,
135	enable_codegen=True,
136	),
137	**implicitly(),
138	)
139	for tgt in codegen_targets
140	)
141	source_root_requests = [	×
142	*(SourceRootRequest.for_target(tgt) for tgt in python_and_resources_targets),
143	*(
144	SourceRootRequest.for_file(f)
145	for sources in codegen_sources
146	for f in sources.snapshot.files
147	),
148	]
149
150	source_root_objs = await concurrently(get_source_root(req) for req in source_root_requests)	×
151	source_root_paths = {source_root_obj.path for source_root_obj in source_root_objs}	×
152	return PythonSourceFiles(	×
153	SourceFiles(init_injected, sources.unrooted_files), tuple(sorted(source_root_paths))
154	)
155
156
157	@rule(level=LogLevel.DEBUG)	11✔
158	async def strip_python_sources(python_sources: PythonSourceFiles) -> StrippedPythonSourceFiles:	11✔
159	stripped = await strip_source_roots(python_sources.source_files)	×
160	return StrippedPythonSourceFiles(stripped)	×
161
162
163	def rules():	11✔
164	return [	10✔
165	*collect_rules(),
166	*ancestor_files.rules(),
167	*source_files.rules(),
168	*stripped_source_files.rules(),
169	]

pantsbuild / pants / 19250292619

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