19018388440

Committed 02 Nov 2025 09:26PM UTC coverage: 80.301% (+0.001%) from 80.3%

Build # 19018388440

Build Type

Pull #22808

github

Committed by

web-flow

Commit Message

Merge ac7b36d30 into 6c024e162

Pull Request Pull Request #22808: A new "pants next-gen" command-line parser.

Run Details

77995 of 97128 relevant lines covered (80.3%)

3.35 hits per line

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

71.43

/src/python/pants/util/osutil.py

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

from __future__ import annotations

import errno
import getpass
import logging
import os
import platform
import posix
from functools import reduce

logger = logging.getLogger(__name__)


def _compute_cpu_count() -> int:
    # We use `sched_getaffinity()` to get the number of cores available to the process, rather than
    # the raw number of cores. This sometimes helps for containers to accurately report their # of
    # cores, rather than the host's.
    sched_getaffinity = getattr(os, "sched_getaffinity", None)
    if sched_getaffinity:
        return len(sched_getaffinity(0))
    cpu_count = os.cpu_count()
    if cpu_count:
        return cpu_count
    return 2


CPU_COUNT = _compute_cpu_count()


OS_ALIASES = {
    "macos": {"macos", "darwin", "macosx", "mac os x", "mac"},
    "linux": {"linux", "linux2"},
}

ARCH_ALIASES = {
    "x86_64": {"x86_64", "x86-64", "amd64"},
    "arm64": {"arm64", "aarch64"},
}

Pid = int


def get_arch_name(uname_result: posix.uname_result | None = None) -> str:
    """
    :API: public
    """
    if uname_result is None:
        uname_result = os.uname()
    return uname_result.machine.lower()


def get_os_name(uname_result: posix.uname_result | None = None) -> str:
    """
    :API: public
    """
    if uname_result is None:
        uname_result = os.uname()
    return uname_result.sysname.lower()


def normalize_arch_name(arch_name: str) -> str:
    """
    :API: public
    """
    return _normalize(arch_name, ARCH_ALIASES, "architecture")


def normalize_os_name(os_name: str) -> str:
    """
    :API: public
    """
    return _normalize(os_name, OS_ALIASES, "operating system")


def _normalize(name: str, aliases: dict[str, set[str]], warning_hint: str) -> str:
    for proper_name, alias_set in aliases.items():
        if name in alias_set:
            return proper_name
    else:
        logger.warning(
            "Unknown {hint} name: {bad}, known names are: {known}".format(
                hint=warning_hint, bad=name, known=", ".join(sorted(_values(aliases)))
            )
        )
        return name


def get_normalized_os_name() -> str:
    return normalize_os_name(get_os_name())


def get_normalized_arch_name() -> str:
    return normalize_arch_name(get_arch_name())


def macos_major_version() -> None | int:
    if not hasattr(platform, "mac_ver"):
        return None

    version = platform.mac_ver()[0]
    if not version:
        return None

    return int(version.split(".", 1)[0])


def is_macos_big_sur() -> bool:
    return macos_major_version() == 11


def getuser() -> str:
    try:
        return getpass.getuser()
    except KeyError:
        # Work when running with a uid not associated with a user,
        # e.g., in a docker container with a host uid.
        return str(os.getuid())


def _values(aliases: dict[str, set[str]]) -> set[str]:
    return reduce(set.union, aliases.values())


# From kill(2) on OSX 10.13:
#     [EINVAL]           Sig is not a valid, supported signal number.
#
#     [EPERM]            The sending process is not the super-user and its effective user id does not match the effective user-id of the receiving process.  When signaling a process group, this error is returned if
#                        any members of the group could not be signaled.
#
#     [ESRCH]            No process or process group can be found corresponding to that specified by pid.
#
#     [ESRCH]            The process id was given as 0, but the sending process does not have a process group.
def safe_kill(pid: Pid, signum: int) -> None:
    """Kill a process with the specified signal, catching nonfatal errors."""
    assert isinstance(pid, Pid)
    assert isinstance(signum, int)
    try:
        os.kill(pid, signum)
    except OSError as e:
        if e.errno in [errno.ESRCH, errno.EPERM]:
            pass
        elif e.errno == errno.EINVAL:
            raise ValueError(f"Invalid signal number {signum}: {e}", e)
        else:
            raise

1	# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	12✔
5
6	import errno	12✔
7	import getpass	12✔
8	import logging	12✔
9	import os	12✔
10	import platform	12✔
11	import posix	12✔
12	from functools import reduce	12✔
13
14	logger = logging.getLogger(__name__)	12✔
15
16
17	def _compute_cpu_count() -> int:	12✔
18	# We use `sched_getaffinity()` to get the number of cores available to the process, rather than
19	# the raw number of cores. This sometimes helps for containers to accurately report their # of
20	# cores, rather than the host's.
21	sched_getaffinity = getattr(os, "sched_getaffinity", None)	12✔
22	if sched_getaffinity:	12✔
23	return len(sched_getaffinity(0))	11✔
24	cpu_count = os.cpu_count()	1✔
25	if cpu_count:	1✔
26	return cpu_count	1✔
27	return 2	×
28
29
30	CPU_COUNT = _compute_cpu_count()	12✔
31
32
33	OS_ALIASES = {	12✔
34	"macos": {"macos", "darwin", "macosx", "mac os x", "mac"},
35	"linux": {"linux", "linux2"},
36	}
37
38	ARCH_ALIASES = {	12✔
39	"x86_64": {"x86_64", "x86-64", "amd64"},
40	"arm64": {"arm64", "aarch64"},
41	}
42
43	Pid = int	12✔
44
45
46	def get_arch_name(uname_result: posix.uname_result \| None = None) -> str:	12✔
47	"""
48	:API: public
49	"""
50	if uname_result is None:	11✔
51	uname_result = os.uname()	11✔
52	return uname_result.machine.lower()	11✔
53
54
55	def get_os_name(uname_result: posix.uname_result \| None = None) -> str:	12✔
56	"""
57	:API: public
58	"""
59	if uname_result is None:	11✔
60	uname_result = os.uname()	11✔
61	return uname_result.sysname.lower()	11✔
62
63
64	def normalize_arch_name(arch_name: str) -> str:	12✔
65	"""
66	:API: public
67	"""
68	return _normalize(arch_name, ARCH_ALIASES, "architecture")	11✔
69
70
71	def normalize_os_name(os_name: str) -> str:	12✔
72	"""
73	:API: public
74	"""
75	return _normalize(os_name, OS_ALIASES, "operating system")	11✔
76
77
78	def _normalize(name: str, aliases: dict[str, set[str]], warning_hint: str) -> str:	12✔
79	for proper_name, alias_set in aliases.items():	11✔
80	if name in alias_set:	11✔
81	return proper_name	11✔
82	else:
83	logger.warning(	1✔
84	"Unknown {hint} name: {bad}, known names are: {known}".format(
85	hint=warning_hint, bad=name, known=", ".join(sorted(_values(aliases)))
86	)
87	)
88	return name	1✔
89
90
91	def get_normalized_os_name() -> str:	12✔
92	return normalize_os_name(get_os_name())	11✔
93
94
95	def get_normalized_arch_name() -> str:	12✔
96	return normalize_arch_name(get_arch_name())	11✔
97
98
99	def macos_major_version() -> None \| int:	12✔
100	if not hasattr(platform, "mac_ver"):	×
101	return None	×
102
103	version = platform.mac_ver()[0]	×
104	if not version:	×
105	return None	×
106
107	return int(version.split(".", 1)[0])	×
108
109
110	def is_macos_big_sur() -> bool:	12✔
111	return macos_major_version() == 11	×
112
113
114	def getuser() -> str:	12✔
115	try:	1✔
116	return getpass.getuser()	1✔
117	except KeyError:	×
118	# Work when running with a uid not associated with a user,
119	# e.g., in a docker container with a host uid.
120	return str(os.getuid())	×
121
122
123	def _values(aliases: dict[str, set[str]]) -> set[str]:	12✔
124	return reduce(set.union, aliases.values())	1✔
125
126
127	# From kill(2) on OSX 10.13:
128	# [EINVAL] Sig is not a valid, supported signal number.
129	#
130	# [EPERM] The sending process is not the super-user and its effective user id does not match the effective user-id of the receiving process. When signaling a process group, this error is returned if
131	# any members of the group could not be signaled.
132	#
133	# [ESRCH] No process or process group can be found corresponding to that specified by pid.
134	#
135	# [ESRCH] The process id was given as 0, but the sending process does not have a process group.
136	def safe_kill(pid: Pid, signum: int) -> None:	12✔
137	"""Kill a process with the specified signal, catching nonfatal errors."""
138	assert isinstance(pid, Pid)	×
139	assert isinstance(signum, int)	×
140	try:	×
141	os.kill(pid, signum)	×
142	except OSError as e:	×
143	if e.errno in [errno.ESRCH, errno.EPERM]:	×
144	pass	×
145	elif e.errno == errno.EINVAL:	×
146	raise ValueError(f"Invalid signal number {signum}: {e}", e)	×
147	else:
148	raise	×

pantsbuild / pants / 19018388440

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