18252174847

Committed 05 Oct 2025 01:36AM UTC coverage: 43.382% (-36.9%) from 80.261%

Build # 18252174847

Build Type

push

github

Committed by

web-flow

Commit Message

run tests on mac arm (#22717)

Just doing the minimal to pull forward the x86_64 pattern.

ref #20993

Run Details

25776 of 59416 relevant lines covered (43.38%)

1.3 hits per line

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

74.73

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

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

from __future__ import annotations

from pathlib import Path
from textwrap import dedent

import pytest

from pants.backend.adhoc import run_system_binary
from pants.backend.adhoc.target_types import SystemBinaryTarget
from pants.backend.shell.goals import test
from pants.backend.shell.goals.test import ShellTestRequest, TestShellCommandFieldSet
from pants.backend.shell.target_types import (
    ShellCommandTarget,
    ShellCommandTestTarget,
    ShellSourcesGeneratorTarget,
)
from pants.build_graph.address import Address
from pants.core.goals import package
from pants.core.goals.test import TestDebugRequest, TestResult, get_filtered_environment
from pants.core.util_rules import archive, source_files, system_binaries
from pants.engine.fs import DigestContents, FileContent
from pants.engine.internals.scheduler import ExecutionError
from pants.engine.rules import QueryRule
from pants.engine.target import Target
from pants.testutil.rule_runner import RuleRunner, mock_console

ATTEMPTS_DEFAULT_OPTION = 2


@pytest.fixture
def rule_runner() -> RuleRunner:
    rule_runner = RuleRunner(
        rules=[
            *test.rules(),
            *source_files.rules(),
            *archive.rules(),
            *package.rules(),
            *system_binaries.rules(),
            *run_system_binary.rules(),
            get_filtered_environment,
            QueryRule(TestResult, (ShellTestRequest.Batch,)),
            QueryRule(TestDebugRequest, [ShellTestRequest.Batch]),
        ],
        target_types=[
            ShellSourcesGeneratorTarget,
            ShellCommandTarget,
            ShellCommandTestTarget,
            SystemBinaryTarget,
        ],
    )
    rule_runner.set_options(
        [f"--test-attempts-default={ATTEMPTS_DEFAULT_OPTION}"], env_inherit={"PATH"}
    )
    return rule_runner


@pytest.mark.platform_specific_behavior
def test_basic_usage_of_test_shell_command(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "BUILD": dedent(
                """\
                shell_sources(name="src")

                shell_command(
                  name="msg-gen",
                  command="echo message > msg.txt",
                  tools=["echo"],
                  output_files=["msg.txt"],
                )

                test_shell_command(
                  name="pass",
                  execution_dependencies=[":msg-gen", ":src"],
                  tools=["echo"],
                  command="./test.sh msg.txt message",
                )

                test_shell_command(
                  name="fail",
                  execution_dependencies=[":msg-gen", ":src"],
                  tools=["echo"],
                  command="./test.sh msg.txt xyzzy",
                )

                # Check whether `runnable_dependencies` works.
                system_binary(
                    name="cat",
                    binary_name="cat",
                )
                system_binary(
                    name="test",
                    binary_name="test",
                    fingerprint_args=["1", "=", "1"]
                )
                test_shell_command(
                  name="pass_with_runnable_dependency",
                  execution_dependencies=[":msg-gen", ":src"],
                  tools=["echo"],
                  runnable_dependencies=[":cat", ":test"],
                  command="value=$(cat msg.txt) && test $value = message",
                )
                """
            ),
            "test.sh": dedent(
                """\
                contents="$(<$1)"
                if [ "$contents" = "$2" ]; then
                  echo "contains '$2'"
                  exit 0
                else
                  echo "does not contain '$2'"
                  exit 1
                fi
                """
            ),
        }
    )
    (Path(rule_runner.build_root) / "test.sh").chmod(0o555)

    def test_batch_for_target(test_target: Target) -> ShellTestRequest.Batch:
        return ShellTestRequest.Batch("", (TestShellCommandFieldSet.create(test_target),), None)

    def run_test(test_target: Target) -> TestResult:
        return rule_runner.request(TestResult, [test_batch_for_target(test_target)])

    pass_target = rule_runner.get_target(Address("", target_name="pass"))
    pass_result = run_test(pass_target)
    assert pass_result.exit_code == 0
    assert pass_result.stdout_bytes == b"contains 'message'\n"

    fail_target = rule_runner.get_target(Address("", target_name="fail"))
    fail_result = run_test(fail_target)
    assert fail_result.exit_code == 1
    assert fail_result.stdout_bytes == b"does not contain 'xyzzy'\n"
    assert len(fail_result.process_results) == ATTEMPTS_DEFAULT_OPTION

    # Check whether interactive execution via the `test` goal's `--debug` flags succeeds.
    pass_debug_request = rule_runner.request(TestDebugRequest, [test_batch_for_target(pass_target)])
    with mock_console(rule_runner.options_bootstrapper):
        pass_debug_result = rule_runner.run_interactive_process(pass_debug_request.process)
        assert pass_debug_result.exit_code == 0

    fail_debug_request = rule_runner.request(TestDebugRequest, [test_batch_for_target(fail_target)])
    with mock_console(rule_runner.options_bootstrapper):
        fail_debug_result = rule_runner.run_interactive_process(fail_debug_request.process)
        assert fail_debug_result.exit_code == 1

    pass_for_runnable_dependency_target = rule_runner.get_target(
        Address("", target_name="pass_with_runnable_dependency")
    )
    pass_for_runnable_dependency_result = run_test(pass_for_runnable_dependency_target)
    assert pass_for_runnable_dependency_result.exit_code == 0


@pytest.mark.platform_specific_behavior
def test_extra_outputs_support(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "BUILD": dedent(
                """\
                shell_sources(name="src")

                test_shell_command(
                  name="test",
                  execution_dependencies=[":src"],
                  tools=["echo", "mkdir"],
                  command="./test.sh msg.txt message",
                  output_files=["world.txt"],
                  output_directories=["some-dir"],
                )
                """
            ),
            "test.sh": dedent(
                """\
                mkdir -p some-dir
                echo "xyzzy" > some-dir/foo.txt
                echo "hello" > world.txt
                """
            ),
        }
    )
    (Path(rule_runner.build_root) / "test.sh").chmod(0o555)

    def test_batch_for_target(test_target: Target) -> ShellTestRequest.Batch:
        return ShellTestRequest.Batch("", (TestShellCommandFieldSet.create(test_target),), None)

    def run_test(test_target: Target) -> TestResult:
        return rule_runner.request(TestResult, [test_batch_for_target(test_target)])

    result = run_test(rule_runner.get_target(Address("", target_name="test")))
    assert result.extra_output is not None
    digest_contents = rule_runner.request(DigestContents, [result.extra_output.digest])
    digest_contents_sorted = sorted(digest_contents, key=lambda x: x.path)
    assert len(digest_contents_sorted) == 2
    assert digest_contents_sorted[0] == FileContent("some-dir/foo.txt", b"xyzzy\n")
    assert digest_contents_sorted[1] == FileContent("world.txt", b"hello\n")


def test_outputs_match_mode_support(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "BUILD": dedent(
                """\
            test_shell_command(
                name="allow_empty",
                command="true",
                output_files=["non-existent-file"],
                output_directories=["non-existent-dir"],
                outputs_match_mode="allow_empty",
            )
            test_shell_command(
                name="all_with_present_file",
                command="touch some-file",
                tools=["touch"],
                output_files=["some-file"],
                output_directories=["some-directory"],
                outputs_match_mode="all",
            )
            test_shell_command(
                name="all_with_present_directory",
                command="mkdir some-directory",
                tools=["mkdir"],
                output_files=["some-file"],
                output_directories=["some-directory"],
                outputs_match_mode="all",
            )
            test_shell_command(
                name="at_least_one_with_present_file",
                command="touch some-file",
                tools=["touch"],
                output_files=["some-file"],
                output_directories=["some-directory"],
                outputs_match_mode="at_least_one",
            )
            test_shell_command(
                name="at_least_one_with_present_directory",
                command="mkdir some-directory && touch some-directory/foo.txt",
                tools=["mkdir", "touch"],
                output_files=["some-file"],
                output_directories=["some-directory"],
                outputs_match_mode="at_least_one",
            )
            """
            )
        }
    )

    def test_batch_for_target(test_target: Target) -> ShellTestRequest.Batch:
        return ShellTestRequest.Batch("", (TestShellCommandFieldSet.create(test_target),), None)

    def run_test(address: Address) -> TestResult:
        test_target = rule_runner.get_target(address)
        return rule_runner.request(TestResult, [test_batch_for_target(test_target)])

    def assert_result(
        address: Address,
        expected_contents: dict[str, str],
    ) -> None:
        result = run_test(address)
        if expected_contents:
            assert result.extra_output
            assert result.extra_output.files == tuple(expected_contents)

            contents = rule_runner.request(DigestContents, [result.extra_output.digest])
            for fc in contents:
                assert fc.content == expected_contents[fc.path].encode()

    assert_result(Address("", target_name="allow_empty"), {})

    with pytest.raises(ExecutionError) as exc_info:
        run_test(Address("", target_name="all_with_present_file"))
    assert "some-directory" in str(exc_info)

    with pytest.raises(ExecutionError) as exc_info:
        run_test(Address("", target_name="all_with_present_directory"))
    assert "some-file" in str(exc_info)

    assert_result(Address("", target_name="at_least_one_with_present_file"), {"some-file": ""})
    assert_result(
        Address("", target_name="at_least_one_with_present_directory"),
        {"some-directory/foo.txt": ""},
    )

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	3✔
5
6	from pathlib import Path	3✔
7	from textwrap import dedent	3✔
8
9	import pytest	3✔
10
11	from pants.backend.adhoc import run_system_binary	3✔
12	from pants.backend.adhoc.target_types import SystemBinaryTarget	3✔
13	from pants.backend.shell.goals import test	3✔
14	from pants.backend.shell.goals.test import ShellTestRequest, TestShellCommandFieldSet	3✔
15	from pants.backend.shell.target_types import (	3✔
16	ShellCommandTarget,
17	ShellCommandTestTarget,
18	ShellSourcesGeneratorTarget,
19	)
20	from pants.build_graph.address import Address	3✔
21	from pants.core.goals import package	3✔
22	from pants.core.goals.test import TestDebugRequest, TestResult, get_filtered_environment	3✔
23	from pants.core.util_rules import archive, source_files, system_binaries	3✔
24	from pants.engine.fs import DigestContents, FileContent	3✔
25	from pants.engine.internals.scheduler import ExecutionError	3✔
26	from pants.engine.rules import QueryRule	3✔
27	from pants.engine.target import Target	3✔
28	from pants.testutil.rule_runner import RuleRunner, mock_console	3✔
29
30	ATTEMPTS_DEFAULT_OPTION = 2	3✔
31
32
33	@pytest.fixture	3✔
34	def rule_runner() -> RuleRunner:	3✔
35	rule_runner = RuleRunner(	3✔
36	rules=[
37	*test.rules(),
38	*source_files.rules(),
39	*archive.rules(),
40	*package.rules(),
41	*system_binaries.rules(),
42	*run_system_binary.rules(),
43	get_filtered_environment,
44	QueryRule(TestResult, (ShellTestRequest.Batch,)),
45	QueryRule(TestDebugRequest, [ShellTestRequest.Batch]),
46	],
47	target_types=[
48	ShellSourcesGeneratorTarget,
49	ShellCommandTarget,
50	ShellCommandTestTarget,
51	SystemBinaryTarget,
52	],
53	)
54	rule_runner.set_options(	3✔
55	[f"--test-attempts-default={ATTEMPTS_DEFAULT_OPTION}"], env_inherit={"PATH"}
56	)
57	return rule_runner	3✔
58
59
60	@pytest.mark.platform_specific_behavior	3✔
61	def test_basic_usage_of_test_shell_command(rule_runner: RuleRunner) -> None:	3✔
62	rule_runner.write_files(	3✔
63	{
64	"BUILD": dedent(
65	"""\
66	shell_sources(name="src")
67
68	shell_command(
69	name="msg-gen",
70	command="echo message > msg.txt",
71	tools=["echo"],
72	output_files=["msg.txt"],
73	)
74
75	test_shell_command(
76	name="pass",
77	execution_dependencies=[":msg-gen", ":src"],
78	tools=["echo"],
79	command="./test.sh msg.txt message",
80	)
81
82	test_shell_command(
83	name="fail",
84	execution_dependencies=[":msg-gen", ":src"],
85	tools=["echo"],
86	command="./test.sh msg.txt xyzzy",
87	)
88
89	# Check whether `runnable_dependencies` works.
90	system_binary(
91	name="cat",
92	binary_name="cat",
93	)
94	system_binary(
95	name="test",
96	binary_name="test",
97	fingerprint_args=["1", "=", "1"]
98	)
99	test_shell_command(
100	name="pass_with_runnable_dependency",
101	execution_dependencies=[":msg-gen", ":src"],
102	tools=["echo"],
103	runnable_dependencies=[":cat", ":test"],
104	command="value=$(cat msg.txt) && test $value = message",
105	)
106	"""
107	),
108	"test.sh": dedent(
109	"""\
110	contents="$(<$1)"
111	if [ "$contents" = "$2" ]; then
112	echo "contains '$2'"
113	exit 0
114	else
115	echo "does not contain '$2'"
116	exit 1
117	fi
118	"""
119	),
120	}
121	)
122	(Path(rule_runner.build_root) / "test.sh").chmod(0o555)	3✔
123
124	def test_batch_for_target(test_target: Target) -> ShellTestRequest.Batch:	3✔
125	return ShellTestRequest.Batch("", (TestShellCommandFieldSet.create(test_target),), None)	3✔
126
127	def run_test(test_target: Target) -> TestResult:	3✔
128	return rule_runner.request(TestResult, [test_batch_for_target(test_target)])	3✔
129
130	pass_target = rule_runner.get_target(Address("", target_name="pass"))	3✔
131	pass_result = run_test(pass_target)	3✔
132	assert pass_result.exit_code == 0	3✔
133	assert pass_result.stdout_bytes == b"contains 'message'\n"	3✔
134
135	fail_target = rule_runner.get_target(Address("", target_name="fail"))	3✔
136	fail_result = run_test(fail_target)	3✔
137	assert fail_result.exit_code == 1	3✔
138	assert fail_result.stdout_bytes == b"does not contain 'xyzzy'\n"	3✔
139	assert len(fail_result.process_results) == ATTEMPTS_DEFAULT_OPTION	3✔
140
141	# Check whether interactive execution via the `test` goal's `--debug` flags succeeds.
142	pass_debug_request = rule_runner.request(TestDebugRequest, [test_batch_for_target(pass_target)])	3✔
143	with mock_console(rule_runner.options_bootstrapper):	3✔
144	pass_debug_result = rule_runner.run_interactive_process(pass_debug_request.process)	3✔
145	assert pass_debug_result.exit_code == 0	3✔
146
147	fail_debug_request = rule_runner.request(TestDebugRequest, [test_batch_for_target(fail_target)])	3✔
148	with mock_console(rule_runner.options_bootstrapper):	3✔
149	fail_debug_result = rule_runner.run_interactive_process(fail_debug_request.process)	3✔
150	assert fail_debug_result.exit_code == 1	3✔
151
152	pass_for_runnable_dependency_target = rule_runner.get_target(	3✔
153	Address("", target_name="pass_with_runnable_dependency")
154	)
155	pass_for_runnable_dependency_result = run_test(pass_for_runnable_dependency_target)	3✔
156	assert pass_for_runnable_dependency_result.exit_code == 0	3✔
157
158
159	@pytest.mark.platform_specific_behavior	3✔
160	def test_extra_outputs_support(rule_runner: RuleRunner) -> None:	3✔
161	rule_runner.write_files(	3✔
162	{
163	"BUILD": dedent(
164	"""\
165	shell_sources(name="src")
166
167	test_shell_command(
168	name="test",
169	execution_dependencies=[":src"],
170	tools=["echo", "mkdir"],
171	command="./test.sh msg.txt message",
172	output_files=["world.txt"],
173	output_directories=["some-dir"],
174	)
175	"""
176	),
177	"test.sh": dedent(
178	"""\
179	mkdir -p some-dir
180	echo "xyzzy" > some-dir/foo.txt
181	echo "hello" > world.txt
182	"""
183	),
184	}
185	)
186	(Path(rule_runner.build_root) / "test.sh").chmod(0o555)	3✔
187
188	def test_batch_for_target(test_target: Target) -> ShellTestRequest.Batch:	3✔
189	return ShellTestRequest.Batch("", (TestShellCommandFieldSet.create(test_target),), None)	3✔
190
191	def run_test(test_target: Target) -> TestResult:	3✔
192	return rule_runner.request(TestResult, [test_batch_for_target(test_target)])	3✔
193
194	result = run_test(rule_runner.get_target(Address("", target_name="test")))	3✔
195	assert result.extra_output is not None	3✔
196	digest_contents = rule_runner.request(DigestContents, [result.extra_output.digest])	3✔
197	digest_contents_sorted = sorted(digest_contents, key=lambda x: x.path)	3✔
198	assert len(digest_contents_sorted) == 2	3✔
199	assert digest_contents_sorted[0] == FileContent("some-dir/foo.txt", b"xyzzy\n")	3✔
200	assert digest_contents_sorted[1] == FileContent("world.txt", b"hello\n")	3✔
201
202
203	def test_outputs_match_mode_support(rule_runner: RuleRunner) -> None:	3✔
204	rule_runner.write_files(	×
205	{
206	"BUILD": dedent(
207	"""\
208	test_shell_command(
209	name="allow_empty",
210	command="true",
211	output_files=["non-existent-file"],
212	output_directories=["non-existent-dir"],
213	outputs_match_mode="allow_empty",
214	)
215	test_shell_command(
216	name="all_with_present_file",
217	command="touch some-file",
218	tools=["touch"],
219	output_files=["some-file"],
220	output_directories=["some-directory"],
221	outputs_match_mode="all",
222	)
223	test_shell_command(
224	name="all_with_present_directory",
225	command="mkdir some-directory",
226	tools=["mkdir"],
227	output_files=["some-file"],
228	output_directories=["some-directory"],
229	outputs_match_mode="all",
230	)
231	test_shell_command(
232	name="at_least_one_with_present_file",
233	command="touch some-file",
234	tools=["touch"],
235	output_files=["some-file"],
236	output_directories=["some-directory"],
237	outputs_match_mode="at_least_one",
238	)
239	test_shell_command(
240	name="at_least_one_with_present_directory",
241	command="mkdir some-directory && touch some-directory/foo.txt",
242	tools=["mkdir", "touch"],
243	output_files=["some-file"],
244	output_directories=["some-directory"],
245	outputs_match_mode="at_least_one",
246	)
247	"""
248	)
249	}
250	)
251
252	def test_batch_for_target(test_target: Target) -> ShellTestRequest.Batch:	×
253	return ShellTestRequest.Batch("", (TestShellCommandFieldSet.create(test_target),), None)	×
254
255	def run_test(address: Address) -> TestResult:	×
256	test_target = rule_runner.get_target(address)	×
257	return rule_runner.request(TestResult, [test_batch_for_target(test_target)])	×
258
259	def assert_result(	×
260	address: Address,
261	expected_contents: dict[str, str],
262	) -> None:
263	result = run_test(address)	×
264	if expected_contents:	×
265	assert result.extra_output	×
266	assert result.extra_output.files == tuple(expected_contents)	×
267
268	contents = rule_runner.request(DigestContents, [result.extra_output.digest])	×
269	for fc in contents:	×
270	assert fc.content == expected_contents[fc.path].encode()	×
271
272	assert_result(Address("", target_name="allow_empty"), {})	×
273
274	with pytest.raises(ExecutionError) as exc_info:	×
275	run_test(Address("", target_name="all_with_present_file"))	×
276	assert "some-directory" in str(exc_info)	×
277
278	with pytest.raises(ExecutionError) as exc_info:	×
279	run_test(Address("", target_name="all_with_present_directory"))	×
280	assert "some-file" in str(exc_info)	×
281
282	assert_result(Address("", target_name="at_least_one_with_present_file"), {"some-file": ""})	×
283	assert_result(	×
284	Address("", target_name="at_least_one_with_present_directory"),
285	{"some-directory/foo.txt": ""},
286	)

pantsbuild / pants / 18252174847

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