19381742489

Committed 15 Nov 2025 12:52AM UTC coverage: 49.706% (-30.6%) from 80.29%

Build # 19381742489

Build Type

Pull #22890

github

Committed by

web-flow

Commit Message

Merge d961abf79 into 42e1ebd41

Pull Request Pull Request #22890: Updated all python subsystem constraints to 3.14

Run Details

4 of 5 new or added lines in 5 files covered. (80.0%)

14659 existing lines in 485 files now uncovered.

31583 of 63540 relevant lines covered (49.71%)

0.79 hits per line

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

76.39

/src/python/pants/engine/internals/platform_rules_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 textwrap import dedent
from unittest.mock import Mock

import pytest

from pants.build_graph.address import Address
from pants.core.environments.subsystems import EnvironmentsSubsystem
from pants.core.environments.target_types import (
    DockerEnvironmentTarget,
    DockerImageField,
    DockerPlatformField,
    EnvironmentTarget,
    LocalEnvironmentTarget,
    RemoteEnvironmentTarget,
    RemotePlatformField,
)
from pants.engine.env_vars import CompleteEnvironmentVars, EnvironmentVars, EnvironmentVarsRequest
from pants.engine.environment import EnvironmentName
from pants.engine.internals.platform_rules import (
    complete_environment_vars,
    current_platform,
    environment_path_variable,
)
from pants.engine.internals.session import SessionValues
from pants.engine.platform import Platform
from pants.engine.process import Process, ProcessResult
from pants.option.global_options import GlobalOptions
from pants.testutil.option_util import create_subsystem
from pants.testutil.rule_runner import QueryRule, RuleRunner, run_rule_with_mocks


@pytest.mark.platform_specific_behavior
def test_current_platform() -> None:
    def assert_platform(
        *,
        envs_enabled: bool = True,
        env_tgt: LocalEnvironmentTarget | RemoteEnvironmentTarget | DockerEnvironmentTarget | None,
        remote_execution: bool,
        expected: Platform,
    ) -> None:
        global_options = create_subsystem(GlobalOptions, remote_execution=remote_execution)
        name = "name"
        env_subsystem = create_subsystem(
            EnvironmentsSubsystem,
            names={name: "addr"} if envs_enabled else {},
        )
        result = run_rule_with_mocks(
            current_platform,
            rule_args=[EnvironmentTarget(name, env_tgt), global_options, env_subsystem],
        )
        assert result == expected

    assert_platform(env_tgt=None, remote_execution=False, expected=Platform.create_for_localhost())
    assert_platform(
        env_tgt=None, envs_enabled=False, remote_execution=True, expected=Platform.linux_x86_64
    )
    assert_platform(
        env_tgt=None,
        envs_enabled=True,
        remote_execution=True,
        expected=Platform.create_for_localhost(),
    )

    for re in (False, True):
        assert_platform(
            env_tgt=LocalEnvironmentTarget({}, Address("dir")),
            remote_execution=re,
            expected=Platform.create_for_localhost(),
        )

    for re in (False, True):
        assert_platform(
            env_tgt=DockerEnvironmentTarget(
                {
                    DockerImageField.alias: "my_img",
                    DockerPlatformField.alias: Platform.linux_arm64.value,
                },
                Address("dir"),
            ),
            remote_execution=re,
            expected=Platform.linux_arm64,
        )

    for re in (False, True):
        assert_platform(
            env_tgt=RemoteEnvironmentTarget(
                {RemotePlatformField.alias: Platform.linux_arm64.value}, Address("dir")
            ),
            remote_execution=re,
            expected=Platform.linux_arm64,
        )


@pytest.mark.platform_specific_behavior
def test_complete_env_vars() -> None:
    def assert_env_vars(
        *,
        envs_enabled: bool = True,
        env_tgt: LocalEnvironmentTarget | RemoteEnvironmentTarget | DockerEnvironmentTarget | None,
        remote_execution: bool,
        expected_env: str,
    ) -> None:
        global_options = create_subsystem(GlobalOptions, remote_execution=remote_execution)
        name = "name"
        env_subsystem = create_subsystem(
            EnvironmentsSubsystem,
            names={name: "addr"} if envs_enabled else {},
        )

        def mock_env_process(process: Process) -> ProcessResult:
            return Mock(
                stdout=b"DOCKER=true" if "Docker" in process.description else b"REMOTE=true"
            )

        result = run_rule_with_mocks(
            complete_environment_vars,
            rule_args=[
                SessionValues(
                    {CompleteEnvironmentVars: CompleteEnvironmentVars({"LOCAL": "true"})}
                ),
                EnvironmentTarget(name, env_tgt),
                global_options,
                env_subsystem,
            ],
            mock_calls={
                "pants.engine.process.fallible_to_exec_result_or_raise": mock_env_process,
            },
        )
        assert dict(result) == {expected_env: "true"}

    assert_env_vars(env_tgt=None, remote_execution=False, expected_env="LOCAL")
    assert_env_vars(env_tgt=None, envs_enabled=False, remote_execution=True, expected_env="REMOTE")
    assert_env_vars(env_tgt=None, envs_enabled=True, remote_execution=True, expected_env="LOCAL")

    for re in (False, True):
        assert_env_vars(
            env_tgt=LocalEnvironmentTarget({}, Address("dir")),
            remote_execution=re,
            expected_env="LOCAL",
        )

    for re in (False, True):
        assert_env_vars(
            env_tgt=DockerEnvironmentTarget({DockerImageField.alias: "my_img"}, Address("dir")),
            remote_execution=re,
            expected_env="DOCKER",
        )

    for re in (False, True):
        assert_env_vars(
            env_tgt=RemoteEnvironmentTarget({}, Address("dir")),
            remote_execution=re,
            expected_env="REMOTE",
        )


@pytest.mark.parametrize(
    ("env", "expected_entries"),
    [
        pytest.param(
            {"PATH": "foo/bar:baz:/qux/quux"}, ["foo/bar", "baz", "/qux/quux"], id="populated_PATH"
        ),
        pytest.param({"PATH": ""}, [], id="empty_PATH"),
        pytest.param({}, [], id="unset_PATH"),
    ],
)
def test_get_environment_paths(env: dict[str, str], expected_entries: list[str]) -> None:
    def mock_environment_vars_subset(_req: EnvironmentVarsRequest) -> EnvironmentVars:
        return EnvironmentVars(env)

    paths = run_rule_with_mocks(
        environment_path_variable,
        mock_calls={
            "pants.core.util_rules.env_vars.environment_vars_subset": mock_environment_vars_subset,
        },
    )
    assert list(paths) == expected_entries


def test_docker_complete_env_vars() -> None:
    rule_runner = RuleRunner(
        rules=[QueryRule(CompleteEnvironmentVars, [EnvironmentName])],
        target_types=[DockerEnvironmentTarget],
        inherent_environment=EnvironmentName("docker"),
    )
    localhost_platform = Platform.create_for_localhost()
    if localhost_platform == Platform.linux_arm64:
        image_sha = "65a4aad1156d8a0679537cb78519a17eb7142e05a968b26a5361153006224fdc"
        platform = Platform.linux_arm64.value
    else:
        image_sha = "a1801b843b1bfaf77c501e7a6d3f709401a1e0c83863037fa3aab063a7fdb9dc"
        platform = Platform.linux_x86_64.value

    rule_runner.write_files(
        {
            "BUILD": dedent(
                f"""\
                docker_environment(
                    name='docker',
                    image='centos@sha256:{image_sha}',
                    platform='{platform}',
                )
                """
            )
        }
    )
    rule_runner.set_options(["--environments-preview-names={'docker': '//:docker'}"])
    result = dict(rule_runner.request(CompleteEnvironmentVars, []))

    # HOSTNAME is not deterministic across machines, so we don't care about the value.
    assert "HOSTNAME" in result
    result.pop("HOSTNAME")
    assert dict(result) == {
        "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
        "HOME": "/root",
    }

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	1✔
5
6	from textwrap import dedent	1✔
7	from unittest.mock import Mock	1✔
8
9	import pytest	1✔
10
11	from pants.build_graph.address import Address	1✔
12	from pants.core.environments.subsystems import EnvironmentsSubsystem	1✔
13	from pants.core.environments.target_types import (	1✔
14	DockerEnvironmentTarget,
15	DockerImageField,
16	DockerPlatformField,
17	EnvironmentTarget,
18	LocalEnvironmentTarget,
19	RemoteEnvironmentTarget,
20	RemotePlatformField,
21	)
22	from pants.engine.env_vars import CompleteEnvironmentVars, EnvironmentVars, EnvironmentVarsRequest	1✔
23	from pants.engine.environment import EnvironmentName	1✔
24	from pants.engine.internals.platform_rules import (	1✔
25	complete_environment_vars,
26	current_platform,
27	environment_path_variable,
28	)
29	from pants.engine.internals.session import SessionValues	1✔
30	from pants.engine.platform import Platform	1✔
31	from pants.engine.process import Process, ProcessResult	1✔
32	from pants.option.global_options import GlobalOptions	1✔
33	from pants.testutil.option_util import create_subsystem	1✔
34	from pants.testutil.rule_runner import QueryRule, RuleRunner, run_rule_with_mocks	1✔
35
36
37	@pytest.mark.platform_specific_behavior	1✔
38	def test_current_platform() -> None:	1✔
39	def assert_platform(	1✔
40	*,
41	envs_enabled: bool = True,
42	env_tgt: LocalEnvironmentTarget \| RemoteEnvironmentTarget \| DockerEnvironmentTarget \| None,
43	remote_execution: bool,
44	expected: Platform,
45	) -> None:
46	global_options = create_subsystem(GlobalOptions, remote_execution=remote_execution)	1✔
47	name = "name"	1✔
48	env_subsystem = create_subsystem(	1✔
49	EnvironmentsSubsystem,
50	names={name: "addr"} if envs_enabled else {},
51	)
52	result = run_rule_with_mocks(	1✔
53	current_platform,
54	rule_args=[EnvironmentTarget(name, env_tgt), global_options, env_subsystem],
55	)
56	assert result == expected	1✔
57
58	assert_platform(env_tgt=None, remote_execution=False, expected=Platform.create_for_localhost())	1✔
59	assert_platform(	1✔
60	env_tgt=None, envs_enabled=False, remote_execution=True, expected=Platform.linux_x86_64
61	)
62	assert_platform(	1✔
63	env_tgt=None,
64	envs_enabled=True,
65	remote_execution=True,
66	expected=Platform.create_for_localhost(),
67	)
68
69	for re in (False, True):	1✔
70	assert_platform(	1✔
71	env_tgt=LocalEnvironmentTarget({}, Address("dir")),
72	remote_execution=re,
73	expected=Platform.create_for_localhost(),
74	)
75
76	for re in (False, True):	1✔
77	assert_platform(	1✔
78	env_tgt=DockerEnvironmentTarget(
79	{
80	DockerImageField.alias: "my_img",
81	DockerPlatformField.alias: Platform.linux_arm64.value,
82	},
83	Address("dir"),
84	),
85	remote_execution=re,
86	expected=Platform.linux_arm64,
87	)
88
89	for re in (False, True):	1✔
90	assert_platform(	1✔
91	env_tgt=RemoteEnvironmentTarget(
92	{RemotePlatformField.alias: Platform.linux_arm64.value}, Address("dir")
93	),
94	remote_execution=re,
95	expected=Platform.linux_arm64,
96	)
97
98
99	@pytest.mark.platform_specific_behavior	1✔
100	def test_complete_env_vars() -> None:	1✔
101	def assert_env_vars(	1✔
102	*,
103	envs_enabled: bool = True,
104	env_tgt: LocalEnvironmentTarget \| RemoteEnvironmentTarget \| DockerEnvironmentTarget \| None,
105	remote_execution: bool,
106	expected_env: str,
107	) -> None:
108	global_options = create_subsystem(GlobalOptions, remote_execution=remote_execution)	1✔
109	name = "name"	1✔
110	env_subsystem = create_subsystem(	1✔
111	EnvironmentsSubsystem,
112	names={name: "addr"} if envs_enabled else {},
113	)
114
115	def mock_env_process(process: Process) -> ProcessResult:	1✔
116	return Mock(	1✔
117	stdout=b"DOCKER=true" if "Docker" in process.description else b"REMOTE=true"
118	)
119
120	result = run_rule_with_mocks(	1✔
121	complete_environment_vars,
122	rule_args=[
123	SessionValues(
124	{CompleteEnvironmentVars: CompleteEnvironmentVars({"LOCAL": "true"})}
125	),
126	EnvironmentTarget(name, env_tgt),
127	global_options,
128	env_subsystem,
129	],
130	mock_calls={
131	"pants.engine.process.fallible_to_exec_result_or_raise": mock_env_process,
132	},
133	)
134	assert dict(result) == {expected_env: "true"}	1✔
135
136	assert_env_vars(env_tgt=None, remote_execution=False, expected_env="LOCAL")	1✔
137	assert_env_vars(env_tgt=None, envs_enabled=False, remote_execution=True, expected_env="REMOTE")	1✔
138	assert_env_vars(env_tgt=None, envs_enabled=True, remote_execution=True, expected_env="LOCAL")	1✔
139
140	for re in (False, True):	1✔
141	assert_env_vars(	1✔
142	env_tgt=LocalEnvironmentTarget({}, Address("dir")),
143	remote_execution=re,
144	expected_env="LOCAL",
145	)
146
147	for re in (False, True):	1✔
148	assert_env_vars(	1✔
149	env_tgt=DockerEnvironmentTarget({DockerImageField.alias: "my_img"}, Address("dir")),
150	remote_execution=re,
151	expected_env="DOCKER",
152	)
153
154	for re in (False, True):	1✔
155	assert_env_vars(	1✔
156	env_tgt=RemoteEnvironmentTarget({}, Address("dir")),
157	remote_execution=re,
158	expected_env="REMOTE",
159	)
160
161
162	@pytest.mark.parametrize(	1✔
163	("env", "expected_entries"),
164	[
165	pytest.param(
166	{"PATH": "foo/bar:baz:/qux/quux"}, ["foo/bar", "baz", "/qux/quux"], id="populated_PATH"
167	),
168	pytest.param({"PATH": ""}, [], id="empty_PATH"),
169	pytest.param({}, [], id="unset_PATH"),
170	],
171	)
172	def test_get_environment_paths(env: dict[str, str], expected_entries: list[str]) -> None:	1✔
UNCOV 173	def mock_environment_vars_subset(_req: EnvironmentVarsRequest) -> EnvironmentVars:	×
UNCOV 174	return EnvironmentVars(env)	×
175
UNCOV 176	paths = run_rule_with_mocks(	×
177	environment_path_variable,
178	mock_calls={
179	"pants.core.util_rules.env_vars.environment_vars_subset": mock_environment_vars_subset,
180	},
181	)
UNCOV 182	assert list(paths) == expected_entries	×
183
184
185	def test_docker_complete_env_vars() -> None:	1✔
UNCOV 186	rule_runner = RuleRunner(	×
187	rules=[QueryRule(CompleteEnvironmentVars, [EnvironmentName])],
188	target_types=[DockerEnvironmentTarget],
189	inherent_environment=EnvironmentName("docker"),
190	)
UNCOV 191	localhost_platform = Platform.create_for_localhost()	×
UNCOV 192	if localhost_platform == Platform.linux_arm64:	×
193	image_sha = "65a4aad1156d8a0679537cb78519a17eb7142e05a968b26a5361153006224fdc"	×
194	platform = Platform.linux_arm64.value	×
195	else:
UNCOV 196	image_sha = "a1801b843b1bfaf77c501e7a6d3f709401a1e0c83863037fa3aab063a7fdb9dc"	×
UNCOV 197	platform = Platform.linux_x86_64.value	×
198
UNCOV 199	rule_runner.write_files(	×
200	{
201	"BUILD": dedent(
202	f"""\
203	docker_environment(
204	name='docker',
205	image='centos@sha256:{image_sha}',
206	platform='{platform}',
207	)
208	"""
209	)
210	}
211	)
UNCOV 212	rule_runner.set_options(["--environments-preview-names={'docker': '//:docker'}"])	×
UNCOV 213	result = dict(rule_runner.request(CompleteEnvironmentVars, []))	×
214
215	# HOSTNAME is not deterministic across machines, so we don't care about the value.
UNCOV 216	assert "HOSTNAME" in result	×
UNCOV 217	result.pop("HOSTNAME")	×
UNCOV 218	assert dict(result) == {	×
219	"PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
220	"HOME": "/root",
221	}

pantsbuild / pants / 19381742489

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