21570473470

Committed 01 Feb 2026 09:21PM UTC coverage: 80.277% (-0.05%) from 80.324%

Build # 21570473470

Build Type

Pull #23062

github

Committed by

web-flow

Commit Message

Merge a72d2193e into d74eb26d0

Pull Request Pull Request #23062: Remove support for Get

Run Details

22 of 25 new or added lines in 4 files covered. (88.0%)

53 existing lines in 6 files now uncovered.

78278 of 97510 relevant lines covered (80.28%)

3.36 hits per line

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

96.0

/src/python/pants/core/goals/run_test.py

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

from __future__ import annotations

import os
import sys
from collections.abc import Iterable, Mapping
from typing import cast

import pytest
from _pytest.monkeypatch import MonkeyPatch

from pants.core.environments.rules import EnvironmentNameRequest
from pants.core.environments.target_types import EnvironmentTarget
from pants.core.goals.run import (
    Run,
    RunDebugAdapterRequest,
    RunFieldSet,
    RunInSandboxBehavior,
    RunRequest,
    RunSubsystem,
    run,
)
from pants.core.subsystems.debug_adapter import DebugAdapterSubsystem
from pants.engine.addresses import Address
from pants.engine.fs import CreateDigest, Digest, FileContent, Workspace
from pants.engine.internals.specs_rules import (
    AmbiguousImplementationsException,
    TooManyTargetsException,
)
from pants.engine.process import InteractiveProcess, InteractiveProcessResult
from pants.engine.target import FieldSet, Target, TargetRootsToFieldSets
from pants.engine.unions import UnionMembership, UnionRule
from pants.option.global_options import GlobalOptions, KeepSandboxes
from pants.testutil.option_util import create_goal_subsystem, create_subsystem
from pants.testutil.rule_runner import MockEffect, RuleRunner, mock_console, run_rule_with_mocks
from pants.util.frozendict import FrozenDict


@pytest.fixture
def rule_runner() -> RuleRunner:
    return RuleRunner()


def create_mock_run_request(rule_runner: RuleRunner, program_text: bytes) -> RunRequest:
    digest = rule_runner.request(
        Digest,
        [CreateDigest([FileContent(path="program.py", content=program_text, is_executable=True)])],
    )
    return RunRequest(digest=digest, args=(os.path.join("{chroot}", "program.py"),))


def create_mock_run_debug_adapter_request(
    rule_runner: RuleRunner, program_text: bytes
) -> RunDebugAdapterRequest:
    return cast(RunDebugAdapterRequest, create_mock_run_request(rule_runner, program_text))


class TestRunFieldSet(RunFieldSet):
    __test__ = False

    required_fields = ()
    run_in_sandbox_behavior = RunInSandboxBehavior.NOT_SUPPORTED


class TestBinaryTarget(Target):
    __test__ = False

    alias = "binary"
    core_fields = ()


A_TARGET = TestBinaryTarget({}, Address("some/addr"))
A_FIELD_SET = TestRunFieldSet.create(A_TARGET)


def single_target_run(
    rule_runner: RuleRunner,
    monkeypatch: MonkeyPatch,
    *,
    program_text: bytes,
    targets_to_field_sets: Mapping[Target, Iterable[FieldSet]] = FrozenDict(
        {A_TARGET: (A_FIELD_SET,)}
    ),
) -> Run:
    workspace = Workspace(rule_runner.scheduler, _enforce_effects=False)

    def noop():
        pass

    monkeypatch.setattr("pants.engine.intrinsics.task_side_effected", noop)
    with mock_console(rule_runner.options_bootstrapper) as (console, _):
        return run_rule_with_mocks(
            run,
            rule_args=[
                create_goal_subsystem(RunSubsystem, args=[], cleanup=True, debug_adapter=False),
                create_subsystem(
                    DebugAdapterSubsystem,
                    host="127.0.0.1",
                    port="5678",
                ),
                create_subsystem(
                    GlobalOptions,
                    pants_workdir=rule_runner.pants_workdir,
                    keep_sandboxes=KeepSandboxes.never,
                ),
                workspace,
                rule_runner.environment,
            ],
            mock_calls={
                "pants.engine.internals.specs_rules.find_valid_field_sets_for_target_roots": lambda _: TargetRootsToFieldSets(
                    targets_to_field_sets
                ),
                "pants.core.goals.run.generate_run_request": lambda _: create_mock_run_request(
                    rule_runner, program_text
                ),
                "pants.core.goals.run.generate_run_debug_adapter_request": lambda _: create_mock_run_request(
                    rule_runner, program_text
                ),
            },
            mock_gets=[
                rule_runner.do_not_use_mock(EnvironmentTarget, (EnvironmentNameRequest,)),
                MockEffect(
                    output_type=InteractiveProcessResult,
                    input_types=(InteractiveProcess,),
                    mock=rule_runner.run_interactive_process,
                ),
            ],
            union_membership=UnionMembership.from_rules(
                {UnionRule(RunFieldSet, TestRunFieldSet)},
            ),
        )


def test_normal_run(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:
    program_text = f'#!{sys.executable}\nprint("hello")'.encode()
    res = single_target_run(
        rule_runner,
        monkeypatch,
        program_text=program_text,
    )
    assert res.exit_code == 0


def test_materialize_input_files(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:
    program_text = f'#!{sys.executable}\nprint("hello")'.encode()
    binary = create_mock_run_request(rule_runner, program_text)
    with mock_console(rule_runner.options_bootstrapper):
        result = rule_runner.run_interactive_process(
            InteractiveProcess(
                argv=("./program.py",),
                run_in_workspace=False,
                input_digest=binary.digest,
            )
        )
    assert result.exit_code == 0


def test_failed_run(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:
    program_text = f'#!{sys.executable}\nraise RuntimeError("foo")'.encode()
    res = single_target_run(rule_runner, monkeypatch, program_text=program_text)
    assert res.exit_code == 1


def test_multi_target_error(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:
    program_text = f'#!{sys.executable}\nprint("hello")'.encode()
    t1 = TestBinaryTarget({}, Address("some/addr"))
    t1_fs = TestRunFieldSet.create(t1)
    t2 = TestBinaryTarget({}, Address("some/other_addr"))
    t2_fs = TestRunFieldSet.create(t2)
    with pytest.raises(TooManyTargetsException):
        single_target_run(
            rule_runner,
            monkeypatch,
            program_text=program_text,
            targets_to_field_sets={t1: [t1_fs], t2: [t2_fs]},
        )


def test_multi_field_set_error(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:
    program_text = f'#!{sys.executable}\nprint("hello")'.encode()
    target = TestBinaryTarget({}, Address("some/addr"))
    fs1 = TestRunFieldSet.create(target)
    fs2 = TestRunFieldSet.create(target)
    with pytest.raises(AmbiguousImplementationsException):
        single_target_run(
            rule_runner,
            monkeypatch,
            program_text=program_text,
            targets_to_field_sets={target: [fs1, fs2]},
        )

1	# Copyright 2019 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 os	1✔
7	import sys	1✔
8	from collections.abc import Iterable, Mapping	1✔
9	from typing import cast	1✔
10
11	import pytest	1✔
12	from _pytest.monkeypatch import MonkeyPatch	1✔
13
14	from pants.core.environments.rules import EnvironmentNameRequest	1✔
15	from pants.core.environments.target_types import EnvironmentTarget	1✔
16	from pants.core.goals.run import (	1✔
17	Run,
18	RunDebugAdapterRequest,
19	RunFieldSet,
20	RunInSandboxBehavior,
21	RunRequest,
22	RunSubsystem,
23	run,
24	)
25	from pants.core.subsystems.debug_adapter import DebugAdapterSubsystem	1✔
26	from pants.engine.addresses import Address	1✔
27	from pants.engine.fs import CreateDigest, Digest, FileContent, Workspace	1✔
28	from pants.engine.internals.specs_rules import (	1✔
29	AmbiguousImplementationsException,
30	TooManyTargetsException,
31	)
32	from pants.engine.process import InteractiveProcess, InteractiveProcessResult	1✔
33	from pants.engine.target import FieldSet, Target, TargetRootsToFieldSets	1✔
34	from pants.engine.unions import UnionMembership, UnionRule	1✔
35	from pants.option.global_options import GlobalOptions, KeepSandboxes	1✔
36	from pants.testutil.option_util import create_goal_subsystem, create_subsystem	1✔
37	from pants.testutil.rule_runner import MockEffect, RuleRunner, mock_console, run_rule_with_mocks	1✔
38	from pants.util.frozendict import FrozenDict	1✔
39
40
41	@pytest.fixture	1✔
42	def rule_runner() -> RuleRunner:	1✔
43	return RuleRunner()	1✔
44
45
46	def create_mock_run_request(rule_runner: RuleRunner, program_text: bytes) -> RunRequest:	1✔
47	digest = rule_runner.request(	1✔
48	Digest,
49	[CreateDigest([FileContent(path="program.py", content=program_text, is_executable=True)])],
50	)
51	return RunRequest(digest=digest, args=(os.path.join("{chroot}", "program.py"),))	1✔
52
53
54	def create_mock_run_debug_adapter_request(	1✔
55	rule_runner: RuleRunner, program_text: bytes
56	) -> RunDebugAdapterRequest:
57	return cast(RunDebugAdapterRequest, create_mock_run_request(rule_runner, program_text))	×
58
59
60	class TestRunFieldSet(RunFieldSet):	1✔
61	__test__ = False	1✔
62
63	required_fields = ()	1✔
64	run_in_sandbox_behavior = RunInSandboxBehavior.NOT_SUPPORTED	1✔
65
66
67	class TestBinaryTarget(Target):	1✔
68	__test__ = False	1✔
69
70	alias = "binary"	1✔
71	core_fields = ()	1✔
72
73
74	A_TARGET = TestBinaryTarget({}, Address("some/addr"))	1✔
75	A_FIELD_SET = TestRunFieldSet.create(A_TARGET)	1✔
76
77
78	def single_target_run(	1✔
79	rule_runner: RuleRunner,
80	monkeypatch: MonkeyPatch,
81	*,
82	program_text: bytes,
83	targets_to_field_sets: Mapping[Target, Iterable[FieldSet]] = FrozenDict(
84	{A_TARGET: (A_FIELD_SET,)}
85	),
86	) -> Run:
87	workspace = Workspace(rule_runner.scheduler, _enforce_effects=False)	1✔
88
89	def noop():	1✔
90	pass	1✔
91
92	monkeypatch.setattr("pants.engine.intrinsics.task_side_effected", noop)	1✔
93	with mock_console(rule_runner.options_bootstrapper) as (console, _):	1✔
94	return run_rule_with_mocks(	1✔
95	run,
96	rule_args=[
97	create_goal_subsystem(RunSubsystem, args=[], cleanup=True, debug_adapter=False),
98	create_subsystem(
99	DebugAdapterSubsystem,
100	host="127.0.0.1",
101	port="5678",
102	),
103	create_subsystem(
104	GlobalOptions,
105	pants_workdir=rule_runner.pants_workdir,
106	keep_sandboxes=KeepSandboxes.never,
107	),
108	workspace,
109	rule_runner.environment,
110	],
111	mock_calls={
112	"pants.engine.internals.specs_rules.find_valid_field_sets_for_target_roots": lambda _: TargetRootsToFieldSets(
113	targets_to_field_sets
114	),
115	"pants.core.goals.run.generate_run_request": lambda _: create_mock_run_request(
116	rule_runner, program_text
117	),
118	"pants.core.goals.run.generate_run_debug_adapter_request": lambda _: create_mock_run_request(
119	rule_runner, program_text
120	),
121	},
122	mock_gets=[
123	rule_runner.do_not_use_mock(EnvironmentTarget, (EnvironmentNameRequest,)),
124	MockEffect(
125	output_type=InteractiveProcessResult,
126	input_types=(InteractiveProcess,),
127	mock=rule_runner.run_interactive_process,
128	),
129	],
130	union_membership=UnionMembership.from_rules(
131	{UnionRule(RunFieldSet, TestRunFieldSet)},
132	),
133	)
134
135
136	def test_normal_run(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:	1✔
137	program_text = f'#!{sys.executable}\nprint("hello")'.encode()	1✔
138	res = single_target_run(	1✔
139	rule_runner,
140	monkeypatch,
141	program_text=program_text,
142	)
UNCOV 143	assert res.exit_code == 0	×
144
145
146	def test_materialize_input_files(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:	1✔
147	program_text = f'#!{sys.executable}\nprint("hello")'.encode()	1✔
148	binary = create_mock_run_request(rule_runner, program_text)	1✔
149	with mock_console(rule_runner.options_bootstrapper):	1✔
150	result = rule_runner.run_interactive_process(	1✔
151	InteractiveProcess(
152	argv=("./program.py",),
153	run_in_workspace=False,
154	input_digest=binary.digest,
155	)
156	)
157	assert result.exit_code == 0	1✔
158
159
160	def test_failed_run(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:	1✔
161	program_text = f'#!{sys.executable}\nraise RuntimeError("foo")'.encode()	1✔
162	res = single_target_run(rule_runner, monkeypatch, program_text=program_text)	1✔
UNCOV 163	assert res.exit_code == 1	×
164
165
166	def test_multi_target_error(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:	1✔
167	program_text = f'#!{sys.executable}\nprint("hello")'.encode()	1✔
168	t1 = TestBinaryTarget({}, Address("some/addr"))	1✔
169	t1_fs = TestRunFieldSet.create(t1)	1✔
170	t2 = TestBinaryTarget({}, Address("some/other_addr"))	1✔
171	t2_fs = TestRunFieldSet.create(t2)	1✔
172	with pytest.raises(TooManyTargetsException):	1✔
173	single_target_run(	1✔
174	rule_runner,
175	monkeypatch,
176	program_text=program_text,
177	targets_to_field_sets={t1: [t1_fs], t2: [t2_fs]},
178	)
179
180
181	def test_multi_field_set_error(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:	1✔
182	program_text = f'#!{sys.executable}\nprint("hello")'.encode()	1✔
183	target = TestBinaryTarget({}, Address("some/addr"))	1✔
184	fs1 = TestRunFieldSet.create(target)	1✔
185	fs2 = TestRunFieldSet.create(target)	1✔
186	with pytest.raises(AmbiguousImplementationsException):	1✔
187	single_target_run(	1✔
188	rule_runner,
189	monkeypatch,
190	program_text=program_text,
191	targets_to_field_sets={target: [fs1, fs2]},
192	)

pantsbuild / pants / 21570473470

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