18810350233

Committed 26 Oct 2025 12:16AM UTC coverage: 80.283% (+0.004%) from 80.279%

Build # 18810350233

Build Type

Pull #22804

github

Committed by

web-flow

Commit Message

Merge 49c033a8a into 4834308dc

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

Run Details

30 of 32 new or added lines in 2 files covered. (93.75%)

11 existing lines in 1 file now uncovered.

77899 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

62.71

/src/python/pants/backend/shell/goals/test.py

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

from __future__ import annotations

import dataclasses
from typing import Any

from pants.backend.shell.subsystems.shell_test_subsys import ShellTestSubsystem
from pants.backend.shell.target_types import (
    ShellCommandCacheScopeField,
    ShellCommandCommandField,
    ShellCommandTestDependenciesField,
    SkipShellCommandTestsField,
)
from pants.backend.shell.util_rules import shell_command
from pants.backend.shell.util_rules.shell_command import (
    ShellCommandProcessFromTargetRequest,
    prepare_process_request_from_target,
)
from pants.core.environments.target_types import EnvironmentField, EnvironmentTarget
from pants.core.goals.test import (
    TestDebugRequest,
    TestExtraEnv,
    TestFieldSet,
    TestRequest,
    TestResult,
    TestSubsystem,
)
from pants.core.util_rules.adhoc_process_support import (
    AdhocProcessRequest,
    FallibleAdhocProcessResult,
    prepare_adhoc_process,
)
from pants.core.util_rules.adhoc_process_support import rules as adhoc_process_support_rules
from pants.core.util_rules.adhoc_process_support import run_prepared_adhoc_process
from pants.engine.fs import EMPTY_DIGEST, Snapshot
from pants.engine.internals.graph import resolve_target
from pants.engine.intrinsics import digest_to_snapshot
from pants.engine.process import InteractiveProcess, ProcessCacheScope
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import Target, WrappedTargetRequest
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel


@dataclasses.dataclass(frozen=True)
class TestShellCommandFieldSet(TestFieldSet):
    required_fields = (
        ShellCommandCommandField,
        ShellCommandTestDependenciesField,
    )

    environment: EnvironmentField
    cache_scope: ShellCommandCacheScopeField

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


class ShellTestRequest(TestRequest):
    tool_subsystem = ShellTestSubsystem
    field_set_type = TestShellCommandFieldSet
    supports_debug = True


@rule(desc="Test with shell command", level=LogLevel.DEBUG)
async def test_shell_command(
    batch: ShellTestRequest.Batch[TestShellCommandFieldSet, Any],
    test_subsystem: TestSubsystem,
    test_extra_env: TestExtraEnv,
    env_target: EnvironmentTarget,
) -> TestResult:
    field_set = batch.single_element
    wrapped_tgt = await resolve_target(
        WrappedTargetRequest(field_set.address, description_of_origin="<infallible>"),
        **implicitly(),
    )

    shell_process = await prepare_process_request_from_target(
        ShellCommandProcessFromTargetRequest(wrapped_tgt.target), **implicitly()
    )

    shell_process = dataclasses.replace(
        shell_process,
        env_vars=FrozenDict(
            {
                **test_extra_env.env,
                **shell_process.env_vars,
            }
        ),
    )

    if field_set.cache_scope.value is None:
        shell_process = dataclasses.replace(
            shell_process,
            cache_scope=(
                ProcessCacheScope.PER_SESSION
                if test_subsystem.force
                else env_target.default_cache_scope
            ),
        )

    results: list[FallibleAdhocProcessResult] = []
    for _ in range(test_subsystem.attempts_default):
        result = await run_prepared_adhoc_process(
            **implicitly({shell_process: AdhocProcessRequest})
        )  # noqa: PNT30: retry loop
        results.append(result)
        if result.process_result.exit_code == 0:
            break

    extra_output: Snapshot | None = None
    if results[-1].adjusted_digest != EMPTY_DIGEST:
        extra_output = await digest_to_snapshot(results[-1].adjusted_digest)

    return TestResult.from_fallible_process_result(
        process_results=tuple(r.process_result for r in results),
        address=field_set.address,
        output_setting=test_subsystem.output,
        extra_output=extra_output,
        log_extra_output=extra_output is not None,
    )


@rule(desc="Test with shell command (interactively)", level=LogLevel.DEBUG)
async def test_shell_command_interactively(
    batch: ShellTestRequest.Batch[TestShellCommandFieldSet, Any],
) -> TestDebugRequest:
    field_set = batch.single_element
    wrapped_tgt = await resolve_target(
        WrappedTargetRequest(field_set.address, description_of_origin="<infallible>"),
        **implicitly(),
    )

    prepared_request = await prepare_adhoc_process(
        **implicitly(ShellCommandProcessFromTargetRequest(wrapped_tgt.target))
    )

    # This is probably not strictly necessary given the use of `InteractiveProcess` but good to be correct in any event.
    shell_process = dataclasses.replace(
        prepared_request.process, cache_scope=ProcessCacheScope.PER_SESSION
    )

    return TestDebugRequest(
        InteractiveProcess.from_process(
            shell_process, forward_signals_to_process=False, restartable=True
        )
    )


def rules():
    return (
        *collect_rules(),
        *shell_command.rules(),
        *ShellTestRequest.rules(),
        *adhoc_process_support_rules(),
    )

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	4✔
5
6	import dataclasses	4✔
7	from typing import Any	4✔
8
9	from pants.backend.shell.subsystems.shell_test_subsys import ShellTestSubsystem	4✔
10	from pants.backend.shell.target_types import (	4✔
11	ShellCommandCacheScopeField,
12	ShellCommandCommandField,
13	ShellCommandTestDependenciesField,
14	SkipShellCommandTestsField,
15	)
16	from pants.backend.shell.util_rules import shell_command	4✔
17	from pants.backend.shell.util_rules.shell_command import (	4✔
18	ShellCommandProcessFromTargetRequest,
19	prepare_process_request_from_target,
20	)
21	from pants.core.environments.target_types import EnvironmentField, EnvironmentTarget	4✔
22	from pants.core.goals.test import (	4✔
23	TestDebugRequest,
24	TestExtraEnv,
25	TestFieldSet,
26	TestRequest,
27	TestResult,
28	TestSubsystem,
29	)
30	from pants.core.util_rules.adhoc_process_support import (	4✔
31	AdhocProcessRequest,
32	FallibleAdhocProcessResult,
33	prepare_adhoc_process,
34	)
35	from pants.core.util_rules.adhoc_process_support import rules as adhoc_process_support_rules	4✔
36	from pants.core.util_rules.adhoc_process_support import run_prepared_adhoc_process	4✔
37	from pants.engine.fs import EMPTY_DIGEST, Snapshot	4✔
38	from pants.engine.internals.graph import resolve_target	4✔
39	from pants.engine.intrinsics import digest_to_snapshot	4✔
40	from pants.engine.process import InteractiveProcess, ProcessCacheScope	4✔
41	from pants.engine.rules import collect_rules, implicitly, rule	4✔
42	from pants.engine.target import Target, WrappedTargetRequest	4✔
43	from pants.util.frozendict import FrozenDict	4✔
44	from pants.util.logging import LogLevel	4✔
45
46
47	@dataclasses.dataclass(frozen=True)	4✔
48	class TestShellCommandFieldSet(TestFieldSet):	4✔
49	required_fields = (	4✔
50	ShellCommandCommandField,
51	ShellCommandTestDependenciesField,
52	)
53
54	environment: EnvironmentField	4✔
55	cache_scope: ShellCommandCacheScopeField	4✔
56
57	@classmethod	4✔
58	def opt_out(cls, tgt: Target) -> bool:	4✔
59	return tgt.get(SkipShellCommandTestsField).value	×
60
61
62	class ShellTestRequest(TestRequest):	4✔
63	tool_subsystem = ShellTestSubsystem	4✔
64	field_set_type = TestShellCommandFieldSet	4✔
65	supports_debug = True	4✔
66
67
68	@rule(desc="Test with shell command", level=LogLevel.DEBUG)	4✔
69	async def test_shell_command(	4✔
70	batch: ShellTestRequest.Batch[TestShellCommandFieldSet, Any],
71	test_subsystem: TestSubsystem,
72	test_extra_env: TestExtraEnv,
73	env_target: EnvironmentTarget,
74	) -> TestResult:
75	field_set = batch.single_element	×
UNCOV 76	wrapped_tgt = await resolve_target(	×
77	WrappedTargetRequest(field_set.address, description_of_origin="<infallible>"),
78	**implicitly(),
79	)
80
UNCOV 81	shell_process = await prepare_process_request_from_target(	×
82	ShellCommandProcessFromTargetRequest(wrapped_tgt.target), **implicitly()
83	)
84
UNCOV 85	shell_process = dataclasses.replace(	×
86	shell_process,
87	env_vars=FrozenDict(
88	{
89	**test_extra_env.env,
90	**shell_process.env_vars,
91	}
92	),
93	)
94
NEW 95	if field_set.cache_scope.value is None:	×
NEW 96	shell_process = dataclasses.replace(	×
97	shell_process,
98	cache_scope=(
99	ProcessCacheScope.PER_SESSION
100	if test_subsystem.force
101	else env_target.default_cache_scope
102	),
103	)
104
105	results: list[FallibleAdhocProcessResult] = []	×
106	for _ in range(test_subsystem.attempts_default):	×
107	result = await run_prepared_adhoc_process(	×
108	**implicitly({shell_process: AdhocProcessRequest})
109	) # noqa: PNT30: retry loop
110	results.append(result)	×
111	if result.process_result.exit_code == 0:	×
UNCOV 112	break	×
113
UNCOV 114	extra_output: Snapshot \| None = None	×
UNCOV 115	if results[-1].adjusted_digest != EMPTY_DIGEST:	×
UNCOV 116	extra_output = await digest_to_snapshot(results[-1].adjusted_digest)	×
117
UNCOV 118	return TestResult.from_fallible_process_result(	×
119	process_results=tuple(r.process_result for r in results),
120	address=field_set.address,
121	output_setting=test_subsystem.output,
122	extra_output=extra_output,
123	log_extra_output=extra_output is not None,
124	)
125
126
127	@rule(desc="Test with shell command (interactively)", level=LogLevel.DEBUG)	4✔
128	async def test_shell_command_interactively(	4✔
129	batch: ShellTestRequest.Batch[TestShellCommandFieldSet, Any],
130	) -> TestDebugRequest:
UNCOV 131	field_set = batch.single_element	×
132	wrapped_tgt = await resolve_target(	×
133	WrappedTargetRequest(field_set.address, description_of_origin="<infallible>"),
134	**implicitly(),
135	)
136
137	prepared_request = await prepare_adhoc_process(	×
138	**implicitly(ShellCommandProcessFromTargetRequest(wrapped_tgt.target))
139	)
140
141	# This is probably not strictly necessary given the use of `InteractiveProcess` but good to be correct in any event.
UNCOV 142	shell_process = dataclasses.replace(	×
143	prepared_request.process, cache_scope=ProcessCacheScope.PER_SESSION
144	)
145
UNCOV 146	return TestDebugRequest(	×
147	InteractiveProcess.from_process(
148	shell_process, forward_signals_to_process=False, restartable=True
149	)
150	)
151
152
153	def rules():	4✔
154	return (	4✔
155	*collect_rules(),
156	*shell_command.rules(),
157	*ShellTestRequest.rules(),
158	*adhoc_process_support_rules(),
159	)

pantsbuild / pants / 18810350233

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