18812500213

Committed 26 Oct 2025 03:42AM UTC coverage: 80.284% (+0.005%) from 80.279%

Build # 18812500213

Build Type

Pull #22804

github

Committed by

web-flow

Commit Message

Merge 2a56fdb46 into 4834308dc

Pull Request Pull Request #22804: test_shell_command: use correct default cache scope for a test's environment

Run Details

29 of 31 new or added lines in 2 files covered. (93.55%)

1314 existing lines in 64 files now uncovered.

77900 of 97030 relevant lines covered (80.28%)

3.35 hits per line

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

73.17

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

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

from __future__ import annotations

import os
from dataclasses import dataclass

from pants.engine.fs import EMPTY_SNAPSHOT, PathGlobs, Snapshot
from pants.engine.intrinsics import digest_to_snapshot, get_digest_contents
from pants.engine.rules import collect_rules, implicitly, rule


@dataclass(frozen=True)
class AncestorFilesRequest:
    """A request for ancestor files of the given names.

    "Ancestor files" means all files with one of the given names that are siblings of, or in parent
    directories of, a `.py` or `.pyi` file in the input_files.
    """

    input_files: tuple[str, ...]
    requested: tuple[str, ...]
    ignore_empty_files: bool = False


@dataclass(frozen=True)
class AncestorFiles:
    """Any ancestor files found."""

    snapshot: Snapshot


def putative_ancestor_files(input_files: tuple[str, ...], requested: tuple[str, ...]) -> set[str]:
    """Return the paths of potentially missing ancestor files.

    NB: The sources are expected to not have had their source roots stripped.
    Therefore this function will consider superfluous files at and above the source roots,
    (e.g., src/python/<name>, src/<name>). It is the caller's responsibility to filter these
    out if necessary.
    """
    packages: set[str] = set()
    for input_file in input_files:
        if not input_file.endswith((".py", ".pyi")):
            continue
        pkg_dir = os.path.dirname(input_file)
        if pkg_dir in packages:
            continue
        package = ""
        packages.add(package)
        for component in pkg_dir.split(os.sep):
            package = os.path.join(package, component)
            packages.add(package)

    return {
        os.path.join(package, requested_f) for package in packages for requested_f in requested
    } - set(input_files)


@rule
async def find_ancestor_files(request: AncestorFilesRequest) -> AncestorFiles:
    putative = putative_ancestor_files(request.input_files, request.requested)
    if not putative:
        return AncestorFiles(EMPTY_SNAPSHOT)

    # NB: This will intentionally _not_ error on any unmatched globs.
    globs = PathGlobs(putative)
    if request.ignore_empty_files:
        digest_contents = await get_digest_contents(**implicitly({globs: PathGlobs}))
        snapshot = await digest_to_snapshot(
            **implicitly(PathGlobs([fc.path for fc in digest_contents if fc.content.strip()]))
        )
    else:
        snapshot = await digest_to_snapshot(**implicitly({globs: PathGlobs}))

    return AncestorFiles(snapshot)


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 __future__ import annotations	12✔
5
6	import os	12✔
7	from dataclasses import dataclass	12✔
8
9	from pants.engine.fs import EMPTY_SNAPSHOT, PathGlobs, Snapshot	12✔
10	from pants.engine.intrinsics import digest_to_snapshot, get_digest_contents	12✔
11	from pants.engine.rules import collect_rules, implicitly, rule	12✔
12
13
14	@dataclass(frozen=True)	12✔
15	class AncestorFilesRequest:	12✔
16	"""A request for ancestor files of the given names.
17
18	"Ancestor files" means all files with one of the given names that are siblings of, or in parent
19	directories of, a `.py` or `.pyi` file in the input_files.
20	"""
21
22	input_files: tuple[str, ...]	12✔
23	requested: tuple[str, ...]	12✔
24	ignore_empty_files: bool = False	12✔
25
26
27	@dataclass(frozen=True)	12✔
28	class AncestorFiles:	12✔
29	"""Any ancestor files found."""
30
31	snapshot: Snapshot	12✔
32
33
34	def putative_ancestor_files(input_files: tuple[str, ...], requested: tuple[str, ...]) -> set[str]:	12✔
35	"""Return the paths of potentially missing ancestor files.
36
37	NB: The sources are expected to not have had their source roots stripped.
38	Therefore this function will consider superfluous files at and above the source roots,
39	(e.g., src/python/<name>, src/<name>). It is the caller's responsibility to filter these
40	out if necessary.
41	"""
UNCOV 42	packages: set[str] = set()	1✔
UNCOV 43	for input_file in input_files:	1✔
UNCOV 44	if not input_file.endswith((".py", ".pyi")):	1✔
45	continue	×
UNCOV 46	pkg_dir = os.path.dirname(input_file)	1✔
UNCOV 47	if pkg_dir in packages:	1✔
48	continue	×
UNCOV 49	package = ""	1✔
UNCOV 50	packages.add(package)	1✔
UNCOV 51	for component in pkg_dir.split(os.sep):	1✔
UNCOV 52	package = os.path.join(package, component)	1✔
UNCOV 53	packages.add(package)	1✔
54
UNCOV 55	return {	1✔
56	os.path.join(package, requested_f) for package in packages for requested_f in requested
57	} - set(input_files)
58
59
60	@rule	12✔
61	async def find_ancestor_files(request: AncestorFilesRequest) -> AncestorFiles:	12✔
62	putative = putative_ancestor_files(request.input_files, request.requested)	×
63	if not putative:	×
64	return AncestorFiles(EMPTY_SNAPSHOT)	×
65
66	# NB: This will intentionally _not_ error on any unmatched globs.
67	globs = PathGlobs(putative)	×
68	if request.ignore_empty_files:	×
69	digest_contents = await get_digest_contents(**implicitly({globs: PathGlobs}))	×
70	snapshot = await digest_to_snapshot(	×
71	**implicitly(PathGlobs([fc.path for fc in digest_contents if fc.content.strip()]))
72	)
73	else:
74	snapshot = await digest_to_snapshot(**implicitly({globs: PathGlobs}))	×
75
76	return AncestorFiles(snapshot)	×
77
78
79	def rules():	12✔
80	return collect_rules()	11✔

pantsbuild / pants / 18812500213

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