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

51.48

/src/python/pants/source/source_root.py

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

from __future__ import annotations

import itertools
import logging
import os
from collections.abc import Iterable
from dataclasses import dataclass
from pathlib import PurePath

from pants.build_graph.address import Address
from pants.engine.collection import DeduplicatedCollection
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.fs import PathGlobs
from pants.engine.intrinsics import path_globs_to_paths
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import Target
from pants.option.option_types import StrListOption
from pants.option.subsystem import Subsystem
from pants.util.docutil import doc_url
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.memo import memoized_method
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)


@dataclass(frozen=True, order=True)
class SourceRoot:
    # Relative path from the buildroot.  Note that a source root at the buildroot
    # is represented as ".".
    path: str


@dataclass(frozen=True)
class OptionalSourceRoot:
    source_root: SourceRoot | None


class SourceRootError(Exception):
    """An error related to SourceRoot computation."""

    def __init__(self, msg: str):
        super().__init__(
            f"{msg}See {doc_url('docs/using-pants/key-concepts/source-roots')} for how to define source roots."
        )


class InvalidSourceRootPatternError(SourceRootError):
    """Indicates an invalid pattern was provided."""


class InvalidMarkerFileError(SourceRootError):
    """Indicates an invalid marker file was provided."""


class NoSourceRootError(SourceRootError):
    """Indicates we failed to map a source file to a source root."""

    def __init__(self, path: str | PurePath, extra_msg: str = ""):
        super().__init__(f"No source root found for `{path}`. {extra_msg}")


# We perform pattern matching against absolute paths, where "/" represents the repo root.
_repo_root = PurePath(os.path.sep)


@dataclass(frozen=True)
class SourceRootPatternMatcher:
    root_patterns: tuple[str, ...]

    def __post_init__(self) -> None:
        for root_pattern in self.root_patterns:
            if ".." in root_pattern.split(os.path.sep):
                raise InvalidSourceRootPatternError(
                    f"`..` disallowed in source root pattern: {root_pattern}."
                )

    def get_patterns(self) -> tuple[str, ...]:
        return tuple(self.root_patterns)

    def matches_root_patterns(self, relpath: PurePath) -> bool:
        """Does this putative root match a pattern?"""
        # Note: This is currently O(n) where n is the number of patterns, which
        # we expect to be small.  We can optimize if it becomes necessary.
        putative_root = _repo_root / relpath
        for pattern in self.root_patterns:
            if putative_root.match(pattern):
                return True
        return False


class SourceRootConfig(Subsystem):
    options_scope = "source"
    help = "Configuration for roots of source trees."

    DEFAULT_ROOT_PATTERNS = [
        "/",
        "src",
        "src/python",
        "src/py",
        "src/thrift",
        "src/protobuf",
        "src/protos",
        "src/scala",
        "src/java",
    ]

    root_patterns = StrListOption(
        default=DEFAULT_ROOT_PATTERNS,
        help=softwrap(
            f"""
            A list of source root suffixes.

            A directory with this suffix will be considered a potential source root.
            E.g., `src/python` will match `<buildroot>/src/python`, `<buildroot>/project1/src/python`
            etc.

            Prepend a `/` to anchor the match at the buildroot.
            E.g., `/src/python` will match `<buildroot>/src/python` but not `<buildroot>/project1/src/python`.

            A `*` wildcard will match a single path segment,
            E.g., `src/*` will match `<buildroot>/src/python` and `<buildroot>/src/rust`.

            Use `/` to signify that the buildroot itself is a source root.

            See {doc_url("docs/using-pants/key-concepts/source-roots")}.
            """
        ),
        advanced=True,
        metavar='["pattern1", "pattern2", ...]',
    )
    marker_filenames = StrListOption(
        help=softwrap(
            """
            The presence of a file of this name in a directory indicates that the directory
            is a source root. The content of the file doesn't matter, and may be empty.
            Useful when you can't or don't wish to centrally enumerate source roots via
            `root_patterns`.
            """
        ),
        advanced=True,
        metavar="filename",
    )

    @memoized_method
    def get_pattern_matcher(self) -> SourceRootPatternMatcher:
        return SourceRootPatternMatcher(self.root_patterns)


@dataclass(frozen=True)
class SourceRootsRequest:
    """Find the source roots for the given files and/or dirs."""

    files: tuple[PurePath, ...]
    dirs: tuple[PurePath, ...]

    def __init__(self, files: Iterable[PurePath], dirs: Iterable[PurePath]) -> None:
        object.__setattr__(self, "files", tuple(sorted(files)))
        object.__setattr__(self, "dirs", tuple(sorted(dirs)))

        self.__post_init__()

    def __post_init__(self) -> None:
        for path in itertools.chain(self.files, self.dirs):
            if ".." in str(path).split(os.path.sep):
                raise ValueError(f"SourceRootRequest cannot contain `..` segment: {path}")
            if path.is_absolute():
                raise ValueError(f"SourceRootRequest path must be relative: {path}")

    @classmethod
    def for_files(cls, file_paths: Iterable[str]) -> SourceRootsRequest:
        """Create a request for the source root for the given file."""
        return cls({PurePath(file_path) for file_path in file_paths}, ())


@dataclass(frozen=True)
class SourceRootRequest(EngineAwareParameter):
    """Find the source root for the given path.

    If you have multiple paths, particularly if many of them share parent directories, you'll get
    better performance with a `SourceRootsRequest` (see above) instead.
    """

    path: PurePath

    def __post_init__(self) -> None:
        if ".." in str(self.path).split(os.path.sep):
            raise ValueError(f"SourceRootRequest cannot contain `..` segment: {self.path}")
        if self.path.is_absolute():
            raise ValueError(f"SourceRootRequest path must be relative: {self.path}")

    @classmethod
    def for_file(cls, file_path: str) -> SourceRootRequest:
        """Create a request for the source root for the given file."""
        # The file itself cannot be a source root, so we may as well start the search
        # from its enclosing directory, and save on some superfluous checking.
        return cls(PurePath(file_path).parent)

    @classmethod
    def for_address(cls, address: Address) -> SourceRootRequest:
        # Note that we don't use for_file() here because the spec_path is a directory.
        return cls(PurePath(address.spec_path))

    @classmethod
    def for_target(cls, target: Target) -> SourceRootRequest:
        return cls.for_address(target.address)

    def debug_hint(self) -> str:
        return str(self.path)


@dataclass(frozen=True)
class SourceRootsResult:
    path_to_root: FrozenDict[PurePath, SourceRoot]


@dataclass(frozen=True)
class OptionalSourceRootsResult:
    path_to_optional_root: FrozenDict[PurePath, OptionalSourceRoot]


@rule
async def get_optional_source_root(
    source_root_request: SourceRootRequest, source_root_config: SourceRootConfig
) -> OptionalSourceRoot:
    """Rule to request a SourceRoot that may not exist."""
    pattern_matcher = source_root_config.get_pattern_matcher()
    path = source_root_request.path

    # Check if the requested path itself is a source root.

    # A) Does it match a pattern?
    if pattern_matcher.matches_root_patterns(path):
        return OptionalSourceRoot(SourceRoot(str(path)))

    # B) Does it contain a marker file?
    marker_filenames = source_root_config.marker_filenames
    if marker_filenames:
        for marker_filename in marker_filenames:
            if (
                os.path.basename(marker_filename) != marker_filename
                or "*" in marker_filename
                or "!" in marker_filename
            ):
                raise InvalidMarkerFileError(
                    f"Marker filename must be a base name: {marker_filename}"
                )
        paths = await path_globs_to_paths(PathGlobs([str(path / mf) for mf in marker_filenames]))
        if len(paths.files) > 0:
            return OptionalSourceRoot(SourceRoot(str(path)))

    # The requested path itself is not a source root, but maybe its parent is.
    if str(path) != ".":
        return await get_optional_source_root(SourceRootRequest(path.parent), **implicitly())

    # The requested path is not under a source root.
    return OptionalSourceRoot(None)


@rule
async def get_optional_source_roots(
    source_roots_request: SourceRootsRequest,
) -> OptionalSourceRootsResult:
    """Rule to request source roots that may not exist."""
    # A file cannot be a source root, so request for its parent.
    # In the typical case, where we have multiple files with the same parent, this can
    # dramatically cut down on the number of engine requests.
    dirs: set[PurePath] = set(source_roots_request.dirs)
    file_to_dir: dict[PurePath, PurePath] = {
        file: file.parent for file in source_roots_request.files
    }
    dirs.update(file_to_dir.values())

    roots = await concurrently(
        get_optional_source_root(SourceRootRequest(d), **implicitly()) for d in dirs
    )
    dir_to_root = dict(zip(dirs, roots))

    path_to_optional_root: dict[PurePath, OptionalSourceRoot] = {}
    for d in source_roots_request.dirs:
        path_to_optional_root[d] = dir_to_root[d]
    for f, d in file_to_dir.items():
        path_to_optional_root[f] = dir_to_root[d]

    return OptionalSourceRootsResult(path_to_optional_root=FrozenDict(path_to_optional_root))


@rule
async def get_source_roots(source_roots_request: SourceRootsRequest) -> SourceRootsResult:
    """Convenience rule to allow callers to request SourceRoots that must exist.

    That way callers don't have to unpack OptionalSourceRoots if they know they expect a SourceRoot
    to exist and are willing to error if it doesn't.
    """
    osrr = await get_optional_source_roots(source_roots_request)
    path_to_root = {}
    for path, osr in osrr.path_to_optional_root.items():
        if osr.source_root is None:
            raise NoSourceRootError(path)
        path_to_root[path] = osr.source_root
    return SourceRootsResult(path_to_root=FrozenDict(path_to_root))


@rule
async def get_source_root(source_root_request: SourceRootRequest) -> SourceRoot:
    """Convenience rule to allow callers to request a SourceRoot directly.

    That way callers don't have to unpack an OptionalSourceRoot if they know they expect a
    SourceRoot to exist and are willing to error if it doesn't.
    """
    optional_source_root = await get_optional_source_root(source_root_request, **implicitly())
    if optional_source_root.source_root is None:
        raise NoSourceRootError(source_root_request.path)
    return optional_source_root.source_root


class AllSourceRoots(DeduplicatedCollection[SourceRoot]):
    sort_input = True


@rule(desc="Compute all source roots", level=LogLevel.DEBUG)
async def all_roots(source_root_config: SourceRootConfig) -> AllSourceRoots:
    source_root_pattern_matcher = source_root_config.get_pattern_matcher()

    # Create globs corresponding to all source root patterns.
    pattern_matches: set[str] = set()
    for path in source_root_pattern_matcher.get_patterns():
        if path == "/":
            pattern_matches.add("**")
        elif path.startswith("/"):
            pattern_matches.add(f"{path[1:]}/")
        else:
            pattern_matches.add(f"**/{path}/")

    # Create globs for any marker files.
    marker_file_matches: set[str] = set()
    for marker_filename in source_root_config.marker_filenames:
        marker_file_matches.add(f"**/{marker_filename}")

    # Match the patterns against actual files, to find the roots that actually exist.
    pattern_paths, marker_paths = await concurrently(
        path_globs_to_paths(PathGlobs(globs=sorted(pattern_matches))),
        path_globs_to_paths(PathGlobs(globs=sorted(marker_file_matches))),
    )

    responses = await concurrently(
        itertools.chain(
            (
                get_optional_source_root(SourceRootRequest(PurePath(d)), **implicitly())
                for d in pattern_paths.dirs
            ),
            # We don't technically need to issue a SourceRootRequest for the marker files,
            # since we know that their immediately enclosing dir is a source root by definition.
            # However we may as well verify this formally, so that we're not replicating that
            # logic here.
            (
                get_optional_source_root(SourceRootRequest(PurePath(f)), **implicitly())
                for f in marker_paths.files
            ),
        )
    )
    all_source_roots = {
        response.source_root for response in responses if response.source_root is not None
    }
    return AllSourceRoots(all_source_roots)


def rules():
    return collect_rules()

1	# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	1✔
5
6	import itertools	1✔
7	import logging	1✔
8	import os	1✔
9	from collections.abc import Iterable	1✔
10	from dataclasses import dataclass	1✔
11	from pathlib import PurePath	1✔
12
13	from pants.build_graph.address import Address	1✔
14	from pants.engine.collection import DeduplicatedCollection	1✔
15	from pants.engine.engine_aware import EngineAwareParameter	1✔
16	from pants.engine.fs import PathGlobs	1✔
17	from pants.engine.intrinsics import path_globs_to_paths	1✔
18	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	1✔
19	from pants.engine.target import Target	1✔
20	from pants.option.option_types import StrListOption	1✔
21	from pants.option.subsystem import Subsystem	1✔
22	from pants.util.docutil import doc_url	1✔
23	from pants.util.frozendict import FrozenDict	1✔
24	from pants.util.logging import LogLevel	1✔
25	from pants.util.memo import memoized_method	1✔
26	from pants.util.strutil import softwrap	1✔
27
28	logger = logging.getLogger(__name__)	1✔
29
30
31	@dataclass(frozen=True, order=True)	1✔
32	class SourceRoot:	1✔
33	# Relative path from the buildroot. Note that a source root at the buildroot
34	# is represented as ".".
35	path: str	1✔
36
37
38	@dataclass(frozen=True)	1✔
39	class OptionalSourceRoot:	1✔
40	source_root: SourceRoot \| None	1✔
41
42
43	class SourceRootError(Exception):	1✔
44	"""An error related to SourceRoot computation."""
45
46	def __init__(self, msg: str):	1✔
47	super().__init__(	×
48	f"{msg}See {doc_url('docs/using-pants/key-concepts/source-roots')} for how to define source roots."
49	)
50
51
52	class InvalidSourceRootPatternError(SourceRootError):	1✔
53	"""Indicates an invalid pattern was provided."""
54
55
56	class InvalidMarkerFileError(SourceRootError):	1✔
57	"""Indicates an invalid marker file was provided."""
58
59
60	class NoSourceRootError(SourceRootError):	1✔
61	"""Indicates we failed to map a source file to a source root."""
62
63	def __init__(self, path: str \| PurePath, extra_msg: str = ""):	1✔
64	super().__init__(f"No source root found for `{path}`. {extra_msg}")	×
65
66
67	# We perform pattern matching against absolute paths, where "/" represents the repo root.
68	_repo_root = PurePath(os.path.sep)	1✔
69
70
71	@dataclass(frozen=True)	1✔
72	class SourceRootPatternMatcher:	1✔
73	root_patterns: tuple[str, ...]	1✔
74
75	def __post_init__(self) -> None:	1✔
UNCOV 76	for root_pattern in self.root_patterns:	×
UNCOV 77	if ".." in root_pattern.split(os.path.sep):	×
78	raise InvalidSourceRootPatternError(	×
79	f"`..` disallowed in source root pattern: {root_pattern}."
80	)
81
82	def get_patterns(self) -> tuple[str, ...]:	1✔
UNCOV 83	return tuple(self.root_patterns)	×
84
85	def matches_root_patterns(self, relpath: PurePath) -> bool:	1✔
86	"""Does this putative root match a pattern?"""
87	# Note: This is currently O(n) where n is the number of patterns, which
88	# we expect to be small. We can optimize if it becomes necessary.
UNCOV 89	putative_root = _repo_root / relpath	×
UNCOV 90	for pattern in self.root_patterns:	×
UNCOV 91	if putative_root.match(pattern):	×
UNCOV 92	return True	×
UNCOV 93	return False	×
94
95
96	class SourceRootConfig(Subsystem):	1✔
97	options_scope = "source"	1✔
98	help = "Configuration for roots of source trees."	1✔
99
100	DEFAULT_ROOT_PATTERNS = [	1✔
101	"/",
102	"src",
103	"src/python",
104	"src/py",
105	"src/thrift",
106	"src/protobuf",
107	"src/protos",
108	"src/scala",
109	"src/java",
110	]
111
112	root_patterns = StrListOption(	1✔
113	default=DEFAULT_ROOT_PATTERNS,
114	help=softwrap(
115	f"""
116	A list of source root suffixes.
117
118	A directory with this suffix will be considered a potential source root.
119	E.g., `src/python` will match `<buildroot>/src/python`, `<buildroot>/project1/src/python`
120	etc.
121
122	Prepend a `/` to anchor the match at the buildroot.
123	E.g., `/src/python` will match `<buildroot>/src/python` but not `<buildroot>/project1/src/python`.
124
125	A `*` wildcard will match a single path segment,
126	E.g., `src/*` will match `<buildroot>/src/python` and `<buildroot>/src/rust`.
127
128	Use `/` to signify that the buildroot itself is a source root.
129
130	See {doc_url("docs/using-pants/key-concepts/source-roots")}.
131	"""
132	),
133	advanced=True,
134	metavar='["pattern1", "pattern2", ...]',
135	)
136	marker_filenames = StrListOption(	1✔
137	help=softwrap(
138	"""
139	The presence of a file of this name in a directory indicates that the directory
140	is a source root. The content of the file doesn't matter, and may be empty.
141	Useful when you can't or don't wish to centrally enumerate source roots via
142	`root_patterns`.
143	"""
144	),
145	advanced=True,
146	metavar="filename",
147	)
148
149	@memoized_method	1✔
150	def get_pattern_matcher(self) -> SourceRootPatternMatcher:	1✔
UNCOV 151	return SourceRootPatternMatcher(self.root_patterns)	×
152
153
154	@dataclass(frozen=True)	1✔
155	class SourceRootsRequest:	1✔
156	"""Find the source roots for the given files and/or dirs."""
157
158	files: tuple[PurePath, ...]	1✔
159	dirs: tuple[PurePath, ...]	1✔
160
161	def __init__(self, files: Iterable[PurePath], dirs: Iterable[PurePath]) -> None:	1✔
UNCOV 162	object.__setattr__(self, "files", tuple(sorted(files)))	×
UNCOV 163	object.__setattr__(self, "dirs", tuple(sorted(dirs)))	×
164
UNCOV 165	self.__post_init__()	×
166
167	def __post_init__(self) -> None:	1✔
UNCOV 168	for path in itertools.chain(self.files, self.dirs):	×
UNCOV 169	if ".." in str(path).split(os.path.sep):	×
170	raise ValueError(f"SourceRootRequest cannot contain `..` segment: {path}")	×
UNCOV 171	if path.is_absolute():	×
172	raise ValueError(f"SourceRootRequest path must be relative: {path}")	×
173
174	@classmethod	1✔
175	def for_files(cls, file_paths: Iterable[str]) -> SourceRootsRequest:	1✔
176	"""Create a request for the source root for the given file."""
177	return cls({PurePath(file_path) for file_path in file_paths}, ())	×
178
179
180	@dataclass(frozen=True)	1✔
181	class SourceRootRequest(EngineAwareParameter):	1✔
182	"""Find the source root for the given path.
183
184	If you have multiple paths, particularly if many of them share parent directories, you'll get
185	better performance with a `SourceRootsRequest` (see above) instead.
186	"""
187
188	path: PurePath	1✔
189
190	def __post_init__(self) -> None:	1✔
UNCOV 191	if ".." in str(self.path).split(os.path.sep):	×
UNCOV 192	raise ValueError(f"SourceRootRequest cannot contain `..` segment: {self.path}")	×
UNCOV 193	if self.path.is_absolute():	×
194	raise ValueError(f"SourceRootRequest path must be relative: {self.path}")	×
195
196	@classmethod	1✔
197	def for_file(cls, file_path: str) -> SourceRootRequest:	1✔
198	"""Create a request for the source root for the given file."""
199	# The file itself cannot be a source root, so we may as well start the search
200	# from its enclosing directory, and save on some superfluous checking.
201	return cls(PurePath(file_path).parent)	×
202
203	@classmethod	1✔
204	def for_address(cls, address: Address) -> SourceRootRequest:	1✔
205	# Note that we don't use for_file() here because the spec_path is a directory.
206	return cls(PurePath(address.spec_path))	×
207
208	@classmethod	1✔
209	def for_target(cls, target: Target) -> SourceRootRequest:	1✔
210	return cls.for_address(target.address)	×
211
212	def debug_hint(self) -> str:	1✔
213	return str(self.path)	×
214
215
216	@dataclass(frozen=True)	1✔
217	class SourceRootsResult:	1✔
218	path_to_root: FrozenDict[PurePath, SourceRoot]	1✔
219
220
221	@dataclass(frozen=True)	1✔
222	class OptionalSourceRootsResult:	1✔
223	path_to_optional_root: FrozenDict[PurePath, OptionalSourceRoot]	1✔
224
225
226	@rule	1✔
227	async def get_optional_source_root(	1✔
228	source_root_request: SourceRootRequest, source_root_config: SourceRootConfig
229	) -> OptionalSourceRoot:
230	"""Rule to request a SourceRoot that may not exist."""
UNCOV 231	pattern_matcher = source_root_config.get_pattern_matcher()	×
UNCOV 232	path = source_root_request.path	×
233
234	# Check if the requested path itself is a source root.
235
236	# A) Does it match a pattern?
UNCOV 237	if pattern_matcher.matches_root_patterns(path):	×
UNCOV 238	return OptionalSourceRoot(SourceRoot(str(path)))	×
239
240	# B) Does it contain a marker file?
UNCOV 241	marker_filenames = source_root_config.marker_filenames	×
UNCOV 242	if marker_filenames:	×
UNCOV 243	for marker_filename in marker_filenames:	×
UNCOV 244	if (	×
245	os.path.basename(marker_filename) != marker_filename
246	or "*" in marker_filename
247	or "!" in marker_filename
248	):
249	raise InvalidMarkerFileError(	×
250	f"Marker filename must be a base name: {marker_filename}"
251	)
UNCOV 252	paths = await path_globs_to_paths(PathGlobs([str(path / mf) for mf in marker_filenames]))	×
UNCOV 253	if len(paths.files) > 0:	×
UNCOV 254	return OptionalSourceRoot(SourceRoot(str(path)))	×
255
256	# The requested path itself is not a source root, but maybe its parent is.
UNCOV 257	if str(path) != ".":	×
UNCOV 258	return await get_optional_source_root(SourceRootRequest(path.parent), **implicitly())	×
259
260	# The requested path is not under a source root.
UNCOV 261	return OptionalSourceRoot(None)	×
262
263
264	@rule	1✔
265	async def get_optional_source_roots(	1✔
266	source_roots_request: SourceRootsRequest,
267	) -> OptionalSourceRootsResult:
268	"""Rule to request source roots that may not exist."""
269	# A file cannot be a source root, so request for its parent.
270	# In the typical case, where we have multiple files with the same parent, this can
271	# dramatically cut down on the number of engine requests.
272	dirs: set[PurePath] = set(source_roots_request.dirs)	×
273	file_to_dir: dict[PurePath, PurePath] = {	×
274	file: file.parent for file in source_roots_request.files
275	}
276	dirs.update(file_to_dir.values())	×
277
278	roots = await concurrently(	×
279	get_optional_source_root(SourceRootRequest(d), **implicitly()) for d in dirs
280	)
281	dir_to_root = dict(zip(dirs, roots))	×
282
283	path_to_optional_root: dict[PurePath, OptionalSourceRoot] = {}	×
284	for d in source_roots_request.dirs:	×
285	path_to_optional_root[d] = dir_to_root[d]	×
286	for f, d in file_to_dir.items():	×
287	path_to_optional_root[f] = dir_to_root[d]	×
288
289	return OptionalSourceRootsResult(path_to_optional_root=FrozenDict(path_to_optional_root))	×
290
291
292	@rule	1✔
293	async def get_source_roots(source_roots_request: SourceRootsRequest) -> SourceRootsResult:	1✔
294	"""Convenience rule to allow callers to request SourceRoots that must exist.
295
296	That way callers don't have to unpack OptionalSourceRoots if they know they expect a SourceRoot
297	to exist and are willing to error if it doesn't.
298	"""
299	osrr = await get_optional_source_roots(source_roots_request)	×
300	path_to_root = {}	×
301	for path, osr in osrr.path_to_optional_root.items():	×
302	if osr.source_root is None:	×
303	raise NoSourceRootError(path)	×
304	path_to_root[path] = osr.source_root	×
305	return SourceRootsResult(path_to_root=FrozenDict(path_to_root))	×
306
307
308	@rule	1✔
309	async def get_source_root(source_root_request: SourceRootRequest) -> SourceRoot:	1✔
310	"""Convenience rule to allow callers to request a SourceRoot directly.
311
312	That way callers don't have to unpack an OptionalSourceRoot if they know they expect a
313	SourceRoot to exist and are willing to error if it doesn't.
314	"""
315	optional_source_root = await get_optional_source_root(source_root_request, **implicitly())	×
316	if optional_source_root.source_root is None:	×
317	raise NoSourceRootError(source_root_request.path)	×
318	return optional_source_root.source_root	×
319
320
321	class AllSourceRoots(DeduplicatedCollection[SourceRoot]):	1✔
322	sort_input = True	1✔
323
324
325	@rule(desc="Compute all source roots", level=LogLevel.DEBUG)	1✔
326	async def all_roots(source_root_config: SourceRootConfig) -> AllSourceRoots:	1✔
UNCOV 327	source_root_pattern_matcher = source_root_config.get_pattern_matcher()	×
328
329	# Create globs corresponding to all source root patterns.
UNCOV 330	pattern_matches: set[str] = set()	×
UNCOV 331	for path in source_root_pattern_matcher.get_patterns():	×
UNCOV 332	if path == "/":	×
UNCOV 333	pattern_matches.add("**")	×
UNCOV 334	elif path.startswith("/"):	×
335	pattern_matches.add(f"{path[1:]}/")	×
336	else:
UNCOV 337	pattern_matches.add(f"**/{path}/")	×
338
339	# Create globs for any marker files.
UNCOV 340	marker_file_matches: set[str] = set()	×
UNCOV 341	for marker_filename in source_root_config.marker_filenames:	×
342	marker_file_matches.add(f"**/{marker_filename}")	×
343
344	# Match the patterns against actual files, to find the roots that actually exist.
UNCOV 345	pattern_paths, marker_paths = await concurrently(	×
346	path_globs_to_paths(PathGlobs(globs=sorted(pattern_matches))),
347	path_globs_to_paths(PathGlobs(globs=sorted(marker_file_matches))),
348	)
349
UNCOV 350	responses = await concurrently(	×
351	itertools.chain(
352	(
353	get_optional_source_root(SourceRootRequest(PurePath(d)), **implicitly())
354	for d in pattern_paths.dirs
355	),
356	# We don't technically need to issue a SourceRootRequest for the marker files,
357	# since we know that their immediately enclosing dir is a source root by definition.
358	# However we may as well verify this formally, so that we're not replicating that
359	# logic here.
360	(
361	get_optional_source_root(SourceRootRequest(PurePath(f)), **implicitly())
362	for f in marker_paths.files
363	),
364	)
365	)
UNCOV 366	all_source_roots = {	×
367	response.source_root for response in responses if response.source_root is not None
368	}
UNCOV 369	return AllSourceRoots(all_source_roots)	×
370
371
372	def rules():	1✔
UNCOV 373	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