pw_rpc.console_tools: Support aliasing renamed commands
- The alias_deprecated_command function creates an alias that redirects
from an old RPC command name to a new name.
- Move console_tools.Watchdog tests to their own file.
Change-Id: Ie7071462e5c17e5782fb549cc2f15aa19e64bbf0
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/44120
Reviewed-by: Alexei Frolov <frolv@google.com>
Commit-Queue: Wyatt Hepler <hepler@google.com>
diff --git a/pw_rpc/py/BUILD.gn b/pw_rpc/py/BUILD.gn
index 71faaee..8b729a0 100644
--- a/pw_rpc/py/BUILD.gn
+++ b/pw_rpc/py/BUILD.gn
@@ -46,6 +46,7 @@
"tests/client_test.py",
"tests/console_tools/console_tools_test.py",
"tests/console_tools/functions_test.py",
+ "tests/console_tools/watchdog_test.py",
"tests/descriptors_test.py",
"tests/ids_test.py",
"tests/packets_test.py",
diff --git a/pw_rpc/py/docs.rst b/pw_rpc/py/docs.rst
index 5aeea46..3e6ef22 100644
--- a/pw_rpc/py/docs.rst
+++ b/pw_rpc/py/docs.rst
@@ -20,4 +20,4 @@
pw_rpc.console_tools
====================
.. automodule:: pw_rpc.console_tools
- :members: Context, ClientInfo, Watchdog, help_as_repr
+ :members: Context, ClientInfo, Watchdog, alias_deprecated_command, help_as_repr
diff --git a/pw_rpc/py/pw_rpc/console_tools/__init__.py b/pw_rpc/py/pw_rpc/console_tools/__init__.py
index 3f157f1..c714962 100644
--- a/pw_rpc/py/pw_rpc/console_tools/__init__.py
+++ b/pw_rpc/py/pw_rpc/console_tools/__init__.py
@@ -13,6 +13,7 @@
# the License.
"""Utilities for building tools that interact with pw_rpc."""
-from pw_rpc.console_tools.console import Context, CommandHelper, ClientInfo
+from pw_rpc.console_tools.console import (Context, CommandHelper, ClientInfo,
+ alias_deprecated_command)
from pw_rpc.console_tools.functions import help_as_repr
from pw_rpc.console_tools.watchdog import Watchdog
diff --git a/pw_rpc/py/pw_rpc/console_tools/console.py b/pw_rpc/py/pw_rpc/console_tools/console.py
index 29797f5..6ccdb63 100644
--- a/pw_rpc/py/pw_rpc/console_tools/console.py
+++ b/pw_rpc/py/pw_rpc/console_tools/console.py
@@ -14,6 +14,7 @@
"""Utilities for creating an interactive console."""
from collections import defaultdict
+import functools
from itertools import chain
import inspect
import textwrap
@@ -217,3 +218,71 @@
self.protos.packages[package_name]._add_item(rpcs) # pylint: disable=protected-access
self.current_client = selected_client
+
+
+def _create_command_alias(command: Any, name: str, message: str) -> object:
+ """Wraps __call__, __getattr__, and __repr__ to print a message."""
+ @functools.wraps(command.__call__)
+ def print_message_and_call(_, *args, **kwargs):
+ print(message)
+ return command(*args, **kwargs)
+
+ def getattr_and_print_message(_, name: str) -> Any:
+ attr = getattr(command, name)
+ print(message)
+ return attr
+
+ return type(
+ name, (),
+ dict(__call__=print_message_and_call,
+ __getattr__=getattr_and_print_message,
+ __repr__=lambda _: message))()
+
+
+def _access_in_dict_or_namespace(item, name: str, create_if_missing: bool):
+ """Gets name as either a key or attribute on item."""
+ try:
+ return item[name]
+ except KeyError:
+ if create_if_missing:
+ try:
+ item[name] = types.SimpleNamespace()
+ return item[name]
+ except TypeError:
+ pass
+ except TypeError:
+ pass
+
+ if create_if_missing and not hasattr(item, name):
+ setattr(item, name, types.SimpleNamespace())
+
+ return getattr(item, name)
+
+
+def _access_names(item, names: Iterable[str], create_if_missing: bool):
+ for name in names:
+ item = _access_in_dict_or_namespace(item, name, create_if_missing)
+
+ return item
+
+
+def alias_deprecated_command(variables: Any, old_name: str,
+ new_name: str) -> None:
+ """Adds an alias for an old command that redirects to the new command.
+
+ The deprecated command prints a message then invokes the new command.
+ """
+ # Get the new command.
+ item = _access_names(variables,
+ new_name.split('.'),
+ create_if_missing=False)
+
+ # Create a wrapper to the new comamnd with the old name.
+ wrapper = _create_command_alias(
+ item, old_name,
+ f'WARNING: {old_name} is DEPRECATED; use {new_name} instead')
+
+ # Add the wrapper to the variables with the old command's name.
+ name_parts = old_name.split('.')
+ item = _access_names(variables, name_parts[:-1], create_if_missing=True)
+ setattr(item, name_parts[-1], wrapper)
diff --git a/pw_rpc/py/tests/console_tools/console_tools_test.py b/pw_rpc/py/tests/console_tools/console_tools_test.py
index 3dbb5ce..36285db 100755
--- a/pw_rpc/py/tests/console_tools/console_tools_test.py
+++ b/pw_rpc/py/tests/console_tools/console_tools_test.py
@@ -12,72 +12,18 @@
# 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 encoding HDLC frames."""
+"""Tests the pw_rpc.console_tools.console module."""
+import types
import unittest
-from unittest import mock
import pw_status
from pw_protobuf_compiler import python_protos
import pw_rpc
from pw_rpc import callback_client
-from pw_rpc.console_tools import CommandHelper, Context, ClientInfo, Watchdog
-
-
-class TestWatchdog(unittest.TestCase):
- """Tests the Watchdog class."""
- def setUp(self) -> None:
- self._reset = mock.Mock()
- self._expiration = mock.Mock()
- self._while_expired = mock.Mock()
-
- self._watchdog = Watchdog(self._reset, self._expiration,
- self._while_expired, 99999)
-
- def _trigger_timeout(self) -> None:
- # Don't wait for the timeout -- that's too flaky. Call the internal
- # timeout function instead.
- self._watchdog._timeout_expired() # pylint: disable=protected-access
-
- def test_expiration_callbacks(self) -> None:
- self._watchdog.start()
-
- self._expiration.not_called()
-
- self._trigger_timeout()
-
- self._expiration.assert_called_once_with()
- self._while_expired.assert_not_called()
-
- self._trigger_timeout()
-
- self._expiration.assert_called_once_with()
- self._while_expired.assert_called_once_with()
-
- self._trigger_timeout()
-
- self._expiration.assert_called_once_with()
- self._while_expired.assert_called()
-
- def test_reset_not_called_unless_expires(self) -> None:
- self._watchdog.start()
- self._watchdog.reset()
-
- self._reset.assert_not_called()
- self._expiration.assert_not_called()
- self._while_expired.assert_not_called()
-
- def test_reset_called_if_expired(self) -> None:
- self._watchdog.start()
- self._trigger_timeout()
-
- self._watchdog.reset()
-
- self._trigger_timeout()
-
- self._reset.assert_called_once_with()
- self._expiration.assert_called()
+from pw_rpc.console_tools.console import (CommandHelper, Context, ClientInfo,
+ alias_deprecated_command)
class TestCommandHelper(unittest.TestCase):
@@ -229,5 +175,31 @@
self.assertTrue(called_derived_set_target)
+class TestAliasDeprecatedCommand(unittest.TestCase):
+ def test_wraps_command_to_new_package(self) -> None:
+ variables = {'abc': types.SimpleNamespace(command=lambda: 123)}
+ alias_deprecated_command(variables, 'xyz.one.two.three', 'abc.command')
+
+ self.assertEqual(variables['xyz'].one.two.three(), 123)
+
+ def test_wraps_command_to_existing_package(self) -> None:
+ variables = {
+ 'abc': types.SimpleNamespace(NewCmd=lambda: 456),
+ 'one': types.SimpleNamespace(),
+ }
+ alias_deprecated_command(variables, 'one.two.OldCmd', 'abc.NewCmd')
+
+ self.assertEqual(variables['one'].two.OldCmd(), 456)
+
+ def test_error_if_new_command_does_not_exist(self) -> None:
+ variables = {
+ 'abc': types.SimpleNamespace(),
+ 'one': types.SimpleNamespace(),
+ }
+
+ with self.assertRaises(AttributeError):
+ alias_deprecated_command(variables, 'one.two.OldCmd', 'abc.NewCmd')
+
+
if __name__ == '__main__':
unittest.main()
diff --git a/pw_rpc/py/tests/console_tools/watchdog_test.py b/pw_rpc/py/tests/console_tools/watchdog_test.py
new file mode 100644
index 0000000..9bc203c
--- /dev/null
+++ b/pw_rpc/py/tests/console_tools/watchdog_test.py
@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+# Copyright 2021 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 the Watchdog module."""
+
+import unittest
+from unittest import mock
+
+from pw_rpc.console_tools import Watchdog
+
+
+class TestWatchdog(unittest.TestCase):
+ """Tests the Watchdog class."""
+ def setUp(self) -> None:
+ self._reset = mock.Mock()
+ self._expiration = mock.Mock()
+ self._while_expired = mock.Mock()
+
+ self._watchdog = Watchdog(self._reset, self._expiration,
+ self._while_expired, 99999)
+
+ def _trigger_timeout(self) -> None:
+ # Don't wait for the timeout -- that's too flaky. Call the internal
+ # timeout function instead.
+ self._watchdog._timeout_expired() # pylint: disable=protected-access
+
+ def test_expiration_callbacks(self) -> None:
+ self._watchdog.start()
+
+ self._expiration.not_called()
+
+ self._trigger_timeout()
+
+ self._expiration.assert_called_once_with()
+ self._while_expired.assert_not_called()
+
+ self._trigger_timeout()
+
+ self._expiration.assert_called_once_with()
+ self._while_expired.assert_called_once_with()
+
+ self._trigger_timeout()
+
+ self._expiration.assert_called_once_with()
+ self._while_expired.assert_called()
+
+ def test_reset_not_called_unless_expires(self) -> None:
+ self._watchdog.start()
+ self._watchdog.reset()
+
+ self._reset.assert_not_called()
+ self._expiration.assert_not_called()
+ self._while_expired.assert_not_called()
+
+ def test_reset_called_if_expired(self) -> None:
+ self._watchdog.start()
+ self._trigger_timeout()
+
+ self._watchdog.reset()
+
+ self._trigger_timeout()
+
+ self._reset.assert_called_once_with()
+ self._expiration.assert_called()
+
+
+if __name__ == '__main__':
+ unittest.main()
