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

38.0

/src/python/pants/engine/internals/platform_rules.py

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

from __future__ import annotations

import os

from pants.core.environments.subsystems import EnvironmentsSubsystem
from pants.core.environments.target_types import (
    DockerImageField,
    DockerPlatformField,
    EnvironmentTarget,
    RemotePlatformField,
)
from pants.core.util_rules.env_vars import environment_vars_subset
from pants.engine.env_vars import (
    CompleteEnvironmentVars,
    EnvironmentVarsRequest,
    PathEnvironmentVariable,
)
from pants.engine.internals.session import SessionValues
from pants.engine.platform import Platform
from pants.engine.process import Process, execute_process_or_raise
from pants.engine.rules import collect_rules, implicitly, rule
from pants.option.global_options import GlobalOptions
from pants.util.logging import LogLevel


@rule
async def current_platform(
    env_tgt: EnvironmentTarget,
    global_options: GlobalOptions,
    environments_subsystem: EnvironmentsSubsystem,
) -> Platform:
    if environments_subsystem.remote_execution_used_globally(global_options):
        return Platform.linux_x86_64

    if env_tgt.val:
        if env_tgt.val.has_field(DockerPlatformField):
            return Platform(env_tgt.val[DockerPlatformField].normalized_value)
        if env_tgt.val.has_field(RemotePlatformField):
            remote_platform = env_tgt.val[RemotePlatformField].value
            if remote_platform:
                return Platform(remote_platform)
    return Platform.create_for_localhost()


@rule
async def complete_environment_vars(
    session_values: SessionValues,
    env_tgt: EnvironmentTarget,
    global_options: GlobalOptions,
    environments_subsystem: EnvironmentsSubsystem,
) -> CompleteEnvironmentVars:
    # If a local environment is used, we simply use SessionValues. Otherwise, we need to run `env`
    # and parse the output.
    #
    # Note that running `env` works for both Docker and Remote Execution because we intentionally
    # do not strip the environment from either runtime. It is reasonable to not strip because
    # every user will have the same consistent Docker image or Remote Execution environment, unlike
    # local environments.
    if env_tgt.val:
        if env_tgt.val.has_field(DockerImageField):
            description_of_env_source = f"the Docker image {env_tgt.val[DockerImageField].value}"
        elif env_tgt.val.has_field(RemotePlatformField):
            description_of_env_source = "the remote execution environment"
        else:
            # Else, it's a local environment.
            return session_values[CompleteEnvironmentVars]
    else:
        if environments_subsystem.remote_execution_used_globally(global_options):
            description_of_env_source = "the remote execution environment"
        else:
            return session_values[CompleteEnvironmentVars]

    env_process_result = await execute_process_or_raise(
        **implicitly(
            Process(
                ["env", "-0"],
                description=f"Extract environment variables from {description_of_env_source}",
                level=LogLevel.DEBUG,
                cache_scope=env_tgt.executable_search_path_cache_scope(),
            )
        )
    )
    result = {}
    for line in env_process_result.stdout.decode("utf-8").rstrip().split("\0"):
        if not line:
            continue
        k, v = line.split("=", maxsplit=1)
        result[k] = v
    return CompleteEnvironmentVars(result)


@rule
async def environment_path_variable() -> PathEnvironmentVariable:
    env = await environment_vars_subset(EnvironmentVarsRequest(("PATH",)), **implicitly())
    path = env.get("PATH", None)
    return PathEnvironmentVariable(path.split(os.pathsep) if path else ())


def rules():
    return collect_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	1✔
5
6	import os	1✔
7
8	from pants.core.environments.subsystems import EnvironmentsSubsystem	1✔
9	from pants.core.environments.target_types import (	1✔
10	DockerImageField,
11	DockerPlatformField,
12	EnvironmentTarget,
13	RemotePlatformField,
14	)
15	from pants.core.util_rules.env_vars import environment_vars_subset	1✔
16	from pants.engine.env_vars import (	1✔
17	CompleteEnvironmentVars,
18	EnvironmentVarsRequest,
19	PathEnvironmentVariable,
20	)
21	from pants.engine.internals.session import SessionValues	1✔
22	from pants.engine.platform import Platform	1✔
23	from pants.engine.process import Process, execute_process_or_raise	1✔
24	from pants.engine.rules import collect_rules, implicitly, rule	1✔
25	from pants.option.global_options import GlobalOptions	1✔
26	from pants.util.logging import LogLevel	1✔
27
28
29	@rule	1✔
30	async def current_platform(	1✔
31	env_tgt: EnvironmentTarget,
32	global_options: GlobalOptions,
33	environments_subsystem: EnvironmentsSubsystem,
34	) -> Platform:
UNCOV 35	if environments_subsystem.remote_execution_used_globally(global_options):	×
UNCOV 36	return Platform.linux_x86_64	×
37
UNCOV 38	if env_tgt.val:	×
UNCOV 39	if env_tgt.val.has_field(DockerPlatformField):	×
UNCOV 40	return Platform(env_tgt.val[DockerPlatformField].normalized_value)	×
UNCOV 41	if env_tgt.val.has_field(RemotePlatformField):	×
UNCOV 42	remote_platform = env_tgt.val[RemotePlatformField].value	×
UNCOV 43	if remote_platform:	×
UNCOV 44	return Platform(remote_platform)	×
UNCOV 45	return Platform.create_for_localhost()	×
46
47
48	@rule	1✔
49	async def complete_environment_vars(	1✔
50	session_values: SessionValues,
51	env_tgt: EnvironmentTarget,
52	global_options: GlobalOptions,
53	environments_subsystem: EnvironmentsSubsystem,
54	) -> CompleteEnvironmentVars:
55	# If a local environment is used, we simply use SessionValues. Otherwise, we need to run `env`
56	# and parse the output.
57	#
58	# Note that running `env` works for both Docker and Remote Execution because we intentionally
59	# do not strip the environment from either runtime. It is reasonable to not strip because
60	# every user will have the same consistent Docker image or Remote Execution environment, unlike
61	# local environments.
UNCOV 62	if env_tgt.val:	×
UNCOV 63	if env_tgt.val.has_field(DockerImageField):	×
UNCOV 64	description_of_env_source = f"the Docker image {env_tgt.val[DockerImageField].value}"	×
UNCOV 65	elif env_tgt.val.has_field(RemotePlatformField):	×
UNCOV 66	description_of_env_source = "the remote execution environment"	×
67	else:
68	# Else, it's a local environment.
UNCOV 69	return session_values[CompleteEnvironmentVars]	×
70	else:
UNCOV 71	if environments_subsystem.remote_execution_used_globally(global_options):	×
UNCOV 72	description_of_env_source = "the remote execution environment"	×
73	else:
UNCOV 74	return session_values[CompleteEnvironmentVars]	×
75
UNCOV 76	env_process_result = await execute_process_or_raise(	×
77	**implicitly(
78	Process(
79	["env", "-0"],
80	description=f"Extract environment variables from {description_of_env_source}",
81	level=LogLevel.DEBUG,
82	cache_scope=env_tgt.executable_search_path_cache_scope(),
83	)
84	)
85	)
UNCOV 86	result = {}	×
UNCOV 87	for line in env_process_result.stdout.decode("utf-8").rstrip().split("\0"):	×
UNCOV 88	if not line:	×
89	continue	×
UNCOV 90	k, v = line.split("=", maxsplit=1)	×
UNCOV 91	result[k] = v	×
UNCOV 92	return CompleteEnvironmentVars(result)	×
93
94
95	@rule	1✔
96	async def environment_path_variable() -> PathEnvironmentVariable:	1✔
UNCOV 97	env = await environment_vars_subset(EnvironmentVarsRequest(("PATH",)), **implicitly())	×
UNCOV 98	path = env.get("PATH", None)	×
UNCOV 99	return PathEnvironmentVariable(path.split(os.pathsep) if path else ())	×
100
101
102	def rules():	1✔
UNCOV 103	return collect_rules()	×

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