aboutsummaryrefslogtreecommitdiff
path: root/pw_cli/py
diff options
context:
space:
mode:
Diffstat (limited to 'pw_cli/py')
-rw-r--r--pw_cli/py/BUILD.bazel10
-rw-r--r--pw_cli/py/BUILD.gn15
-rw-r--r--pw_cli/py/envparse_test.py44
-rw-r--r--pw_cli/py/plugins_test.py57
-rw-r--r--pw_cli/py/process_integration_test.py73
-rw-r--r--pw_cli/py/pw_cli/__main__.py2
-rw-r--r--pw_cli/py/pw_cli/argument_types.py3
-rw-r--r--pw_cli/py/pw_cli/arguments.py43
-rw-r--r--pw_cli/py/pw_cli/branding.py10
-rw-r--r--pw_cli/py/pw_cli/color.py9
-rw-r--r--pw_cli/py/pw_cli/env.py50
-rw-r--r--pw_cli/py/pw_cli/envparse.py87
-rw-r--r--pw_cli/py/pw_cli/log.py87
-rw-r--r--pw_cli/py/pw_cli/plugins.py187
-rw-r--r--pw_cli/py/pw_cli/process.py129
-rw-r--r--pw_cli/py/pw_cli/pw_command_plugins.py22
-rwxr-xr-xpw_cli/py/pw_cli/requires.py28
-rw-r--r--pw_cli/py/pw_cli/toml_config_loader_mixin.py50
-rw-r--r--pw_cli/py/pw_cli/yaml_config_loader_mixin.py166
-rw-r--r--pw_cli/py/setup.cfg4
20 files changed, 851 insertions, 225 deletions
diff --git a/pw_cli/py/BUILD.bazel b/pw_cli/py/BUILD.bazel
index eac6f4c3e..c77ce2ca2 100644
--- a/pw_cli/py/BUILD.bazel
+++ b/pw_cli/py/BUILD.bazel
@@ -34,6 +34,8 @@ py_library(
"pw_cli/process.py",
"pw_cli/pw_command_plugins.py",
"pw_cli/requires.py",
+ "pw_cli/toml_config_loader_mixin.py",
+ "pw_cli/yaml_config_loader_mixin.py",
],
imports = ["."],
)
@@ -49,10 +51,10 @@ py_binary(
)
py_test(
- name = "plugins_test",
+ name = "envparse_test",
size = "small",
srcs = [
- "plugins_test.py",
+ "envparse_test.py",
],
deps = [
":pw_cli",
@@ -60,10 +62,10 @@ py_test(
)
py_test(
- name = "envparse_test",
+ name = "plugins_test",
size = "small",
srcs = [
- "envparse_test.py",
+ "plugins_test.py",
],
deps = [
":pw_cli",
diff --git a/pw_cli/py/BUILD.gn b/pw_cli/py/BUILD.gn
index 0057e8978..4bb4cd2e3 100644
--- a/pw_cli/py/BUILD.gn
+++ b/pw_cli/py/BUILD.gn
@@ -36,10 +36,25 @@ pw_python_package("py") {
"pw_cli/process.py",
"pw_cli/pw_command_plugins.py",
"pw_cli/requires.py",
+ "pw_cli/toml_config_loader_mixin.py",
+ "pw_cli/yaml_config_loader_mixin.py",
]
tests = [
"envparse_test.py",
"plugins_test.py",
]
pylintrc = "$dir_pigweed/.pylintrc"
+ mypy_ini = "$dir_pigweed/.mypy.ini"
+}
+
+pw_python_script("process_integration_test") {
+ sources = [ "process_integration_test.py" ]
+ python_deps = [ ":py" ]
+
+ pylintrc = "$dir_pigweed/.pylintrc"
+ mypy_ini = "$dir_pigweed/.mypy.ini"
+
+ action = {
+ stamp = true
+ }
}
diff --git a/pw_cli/py/envparse_test.py b/pw_cli/py/envparse_test.py
index cbeb332b6..e1e9b5497 100644
--- a/pw_cli/py/envparse_test.py
+++ b/pw_cli/py/envparse_test.py
@@ -31,6 +31,7 @@ def error(value: str):
class TestEnvironmentParser(unittest.TestCase):
"""Tests for envparse.EnvironmentParser."""
+
def setUp(self):
self.raw_env = {
'PATH': '/bin:/usr/bin:/usr/local/bin',
@@ -91,6 +92,7 @@ class TestEnvironmentParser(unittest.TestCase):
class TestEnvironmentParserWithPrefix(unittest.TestCase):
"""Tests for envparse.EnvironmentParser using a prefix."""
+
def setUp(self):
self.raw_env = {
'PW_FOO': '001',
@@ -100,8 +102,9 @@ class TestEnvironmentParserWithPrefix(unittest.TestCase):
}
def test_parse_unrecognized_variable(self):
- parser = envparse.EnvironmentParser(prefix='PW_',
- error_on_unrecognized=True)
+ parser = envparse.EnvironmentParser(
+ prefix='PW_', error_on_unrecognized=True
+ )
parser.add_var('PW_FOO')
parser.add_var('PW_BAR')
@@ -109,24 +112,27 @@ class TestEnvironmentParserWithPrefix(unittest.TestCase):
parser.parse_env(env=self.raw_env)
def test_parse_unrecognized_but_allowed_suffix(self):
- parser = envparse.EnvironmentParser(prefix='PW_',
- error_on_unrecognized=True)
+ parser = envparse.EnvironmentParser(
+ prefix='PW_', error_on_unrecognized=True
+ )
parser.add_allowed_suffix('_ALLOWED_SUFFIX')
env = parser.parse_env(env={'PW_FOO_ALLOWED_SUFFIX': '001'})
self.assertEqual(env.PW_FOO_ALLOWED_SUFFIX, '001')
def test_parse_allowed_suffix_but_not_suffix(self):
- parser = envparse.EnvironmentParser(prefix='PW_',
- error_on_unrecognized=True)
+ parser = envparse.EnvironmentParser(
+ prefix='PW_', error_on_unrecognized=True
+ )
parser.add_allowed_suffix('_ALLOWED_SUFFIX')
with self.assertRaises(ValueError):
parser.parse_env(env={'PW_FOO_ALLOWED_SUFFIX_FOO': '001'})
def test_parse_ignore_unrecognized(self):
- parser = envparse.EnvironmentParser(prefix='PW_',
- error_on_unrecognized=False)
+ parser = envparse.EnvironmentParser(
+ prefix='PW_', error_on_unrecognized=False
+ )
parser.add_var('PW_FOO')
parser.add_var('PW_BAR')
@@ -135,27 +141,39 @@ class TestEnvironmentParserWithPrefix(unittest.TestCase):
self.assertEqual(env.PW_BAR, self.raw_env['PW_BAR'])
def test_add_var_without_prefix(self):
- parser = envparse.EnvironmentParser(prefix='PW_',
- error_on_unrecognized=True)
+ parser = envparse.EnvironmentParser(
+ prefix='PW_', error_on_unrecognized=True
+ )
with self.assertRaises(ValueError):
parser.add_var('FOO')
class TestStrictBool(unittest.TestCase):
"""Tests for envparse.strict_bool."""
+
def setUp(self):
self.good_bools = ['true', '1', 'TRUE', 'tRuE']
self.bad_bools = [
- '', 'false', '0', 'foo', '2', '999', 'ok', 'yes', 'no'
+ '',
+ 'false',
+ '0',
+ 'foo',
+ '2',
+ '999',
+ 'ok',
+ 'yes',
+ 'no',
]
def test_good_bools(self):
self.assertTrue(
- all(envparse.strict_bool(val) for val in self.good_bools))
+ all(envparse.strict_bool(val) for val in self.good_bools)
+ )
def test_bad_bools(self):
self.assertFalse(
- any(envparse.strict_bool(val) for val in self.bad_bools))
+ any(envparse.strict_bool(val) for val in self.bad_bools)
+ )
if __name__ == '__main__':
diff --git a/pw_cli/py/plugins_test.py b/pw_cli/py/plugins_test.py
index c2063996d..a3f35132d 100644
--- a/pw_cli/py/plugins_test.py
+++ b/pw_cli/py/plugins_test.py
@@ -42,10 +42,12 @@ def _create_files(directory: str, files: Dict[str, str]) -> Iterator[Path]:
class TestPlugin(unittest.TestCase):
"""Tests for plugins.Plugins."""
+
def test_target_name_attribute(self) -> None:
self.assertEqual(
plugins.Plugin('abc', _no_docstring).target_name,
- f'{__name__}._no_docstring')
+ f'{__name__}._no_docstring',
+ )
def test_target_name_no_name_attribute(self) -> None:
has_no_name = 'no __name__'
@@ -53,42 +55,49 @@ class TestPlugin(unittest.TestCase):
self.assertEqual(
plugins.Plugin('abc', has_no_name).target_name,
- '<unknown>.no __name__')
+ '<unknown>.no __name__',
+ )
_TEST_PLUGINS = {
- 'TEST_PLUGINS': (f'test_plugin {__name__} _with_docstring\n'
- f'other_plugin {__name__} _no_docstring\n'),
- 'nested/in/dirs/TEST_PLUGINS':
- f'test_plugin {__name__} _no_docstring\n',
+ 'TEST_PLUGINS': (
+ f'test_plugin {__name__} _with_docstring\n'
+ f'other_plugin {__name__} _no_docstring\n'
+ ),
+ 'nested/in/dirs/TEST_PLUGINS': f'test_plugin {__name__} _no_docstring\n',
}
class TestPluginRegistry(unittest.TestCase):
"""Tests for plugins.Registry."""
+
def setUp(self) -> None:
self._registry = plugins.Registry(
- validator=plugins.callable_with_no_args)
+ validator=plugins.callable_with_no_args
+ )
def test_register(self) -> None:
- self.assertIsNotNone(self._registry.register('a_plugin',
- _no_docstring))
+ self.assertIsNotNone(self._registry.register('a_plugin', _no_docstring))
self.assertIs(self._registry['a_plugin'].target, _no_docstring)
def test_register_by_name(self) -> None:
self.assertIsNotNone(
- self._registry.register_by_name('plugin_one', __name__,
- '_no_docstring'))
+ self._registry.register_by_name(
+ 'plugin_one', __name__, '_no_docstring'
+ )
+ )
self.assertIsNotNone(
- self._registry.register('plugin_two', _no_docstring))
+ self._registry.register('plugin_two', _no_docstring)
+ )
self.assertIs(self._registry['plugin_one'].target, _no_docstring)
self.assertIs(self._registry['plugin_two'].target, _no_docstring)
def test_register_by_name_undefined_module(self) -> None:
with self.assertRaisesRegex(plugins.Error, 'No module named'):
- self._registry.register_by_name('plugin_two', 'not a module',
- 'something')
+ self._registry.register_by_name(
+ 'plugin_two', 'not a module', 'something'
+ )
def test_register_by_name_undefined_function(self) -> None:
with self.assertRaisesRegex(plugins.Error, 'does not exist'):
@@ -116,7 +125,8 @@ class TestPluginRegistry(unittest.TestCase):
def test_register_cannot_overwrite(self) -> None:
self.assertIsNotNone(self._registry.register('foo', lambda: None))
self.assertIsNotNone(
- self._registry.register_by_name('bar', __name__, '_no_docstring'))
+ self._registry.register_by_name('bar', __name__, '_no_docstring')
+ )
with self.assertRaises(plugins.Error):
self._registry.register('foo', lambda: None)
@@ -141,8 +151,9 @@ class TestPluginRegistry(unittest.TestCase):
def test_register_directory_with_restriction(self) -> None:
with tempfile.TemporaryDirectory() as tempdir:
paths = list(_create_files(tempdir, _TEST_PLUGINS))
- self._registry.register_directory(paths[0].parent, 'TEST_PLUGINS',
- Path(tempdir, 'nested', 'in'))
+ self._registry.register_directory(
+ paths[0].parent, 'TEST_PLUGINS', Path(tempdir, 'nested', 'in')
+ )
self.assertNotIn('other_plugin', self._registry)
@@ -161,10 +172,13 @@ class TestPluginRegistry(unittest.TestCase):
self.assertIn(__doc__, '\n'.join(self._registry.detailed_help(['a'])))
- self.assertNotIn(__doc__,
- '\n'.join(self._registry.detailed_help(['b'])))
- self.assertIn(_with_docstring.__doc__,
- '\n'.join(self._registry.detailed_help(['b'])))
+ self.assertNotIn(
+ __doc__, '\n'.join(self._registry.detailed_help(['b']))
+ )
+ self.assertIn(
+ _with_docstring.__doc__,
+ '\n'.join(self._registry.detailed_help(['b'])),
+ )
def test_empty_string_if_no_help(self) -> None:
fake_module_name = f'{__name__}.fake_module_for_test{id(self)}'
@@ -174,7 +188,6 @@ class TestPluginRegistry(unittest.TestCase):
sys.modules[fake_module_name] = fake_module
try:
-
function = lambda: None
function.__module__ = fake_module_name
self.assertIsNotNone(self._registry.register('a', function))
diff --git a/pw_cli/py/process_integration_test.py b/pw_cli/py/process_integration_test.py
new file mode 100644
index 000000000..3d05b1a1b
--- /dev/null
+++ b/pw_cli/py/process_integration_test.py
@@ -0,0 +1,73 @@
+# Copyright 2023 The Pigweed Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+"""Tests for pw_cli.process."""
+
+import unittest
+import sys
+import textwrap
+
+from pw_cli import process
+
+import psutil # type: ignore
+
+
+FAST_TIMEOUT_SECONDS = 0.1
+KILL_SIGNALS = set({-9, 137})
+PYTHON = sys.executable
+
+
+class RunTest(unittest.TestCase):
+ """Tests for process.run."""
+
+ def test_returns_output(self) -> None:
+ echo_str = 'foobar'
+ print_str = f'print("{echo_str}")'
+ result = process.run(PYTHON, '-c', print_str)
+ self.assertEqual(result.output, b'foobar\n')
+
+ def test_timeout_kills_process(self) -> None:
+ sleep_100_seconds = 'import time; time.sleep(100)'
+ result = process.run(
+ PYTHON, '-c', sleep_100_seconds, timeout=FAST_TIMEOUT_SECONDS
+ )
+ self.assertIn(result.returncode, KILL_SIGNALS)
+
+ def test_timeout_kills_subprocess(self) -> None:
+ # Spawn a subprocess which waits for 100 seconds, print its pid,
+ # then wait for 100 seconds.
+ sleep_in_subprocess = textwrap.dedent(
+ f"""
+ import subprocess
+ import time
+
+ child = subprocess.Popen(
+ ['{PYTHON}', '-c', 'import time; print("booh"); time.sleep(100)']
+ )
+ print(child.pid, flush=True)
+ time.sleep(100)
+ """
+ )
+ result = process.run(
+ PYTHON, '-c', sleep_in_subprocess, timeout=FAST_TIMEOUT_SECONDS
+ )
+ self.assertIn(result.returncode, KILL_SIGNALS)
+ # THe first line of the output is the PID of the child sleep process.
+ child_pid_str, sep, remainder = result.output.partition(b'\n')
+ del sep, remainder
+ child_pid = int(child_pid_str)
+ self.assertFalse(psutil.pid_exists(child_pid))
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/pw_cli/py/pw_cli/__main__.py b/pw_cli/py/pw_cli/__main__.py
index ad6ceeb58..3320a4556 100644
--- a/pw_cli/py/pw_cli/__main__.py
+++ b/pw_cli/py/pw_cli/__main__.py
@@ -29,7 +29,7 @@ def main() -> NoReturn:
args = arguments.parse_args()
- pw_cli.log.install(level=args.loglevel)
+ pw_cli.log.install(level=args.loglevel, debug_log=args.debug_log)
# Start with the most critical part of the Pigweed command line tool.
if not args.no_banner:
diff --git a/pw_cli/py/pw_cli/argument_types.py b/pw_cli/py/pw_cli/argument_types.py
index a719e1693..0db022a19 100644
--- a/pw_cli/py/pw_cli/argument_types.py
+++ b/pw_cli/py/pw_cli/argument_types.py
@@ -31,4 +31,5 @@ def log_level(arg: str) -> int:
return getattr(logging, arg.upper())
except AttributeError:
raise argparse.ArgumentTypeError(
- f'"{arg.upper()}" is not a valid log level')
+ f'"{arg.upper()}" is not a valid log level'
+ )
diff --git a/pw_cli/py/pw_cli/arguments.py b/pw_cli/py/pw_cli/arguments.py
index 86ff6e266..f7fe37b4e 100644
--- a/pw_cli/py/pw_cli/arguments.py
+++ b/pw_cli/py/pw_cli/arguments.py
@@ -26,7 +26,7 @@ _HELP_HEADER = '''The Pigweed command line interface (CLI).
Example uses:
pw logdemo
- pw --loglevel debug watch out/clang
+ pw --loglevel debug watch -C out
'''
@@ -46,6 +46,7 @@ def format_help(registry: plugins.Registry) -> str:
class _ArgumentParserWithBanner(argparse.ArgumentParser):
"""Parser that the Pigweed banner when there are parsing errors."""
+
def error(self, message: str) -> NoReturn:
print_banner()
self.print_usage(sys.stderr)
@@ -58,37 +59,53 @@ def _parser() -> argparse.ArgumentParser:
prog='pw',
add_help=False,
description=_HELP_HEADER,
- formatter_class=argparse.RawDescriptionHelpFormatter)
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ )
# Do not use the built-in help argument so that displaying the help info can
# be deferred until the pw plugins have been registered.
- argparser.add_argument('-h',
- '--help',
- action='store_true',
- help='Display this help message and exit')
+ argparser.add_argument(
+ '-h',
+ '--help',
+ action='store_true',
+ help='Display this help message and exit',
+ )
argparser.add_argument(
'-C',
'--directory',
type=argument_types.directory,
default=Path.cwd(),
- help='Change to this directory before doing anything')
+ help='Change to this directory before doing anything',
+ )
argparser.add_argument(
'-l',
'--loglevel',
type=argument_types.log_level,
default=logging.INFO,
- help='Set the log level (debug, info, warning, error, critical)')
- argparser.add_argument('--no-banner',
- action='store_true',
- help='Do not print the Pigweed banner')
+ help='Set the log level (debug, info, warning, error, critical)',
+ )
+ argparser.add_argument(
+ '--debug-log',
+ help=(
+ 'Additionally log to this file at debug level; does not affect '
+ 'terminal output'
+ ),
+ )
+ argparser.add_argument(
+ '--no-banner',
+ action='store_true',
+ help='Do not print the Pigweed banner',
+ )
argparser.add_argument(
'command',
nargs='?',
- help='Which command to run; see supported commands below')
+ help='Which command to run; see supported commands below',
+ )
argparser.add_argument(
'plugin_args',
metavar='...',
nargs=argparse.REMAINDER,
- help='Remaining arguments are forwarded to the command')
+ help='Remaining arguments are forwarded to the command',
+ )
return argparser
diff --git a/pw_cli/py/pw_cli/branding.py b/pw_cli/py/pw_cli/branding.py
index 81274eced..358e4019b 100644
--- a/pw_cli/py/pw_cli/branding.py
+++ b/pw_cli/py/pw_cli/branding.py
@@ -41,14 +41,18 @@ def banner() -> str:
# Take the banner from the file PW_BRANDING_BANNER; or use the default.
banner_filename = parsed_env.PW_BRANDING_BANNER
- _memoized_banner = (Path(banner_filename).read_text()
- if banner_filename else _PIGWEED_BANNER)
+ _memoized_banner = (
+ Path(banner_filename).read_text()
+ if banner_filename
+ else _PIGWEED_BANNER
+ )
# Color the banner if requested.
banner_color = parsed_env.PW_BRANDING_BANNER_COLOR
if banner_color != '':
set_color = operator.attrgetter(banner_color)(pw_cli.color.colors())
_memoized_banner = '\n'.join(
- set_color(line) for line in _memoized_banner.splitlines())
+ set_color(line) for line in _memoized_banner.splitlines()
+ )
return _memoized_banner
diff --git a/pw_cli/py/pw_cli/color.py b/pw_cli/py/pw_cli/color.py
index ef2563aad..855a8d11c 100644
--- a/pw_cli/py/pw_cli/color.py
+++ b/pw_cli/py/pw_cli/color.py
@@ -35,6 +35,7 @@ class _Color:
# pylint: disable=too-few-public-methods
# pylint: disable=too-many-instance-attributes
"""Helpers to surround text with ASCII color escapes"""
+
def __init__(self):
self.none = str
self.red = _make_color(31, 1)
@@ -49,10 +50,13 @@ class _Color:
self.bold_magenta = _make_color(30, 45)
self.bold_white = _make_color(37, 1)
self.black_on_white = _make_color(30, 47) # black fg white bg
+ self.black_on_green = _make_color(30, 42) # black fg green bg
+ self.black_on_red = _make_color(30, 41) # black fg red bg
class _NoColor:
"""Fake version of the _Color class that doesn't colorize."""
+
def __getattr__(self, _):
return str
@@ -64,8 +68,9 @@ def colors(enabled: Optional[bool] = None) -> Union[_Color, _NoColor]:
"""
if enabled is None:
env = pw_cli.env.pigweed_environment()
- enabled = env.PW_USE_COLOR or (sys.stdout.isatty()
- and sys.stderr.isatty())
+ enabled = env.PW_USE_COLOR or (
+ sys.stdout.isatty() and sys.stderr.isatty()
+ )
if enabled and os.name == 'nt':
# Enable ANSI color codes in Windows cmd.exe.
diff --git a/pw_cli/py/pw_cli/env.py b/pw_cli/py/pw_cli/env.py
index c1e47751c..74deff8ea 100644
--- a/pw_cli/py/pw_cli/env.py
+++ b/pw_cli/py/pw_cli/env.py
@@ -13,6 +13,7 @@
# the License.
"""The env module defines the environment variables used by Pigweed."""
+from pathlib import Path
from typing import Optional
from pw_cli import envparse
@@ -27,15 +28,21 @@ def pigweed_environment_parser() -> envparse.EnvironmentParser:
parser.add_var('PW_EMOJI', type=envparse.strict_bool, default=False)
parser.add_var('PW_ENVSETUP')
parser.add_var('PW_ENVSETUP_FULL')
- parser.add_var('PW_ENVSETUP_NO_BANNER',
- type=envparse.strict_bool,
- default=False)
- parser.add_var('PW_ENVSETUP_QUIET',
- type=envparse.strict_bool,
- default=False)
- parser.add_var('PW_ENVIRONMENT_ROOT')
- parser.add_var('PW_PROJECT_ROOT')
- parser.add_var('PW_ROOT')
+ parser.add_var(
+ 'PW_ENVSETUP_NO_BANNER', type=envparse.strict_bool, default=False
+ )
+ parser.add_var(
+ 'PW_ENVSETUP_QUIET', type=envparse.strict_bool, default=False
+ )
+ parser.add_var('PW_ENVIRONMENT_ROOT', type=Path)
+ parser.add_var('PW_PACKAGE_ROOT', type=Path)
+ parser.add_var('PW_PROJECT_ROOT', type=Path)
+ parser.add_var('PW_ROOT', type=Path)
+ parser.add_var(
+ 'PW_DISABLE_ROOT_GIT_REPO_CHECK',
+ type=envparse.strict_bool,
+ default=False,
+ )
parser.add_var('PW_SKIP_BOOTSTRAP')
parser.add_var('PW_SUBPROCESS', type=envparse.strict_bool, default=False)
parser.add_var('PW_USE_COLOR', type=envparse.strict_bool, default=False)
@@ -43,24 +50,33 @@ def pigweed_environment_parser() -> envparse.EnvironmentParser:
parser.add_allowed_suffix('_CIPD_INSTALL_DIR')
- parser.add_var('PW_ENVSETUP_DISABLE_SPINNER',
- type=envparse.strict_bool,
- default=False)
+ parser.add_var(
+ 'PW_ENVSETUP_DISABLE_SPINNER', type=envparse.strict_bool, default=False
+ )
parser.add_var('PW_DOCTOR_SKIP_CIPD_CHECKS')
- parser.add_var('PW_ACTIVATE_SKIP_CHECKS',
- type=envparse.strict_bool,
- default=False)
+ parser.add_var(
+ 'PW_ACTIVATE_SKIP_CHECKS', type=envparse.strict_bool, default=False
+ )
parser.add_var('PW_BANNER_FUNC')
parser.add_var('PW_BRANDING_BANNER')
parser.add_var('PW_BRANDING_BANNER_COLOR', default='magenta')
- parser.add_var('PW_PRESUBMIT_DISABLE_SUBPROCESS_CAPTURE',
- type=envparse.strict_bool)
+ parser.add_var(
+ 'PW_PRESUBMIT_DISABLE_SUBPROCESS_CAPTURE', type=envparse.strict_bool
+ )
parser.add_var('PW_CONSOLE_CONFIG_FILE')
parser.add_var('PW_ENVIRONMENT_NO_ERROR_ON_UNRECOGNIZED')
+ parser.add_var('PW_CIPD_SERVICE_ACCOUNT_JSON')
+
+ # RBE environment variables
+ parser.add_var('PW_USE_RBE', default=False)
+ parser.add_var('PW_RBE_DEBUG', default=False)
+ parser.add_var('PW_RBE_CLANG_CONFIG', default='')
+ parser.add_var('PW_RBE_ARM_GCC_CONFIG', default='')
+
return parser
diff --git a/pw_cli/py/pw_cli/envparse.py b/pw_cli/py/pw_cli/envparse.py
index e27eb5b8c..d9ed9de47 100644
--- a/pw_cli/py/pw_cli/envparse.py
+++ b/pw_cli/py/pw_cli/envparse.py
@@ -16,11 +16,22 @@
import argparse
from dataclasses import dataclass
import os
-from typing import Callable, Dict, Generic, IO, List, Mapping, Optional, TypeVar
-from typing import Union
-
-
-class EnvNamespace(argparse.Namespace): # pylint: disable=too-few-public-methods
+from typing import (
+ Callable,
+ Dict,
+ Generic,
+ IO,
+ List,
+ Mapping,
+ Optional,
+ TypeVar,
+ Union,
+)
+
+
+class EnvNamespace(
+ argparse.Namespace
+): # pylint: disable=too-few-public-methods
"""Base class for parsed environment variable namespaces."""
@@ -42,11 +53,13 @@ class EnvironmentValueError(Exception):
function through the __cause__ attribute for more detailed information on
the error.
"""
+
def __init__(self, variable: str, value: str):
self.variable: str = variable
self.value: str = value
super().__init__(
- f'Bad value for environment variable {variable}: {value}')
+ f'Bad value for environment variable {variable}: {value}'
+ )
class EnvironmentParser:
@@ -71,9 +84,12 @@ class EnvironmentParser:
configure_logging(env.PW_LOG_LEVEL, env.PW_LOG_FILE)
"""
- def __init__(self,
- prefix: Optional[str] = None,
- error_on_unrecognized: Union[bool, None] = None) -> None:
+
+ def __init__(
+ self,
+ prefix: Optional[str] = None,
+ error_on_unrecognized: Union[bool, None] = None,
+ ) -> None:
self._prefix: Optional[str] = prefix
if error_on_unrecognized is None:
varname = 'PW_ENVIRONMENT_NO_ERROR_ON_UNRECOGNIZED'
@@ -104,7 +120,8 @@ class EnvironmentParser:
"""
if self._prefix is not None and not name.startswith(self._prefix):
raise ValueError(
- f'Variable {name} does not have prefix {self._prefix}')
+ f'Variable {name} does not have prefix {self._prefix}'
+ )
self._variables[name] = VariableDescriptor(name, type, default)
@@ -113,8 +130,9 @@ class EnvironmentParser:
self._allowed_suffixes.append(suffix)
- def parse_env(self,
- env: Optional[Mapping[str, str]] = None) -> EnvNamespace:
+ def parse_env(
+ self, env: Optional[Mapping[str, str]] = None
+ ) -> EnvNamespace:
"""Parses known environment variables into a namespace.
Args:
@@ -141,17 +159,21 @@ class EnvironmentParser:
allowed_suffixes = tuple(self._allowed_suffixes)
for var in env:
- if (not hasattr(namespace, var)
- and (self._prefix is None or var.startswith(self._prefix))
- and var.endswith(allowed_suffixes)):
+ if (
+ not hasattr(namespace, var)
+ and (self._prefix is None or var.startswith(self._prefix))
+ and var.endswith(allowed_suffixes)
+ ):
setattr(namespace, var, env[var])
if self._prefix is not None and self._error_on_unrecognized:
for var in env:
- if (var.startswith(self._prefix) and var not in self._variables
- and not var.endswith(allowed_suffixes)):
- raise ValueError(
- f'Unrecognized environment variable {var}')
+ if (
+ var.startswith(self._prefix)
+ and var not in self._variables
+ and not var.endswith(allowed_suffixes)
+ ):
+ raise ValueError(f'Unrecognized environment variable {var}')
return namespace
@@ -160,21 +182,24 @@ class EnvironmentParser:
# List of emoji which are considered to represent "True".
-_BOOLEAN_TRUE_EMOJI = set([
- '✔️',
- '👍',
- '👍🏻',
- '👍🏼',
- '👍🏽',
- '👍🏾',
- '👍🏿',
- '💯',
-])
+_BOOLEAN_TRUE_EMOJI = set(
+ [
+ '✔️',
+ '👍',
+ '👍🏻',
+ '👍🏼',
+ '👍🏽',
+ '👍🏾',
+ '👍🏿',
+ '💯',
+ ]
+)
def strict_bool(value: str) -> bool:
- return (value == '1' or value.lower() == 'true'
- or value in _BOOLEAN_TRUE_EMOJI)
+ return (
+ value == '1' or value.lower() == 'true' or value in _BOOLEAN_TRUE_EMOJI
+ )
# TODO(mohrr) Switch to Literal when no longer supporting Python 3.7.
diff --git a/pw_cli/py/pw_cli/log.py b/pw_cli/py/pw_cli/log.py
index 3ef755470..c8414d1f5 100644
--- a/pw_cli/py/pw_cli/log.py
+++ b/pw_cli/py/pw_cli/log.py
@@ -15,11 +15,11 @@
import logging
from pathlib import Path
+import sys
from typing import NamedTuple, Optional, Union, Iterator
-import pw_cli.color
-import pw_cli.env
-import pw_cli.plugins
+from pw_cli.color import colors as pw_cli_colors
+from pw_cli.env import pigweed_environment
# Log level used for captured output of a subprocess run through pw.
LOGLEVEL_STDOUT = 21
@@ -37,6 +37,7 @@ class _LogLevel(NamedTuple):
# Shorten all the log levels to 3 characters for column-aligned logs.
# Color the logs using ANSI codes.
+# fmt: off
_LOG_LEVELS = (
_LogLevel(LOGLEVEL_FATAL, 'bold_red', 'FTL', '☠️ '),
_LogLevel(logging.CRITICAL, 'bold_magenta', 'CRT', '‼️ '),
@@ -45,7 +46,8 @@ _LOG_LEVELS = (
_LogLevel(logging.INFO, 'magenta', 'INF', 'ℹ️ '),
_LogLevel(LOGLEVEL_STDOUT, 'cyan', 'OUT', '💬'),
_LogLevel(logging.DEBUG, 'blue', 'DBG', '👾'),
-) # yapf: disable
+)
+# fmt: on
_LOG = logging.getLogger(__name__)
_STDERR_HANDLER = logging.StreamHandler()
@@ -72,18 +74,29 @@ def main() -> None:
_LOG.debug('Adding 1 to i')
-def _setup_handler(handler: logging.Handler, formatter: logging.Formatter,
- level: Union[str, int], logger: logging.Logger) -> None:
+def _setup_handler(
+ handler: logging.Handler,
+ formatter: logging.Formatter,
+ level: Union[str, int],
+ logger: logging.Logger,
+) -> None:
handler.setLevel(level)
handler.setFormatter(formatter)
logger.addHandler(handler)
-def install(level: Union[str, int] = logging.INFO,
- use_color: bool = None,
- hide_timestamp: bool = False,
- log_file: Union[str, Path] = None,
- logger: Optional[logging.Logger] = None) -> None:
+def install(
+ level: Union[str, int] = logging.INFO,
+ use_color: Optional[bool] = None,
+ hide_timestamp: bool = False,
+ log_file: Optional[Union[str, Path]] = None,
+ logger: Optional[logging.Logger] = None,
+ debug_log: Optional[Union[str, Path]] = None,
+ time_format: str = '%Y%m%d %H:%M:%S',
+ msec_format: str = '%s,%03d',
+ include_msec: bool = False,
+ message_format: str = '%(levelname)s %(message)s',
+) -> None:
"""Configures the system logger for the default pw command log format.
If you have Python loggers separate from the root logger you can use
@@ -107,17 +120,27 @@ def install(level: Union[str, int] = logging.INFO,
use_color: When `True` include ANSI escape sequences to colorize log
messages.
hide_timestamp: When `True` omit timestamps from the log formatting.
- log_file: File to save logs into.
+ log_file: File to send logs into instead of the terminal.
logger: Python Logger instance to install Pigweed formatting into.
Defaults to the Python root logger: `logging.getLogger()`.
-
+ debug_log: File to log to from all levels, regardless of chosen log level.
+ Logs will go here in addition to the terminal.
+ time_format: Default time format string.
+ msec_format: Default millisecond format string. This should be a format
+ string that accepts a both a string ``%s`` and an integer ``%d``. The
+ default Python format for this string is ``%s,%03d``.
+ include_msec: Whether or not to include the millisecond part of log
+ timestamps.
+ message_format: The message format string. By default this includes
+ levelname and message. The asctime field is prepended to this unless
+ hide_timestamp=True.
"""
if not logger:
logger = logging.getLogger()
- colors = pw_cli.color.colors(use_color)
+ colors = pw_cli_colors(use_color)
- env = pw_cli.env.pigweed_environment()
+ env = pigweed_environment()
if env.PW_SUBPROCESS or hide_timestamp:
# If the logger is being run in the context of a pw subprocess, the
# time and date are omitted (since pw_cli.process will provide them).
@@ -128,8 +151,20 @@ def install(level: Union[str, int] = logging.INFO,
# colored text.
timestamp_fmt = colors.black_on_white('%(asctime)s') + ' '
- formatter = logging.Formatter(timestamp_fmt + '%(levelname)s %(message)s',
- '%Y%m%d %H:%M:%S')
+ formatter = logging.Formatter(fmt=timestamp_fmt + message_format)
+
+ formatter.default_time_format = time_format
+ if include_msec:
+ formatter.default_msec_format = msec_format
+ else:
+ # Python 3.8 and lower does not check if default_msec_format is set.
+ # https://github.com/python/cpython/blob/3.8/Lib/logging/__init__.py#L611
+ # https://github.com/python/cpython/blob/3.9/Lib/logging/__init__.py#L605
+ if sys.version_info >= (
+ 3,
+ 9,
+ ):
+ formatter.default_msec_format = ''
# Set the log level on the root logger to NOTSET, so that all logs
# propagated from child loggers are handled.
@@ -141,11 +176,25 @@ def install(level: Union[str, int] = logging.INFO,
if log_file:
# Set utf-8 encoding for the log file. Encoding errors may come up on
# Windows if the default system encoding is set to cp1250.
- _setup_handler(logging.FileHandler(log_file, encoding='utf-8'),
- formatter, level, logger)
+ _setup_handler(
+ logging.FileHandler(log_file, encoding='utf-8'),
+ formatter,
+ level,
+ logger,
+ )
# Since we're using a file, filter logs out of the stderr handler.
_STDERR_HANDLER.setLevel(logging.CRITICAL + 1)
+ if debug_log:
+ # Set utf-8 encoding for the log file. Encoding errors may come up on
+ # Windows if the default system encoding is set to cp1250.
+ _setup_handler(
+ logging.FileHandler(debug_log, encoding='utf-8'),
+ formatter,
+ logging.DEBUG,
+ logger,
+ )
+
if env.PW_EMOJI:
name_attr = 'emoji'
colorize = lambda ll: str
diff --git a/pw_cli/py/pw_cli/plugins.py b/pw_cli/py/pw_cli/plugins.py
index ebaf3fb52..b82b7f675 100644
--- a/pw_cli/py/pw_cli/plugins.py
+++ b/pw_cli/py/pw_cli/plugins.py
@@ -46,6 +46,7 @@ _BUILT_IN = '<built-in>'
class Error(Exception):
"""Indicates that a plugin is invalid or cannot be registered."""
+
def __str__(self):
"""Displays the error as a string, including the __cause__ if present.
@@ -54,8 +55,10 @@ class Error(Exception):
if self.__cause__ is None:
return super().__str__()
- return (f'{super().__str__()} '
- f'({type(self.__cause__).__name__}: {self.__cause__})')
+ return (
+ f'{super().__str__()} '
+ f'({type(self.__cause__).__name__}: {self.__cause__})'
+ )
def _get_module(member: object) -> types.ModuleType:
@@ -69,9 +72,15 @@ class Plugin:
Each plugin resolves to a Python object, typically a function.
"""
+
@classmethod
- def from_name(cls, name: str, module_name: str, member_name: str,
- source: Optional[Path]) -> 'Plugin':
+ def from_name(
+ cls,
+ name: str,
+ module_name: str,
+ member_name: str,
+ source: Optional[Path],
+ ) -> 'Plugin':
"""Creates a plugin by module and attribute name.
Args:
@@ -86,17 +95,26 @@ class Plugin:
try:
module = importlib.import_module(module_name)
except Exception as err:
+ _LOG.debug(
+ 'Failed to import module "%s" for "%s" plugin',
+ module_name,
+ name,
+ exc_info=True,
+ )
raise Error(f'Failed to import module "{module_name}"') from err
try:
member = getattr(module, member_name)
except AttributeError as err:
raise Error(
- f'"{module_name}.{member_name}" does not exist') from err
+ f'"{module_name}.{member_name}" does not exist'
+ ) from err
return cls(name, member, source)
- def __init__(self, name: str, target: Any, source: Path = None) -> None:
+ def __init__(
+ self, name: str, target: Any, source: Optional[Path] = None
+ ) -> None:
"""Creates a plugin for the provided target."""
self.name = name
self._module = _get_module(target)
@@ -105,8 +123,10 @@ class Plugin:
@property
def target_name(self) -> str:
- return (f'{self._module.__name__}.'
- f'{getattr(self.target, "__name__", self.target)}')
+ return (
+ f'{self._module.__name__}.'
+ f'{getattr(self.target, "__name__", self.target)}'
+ )
@property
def source_name(self) -> str:
@@ -137,9 +157,11 @@ class Plugin:
yield f'source {self.source_name}'
def __repr__(self) -> str:
- return (f'{self.__class__.__name__}(name={self.name!r}, '
- f'target={self.target_name}'
- f'{f", source={self.source_name!r}" if self.source else ""})')
+ return (
+ f'{self.__class__.__name__}(name={self.name!r}, '
+ f'target={self.target_name}'
+ f'{f", source={self.source_name!r}" if self.source else ""})'
+ )
def callable_with_no_args(plugin: Plugin) -> None:
@@ -150,20 +172,26 @@ def callable_with_no_args(plugin: Plugin) -> None:
try:
params = inspect.signature(plugin.target).parameters
except TypeError:
- raise Error('Plugin functions must be callable, but '
- f'{plugin.target_name} is a '
- f'{type(plugin.target).__name__}')
+ raise Error(
+ 'Plugin functions must be callable, but '
+ f'{plugin.target_name} is a '
+ f'{type(plugin.target).__name__}'
+ )
positional = sum(p.default == p.empty for p in params.values())
if positional:
- raise Error(f'Plugin functions cannot have any required positional '
- f'arguments, but {plugin.target_name} has {positional}')
+ raise Error(
+ f'Plugin functions cannot have any required positional '
+ f'arguments, but {plugin.target_name} has {positional}'
+ )
class Registry(collections.abc.Mapping):
"""Manages a set of plugins from Python modules or plugins files."""
- def __init__(self,
- validator: Callable[[Plugin], Any] = lambda _: None) -> None:
+
+ def __init__(
+ self, validator: Callable[[Plugin], Any] = lambda _: None
+ ) -> None:
"""Creates a new, empty plugins registry.
Args:
@@ -173,8 +201,7 @@ class Registry(collections.abc.Mapping):
self._registry: Dict[str, Plugin] = {}
self._sources: Set[Path] = set() # Paths to plugins files
- self._errors: Dict[str,
- List[Exception]] = collections.defaultdict(list)
+ self._errors: Dict[str, List[Exception]] = collections.defaultdict(list)
self._validate_plugin = validator
def __getitem__(self, name: str) -> Plugin:
@@ -183,8 +210,10 @@ class Registry(collections.abc.Mapping):
return self._registry[name]
if name in self._errors:
- raise KeyError(f'Registration for "{name}" failed: ' +
- ', '.join(str(e) for e in self._errors[name]))
+ raise KeyError(
+ f'Registration for "{name}" failed: '
+ + ', '.join(str(e) for e in self._errors[name])
+ )
raise KeyError(f'The plugin "{name}" has not been registered')
@@ -218,7 +247,8 @@ class Registry(collections.abc.Mapping):
raise Error(
f'Attempted to register built-in plugin "{plugin.name}", but '
'a plugin with that name was previously registered '
- f'({self[plugin.name]})!')
+ f'({self[plugin.name]})!'
+ )
# Run the user-provided validation function, which raises exceptions
# if there are errors.
@@ -230,20 +260,30 @@ class Registry(collections.abc.Mapping):
return True
if existing.source is None:
- _LOG.debug('%s: Overriding built-in plugin "%s" with %s',
- plugin.source_name, plugin.name, plugin.target_name)
+ _LOG.debug(
+ '%s: Overriding built-in plugin "%s" with %s',
+ plugin.source_name,
+ plugin.name,
+ plugin.target_name,
+ )
return True
if plugin.source != existing.source:
_LOG.debug(
'%s: The plugin "%s" was previously registered in %s; '
- 'ignoring registration as %s', plugin.source_name, plugin.name,
- self._registry[plugin.name].source, plugin.target_name)
+ 'ignoring registration as %s',
+ plugin.source_name,
+ plugin.name,
+ self._registry[plugin.name].source,
+ plugin.target_name,
+ )
elif plugin.source not in self._sources:
_LOG.warning(
'%s: "%s" is registered file multiple times in this file! '
- 'Only the first registration takes effect', plugin.source_name,
- plugin.name)
+ 'Only the first registration takes effect',
+ plugin.source_name,
+ plugin.name,
+ )
return False
@@ -251,14 +291,17 @@ class Registry(collections.abc.Mapping):
"""Registers an object as a plugin."""
return self._register(Plugin(name, target, None))
- def register_by_name(self,
- name: str,
- module_name: str,
- member_name: str,
- source: Path = None) -> Optional[Plugin]:
+ def register_by_name(
+ self,
+ name: str,
+ module_name: str,
+ member_name: str,
+ source: Optional[Path] = None,
+ ) -> Optional[Plugin]:
"""Registers an object from its module and name as a plugin."""
return self._register(
- Plugin.from_name(name, module_name, member_name, source))
+ Plugin.from_name(name, module_name, member_name, source)
+ )
def _register(self, plugin: Plugin) -> Optional[Plugin]:
# Prohibit functions not from a plugins file from overriding others.
@@ -266,8 +309,12 @@ class Registry(collections.abc.Mapping):
return None
self._registry[plugin.name] = plugin
- _LOG.debug('%s: Registered plugin "%s" for %s', plugin.source_name,
- plugin.name, plugin.target_name)
+ _LOG.debug(
+ '%s: Registered plugin "%s" for %s',
+ plugin.source_name,
+ plugin.name,
+ plugin.target_name,
+ )
return plugin
@@ -289,22 +336,33 @@ class Registry(collections.abc.Mapping):
_LOG.error(
'%s:%d: Failed to parse plugin entry "%s": '
'Expected 3 items (name, module, function), '
- 'got %d', path, lineno, line, len(line.split()))
+ 'got %d',
+ path,
+ lineno,
+ line,
+ len(line.split()),
+ )
continue
try:
self.register_by_name(name, module, function, path)
except Error as err:
self._errors[name].append(err)
- _LOG.error('%s: Failed to register plugin "%s": %s', path,
- name, err)
+ _LOG.error(
+ '%s: Failed to register plugin "%s": %s',
+ path,
+ name,
+ err,
+ )
self._sources.add(path)
- def register_directory(self,
- directory: Path,
- file_name: str,
- restrict_to: Path = None) -> None:
+ def register_directory(
+ self,
+ directory: Path,
+ file_name: str,
+ restrict_to: Optional[Path] = None,
+ ) -> None:
"""Finds and registers plugins from plugins files in a directory.
Args:
@@ -319,7 +377,9 @@ class Registry(collections.abc.Mapping):
if restrict_to is not None and restrict_to not in path.parents:
_LOG.debug(
"Skipping plugins file %s because it's outside of %s",
- path, restrict_to)
+ path,
+ restrict_to,
+ )
continue
_LOG.debug('Found plugins file %s', path)
@@ -327,11 +387,15 @@ class Registry(collections.abc.Mapping):
def short_help(self) -> str:
"""Returns a help string for the registered plugins."""
- width = max(len(name)
- for name in self._registry) + 1 if self._registry else 1
+ width = (
+ max(len(name) for name in self._registry) + 1
+ if self._registry
+ else 1
+ )
help_items = '\n'.join(
f' {name:{width}} {plugin.help()}'
- for name, plugin in sorted(self._registry.items()))
+ for name, plugin in sorted(self._registry.items())
+ )
return f'supported plugins:\n{help_items}'
def detailed_help(self, plugins: Iterable[str] = ()) -> Iterator[str]:
@@ -341,9 +405,9 @@ class Registry(collections.abc.Mapping):
yield '\ndetailed plugin information:'
- wrapper = TextWrapper(width=80,
- initial_indent=' ',
- subsequent_indent=' ' * 11)
+ wrapper = TextWrapper(
+ width=80, initial_indent=' ', subsequent_indent=' ' * 11
+ )
plugins = sorted(plugins)
for plugin in plugins:
@@ -360,19 +424,19 @@ class Registry(collections.abc.Mapping):
yield 'Plugins files:'
if self._sources:
- yield from (f' [{i}] {file}'
- for i, file in enumerate(self._sources, 1))
+ yield from (
+ f' [{i}] {file}' for i, file in enumerate(self._sources, 1)
+ )
else:
yield ' (none found)'
- def plugin(self,
- function: Callable = None,
- *,
- name: str = None) -> Callable[[Callable], Callable]:
+ def plugin(
+ self, function: Optional[Callable] = None, *, name: Optional[str] = None
+ ) -> Callable[[Callable], Callable]:
"""Decorator that registers a function with this plugin registry."""
+
def decorator(function: Callable) -> Callable:
- self.register(function.__name__ if name is None else name,
- function)
+ self.register(function.__name__ if name is None else name, function)
return function
if function is None:
@@ -407,8 +471,9 @@ def find_all_in_parents(name: str, path: Path) -> Iterator[Path]:
path = result.parent.parent
-def import_submodules(module: types.ModuleType,
- recursive: bool = False) -> None:
+def import_submodules(
+ module: types.ModuleType, recursive: bool = False
+) -> None:
"""Imports the submodules of a package.
This can be used to collect plugins registered with a decorator from a
diff --git a/pw_cli/py/pw_cli/process.py b/pw_cli/py/pw_cli/process.py
index 2366fc066..98852e35f 100644
--- a/pw_cli/py/pw_cli/process.py
+++ b/pw_cli/py/pw_cli/process.py
@@ -18,11 +18,14 @@ import logging
import os
import shlex
import tempfile
-from typing import IO, Tuple, Union, Optional
+from typing import Dict, IO, Tuple, Union, Optional
import pw_cli.color
import pw_cli.log
+import psutil # type: ignore
+
+
_COLOR = pw_cli.color.colors()
_LOG = logging.getLogger(__name__)
@@ -32,9 +35,18 @@ PW_SUBPROCESS_ENV = 'PW_SUBPROCESS'
class CompletedProcess:
- """Information about a process executed in run_async."""
- def __init__(self, process: 'asyncio.subprocess.Process',
- output: Union[bytes, IO[bytes]]):
+ """Information about a process executed in run_async.
+
+ Attributes:
+ pid: The process identifier.
+ returncode: The return code of the process.
+ """
+
+ def __init__(
+ self,
+ process: 'asyncio.subprocess.Process',
+ output: Union[bytes, IO[bytes]],
+ ):
assert process.returncode is not None
self.returncode: int = process.returncode
self.pid = process.pid
@@ -58,7 +70,8 @@ async def _run_and_log(program: str, args: Tuple[str, ...], env: dict):
*args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
- env=env)
+ env=env,
+ )
output = bytearray()
@@ -69,31 +82,105 @@ async def _run_and_log(program: str, args: Tuple[str, ...], env: dict):
break
output += line
- _LOG.log(pw_cli.log.LOGLEVEL_STDOUT, '[%s] %s',
- _COLOR.bold_white(process.pid),
- line.decode(errors='replace').rstrip())
+ _LOG.log(
+ pw_cli.log.LOGLEVEL_STDOUT,
+ '[%s] %s',
+ _COLOR.bold_white(process.pid),
+ line.decode(errors='replace').rstrip(),
+ )
return process, bytes(output)
-async def run_async(program: str,
- *args: str,
- log_output: bool = False,
- timeout: Optional[float] = None) -> CompletedProcess:
+async def _kill_process_and_children(
+ process: 'asyncio.subprocess.Process',
+) -> None:
+ """Kills child processes of a process with PID `pid`."""
+ # Look up child processes before sending the kill signal to the parent,
+ # as otherwise the child lookup would fail.
+ pid = process.pid
+ try:
+ process_util = psutil.Process(pid=pid)
+ kill_list = list(process_util.children(recursive=True))
+ except psutil.NoSuchProcess:
+ # Creating the kill list raced with parent process completion.
+ #
+ # Don't bother cleaning up child processes of parent processes that
+ # exited on their own.
+ kill_list = []
+
+ for proc in kill_list:
+ try:
+ proc.kill()
+ except psutil.NoSuchProcess:
+ pass
+
+ def wait_for_all() -> None:
+ for proc in kill_list:
+ try:
+ proc.wait()
+ except psutil.NoSuchProcess:
+ pass
+
+ # Wait for process completion on the executor to avoid blocking the
+ # event loop.
+ loop = asyncio.get_event_loop()
+ wait_for_children = loop.run_in_executor(None, wait_for_all)
+
+ # Send a kill signal to the main process before waiting for the children
+ # to complete.
+ try:
+ process.kill()
+ await process.wait()
+ except ProcessLookupError:
+ _LOG.debug(
+ 'Process completed before it could be killed. '
+ 'This may have been caused by the killing one of its '
+ 'child subprocesses.',
+ )
+
+ await wait_for_children
+
+
+async def run_async(
+ program: str,
+ *args: str,
+ env: Optional[Dict[str, str]] = None,
+ log_output: bool = False,
+ timeout: Optional[float] = None,
+) -> CompletedProcess:
"""Runs a command, capturing and optionally logging its output.
+ Args:
+ program: The program to run in a new process.
+ args: The arguments to pass to the program.
+ env: An optional mapping of environment variables within which to run
+ the process.
+ log_output: Whether to log stdout and stderr of the process to this
+ process's stdout (prefixed with the PID of the subprocess from which
+ the output originated). If unspecified, the child process's stdout
+ and stderr will be captured, and both will be stored in the returned
+ `CompletedProcess`'s output`.
+ timeout: An optional floating point number of seconds to allow the
+ subprocess to run before killing it and its children. If unspecified,
+ the subprocess will be allowed to continue exiting until it completes.
+
Returns a CompletedProcess with details from the process.
"""
- _LOG.debug('Running `%s`',
- ' '.join(shlex.quote(arg) for arg in [program, *args]))
+ _LOG.debug(
+ 'Running `%s`', ' '.join(shlex.quote(arg) for arg in [program, *args])
+ )
- env = os.environ.copy()
- env[PW_SUBPROCESS_ENV] = '1'
+ hydrated_env = os.environ.copy()
+ if env is not None:
+ for key, value in env.items():
+ hydrated_env[key] = value
+ hydrated_env[PW_SUBPROCESS_ENV] = '1'
output: Union[bytes, IO[bytes]]
if log_output:
- process, output = await _run_and_log(program, args, env)
+ process, output = await _run_and_log(program, args, hydrated_env)
else:
output = tempfile.TemporaryFile()
process = await asyncio.create_subprocess_exec(
@@ -101,19 +188,19 @@ async def run_async(program: str,
*args,
stdout=output,
stderr=asyncio.subprocess.STDOUT,
- env=env)
+ env=hydrated_env,
+ )
try:
await asyncio.wait_for(process.wait(), timeout)
except asyncio.TimeoutError:
_LOG.error('%s timed out after %d seconds', program, timeout)
- process.kill()
- await process.wait()
+ await _kill_process_and_children(process)
if process.returncode:
_LOG.error('%s exited with status %d', program, process.returncode)
else:
- _LOG.info('%s exited successfully', program)
+ _LOG.error('%s exited successfully', program)
return CompletedProcess(process, output)
diff --git a/pw_cli/py/pw_cli/pw_command_plugins.py b/pw_cli/py/pw_cli/pw_command_plugins.py
index 845e0078f..99b86fdd8 100644
--- a/pw_cli/py/pw_cli/pw_command_plugins.py
+++ b/pw_cli/py/pw_cli/pw_command_plugins.py
@@ -29,12 +29,15 @@ def _register_builtin_plugins(registry: plugins.Registry) -> None:
"""Registers the commands that are included with pw by default."""
# Register these by name to avoid circular dependencies.
+ registry.register_by_name('bloat', 'pw_bloat.__main__', 'main')
registry.register_by_name('doctor', 'pw_doctor.doctor', 'main')
- registry.register_by_name('python-packages',
- 'pw_env_setup.python_packages', 'main')
registry.register_by_name('format', 'pw_presubmit.format_code', 'main')
+ registry.register_by_name('keep-sorted', 'pw_presubmit.keep_sorted', 'main')
registry.register_by_name('logdemo', 'pw_cli.log', 'main')
- registry.register_by_name('module-check', 'pw_module.check', 'main')
+ registry.register_by_name('module', 'pw_module.__main__', 'main')
+ registry.register_by_name(
+ 'python-packages', 'pw_env_setup.python_packages', 'main'
+ )
registry.register_by_name('test', 'pw_unit_test.test_runner', 'main')
registry.register_by_name('watch', 'pw_watch.watch', 'main')
@@ -44,10 +47,12 @@ def _register_builtin_plugins(registry: plugins.Registry) -> None:
def _help_command():
"""Display detailed information about pw commands."""
parser = argparse.ArgumentParser(description=_help_command.__doc__)
- parser.add_argument('plugins',
- metavar='plugin',
- nargs='*',
- help='command for which to display detailed info')
+ parser.add_argument(
+ 'plugins',
+ metavar='plugin',
+ nargs='*',
+ help='command for which to display detailed info',
+ )
print(arguments.format_help(_plugin_registry), file=sys.stderr)
@@ -58,7 +63,8 @@ def _help_command():
def register(directory: Path) -> None:
_register_builtin_plugins(_plugin_registry)
_plugin_registry.register_directory(
- directory, REGISTRY_FILE, Path(os.environ.get('PW_PROJECT_ROOT', '')))
+ directory, REGISTRY_FILE, Path(os.environ.get('PW_PROJECT_ROOT', ''))
+ )
def errors() -> dict:
diff --git a/pw_cli/py/pw_cli/requires.py b/pw_cli/py/pw_cli/requires.py
index bf67dd345..79a2b61de 100755
--- a/pw_cli/py/pw_cli/requires.py
+++ b/pw_cli/py/pw_cli/requires.py
@@ -27,6 +27,7 @@ For more see http://go/pigweed-ci-cq-intro.
"""
import argparse
+import json
import logging
from pathlib import Path
import re
@@ -40,11 +41,13 @@ HELPER_PROJECT = 'requires-helper'
HELPER_REPO = 'sso://{}/{}'.format(HELPER_GERRIT, HELPER_PROJECT)
# Pass checks that look for "DO NOT ..." and block submission.
-_DNS = ' '.join((
- 'DO',
- 'NOT',
- 'SUBMIT',
-))
+_DNS = ' '.join(
+ (
+ 'DO',
+ 'NOT',
+ 'SUBMIT',
+ )
+)
# Subset of the output from pushing to Gerrit.
DEFAULT_OUTPUT = f'''
@@ -100,17 +103,24 @@ def clone(requires_dir: Path) -> None:
def create_commit(requires_dir: Path, requirements) -> None:
+ """Create a commit in the local tree with the given requirements."""
change_id = str(uuid.uuid4()).replace('-', '00')
_LOG.debug('change_id %s', change_id)
- path = requires_dir / change_id
+
+ reqs = []
+ for req in requirements:
+ gerrit_name, number = req.split(':', 1)
+ reqs.append({'gerrit_name': gerrit_name, 'number': number})
+
+ path = requires_dir / 'patches.json'
_LOG.debug('path %s', path)
- with open(path, 'w'):
- pass
+ with open(path, 'w') as outs:
+ json.dump(reqs, outs)
_run_command(['git', 'add', path], cwd=requires_dir)
commit_message = [
- f'{_DNS} {change_id[0:10]}',
+ f'{_DNS} {change_id[0:10]}\n\n',
'',
f'Change-Id: I{change_id}',
]
diff --git a/pw_cli/py/pw_cli/toml_config_loader_mixin.py b/pw_cli/py/pw_cli/toml_config_loader_mixin.py
new file mode 100644
index 000000000..bac57ae24
--- /dev/null
+++ b/pw_cli/py/pw_cli/toml_config_loader_mixin.py
@@ -0,0 +1,50 @@
+# Copyright 2022 The Pigweed Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+"""Toml config file loader mixin."""
+
+from typing import Any, Dict, List
+
+import toml # type: ignore
+
+from pw_cli.yaml_config_loader_mixin import YamlConfigLoaderMixin
+
+
+class TomlConfigLoaderMixin(YamlConfigLoaderMixin):
+ """TOML Config file loader mixin.
+
+ Use this mixin to load toml file settings and save them into
+ ``self._config``. For example:
+
+ ::
+
+ from pw_cli.toml_config_loader_mixin import TomlConfigLoaderMixin
+
+ class PwBloatPrefs(TomlConfigLoaderMixin):
+ def __init__(self) -> None:
+ self.config_init(
+ config_section_title='pw_bloat',
+ project_file=Path('$PW_PROJECT_ROOT/.pw_bloat.toml'),
+ project_user_file=Path(
+ '$PW_PROJECT_ROOT/.pw_bloat.user.toml'),
+ user_file=Path('~/.pw_bloat.toml'),
+ default_config={},
+ environment_var='PW_BLOAT_CONFIG_FILE',
+ )
+
+ """
+
+ def _load_config_from_string( # pylint: disable=no-self-use
+ self, file_contents: str
+ ) -> List[Dict[Any, Any]]:
+ return [toml.loads(file_contents)]
diff --git a/pw_cli/py/pw_cli/yaml_config_loader_mixin.py b/pw_cli/py/pw_cli/yaml_config_loader_mixin.py
new file mode 100644
index 000000000..4bbf47bef
--- /dev/null
+++ b/pw_cli/py/pw_cli/yaml_config_loader_mixin.py
@@ -0,0 +1,166 @@
+# Copyright 2022 The Pigweed Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+"""Yaml config file loader mixin."""
+
+import os
+import logging
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Union
+
+import yaml
+
+_LOG = logging.getLogger(__package__)
+
+
+class MissingConfigTitle(Exception):
+ """Exception for when an existing YAML file is missing config_title."""
+
+
+class YamlConfigLoaderMixin:
+ """Yaml Config file loader mixin.
+
+ Use this mixin to load yaml file settings and save them into
+ ``self._config``. For example:
+
+ ::
+
+ class ConsolePrefs(YamlConfigLoaderMixin):
+ def __init__(self) -> None:
+ self.config_init(
+ config_section_title='pw_console',
+ project_file=Path('project_file.yaml'),
+ project_user_file=Path('project_user_file.yaml'),
+ user_file=Path('~/user_file.yaml'),
+ default_config={},
+ environment_var='PW_CONSOLE_CONFIG_FILE',
+ )
+
+ """
+
+ def config_init(
+ self,
+ config_section_title: str,
+ project_file: Optional[Union[Path, bool]] = None,
+ project_user_file: Optional[Union[Path, bool]] = None,
+ user_file: Optional[Union[Path, bool]] = None,
+ default_config: Optional[Dict[Any, Any]] = None,
+ environment_var: Optional[str] = None,
+ ) -> None:
+ """Call this to load YAML config files in order of precedence.
+
+ The following files are loaded in this order:
+ 1. project_file
+ 2. project_user_file
+ 3. user_file
+
+ Lastly, if a valid file path is specified at
+ ``os.environ[environment_var]`` then load that file overriding all
+ config options.
+
+ Args:
+ config_section_title: String name of this config section. For
+ example: ``pw_console`` or ``pw_watch``. In the YAML file this
+ is represented by a ``config_title`` key.
+
+ ::
+
+ ---
+ config_title: pw_console
+
+ project_file: Project level config file. This is intended to be a
+ file living somewhere under a project folder and is checked into
+ the repo. It serves as a base config all developers can inherit
+ from.
+ project_user_file: User's personal config file for a specific
+ project. This can be a file that lives in a project folder that
+ is git-ignored and not checked into the repo.
+ user_file: A global user based config file. This is typically a file
+ in the users home directory and settings here apply to all
+ projects.
+ default_config: A Python dict representing the base default
+ config. This dict will be applied as a starting point before
+ loading any yaml files.
+ environment_var: The name of an environment variable to check for a
+ config file. If a config file exists there it will be loaded on
+ top of the default_config ignoring project and user files.
+ """
+
+ self._config_section_title: str = config_section_title
+ self.default_config = default_config if default_config else {}
+ self.reset_config()
+
+ if project_file and isinstance(project_file, Path):
+ self.project_file = Path(
+ os.path.expandvars(str(project_file.expanduser()))
+ )
+ self.load_config_file(self.project_file)
+
+ if project_user_file and isinstance(project_user_file, Path):
+ self.project_user_file = Path(
+ os.path.expandvars(str(project_user_file.expanduser()))
+ )
+ self.load_config_file(self.project_user_file)
+
+ if user_file and isinstance(user_file, Path):
+ self.user_file = Path(
+ os.path.expandvars(str(user_file.expanduser()))
+ )
+ self.load_config_file(self.user_file)
+
+ # Check for a config file specified by an environment variable.
+ if environment_var is None:
+ return
+ environment_config = os.environ.get(environment_var, None)
+ if environment_config:
+ env_file_path = Path(environment_config)
+ if not env_file_path.is_file():
+ raise FileNotFoundError(
+ f'Cannot load config file: {env_file_path}'
+ )
+ self.reset_config()
+ self.load_config_file(env_file_path)
+
+ def _update_config(self, cfg: Dict[Any, Any]) -> None:
+ if cfg is None:
+ cfg = {}
+ self._config.update(cfg)
+
+ def reset_config(self) -> None:
+ self._config: Dict[Any, Any] = {}
+ self._update_config(self.default_config)
+
+ def _load_config_from_string( # pylint: disable=no-self-use
+ self, file_contents: str
+ ) -> List[Dict[Any, Any]]:
+ return list(yaml.safe_load_all(file_contents))
+
+ def load_config_file(self, file_path: Path) -> None:
+ if not file_path.is_file():
+ return
+
+ cfgs = self._load_config_from_string(file_path.read_text())
+
+ for cfg in cfgs:
+ if self._config_section_title in cfg:
+ self._update_config(cfg[self._config_section_title])
+
+ elif cfg.get('config_title', False) == self._config_section_title:
+ self._update_config(cfg)
+ else:
+ raise MissingConfigTitle(
+ '\n\nThe config file "{}" is missing the expected '
+ '"config_title: {}" setting.'.format(
+ str(file_path), self._config_section_title
+ )
+ )
diff --git a/pw_cli/py/setup.cfg b/pw_cli/py/setup.cfg
index be8343850..20af8de6f 100644
--- a/pw_cli/py/setup.cfg
+++ b/pw_cli/py/setup.cfg
@@ -21,6 +21,10 @@ description = Pigweed swiss-army knife
[options]
packages = find:
zip_safe = False
+install_requires =
+ psutil
+ pyyaml
+ toml
[options.entry_points]
console_scripts = pw = pw_cli.__main__:main
generated by cgit v1.2.3 (git 2.25.1) at 2024-06-20 19:04:17 +0000