26342152999

Committed 23 May 2026 07:59PM UTC coverage: 91.165% (-1.6%) from 92.792%

Build # 26342152999

Build Type

push

github

Committed by

web-flow

Commit Message

Run Linux ARM CI on Depot runners (#23363)

RunsOn is deprecating their v2 stack, and rather than migrate
to v3 we should use the resources kindly donated by Depot.

GitHub also now has Linux ARM runners, should we need them.

Coverage Stats

87305 of 95766 relevant lines covered (91.16%)

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 / 26342152999

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