19381742489

Committed 15 Nov 2025 12:52AM UTC coverage: 49.706% (-30.6%) from 80.29%

Build # 19381742489

Build Type

Pull #22890

github

Committed by

web-flow

Commit Message

Merge d961abf79 into 42e1ebd41

Pull Request Pull Request #22890: Updated all python subsystem constraints to 3.14

Run Details

4 of 5 new or added lines in 5 files covered. (80.0%)

14659 existing lines in 485 files now uncovered.

31583 of 63540 relevant lines covered (49.71%)

0.79 hits per line

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

77.78

/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	1✔
5
6	import ast	1✔
7	import dataclasses	1✔
8	import os	1✔
9	from dataclasses import dataclass	1✔
10
11	import pytest	1✔
12
13	from pants.build_graph.address import Address	1✔
14	from pants.core.util_rules import system_binaries	1✔
15	from pants.core.util_rules.archive import ExtractedArchive, MaybeExtractArchiveRequest	1✔
16	from pants.core.util_rules.system_binaries import UnzipBinary	1✔
17	from pants.engine.addresses import Addresses	1✔
18	from pants.engine.fs import CreateDigest, DigestContents, PathGlobs, RemovePrefix	1✔
19	from pants.engine.internals.native_engine import Digest, Snapshot	1✔
20	from pants.engine.intrinsics import digest_to_snapshot	1✔
21	from pants.engine.process import Process, execute_process_or_raise	1✔
22	from pants.engine.rules import QueryRule, collect_rules, concurrently, implicitly, rule	1✔
23	from pants.engine.target import CoarsenedTarget, CoarsenedTargets, Targets	1✔
24	from pants.jvm.classpath import Classpath	1✔
25	from pants.jvm.compile import ClasspathEntry	1✔
26	from pants.jvm.resolve.key import CoursierResolveKey	1✔
27	from pants.testutil.rule_runner import RuleRunner	1✔
28
29
30	def maybe_skip_jdk_test(func):	1✔
31	run_jdk_tests = bool(ast.literal_eval(os.environ.get("PANTS_RUN_JDK_TESTS", "True")))	1✔
32	return pytest.mark.skipif(not run_jdk_tests, reason="Skip JDK tests")(func)	1✔
33
34
35	def expect_single_expanded_coarsened_target(	1✔
36	rule_runner: RuleRunner, address: Address
37	) -> CoarsenedTarget:
38	expanded_target = rule_runner.request(Targets, [Addresses([address])]).expect_single()	1✔
39	coarsened_targets = rule_runner.request(	1✔
40	CoarsenedTargets, [Addresses([expanded_target.address])]
41	)
42	assert len(coarsened_targets) == 1	1✔
43	return coarsened_targets[0]	1✔
44
45
46	def make_resolve(	1✔
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])])	1✔
52	return CoursierResolveKey(name=resolve_name, path=resolve_path, digest=digest)	1✔
53
54
55	@dataclass(frozen=True)	1✔
56	class RenderedClasspath:	1✔
57	"""The contents of a classpath, organized as a key per entry with its contained classfiles."""
58
59	content: dict[str, set[str]]	1✔
60
61
62	@rule	1✔
63	async def render_classpath_entry(	1✔
64	classpath_entry: ClasspathEntry, unzip_binary: UnzipBinary
65	) -> RenderedClasspath:
66	dest_dir = "dest"	×
67	process_results = await concurrently(	×
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(	×
87	digest_to_snapshot(**implicitly(RemovePrefix(pr.output_digest, dest_dir)))
88	for pr in process_results
89	)
90
91	return RenderedClasspath(	×
92	{
93	path: set(listing.files)
94	for path, listing in zip(classpath_entry.filenames, listing_snapshots)
95	}
96	)
97
98
99	@rule	1✔
100	async def render_classpath(classpath: Classpath) -> RenderedClasspath:	1✔
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():	1✔
109	return [	1✔
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(	1✔
120	rule_runner: RuleRunner, *, filename: str, digest: Digest
121	) -> Snapshot:
UNCOV 122	contents = rule_runner.request(DigestContents, [digest])	×
UNCOV 123	files_content = [content for content in contents if content.path == filename]	×
UNCOV 124	assert len(files_content) == 1	×
125
UNCOV 126	renamed_digest = rule_runner.request(	×
127	Digest, [CreateDigest([dataclasses.replace(files_content[0], path=f"{filename}.zip")])]
128	)
UNCOV 129	extracted_jar = rule_runner.request(	×
130	ExtractedArchive, [MaybeExtractArchiveRequest(renamed_digest)]
131	)
UNCOV 132	return rule_runner.request(Snapshot, [extracted_jar.digest])	×

pantsbuild / pants / 19381742489

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