25600929628

Committed 09 May 2026 12:18PM UTC coverage: 91.154% (-1.6%) from 92.787%

Build # 25600929628

Build Type

Pull #23341

github

Committed by

web-flow

Commit Message

Merge 0787d1df4 into 60371862f

Pull Request Pull Request #23341: Restore missing-entry guard in CoursierResolvedLockfile.dependencies() (regression from #22906)

Coverage Stats

87247 of 95714 relevant lines covered (91.15%)

3.87 hits per line

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

96.3

/src/python/pants/jvm/testutil.py

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

from __future__ import annotations

import ast
import dataclasses
import os
from dataclasses import dataclass

import pytest

from pants.build_graph.address import Address
from pants.core.util_rules import system_binaries
from pants.core.util_rules.archive import ExtractedArchive, MaybeExtractArchiveRequest
from pants.core.util_rules.system_binaries import UnzipBinary
from pants.engine.addresses import Addresses
from pants.engine.fs import CreateDigest, DigestContents, PathGlobs, RemovePrefix
from pants.engine.internals.native_engine import Digest, Snapshot
from pants.engine.intrinsics import digest_to_snapshot
from pants.engine.process import Process, execute_process_or_raise
from pants.engine.rules import QueryRule, collect_rules, concurrently, implicitly, rule
from pants.engine.target import CoarsenedTarget, CoarsenedTargets, Targets
from pants.jvm.classpath import Classpath
from pants.jvm.compile import ClasspathEntry
from pants.jvm.resolve.key import CoursierResolveKey
from pants.testutil.rule_runner import RuleRunner


def maybe_skip_jdk_test(func):
    run_jdk_tests = bool(ast.literal_eval(os.environ.get("PANTS_RUN_JDK_TESTS", "True")))
    return pytest.mark.skipif(not run_jdk_tests, reason="Skip JDK tests")(func)


def expect_single_expanded_coarsened_target(
    rule_runner: RuleRunner, address: Address
) -> CoarsenedTarget:
    expanded_target = rule_runner.request(Targets, [Addresses([address])]).expect_single()
    coarsened_targets = rule_runner.request(
        CoarsenedTargets, [Addresses([expanded_target.address])]
    )
    assert len(coarsened_targets) == 1
    return coarsened_targets[0]


def make_resolve(
    rule_runner: RuleRunner,
    resolve_name: str = "jvm-default",
    resolve_path: str = "3rdparty/jvm/default.lock",
) -> CoursierResolveKey:
    digest = rule_runner.request(Digest, [PathGlobs([resolve_path])])
    return CoursierResolveKey(name=resolve_name, path=resolve_path, digest=digest)


@dataclass(frozen=True)
class RenderedClasspath:
    """The contents of a classpath, organized as a key per entry with its contained classfiles."""

    content: dict[str, set[str]]


@rule
async def render_classpath_entry(
    classpath_entry: ClasspathEntry, unzip_binary: UnzipBinary
) -> RenderedClasspath:
    dest_dir = "dest"
    process_results = await concurrently(
        execute_process_or_raise(
            **implicitly(
                Process(
                    argv=[
                        unzip_binary.path,
                        "-d",
                        dest_dir,
                        filename,
                    ],
                    input_digest=classpath_entry.digest,
                    output_directories=(dest_dir,),
                    description=f"Extract {filename}",
                ),
            )
        )
        for filename in classpath_entry.filenames
    )

    listing_snapshots = await concurrently(
        digest_to_snapshot(**implicitly(RemovePrefix(pr.output_digest, dest_dir)))
        for pr in process_results
    )

    return RenderedClasspath(
        {
            path: set(listing.files)
            for path, listing in zip(classpath_entry.filenames, listing_snapshots)
        }
    )


@rule
async def render_classpath(classpath: Classpath) -> RenderedClasspath:
    rendered_classpaths = await concurrently(
        render_classpath_entry(cpe, **implicitly())
        for cpe in ClasspathEntry.closure(classpath.entries)
    )
    return RenderedClasspath({k: v for rc in rendered_classpaths for k, v in rc.content.items()})


def rules():
    return [
        *collect_rules(),
        *system_binaries.rules(),
        QueryRule(CoarsenedTargets, (Addresses,)),
        QueryRule(RenderedClasspath, (Classpath,)),
        QueryRule(RenderedClasspath, (ClasspathEntry,)),
        QueryRule(Targets, (Addresses,)),
    ]


def _get_jar_contents_snapshot(
    rule_runner: RuleRunner, *, filename: str, digest: Digest
) -> Snapshot:
    contents = rule_runner.request(DigestContents, [digest])
    files_content = [content for content in contents if content.path == filename]
    assert len(files_content) == 1

    renamed_digest = rule_runner.request(
        Digest, [CreateDigest([dataclasses.replace(files_content[0], path=f"{filename}.zip")])]
    )
    extracted_jar = rule_runner.request(
        ExtractedArchive, [MaybeExtractArchiveRequest(renamed_digest)]
    )
    return rule_runner.request(Snapshot, [extracted_jar.digest])

1	# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	9✔
5
6	import ast	9✔
7	import dataclasses	9✔
8	import os	9✔
9	from dataclasses import dataclass	9✔
10
11	import pytest	9✔
12
13	from pants.build_graph.address import Address	9✔
14	from pants.core.util_rules import system_binaries	9✔
15	from pants.core.util_rules.archive import ExtractedArchive, MaybeExtractArchiveRequest	9✔
16	from pants.core.util_rules.system_binaries import UnzipBinary	9✔
17	from pants.engine.addresses import Addresses	9✔
18	from pants.engine.fs import CreateDigest, DigestContents, PathGlobs, RemovePrefix	9✔
19	from pants.engine.internals.native_engine import Digest, Snapshot	9✔
20	from pants.engine.intrinsics import digest_to_snapshot	9✔
21	from pants.engine.process import Process, execute_process_or_raise	9✔
22	from pants.engine.rules import QueryRule, collect_rules, concurrently, implicitly, rule	9✔
23	from pants.engine.target import CoarsenedTarget, CoarsenedTargets, Targets	9✔
24	from pants.jvm.classpath import Classpath	9✔
25	from pants.jvm.compile import ClasspathEntry	9✔
26	from pants.jvm.resolve.key import CoursierResolveKey	9✔
27	from pants.testutil.rule_runner import RuleRunner	9✔
28
29
30	def maybe_skip_jdk_test(func):	9✔
31	run_jdk_tests = bool(ast.literal_eval(os.environ.get("PANTS_RUN_JDK_TESTS", "True")))	9✔
32	return pytest.mark.skipif(not run_jdk_tests, reason="Skip JDK tests")(func)	9✔
33
34
35	def expect_single_expanded_coarsened_target(	9✔
36	rule_runner: RuleRunner, address: Address
37	) -> CoarsenedTarget:
38	expanded_target = rule_runner.request(Targets, [Addresses([address])]).expect_single()	7✔
39	coarsened_targets = rule_runner.request(	7✔
40	CoarsenedTargets, [Addresses([expanded_target.address])]
41	)
42	assert len(coarsened_targets) == 1	7✔
43	return coarsened_targets[0]	7✔
44
45
46	def make_resolve(	9✔
47	rule_runner: RuleRunner,
48	resolve_name: str = "jvm-default",
49	resolve_path: str = "3rdparty/jvm/default.lock",
50	) -> CoursierResolveKey:
51	digest = rule_runner.request(Digest, [PathGlobs([resolve_path])])	7✔
52	return CoursierResolveKey(name=resolve_name, path=resolve_path, digest=digest)	7✔
53
54
55	@dataclass(frozen=True)	9✔
56	class RenderedClasspath:	9✔
57	"""The contents of a classpath, organized as a key per entry with its contained classfiles."""
58
59	content: dict[str, set[str]]	9✔
60
61
62	@rule	9✔
63	async def render_classpath_entry(	9✔
64	classpath_entry: ClasspathEntry, unzip_binary: UnzipBinary
65	) -> RenderedClasspath:
66	dest_dir = "dest"	7✔
67	process_results = await concurrently(	7✔
68	execute_process_or_raise(
69	**implicitly(
70	Process(
71	argv=[
72	unzip_binary.path,
73	"-d",
74	dest_dir,
75	filename,
76	],
77	input_digest=classpath_entry.digest,
78	output_directories=(dest_dir,),
79	description=f"Extract {filename}",
80	),
81	)
82	)
83	for filename in classpath_entry.filenames
84	)
85
86	listing_snapshots = await concurrently(	7✔
87	digest_to_snapshot(**implicitly(RemovePrefix(pr.output_digest, dest_dir)))
88	for pr in process_results
89	)
90
91	return RenderedClasspath(	7✔
92	{
93	path: set(listing.files)
94	for path, listing in zip(classpath_entry.filenames, listing_snapshots)
95	}
96	)
97
98
99	@rule	9✔
100	async def render_classpath(classpath: Classpath) -> RenderedClasspath:	9✔
101	rendered_classpaths = await concurrently(	×
102	render_classpath_entry(cpe, **implicitly())
103	for cpe in ClasspathEntry.closure(classpath.entries)
104	)
105	return RenderedClasspath({k: v for rc in rendered_classpaths for k, v in rc.content.items()})	×
106
107
108	def rules():	9✔
109	return [	8✔
110	*collect_rules(),
111	*system_binaries.rules(),
112	QueryRule(CoarsenedTargets, (Addresses,)),
113	QueryRule(RenderedClasspath, (Classpath,)),
114	QueryRule(RenderedClasspath, (ClasspathEntry,)),
115	QueryRule(Targets, (Addresses,)),
116	]
117
118
119	def _get_jar_contents_snapshot(	9✔
120	rule_runner: RuleRunner, *, filename: str, digest: Digest
121	) -> Snapshot:
122	contents = rule_runner.request(DigestContents, [digest])	2✔
123	files_content = [content for content in contents if content.path == filename]	2✔
124	assert len(files_content) == 1	2✔
125
126	renamed_digest = rule_runner.request(	2✔
127	Digest, [CreateDigest([dataclasses.replace(files_content[0], path=f"{filename}.zip")])]
128	)
129	extracted_jar = rule_runner.request(	2✔
130	ExtractedArchive, [MaybeExtractArchiveRequest(renamed_digest)]
131	)
132	return rule_runner.request(Snapshot, [extracted_jar.digest])	2✔

pantsbuild / pants / 25600929628

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