mirror of
https://github.com/mongodb/mongo.git
synced 2024-12-01 09:32:32 +01:00
SERVER-63737 Add toolchain data script
This commit is contained in:
Ryan Egesdahl
Evergreen Agent
parent
4beaff69ac
commit
36821093f2
815
buildscripts/toolchains.py
Executable file
815
buildscripts/toolchains.py
Executable file
@ -0,0 +1,815 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Provides data regarding mongodbtoolchain."""
|
||||
|
||||
import argparse
|
||||
import enum
|
||||
import inspect
|
||||
import os
|
||||
import pathlib
|
||||
import string
|
||||
import sys
|
||||
import warnings
|
||||
|
||||
from types import FunctionType
|
||||
from typing import Any, Callable, Dict, Generator, Iterator, List, Mapping
|
||||
from typing import Optional, Set, Sequence, Tuple, Type, Union, overload
|
||||
|
||||
import yaml
|
||||
|
||||
__all__ = [
|
||||
'DEFAULT_DATA_FILE',
|
||||
'Toolchain',
|
||||
'ToolchainConfig',
|
||||
'ToolchainDataException',
|
||||
'ToolchainReleaseName',
|
||||
'ToolchainVersionName',
|
||||
'Toolchains',
|
||||
]
|
||||
|
||||
DEFAULT_DATA_FILE: pathlib.Path = pathlib.Path(__file__).parent / '../etc/toolchains.yaml'
|
||||
|
||||
|
||||
class ToolchainVersionName(enum.Enum):
|
||||
"""Represents a "named" toolchain version, such as "stable" or "testing"."""
|
||||
|
||||
STABLE: str = 'stable'
|
||||
TESTING: str = 'testing'
|
||||
|
||||
# TODO: SERVER-64481 - remove these false positives
|
||||
# pylint: disable=invalid-str-returned
|
||||
def __str__(self) -> str:
|
||||
return self.value
|
||||
|
||||
|
||||
class ToolchainReleaseName(enum.Enum):
|
||||
"""Represents a "named" toolchain release, such as "rollback" or "current"."""
|
||||
|
||||
ROLLBACK: str = 'rollback'
|
||||
CURRENT: str = 'current'
|
||||
LATEST: str = 'latest'
|
||||
|
||||
# TODO: SERVER-64481 - remove these false positives
|
||||
# pylint: disable=invalid-str-returned
|
||||
def __str__(self) -> str:
|
||||
return self.value
|
||||
|
||||
|
||||
class ToolchainDistroName(enum.Enum):
|
||||
"""Represents a distribution for which the toolchain is built."""
|
||||
|
||||
AMAZON1_2012: Tuple[str, ...] = (
|
||||
'amazon1-2012',
|
||||
'linux-64-amzn',
|
||||
)
|
||||
AMAZON1_2018: Tuple[str, ...] = ('amazon1-2018', )
|
||||
AMAZON2: Tuple[str, ...] = ('amazon2', )
|
||||
ARCHLINUX: Tuple[str, ...] = ('archlinux', )
|
||||
CENTOS6: Tuple[str, ...] = ('centos6', )
|
||||
DEBIAN8: Tuple[str, ...] = ('debian81', )
|
||||
DEBIAN9: Tuple[str, ...] = ('debian92', )
|
||||
DEBIAN10: Tuple[str, ...] = ('debian10', )
|
||||
MACOS1012: Tuple[str, ...] = ('macos-1012', )
|
||||
MACOS1014: Tuple[str, ...] = ('macos-1014', )
|
||||
MACOS1100: Tuple[str, ...] = ('macos-1100', )
|
||||
RHEL6: Tuple[str, ...] = ('rhel6', 'rhel62', 'rhel67')
|
||||
RHEL7: Tuple[str, ...] = ('rhel7', 'rhel70', 'rhel71', 'rhel72', 'rhel76', 'ubi7')
|
||||
RHEL8: Tuple[str, ...] = ('rhel8', 'rhel80', 'rhel81', 'rhel82', 'rhel83', 'rhel84', 'ubi8')
|
||||
SUSE12: Tuple[str, ...] = ('suse12', 'suse12-sp5')
|
||||
SUSE15: Tuple[str, ...] = ('suse15', 'suse15-sp0', 'suse15-sp2')
|
||||
UBUNTU1404: Tuple[str, ...] = ('ubuntu1404', )
|
||||
UBUNTU1604: Tuple[str, ...] = ('ubuntu1604', )
|
||||
UBUNTU1804: Tuple[str, ...] = ('ubuntu1804', )
|
||||
UBUNTU2004: Tuple[str, ...] = ('ubuntu2004', )
|
||||
DEFAULT: Tuple[str, ...] = ('default', )
|
||||
|
||||
@classmethod
|
||||
def from_str(cls, text: str) -> 'ToolchainDistroName':
|
||||
"""Return the enumeration object matching a given string."""
|
||||
|
||||
for distro in cls:
|
||||
if text in distro.value:
|
||||
return distro
|
||||
|
||||
raise ValueError(f"Unknown distro string: {text}")
|
||||
|
||||
# TODO: SERVER-64481 - remove these false positives
|
||||
# pylint: disable=unsubscriptable-object
|
||||
def __str__(self) -> str:
|
||||
return self.value[0]
|
||||
|
||||
|
||||
class ToolchainArchName(enum.Enum):
|
||||
"""Represents an architecture for which the toolchain is built."""
|
||||
|
||||
ARM64: Tuple[str, ...] = ('arm64', 'aarch64')
|
||||
PPC64LE: Tuple[str, ...] = ('ppc64le', 'power8')
|
||||
S390X: Tuple[str, ...] = ('s390x', 'zSeries')
|
||||
X86_64: Tuple[str, ...] = ('x86_64', )
|
||||
DEFAULT: Tuple[str, ...] = ('', )
|
||||
|
||||
@classmethod
|
||||
def from_str(cls, text: str) -> 'ToolchainArchName':
|
||||
"""Return the enumeratrion object matching a given string."""
|
||||
|
||||
for arch in cls:
|
||||
if text in arch.value:
|
||||
return arch
|
||||
|
||||
raise ValueError(f"Unknown arch string: {text}")
|
||||
|
||||
# TODO: SERVER-64481 - remove these false positives
|
||||
# pylint: disable=unsubscriptable-object
|
||||
def __str__(self) -> str:
|
||||
return self.value[0]
|
||||
|
||||
|
||||
class ToolchainDataException(Exception):
|
||||
"""Represents a problem encountered while reading or querying toolchain data."""
|
||||
|
||||
|
||||
class ToolchainPlatform:
|
||||
"""Represents a platform for which the toolchain is built."""
|
||||
|
||||
def __init__(self, distro_id: str, arch: Optional[ToolchainArchName] = None) -> None:
|
||||
"""Parse a distro_id into a full toolchain platform."""
|
||||
|
||||
self._distro_id: str = distro_id
|
||||
|
||||
self._distro: Optional[ToolchainDistroName] = None
|
||||
self._arch: Optional[ToolchainArchName] = arch
|
||||
self._tag: Optional[str] = None
|
||||
|
||||
# These two actions are order-dependent!
|
||||
self._distro_length: int = self._find_distro_length()
|
||||
self._arch_span: Tuple[int, int] = self._find_arch_span()
|
||||
|
||||
def _split_distro_id(self, start: int = 0) -> Tuple[str, str]:
|
||||
return self._distro_id[start:].split('-', 1)[0], self._distro_id[start:].split('.')[0]
|
||||
|
||||
def _find_distro_length(self) -> int:
|
||||
for distro in ToolchainDistroName:
|
||||
for name in distro.value:
|
||||
|
||||
if not name:
|
||||
continue
|
||||
|
||||
if name.lower() in self._split_distro_id():
|
||||
return len(name)
|
||||
|
||||
raise ValueError(f"Failed to match distro from distro_id: `{self._distro_id}'")
|
||||
|
||||
def _find_arch_span(self) -> Tuple[int, int]:
|
||||
arch_span: Optional[Tuple[int, int]] = None
|
||||
|
||||
for arch in ToolchainArchName:
|
||||
for name in arch.value:
|
||||
|
||||
if not name:
|
||||
continue
|
||||
|
||||
iter_start = self._distro_length + 1
|
||||
while iter_start < len(self._distro_id):
|
||||
|
||||
if name.lower() in self._split_distro_id(self._distro_length):
|
||||
arch_span = (iter_start, len(name))
|
||||
|
||||
iter_start += len(name)
|
||||
if iter_start < len(self._distro_id) and self._distro_id[iter_start] in ('-',
|
||||
'.'):
|
||||
iter_start += 1
|
||||
|
||||
if arch_span is None:
|
||||
arch_span = (self._distro_length, 0)
|
||||
|
||||
return arch_span
|
||||
|
||||
@property
|
||||
def distro_id(self) -> str:
|
||||
"""Return the distro_id."""
|
||||
|
||||
return self._distro_id
|
||||
|
||||
@property
|
||||
def distro(self) -> ToolchainDistroName:
|
||||
"""Return the distro component of the distro_id."""
|
||||
|
||||
if self._distro is None:
|
||||
distro_length: int = self._distro_length
|
||||
self._distro = ToolchainDistroName.from_str(self._distro_id[:distro_length])
|
||||
|
||||
return self._distro
|
||||
|
||||
@property
|
||||
def arch(self) -> ToolchainArchName:
|
||||
"""Return the architecture component of the distro_id."""
|
||||
|
||||
if self._arch is None:
|
||||
arch_span: Tuple[int, int] = self._arch_span
|
||||
if arch_span[1] == 0:
|
||||
# There is no architecture specified in the distro_id
|
||||
if self.distro == ToolchainDistroName.DEFAULT:
|
||||
# The "default" distro is allowed to have a nonexistent arch
|
||||
self._arch = ToolchainArchName.DEFAULT
|
||||
else:
|
||||
self._arch = ToolchainArchName.X86_64
|
||||
else:
|
||||
self._arch = ToolchainArchName.from_str(
|
||||
self.distro_id[arch_span[0]:arch_span[0] + arch_span[1]])
|
||||
|
||||
return self._arch
|
||||
|
||||
@property
|
||||
def tag(self) -> Optional[str]:
|
||||
"""Return the descriptive tag component of the distro_id."""
|
||||
|
||||
if self._tag is None:
|
||||
arch_span: Tuple[int, int] = self._arch_span
|
||||
if arch_span[0] + arch_span[1] + 1 < len(self._distro_id):
|
||||
self._tag = self._distro_id[arch_span[0] + arch_span[1] + 1:]
|
||||
else:
|
||||
self._tag = ''
|
||||
|
||||
if self._tag:
|
||||
return self._tag
|
||||
|
||||
return None
|
||||
|
||||
def __str__(self) -> str:
|
||||
result: str = f"{self.distro}"
|
||||
|
||||
if self.arch != ToolchainArchName.DEFAULT:
|
||||
result += f".{self.arch}"
|
||||
|
||||
if self.tag is not None:
|
||||
result += f".{self.tag}"
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class ToolchainConfig:
|
||||
"""Represents a toolchain configuration."""
|
||||
|
||||
def __init__(self, data_file: pathlib.Path, platform: ToolchainPlatform) -> None:
|
||||
"""Construct a toolchain configuration from a data file."""
|
||||
|
||||
try:
|
||||
with open(data_file.absolute(), 'r', encoding='utf-8') as yaml_stream:
|
||||
self._data = yaml.safe_load(yaml_stream)
|
||||
except yaml.YAMLError as parent_exc:
|
||||
raise ToolchainDataException(
|
||||
f"Could not read toolchain data file: `{data_file}'") from parent_exc
|
||||
|
||||
self._platform: ToolchainPlatform = platform
|
||||
|
||||
@property
|
||||
def base_path(self) -> pathlib.Path:
|
||||
"""Return the base (installed) path for toolchain releases."""
|
||||
|
||||
return pathlib.Path(self._data['toolchains']['base_path'])
|
||||
|
||||
@property
|
||||
def all_releases(self) -> Dict[str, Dict[str, str]]:
|
||||
"""Return all known releases in the data file."""
|
||||
|
||||
try:
|
||||
return self._data['toolchains']['releases']
|
||||
except (KeyError, TypeError):
|
||||
return {}
|
||||
|
||||
@property
|
||||
def releases(self) -> Dict[str, str]:
|
||||
"""Return all releases for a the current platform."""
|
||||
|
||||
# Successively match the distro and/or platform against available toolchains.
|
||||
# If none is found, use the default entry. If the default entry doesn't exist,
|
||||
# we want to know about it because that's a misconfiguration.
|
||||
platform_section: Optional[Dict[str, str]] = None
|
||||
|
||||
if self._platform.distro_id in self.all_releases:
|
||||
platform_section = self.all_releases[self._platform.distro_id]
|
||||
elif self._platform.distro != ToolchainDistroName.DEFAULT:
|
||||
if str(self._platform) in self.all_releases:
|
||||
platform_section = self.all_releases[str(self._platform)]
|
||||
elif f"{self._platform.distro}.{self._platform.arch}" in self.all_releases:
|
||||
platform_section = self.all_releases[
|
||||
f"{self._platform.distro}.{self._platform.arch}"]
|
||||
elif f"{self._platform.distro}.{self._platform.tag}" in self.all_releases:
|
||||
platform_section = self.all_releases[
|
||||
f"{self._platform.distro}.{self._platform.tag}"]
|
||||
elif f"{self._platform.distro}" in self.all_releases:
|
||||
platform_section = self.all_releases[f"{self._platform.distro}"]
|
||||
|
||||
if not platform_section:
|
||||
try:
|
||||
platform_section = self.all_releases['default']
|
||||
except KeyError:
|
||||
return {}
|
||||
|
||||
return platform_section
|
||||
|
||||
@property
|
||||
def versions(self) -> List[str]:
|
||||
"""Return all known versions in the data file."""
|
||||
|
||||
return self._data['toolchains']['versions']
|
||||
|
||||
@property
|
||||
def aliases(self) -> Dict[str, str]:
|
||||
"""Return all known version aliases in the data file."""
|
||||
|
||||
return self._data['toolchains']['version_aliases']
|
||||
|
||||
@property
|
||||
def revisions_dir(self) -> Optional[pathlib.Path]:
|
||||
"""Return the legacy revisions directory for toolchain releases."""
|
||||
|
||||
warnings.warn(
|
||||
f"This is legacy toolchain usage. Call {self.__class__.__name__}.releases_dir() instead.",
|
||||
DeprecationWarning, stacklevel=2)
|
||||
|
||||
return self.base_path.joinpath('revisions')
|
||||
|
||||
@property
|
||||
def releases_dir(self) -> pathlib.Path:
|
||||
"""Return the directory where toolchain releases are installed."""
|
||||
|
||||
return self.base_path.joinpath('releases')
|
||||
|
||||
def search_releases(self, release_name: str) -> Optional[str]:
|
||||
"""Search configured releases for a given release name."""
|
||||
|
||||
try:
|
||||
return self.releases[release_name]
|
||||
except KeyError:
|
||||
return None
|
||||
|
||||
|
||||
class Toolchain:
|
||||
"""Represents the raw toolchain data."""
|
||||
|
||||
def __init__(self, config: ToolchainConfig, release: str) -> None:
|
||||
"""Construct a toolchain object from a supplied release and version."""
|
||||
|
||||
self._config = config
|
||||
self._release = release
|
||||
|
||||
@property
|
||||
def release(self) -> str:
|
||||
"""Return the toolchain release ID."""
|
||||
|
||||
return self._release
|
||||
|
||||
@property
|
||||
def install_path(self) -> pathlib.Path:
|
||||
"""Return the path to where the toolchain should be installed."""
|
||||
|
||||
# We need to determine if the configured release is a legacy release
|
||||
# that only appears in revisions_dir. We can tell the difference
|
||||
# because legacy releases are all identified by git commit hashes.
|
||||
if len(self.release) == 40 and set(self.release) <= set(string.hexdigits):
|
||||
with warnings.catch_warnings():
|
||||
warnings.simplefilter("ignore")
|
||||
return self._config.revisions_dir / self.release
|
||||
else:
|
||||
return self._config.releases_dir / self.release
|
||||
|
||||
def exec_path(self, version: Union[ToolchainVersionName, str] = None) -> Optional[pathlib.Path]:
|
||||
"""Return a path to a specific toolchain version."""
|
||||
|
||||
install_path = self.install_path
|
||||
|
||||
if isinstance(version, ToolchainVersionName):
|
||||
version = version.value
|
||||
elif version not in self._config.versions:
|
||||
try:
|
||||
version = self._config.aliases[version]
|
||||
except KeyError:
|
||||
raise ValueError(
|
||||
f"Toolchain version `{version}' not defined in data file") from None
|
||||
|
||||
return install_path / version
|
||||
|
||||
|
||||
class Toolchains(Mapping[Union[ToolchainReleaseName, str], Toolchain]):
|
||||
"""Represents a collection of toolchains that may or may not be installed on a system."""
|
||||
|
||||
def __init__(self, config: ToolchainConfig) -> None:
|
||||
"""Manipulate raw toolchain configuration data."""
|
||||
|
||||
self._config = config
|
||||
self._available: Optional[List[str]] = None
|
||||
|
||||
def _search_filesystem(self) -> List[str]:
|
||||
release_dirs: List[pathlib.Path] = []
|
||||
|
||||
releases_dir: Optional[pathlib.Path] = self._config.releases_dir
|
||||
if releases_dir.exists():
|
||||
release_dirs.extend([path for path in releases_dir.iterdir() if path.is_dir()])
|
||||
|
||||
with warnings.catch_warnings():
|
||||
warnings.simplefilter("ignore")
|
||||
revisions_dir: Optional[pathlib.Path] = self._config.revisions_dir
|
||||
|
||||
if revisions_dir.exists():
|
||||
release_dirs.extend([path for path in revisions_dir.iterdir() if path.is_dir()])
|
||||
|
||||
if release_dirs:
|
||||
return [
|
||||
path.name for path in sorted(release_dirs, key=lambda path: path.stat().st_mtime,
|
||||
reverse=True)
|
||||
]
|
||||
|
||||
return []
|
||||
|
||||
@property
|
||||
def available(self) -> List[str]:
|
||||
"""Return a list of all installed toolchain releases ordered from newest to oldest."""
|
||||
|
||||
if self._available is None:
|
||||
self._available = self._search_filesystem()
|
||||
|
||||
return self._available
|
||||
|
||||
@property
|
||||
def configured(self) -> List[str]:
|
||||
"""Return a list of all configured toolchain releases."""
|
||||
|
||||
configured: Set[str] = {
|
||||
self._config.search_releases(name.value)
|
||||
for name in ToolchainReleaseName
|
||||
}
|
||||
|
||||
configured.add(self._config.search_releases('default'))
|
||||
|
||||
return [release for release in configured if release is not None]
|
||||
|
||||
@property
|
||||
def latest(self) -> Optional[str]:
|
||||
"""Return the latest installed toolchain release.
|
||||
|
||||
This can possibly be different from available[0]
|
||||
because we take into account the "latest" symlink present on end-user toolchain
|
||||
installations, which could potentially not be the "newest" for any reason. Therefore, these
|
||||
methods can be used to determine whether the "latest" symlink is out of date.
|
||||
"""
|
||||
|
||||
latest_symlink: Optional[pathlib.Path] = None
|
||||
try:
|
||||
latest_symlink = self._config.releases_dir.joinpath('latest')
|
||||
except AttributeError:
|
||||
latest_symlink = None
|
||||
|
||||
if latest_symlink.exists():
|
||||
return pathlib.Path(os.readlink(latest_symlink.absolute())).name
|
||||
|
||||
return self.available[0] or None
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
for release in set(self.available).union(set(self.configured)):
|
||||
yield release
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(list(self.__iter__()))
|
||||
|
||||
@overload
|
||||
def __getitem__(self, key: ToolchainReleaseName) -> Toolchain:
|
||||
...
|
||||
|
||||
@overload
|
||||
def __getitem__(self, key: str) -> Toolchain:
|
||||
...
|
||||
|
||||
@overload
|
||||
def __getitem__(self, key: Union[ToolchainReleaseName, str]) -> Toolchain:
|
||||
...
|
||||
|
||||
def __getitem__(self, key: Union[ToolchainReleaseName, str]) -> Toolchain:
|
||||
"""Return the named toolchain release.
|
||||
|
||||
This method supports two disjoint use cases:
|
||||
|
||||
1. We don't know the release ID and pass a release name. Here, we are attempting to
|
||||
determine the release ID corresponding to the name for the current platform. If no
|
||||
such release name is configured, we raise an exception to indicate a potential
|
||||
misconfiguration or code error.
|
||||
|
||||
2. We know the release ID and want to determine whether it is installed. In this case,
|
||||
None is returned if the release is not available to indicate it is not installed.
|
||||
"""
|
||||
|
||||
release: Optional[str] = None
|
||||
|
||||
if isinstance(key, ToolchainReleaseName):
|
||||
# We don't know the release ID and are querying by name. This supports use case 1.
|
||||
if key == ToolchainReleaseName.LATEST:
|
||||
latest = self.latest
|
||||
if latest is None:
|
||||
raise KeyError(key)
|
||||
|
||||
release = self.latest
|
||||
|
||||
release = self._config.search_releases(str(key))
|
||||
else:
|
||||
# We know the release ID and want to know whether it's installed. This supports use
|
||||
# case 2. This is the "short path" for the method because any string other than a
|
||||
# real release ID or release name causes a return.
|
||||
if key == str(ToolchainReleaseName.LATEST):
|
||||
release = self.latest
|
||||
if key in self.available:
|
||||
release = key
|
||||
if key in [name.value for name in ToolchainReleaseName]:
|
||||
release = self._config.search_releases(key)
|
||||
|
||||
if release is None:
|
||||
raise KeyError(key)
|
||||
|
||||
return Toolchain(self._config, release)
|
||||
|
||||
|
||||
class _FormatterClass:
|
||||
"""A protocol for a formatter class.
|
||||
|
||||
HelpFormatter is missing this protocol in its inheritance tree in
|
||||
some versions of the mypy typeshed, which causes a spurious mypy error
|
||||
when trying to assign a custom HelpFormatter subclass. This is just
|
||||
here for NicerHelpFormatter to inherit from it and prevent the error.
|
||||
"""
|
||||
|
||||
def __call__(self, prog: str) -> argparse.HelpFormatter:
|
||||
...
|
||||
|
||||
|
||||
# pylint: disable=protected-access
|
||||
class NicerHelpFormatter(argparse.HelpFormatter, _FormatterClass):
|
||||
"""A HelpFormatter with nicer output than the default."""
|
||||
|
||||
def __init__(self, prog: str, indent_increment: int = 2, max_help_position: int = 32,
|
||||
width=None) -> None:
|
||||
super().__init__(prog=prog, indent_increment=indent_increment,
|
||||
max_help_position=max_help_position, width=width)
|
||||
|
||||
def __call__(self, prog: str) -> argparse.HelpFormatter:
|
||||
return NicerHelpFormatter(prog)
|
||||
|
||||
def _format_action(self, action: argparse.Action) -> str:
|
||||
result = super()._format_action(action)
|
||||
if isinstance(action, (argparse._SubParsersAction, DictChoiceAction)):
|
||||
return (" " * self._current_indent) + f"{result.lstrip()}"
|
||||
return result
|
||||
|
||||
def _format_action_invocation(self, action: argparse.Action) -> str:
|
||||
if isinstance(action, (argparse._SubParsersAction, DictChoiceAction)):
|
||||
return ""
|
||||
if not action.option_strings:
|
||||
default = self._get_default_metavar_for_optional(action)
|
||||
metavar, = self._metavar_formatter(action, default)(1)
|
||||
return metavar
|
||||
else:
|
||||
parts: List[str] = []
|
||||
|
||||
if action.nargs == 0:
|
||||
parts.extend(action.option_strings)
|
||||
else:
|
||||
default = self._get_default_metavar_for_optional(action)
|
||||
args_string = self._format_args(action, default)
|
||||
|
||||
for option_string in action.option_strings:
|
||||
parts.append(option_string)
|
||||
|
||||
return f"{' '.join(parts)} {args_string}"
|
||||
|
||||
return ' '.join(parts)
|
||||
|
||||
def _iter_indented_subactions(
|
||||
self, action: argparse.Action) -> Generator[argparse.Action, None, None]:
|
||||
if isinstance(action, (argparse._SubParsersAction, DictChoiceAction)):
|
||||
try:
|
||||
get_subactions = action._get_subactions
|
||||
except AttributeError:
|
||||
pass
|
||||
else:
|
||||
for subaction in get_subactions():
|
||||
yield subaction
|
||||
else:
|
||||
for subaction in super()._iter_indented_subactions(action):
|
||||
yield subaction
|
||||
|
||||
def _metavar_formatter(self, action: argparse.Action,
|
||||
default_metavar: str) -> Callable[[int], Tuple[str, ...]]:
|
||||
if action.metavar is not None:
|
||||
result = action.metavar
|
||||
elif action.choices is not None:
|
||||
choice_strs = [f"{choice}" for choice in action.choices]
|
||||
result = f"{' | '.join(choice_strs)}"
|
||||
if action.required:
|
||||
result = f"({result})"
|
||||
else:
|
||||
result = f"[{result}]"
|
||||
else:
|
||||
result = default_metavar
|
||||
|
||||
def _format(tuple_size: int) -> Tuple[str, ...]:
|
||||
if isinstance(result, tuple):
|
||||
return result
|
||||
else:
|
||||
return (result, ) * tuple_size
|
||||
|
||||
return _format
|
||||
|
||||
|
||||
# pylint: disable=protected-access, redefined-builtin, redefined-outer-name
|
||||
class DictChoiceAction(argparse._StoreAction):
|
||||
"""An action with nicer per-choice formatting."""
|
||||
|
||||
class _ChoicesPseudoAction(argparse.Action):
|
||||
def __init__(self, name: str, aliases: str, help: Optional[str] = None) -> None:
|
||||
|
||||
metavar = dest = name
|
||||
if aliases:
|
||||
metavar += f" {' | '.join(aliases)}"
|
||||
if self.required:
|
||||
metavar = f"({metavar})"
|
||||
else:
|
||||
metavar = f"[{metavar})"
|
||||
|
||||
super().__init__(option_strings=[], dest=dest, help=help, metavar=metavar)
|
||||
|
||||
def __call__(self, parser: argparse.ArgumentParser, namespace: argparse.Namespace,
|
||||
values: Union[str, Sequence[Any]], __: Optional[str] = None) -> None:
|
||||
|
||||
parser.print_help()
|
||||
parser.exit()
|
||||
|
||||
def __init__(self, option_strings: Optional[List[str]] = None, dest: Optional[str] = None,
|
||||
nargs: Optional[int] = None, const: Optional[Any] = None,
|
||||
default: Optional[Any] = None, type: Optional[type] = None,
|
||||
choices: Optional[Dict[str, str]] = None, required: bool = False,
|
||||
help: Optional[str] = None, metavar: Optional[str] = None) -> None:
|
||||
|
||||
super().__init__(option_strings=option_strings, dest=dest, nargs=nargs, const=const,
|
||||
default=default, type=type, choices=choices, required=required, help=help,
|
||||
metavar=metavar)
|
||||
|
||||
def _get_subactions(self):
|
||||
choices_actions: List[argparse.Action] = []
|
||||
for name, help_text in self.choices.items():
|
||||
choice_action = self._ChoicesPseudoAction(name, [], help_text)
|
||||
choices_actions.append(choice_action)
|
||||
return choices_actions
|
||||
|
||||
|
||||
class NicerArgumentParser(argparse.ArgumentParser):
|
||||
"""An argument parser with nicer help output."""
|
||||
|
||||
def __init__(self, prog: Optional[str] = None, usage: Optional[str] = None,
|
||||
description: Optional[str] = None, epilog: Optional[str] = None,
|
||||
parents: List[argparse.ArgumentParser] = None, prefix_chars: Optional[str] = '-',
|
||||
fromfile_prefix_chars: Optional[str] = None,
|
||||
argument_default: Optional[Any] = None, conflict_handler: str = 'error',
|
||||
add_help: bool = True, allow_abbrev: bool = True) -> None:
|
||||
"""Initialize a NicerParser."""
|
||||
|
||||
super().__init__(prog=prog, usage=usage, description=description, epilog=epilog,
|
||||
parents=parents if parents else [], formatter_class=NicerHelpFormatter,
|
||||
prefix_chars=prefix_chars, fromfile_prefix_chars=fromfile_prefix_chars,
|
||||
argument_default=argument_default, conflict_handler=conflict_handler,
|
||||
add_help=add_help, allow_abbrev=allow_abbrev)
|
||||
self._optionals.title = 'Options'
|
||||
self._positionals.title = 'Queries'
|
||||
|
||||
def format_help(self) -> str:
|
||||
formatter = self._get_formatter()
|
||||
formatter.add_text(self.description)
|
||||
formatter.add_usage(self.usage, self._actions, self._mutually_exclusive_groups,
|
||||
prefix='Usage:\n ')
|
||||
|
||||
for action_group in self._action_groups:
|
||||
formatter.start_section(action_group.title)
|
||||
formatter.add_text(action_group.description)
|
||||
formatter.add_arguments(action_group._group_actions)
|
||||
formatter.end_section()
|
||||
|
||||
formatter.add_text(self.epilog)
|
||||
|
||||
return formatter.format_help()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
parser = NicerArgumentParser(
|
||||
description='Tool for querying information about mongodbtoolchain.', add_help=False)
|
||||
|
||||
parser.add_argument('-h', '--help', action='help', default=argparse.SUPPRESS,
|
||||
help='Show this help message and exit')
|
||||
parser.add_argument('-f', '--from-file', help='Specify a toolchain data file', metavar='FILE',
|
||||
type=str, default=str(DEFAULT_DATA_FILE))
|
||||
parser.add_argument('-d', '--distro-id', help='Evergreen distro_id', type=str, required=True)
|
||||
parser.add_argument('-a', '--arch', help='Host architecture', type=str)
|
||||
|
||||
subparsers = parser.add_subparsers(title='Commands', dest='command', required=True)
|
||||
|
||||
show_parser = subparsers.add_parser(
|
||||
'show', description='Shows general toolchain collection info.', add_help=False,
|
||||
help='Show general toolchain collection info')
|
||||
config_parser = subparsers.add_parser('config',
|
||||
description='Shows toolchain configuration info.',
|
||||
add_help=False, help='Show toolchain configuration info')
|
||||
platform_parser = subparsers.add_parser('platform',
|
||||
description='Shows component parts of a distro_id.',
|
||||
add_help=False, help='Show parts of a distro_id')
|
||||
toolchain_parser = subparsers.add_parser('toolchain',
|
||||
description='Shows specific toolchain info.',
|
||||
add_help=False, help='Show specific toolchain info')
|
||||
|
||||
show_parser.add_argument('-h', '--help', action='help', default=argparse.SUPPRESS,
|
||||
help='Show this help message and exit')
|
||||
show_parser.add_argument(
|
||||
'query', action=DictChoiceAction, type=str, choices={
|
||||
'available': 'All installed toolchains',
|
||||
'configured': 'Toolchains configured for the distro_id',
|
||||
'latest': 'The most recent installed toolchain',
|
||||
})
|
||||
|
||||
config_parser.add_argument('-h', '--help', action='help', default=argparse.SUPPRESS,
|
||||
help='Show this help message and exit')
|
||||
config_parser.add_argument(
|
||||
'query', action=DictChoiceAction, type=str, choices={
|
||||
'base_path': 'Toolchain base execution path',
|
||||
'releases': 'All defined release names',
|
||||
'versions': 'All defined version names',
|
||||
'aliases': 'All defined aliases for version names',
|
||||
})
|
||||
|
||||
platform_parser.add_argument('-h', '--help', action='help', default=argparse.SUPPRESS,
|
||||
help='Show this help message and exit')
|
||||
platform_parser.add_argument(
|
||||
'query', action=DictChoiceAction, type=str, choices={
|
||||
'distro': 'Show the "distro" component of the distro_id',
|
||||
'arch': 'Show the "arch" component of the distro_id',
|
||||
'tag': 'Show the information tag component of the distro_id',
|
||||
})
|
||||
|
||||
toolchain_parser.add_argument('-h', '--help', action='help', default=argparse.SUPPRESS,
|
||||
help='Show this help message and exit')
|
||||
toolchain_parser.add_argument('-v', '--toolchain-version', help='Toolchain version', type=str,
|
||||
default=str(ToolchainVersionName.STABLE))
|
||||
toolchain_parser.add_argument('-r', '--release', help="Toolchain release", type=str,
|
||||
default=str(ToolchainReleaseName.CURRENT))
|
||||
toolchain_parser.add_argument(
|
||||
'query', action=DictChoiceAction, type=str, choices={
|
||||
'install_path': 'Toolchain installation path',
|
||||
'exec_path': 'Toolchain execution path',
|
||||
})
|
||||
|
||||
parsed_args = parser.parse_args()
|
||||
|
||||
# Set up the objects required for each command
|
||||
toolchain_platform = ToolchainPlatform(distro_id=parsed_args.distro_id, arch=parsed_args.arch)
|
||||
|
||||
if parsed_args.command in ('show', 'config', 'toolchain'):
|
||||
try:
|
||||
toolchain_config = ToolchainConfig(
|
||||
pathlib.Path(parsed_args.from_file), platform=toolchain_platform)
|
||||
except ToolchainDataException as exc:
|
||||
print(exc, file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
toolchains = Toolchains(config=toolchain_config)
|
||||
|
||||
if parsed_args.command == 'toolchain':
|
||||
toolchain = toolchains[parsed_args.release]
|
||||
|
||||
# Set the object to query for each command
|
||||
obj: object
|
||||
if parsed_args.command == 'platform':
|
||||
obj = toolchain_platform
|
||||
if parsed_args.command == 'show':
|
||||
obj = toolchains
|
||||
elif parsed_args.command == 'config':
|
||||
obj = toolchain_config
|
||||
elif parsed_args.command == 'toolchain':
|
||||
obj = toolchain
|
||||
|
||||
# Get and handle output
|
||||
output: Any
|
||||
attribute = getattr(obj, parsed_args.query)
|
||||
if callable(attribute):
|
||||
if attribute.__name__ == 'exec_path':
|
||||
output = attribute(parsed_args.toolchain_version)
|
||||
else:
|
||||
output = attribute()
|
||||
else:
|
||||
output = attribute # If there is no output, it should indicate an error to the caller.
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
if output is not None:
|
||||
if isinstance(output, (tuple, list)):
|
||||
output = str.join(" ", output)
|
||||
elif isinstance(output, dict):
|
||||
output = '\n'.join([f"{k}: {v}" for k, v in output.items()])
|
||||
elif not isinstance(output, str):
|
||||
output = str(output)
|
||||
|
||||
if output:
|
||||
print(output)
|
||||
else:
|
||||
sys.exit(1)
|
||||
30
etc/toolchains.yaml
Normal file
30
etc/toolchains.yaml
Normal file
@ -0,0 +1,30 @@
|
||||
toolchains:
|
||||
base_path: /opt/mongodbtoolchain
|
||||
versions:
|
||||
- v2
|
||||
- v3
|
||||
- v4
|
||||
version_aliases:
|
||||
stable: v3
|
||||
testing: v4
|
||||
releases:
|
||||
default:
|
||||
current: c6da1cf7f0b4b60d53566305e59857d3d540dcf7
|
||||
rhel80-s390x-playbook:
|
||||
current: 41a191fa8daae48ef6271805fe30c004c69e18d4
|
||||
ubuntu1804-power8-playbook:
|
||||
current: 8391ec859b03d3bebb666821a88e616ae61a567a
|
||||
ubuntu1804-zseries-playbook:
|
||||
current: 32eb70c47bd9e9759dd05654843feb80461aaef3
|
||||
archlinux-packer:
|
||||
current: ca1de958b3e011a973bfdd2cb947360dab348502
|
||||
ubuntu1404-jepsen:
|
||||
current: ca1de958b3e011a973bfdd2cb947360dab348502
|
||||
ubuntu1604-container:
|
||||
current: ca1de958b3e011a973bfdd2cb947360dab348502
|
||||
suse12:
|
||||
current: ca1de958b3e011a973bfdd2cb947360dab348502
|
||||
amazon1-2018:
|
||||
current: ca1de958b3e011a973bfdd2cb947360dab348502
|
||||
macos-1012:
|
||||
current: ca1de958b3e011a973bfdd2cb947360dab348502
|
||||
Loading…
Reference in New Issue
Block a user