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

0.0

/src/python/pants/backend/shell/shunit2_test_runner.py

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

import os
import re
from dataclasses import dataclass
from typing import Any

from pants.backend.shell.subsystems.shell_setup import ShellSetup
from pants.backend.shell.subsystems.shunit2 import Shunit2
from pants.backend.shell.target_types import (
    ShellSourceField,
    Shunit2Shell,
    Shunit2ShellField,
    Shunit2TestsGeneratorTarget,
    Shunit2TestSourceField,
    Shunit2TestTimeoutField,
    SkipShunit2TestsField,
)
from pants.core.goals.test import (
    BuildPackageDependenciesRequest,
    RuntimePackageDependenciesField,
    TestDebugRequest,
    TestExtraEnv,
    TestFieldSet,
    TestRequest,
    TestResult,
    TestSubsystem,
    build_runtime_package_dependencies,
)
from pants.core.target_types import FileSourceField, ResourceSourceField
from pants.core.util_rules.external_tool import download_external_tool
from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files
from pants.core.util_rules.system_binaries import (
    BinaryNotFoundError,
    BinaryPath,
    BinaryPathRequest,
    find_binary,
)
from pants.engine.addresses import Address
from pants.engine.fs import CreateDigest, FileContent, MergeDigests
from pants.engine.internals.graph import transitive_targets as transitive_targets_get
from pants.engine.intrinsics import (
    create_digest,
    execute_process_with_retry,
    get_digest_contents,
    merge_digests,
)
from pants.engine.platform import Platform
from pants.engine.process import InteractiveProcess, Process, ProcessCacheScope, ProcessWithRetries
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import SourcesField, Target, TransitiveTargetsRequest
from pants.option.global_options import GlobalOptions
from pants.util.docutil import bin_name
from pants.util.logging import LogLevel


@dataclass(frozen=True)
class Shunit2FieldSet(TestFieldSet):
    required_fields = (Shunit2TestSourceField,)

    sources: Shunit2TestSourceField
    timeout: Shunit2TestTimeoutField
    shell: Shunit2ShellField
    runtime_package_dependencies: RuntimePackageDependenciesField

    @classmethod
    def opt_out(cls, tgt: Target) -> bool:
        return tgt.get(SkipShunit2TestsField).value


class Shunit2TestRequest(TestRequest):
    tool_subsystem = Shunit2  # type: ignore[assignment]
    field_set_type = Shunit2FieldSet
    supports_debug = True


@dataclass(frozen=True)
class TestSetupRequest:
    field_set: Shunit2FieldSet


@dataclass(frozen=True)
class TestSetup:
    process: Process


class ShellNotConfigured(Exception):
    pass


SOURCE_SHUNIT2_REGEX = re.compile(rb"(?:source|\.)\s+[.${}/'\"\w]*shunit2\b['\"]?")


def add_source_shunit2(fc: FileContent, binary_name: str) -> FileContent:
    # NB: We always run tests from the build root, so we source `shunit2` relative to there.
    source_line = f"source ./{binary_name}".encode()

    lines: list[bytes] = []
    already_had_source = False
    for line in fc.content.splitlines():
        if SOURCE_SHUNIT2_REGEX.search(line):
            lines.append(SOURCE_SHUNIT2_REGEX.sub(source_line, line))
            already_had_source = True
        else:
            lines.append(line)

    if not already_had_source:
        lines.append(source_line)

    return FileContent(fc.path, b"\n".join(lines))


@dataclass(frozen=True)
class Shunit2RunnerRequest:
    address: Address
    test_file_content: FileContent
    shell_field: Shunit2ShellField


@dataclass(frozen=True)
class Shunit2Runner:
    shell: Shunit2Shell
    binary_path: BinaryPath


@rule(desc="Determine shunit2 shell")
async def determine_shunit2_shell(
    request: Shunit2RunnerRequest,
    shell_setup: ShellSetup.EnvironmentAware,
) -> Shunit2Runner:
    if request.shell_field.value is not None:
        tgt_shell = Shunit2Shell(request.shell_field.value)
    else:
        parse_result = Shunit2Shell.parse_shebang(request.test_file_content.content)
        if parse_result is None:
            raise ShellNotConfigured(
                f"Could not determine which shell to use to run shunit2 on {request.address}.\n\n"
                f"Please either specify the `{Shunit2ShellField.alias}` field or add a "
                f"shebang to {request.test_file_content.path} with one of the supported shells in "
                f"the format `!#/path/to/shell` or `!#/path/to/env shell`"
                f"(run `{bin_name()} help {Shunit2TestsGeneratorTarget.alias}` for valid shells)."
            )
        tgt_shell = parse_result

    path_request = BinaryPathRequest(
        binary_name=tgt_shell.name,
        search_path=shell_setup.executable_search_path,
        test=tgt_shell.binary_path_test,
    )
    paths = await find_binary(path_request, **implicitly())
    first_path = paths.first_path
    if not first_path:
        raise BinaryNotFoundError.from_request(
            path_request, rationale=f"run shunit2 on {request.address}"
        )
    return Shunit2Runner(tgt_shell, first_path)


@rule(desc="Setup shunit2", level=LogLevel.DEBUG)
async def setup_shunit2_for_target(
    request: TestSetupRequest,
    shell_setup: ShellSetup.EnvironmentAware,
    test_subsystem: TestSubsystem,
    test_extra_env: TestExtraEnv,
    shunit2: Shunit2,
    platform: Platform,
) -> TestSetup:
    shunit2_script, transitive_targets, built_package_dependencies = await concurrently(
        download_external_tool(shunit2.get_request(platform)),
        transitive_targets_get(
            TransitiveTargetsRequest([request.field_set.address]), **implicitly()
        ),
        build_runtime_package_dependencies(
            BuildPackageDependenciesRequest(request.field_set.runtime_package_dependencies)
        ),
    )

    dependencies_source_files_request = determine_source_files(
        SourceFilesRequest(
            (tgt.get(SourcesField) for tgt in transitive_targets.dependencies),
            for_sources_types=(ShellSourceField, FileSourceField, ResourceSourceField),
            enable_codegen=True,
        )
    )
    dependencies_source_files, field_set_sources = await concurrently(
        dependencies_source_files_request,
        determine_source_files(SourceFilesRequest([request.field_set.sources])),
    )

    field_set_digest_content = await get_digest_contents(field_set_sources.snapshot.digest)
    # `ShellTestSourceField` validates that there's exactly one file.
    test_file_content = field_set_digest_content[0]
    updated_test_file_content = add_source_shunit2(test_file_content, shunit2_script.exe)

    updated_test_digest, runner = await concurrently(
        create_digest(CreateDigest([updated_test_file_content])),
        determine_shunit2_shell(
            Shunit2RunnerRequest(
                request.field_set.address, test_file_content, request.field_set.shell
            ),
            **implicitly(),
        ),
    )

    input_digest = await merge_digests(
        MergeDigests(
            (
                shunit2_script.digest,
                updated_test_digest,
                dependencies_source_files.snapshot.digest,
                *(pkg.digest for pkg in built_package_dependencies),
            )
        )
    )

    env_dict = {
        "PATH": os.pathsep.join(shell_setup.executable_search_path),
        # Always include colors and strip them out for display below (if required), for better cache
        # hit rates
        "SHUNIT_COLOR": "always",
        **test_extra_env.env,
    }
    argv = (
        # Zsh requires extra args. See https://github.com/kward/shunit2/#-zsh.
        [runner.binary_path.path, "-o", "shwordsplit", "--", *field_set_sources.snapshot.files]
        if runner.shell == Shunit2Shell.zsh
        else [runner.binary_path.path, *field_set_sources.snapshot.files]
    )
    cache_scope = (
        ProcessCacheScope.PER_SESSION if test_subsystem.force else ProcessCacheScope.SUCCESSFUL
    )
    process = Process(
        argv=argv,
        input_digest=input_digest,
        description=f"Run shunit2 for {request.field_set.address}.",
        level=LogLevel.DEBUG,
        env=env_dict,
        timeout_seconds=request.field_set.timeout.calculate_from_global_options(test_subsystem),
        cache_scope=cache_scope,
    )
    return TestSetup(process)


@rule(desc="Run tests with Shunit2", level=LogLevel.DEBUG)
async def run_tests_with_shunit2(
    batch: Shunit2TestRequest.Batch[Shunit2FieldSet, Any],
    test_subsystem: TestSubsystem,
    global_options: GlobalOptions,
) -> TestResult:
    field_set = batch.single_element

    setup = await setup_shunit2_for_target(TestSetupRequest(field_set), **implicitly())
    results = await execute_process_with_retry(
        ProcessWithRetries(setup.process, test_subsystem.attempts_default)
    )
    return TestResult.from_fallible_process_result(
        process_results=results.results,
        address=field_set.address,
        output_setting=test_subsystem.output,
        output_simplifier=global_options.output_simplifier(),
    )


@rule(desc="Setup Shunit2 to run interactively", level=LogLevel.DEBUG)
async def setup_shunit2_debug_test(
    batch: Shunit2TestRequest.Batch[Shunit2FieldSet, Any],
) -> TestDebugRequest:
    setup = await setup_shunit2_for_target(TestSetupRequest(batch.single_element), **implicitly())
    return TestDebugRequest(
        InteractiveProcess.from_process(
            setup.process, forward_signals_to_process=False, restartable=True
        )
    )


def rules():
    return (
        *collect_rules(),
        *Shunit2TestRequest.rules(),
    )

1	# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
UNCOV 4	import os	×
UNCOV 5	import re	×
UNCOV 6	from dataclasses import dataclass	×
UNCOV 7	from typing import Any	×
8
UNCOV 9	from pants.backend.shell.subsystems.shell_setup import ShellSetup	×
UNCOV 10	from pants.backend.shell.subsystems.shunit2 import Shunit2	×
UNCOV 11	from pants.backend.shell.target_types import (	×
12	ShellSourceField,
13	Shunit2Shell,
14	Shunit2ShellField,
15	Shunit2TestsGeneratorTarget,
16	Shunit2TestSourceField,
17	Shunit2TestTimeoutField,
18	SkipShunit2TestsField,
19	)
UNCOV 20	from pants.core.goals.test import (	×
21	BuildPackageDependenciesRequest,
22	RuntimePackageDependenciesField,
23	TestDebugRequest,
24	TestExtraEnv,
25	TestFieldSet,
26	TestRequest,
27	TestResult,
28	TestSubsystem,
29	build_runtime_package_dependencies,
30	)
UNCOV 31	from pants.core.target_types import FileSourceField, ResourceSourceField	×
UNCOV 32	from pants.core.util_rules.external_tool import download_external_tool	×
UNCOV 33	from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files	×
UNCOV 34	from pants.core.util_rules.system_binaries import (	×
35	BinaryNotFoundError,
36	BinaryPath,
37	BinaryPathRequest,
38	find_binary,
39	)
UNCOV 40	from pants.engine.addresses import Address	×
UNCOV 41	from pants.engine.fs import CreateDigest, FileContent, MergeDigests	×
UNCOV 42	from pants.engine.internals.graph import transitive_targets as transitive_targets_get	×
UNCOV 43	from pants.engine.intrinsics import (	×
44	create_digest,
45	execute_process_with_retry,
46	get_digest_contents,
47	merge_digests,
48	)
UNCOV 49	from pants.engine.platform import Platform	×
UNCOV 50	from pants.engine.process import InteractiveProcess, Process, ProcessCacheScope, ProcessWithRetries	×
UNCOV 51	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
UNCOV 52	from pants.engine.target import SourcesField, Target, TransitiveTargetsRequest	×
UNCOV 53	from pants.option.global_options import GlobalOptions	×
UNCOV 54	from pants.util.docutil import bin_name	×
UNCOV 55	from pants.util.logging import LogLevel	×
56
57
UNCOV 58	@dataclass(frozen=True)	×
UNCOV 59	class Shunit2FieldSet(TestFieldSet):	×
UNCOV 60	required_fields = (Shunit2TestSourceField,)	×
61
62	sources: Shunit2TestSourceField
63	timeout: Shunit2TestTimeoutField
64	shell: Shunit2ShellField
65	runtime_package_dependencies: RuntimePackageDependenciesField
66
UNCOV 67	@classmethod	×
UNCOV 68	def opt_out(cls, tgt: Target) -> bool:	×
UNCOV 69	return tgt.get(SkipShunit2TestsField).value	×
70
71
UNCOV 72	class Shunit2TestRequest(TestRequest):	×
UNCOV 73	tool_subsystem = Shunit2 # type: ignore[assignment]	×
UNCOV 74	field_set_type = Shunit2FieldSet	×
UNCOV 75	supports_debug = True	×
76
77
UNCOV 78	@dataclass(frozen=True)	×
UNCOV 79	class TestSetupRequest:	×
80	field_set: Shunit2FieldSet
81
82
UNCOV 83	@dataclass(frozen=True)	×
UNCOV 84	class TestSetup:	×
85	process: Process
86
87
UNCOV 88	class ShellNotConfigured(Exception):	×
UNCOV 89	pass	×
90
91
UNCOV 92	SOURCE_SHUNIT2_REGEX = re.compile(rb"(?:source\|\.)\s+[.${}/'\"\w]*shunit2\b['\"]?")	×
93
94
UNCOV 95	def add_source_shunit2(fc: FileContent, binary_name: str) -> FileContent:	×
96	# NB: We always run tests from the build root, so we source `shunit2` relative to there.
UNCOV 97	source_line = f"source ./{binary_name}".encode()	×
98
UNCOV 99	lines: list[bytes] = []	×
UNCOV 100	already_had_source = False	×
UNCOV 101	for line in fc.content.splitlines():	×
UNCOV 102	if SOURCE_SHUNIT2_REGEX.search(line):	×
UNCOV 103	lines.append(SOURCE_SHUNIT2_REGEX.sub(source_line, line))	×
UNCOV 104	already_had_source = True	×
105	else:
UNCOV 106	lines.append(line)	×
107
UNCOV 108	if not already_had_source:	×
UNCOV 109	lines.append(source_line)	×
110
UNCOV 111	return FileContent(fc.path, b"\n".join(lines))	×
112
113
UNCOV 114	@dataclass(frozen=True)	×
UNCOV 115	class Shunit2RunnerRequest:	×
116	address: Address
117	test_file_content: FileContent
118	shell_field: Shunit2ShellField
119
120
UNCOV 121	@dataclass(frozen=True)	×
UNCOV 122	class Shunit2Runner:	×
123	shell: Shunit2Shell
124	binary_path: BinaryPath
125
126
UNCOV 127	@rule(desc="Determine shunit2 shell")	×
UNCOV 128	async def determine_shunit2_shell(	×
129	request: Shunit2RunnerRequest,
130	shell_setup: ShellSetup.EnvironmentAware,
131	) -> Shunit2Runner:
132	if request.shell_field.value is not None:	×
133	tgt_shell = Shunit2Shell(request.shell_field.value)	×
134	else:
135	parse_result = Shunit2Shell.parse_shebang(request.test_file_content.content)	×
136	if parse_result is None:	×
137	raise ShellNotConfigured(	×
138	f"Could not determine which shell to use to run shunit2 on {request.address}.\n\n"
139	f"Please either specify the `{Shunit2ShellField.alias}` field or add a "
140	f"shebang to {request.test_file_content.path} with one of the supported shells in "
141	f"the format `!#/path/to/shell` or `!#/path/to/env shell`"
142	f"(run `{bin_name()} help {Shunit2TestsGeneratorTarget.alias}` for valid shells)."
143	)
144	tgt_shell = parse_result	×
145
146	path_request = BinaryPathRequest(	×
147	binary_name=tgt_shell.name,
148	search_path=shell_setup.executable_search_path,
149	test=tgt_shell.binary_path_test,
150	)
151	paths = await find_binary(path_request, **implicitly())	×
152	first_path = paths.first_path	×
153	if not first_path:	×
154	raise BinaryNotFoundError.from_request(	×
155	path_request, rationale=f"run shunit2 on {request.address}"
156	)
157	return Shunit2Runner(tgt_shell, first_path)	×
158
159
UNCOV 160	@rule(desc="Setup shunit2", level=LogLevel.DEBUG)	×
UNCOV 161	async def setup_shunit2_for_target(	×
162	request: TestSetupRequest,
163	shell_setup: ShellSetup.EnvironmentAware,
164	test_subsystem: TestSubsystem,
165	test_extra_env: TestExtraEnv,
166	shunit2: Shunit2,
167	platform: Platform,
168	) -> TestSetup:
169	shunit2_script, transitive_targets, built_package_dependencies = await concurrently(	×
170	download_external_tool(shunit2.get_request(platform)),
171	transitive_targets_get(
172	TransitiveTargetsRequest([request.field_set.address]), **implicitly()
173	),
174	build_runtime_package_dependencies(
175	BuildPackageDependenciesRequest(request.field_set.runtime_package_dependencies)
176	),
177	)
178
179	dependencies_source_files_request = determine_source_files(	×
180	SourceFilesRequest(
181	(tgt.get(SourcesField) for tgt in transitive_targets.dependencies),
182	for_sources_types=(ShellSourceField, FileSourceField, ResourceSourceField),
183	enable_codegen=True,
184	)
185	)
186	dependencies_source_files, field_set_sources = await concurrently(	×
187	dependencies_source_files_request,
188	determine_source_files(SourceFilesRequest([request.field_set.sources])),
189	)
190
191	field_set_digest_content = await get_digest_contents(field_set_sources.snapshot.digest)	×
192	# `ShellTestSourceField` validates that there's exactly one file.
193	test_file_content = field_set_digest_content[0]	×
194	updated_test_file_content = add_source_shunit2(test_file_content, shunit2_script.exe)	×
195
196	updated_test_digest, runner = await concurrently(	×
197	create_digest(CreateDigest([updated_test_file_content])),
198	determine_shunit2_shell(
199	Shunit2RunnerRequest(
200	request.field_set.address, test_file_content, request.field_set.shell
201	),
202	**implicitly(),
203	),
204	)
205
206	input_digest = await merge_digests(	×
207	MergeDigests(
208	(
209	shunit2_script.digest,
210	updated_test_digest,
211	dependencies_source_files.snapshot.digest,
212	*(pkg.digest for pkg in built_package_dependencies),
213	)
214	)
215	)
216
217	env_dict = {	×
218	"PATH": os.pathsep.join(shell_setup.executable_search_path),
219	# Always include colors and strip them out for display below (if required), for better cache
220	# hit rates
221	"SHUNIT_COLOR": "always",
222	**test_extra_env.env,
223	}
224	argv = (	×
225	# Zsh requires extra args. See https://github.com/kward/shunit2/#-zsh.
226	[runner.binary_path.path, "-o", "shwordsplit", "--", *field_set_sources.snapshot.files]
227	if runner.shell == Shunit2Shell.zsh
228	else [runner.binary_path.path, *field_set_sources.snapshot.files]
229	)
230	cache_scope = (	×
231	ProcessCacheScope.PER_SESSION if test_subsystem.force else ProcessCacheScope.SUCCESSFUL
232	)
233	process = Process(	×
234	argv=argv,
235	input_digest=input_digest,
236	description=f"Run shunit2 for {request.field_set.address}.",
237	level=LogLevel.DEBUG,
238	env=env_dict,
239	timeout_seconds=request.field_set.timeout.calculate_from_global_options(test_subsystem),
240	cache_scope=cache_scope,
241	)
242	return TestSetup(process)	×
243
244
UNCOV 245	@rule(desc="Run tests with Shunit2", level=LogLevel.DEBUG)	×
UNCOV 246	async def run_tests_with_shunit2(	×
247	batch: Shunit2TestRequest.Batch[Shunit2FieldSet, Any],
248	test_subsystem: TestSubsystem,
249	global_options: GlobalOptions,
250	) -> TestResult:
251	field_set = batch.single_element	×
252
253	setup = await setup_shunit2_for_target(TestSetupRequest(field_set), **implicitly())	×
254	results = await execute_process_with_retry(	×
255	ProcessWithRetries(setup.process, test_subsystem.attempts_default)
256	)
257	return TestResult.from_fallible_process_result(	×
258	process_results=results.results,
259	address=field_set.address,
260	output_setting=test_subsystem.output,
261	output_simplifier=global_options.output_simplifier(),
262	)
263
264
UNCOV 265	@rule(desc="Setup Shunit2 to run interactively", level=LogLevel.DEBUG)	×
UNCOV 266	async def setup_shunit2_debug_test(	×
267	batch: Shunit2TestRequest.Batch[Shunit2FieldSet, Any],
268	) -> TestDebugRequest:
269	setup = await setup_shunit2_for_target(TestSetupRequest(batch.single_element), **implicitly())	×
270	return TestDebugRequest(	×
271	InteractiveProcess.from_process(
272	setup.process, forward_signals_to_process=False, restartable=True
273	)
274	)
275
276
UNCOV 277	def rules():	×
UNCOV 278	return (	×
279	*collect_rules(),
280	*Shunit2TestRequest.rules(),
281	)

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