[ros2interface] Allow stdin input for 'ros2 interface show' (ros2#387)

* Allow stdin input for 'ros2 interface show' Signed-off-by: Nursharmin Ramli <[email protected]> * just use help for all the information Signed-off-by: Dirk Thomas <[email protected]> * fix import order Signed-off-by: Dirk Thomas <[email protected]> * simplify logic Signed-off-by: Dirk Thomas <[email protected]> * catch empty values in case stdin doesn't contain output Signed-off-by: Dirk Thomas <[email protected]> * Add test for 'ros2 interface show' with stdin Signed-off-by: Nursharmin Ramli <[email protected]> * Use test_msgs instead of std_msgs for stdin test Signed-off-by: Nursharmin Ramli <[email protected]> * Use example_interfaces in help for show Signed-off-by: Nursharmin Ramli <[email protected]> Co-authored-by: Dirk Thomas <[email protected]> Signed-off-by: Craig <[email protected]>
craigh92 · Jun 11, 2020 · dd9509e · dd9509e
1 parent c280dc4
commit dd9509e
Show file tree

Hide file tree

Showing 2 changed files with 40 additions and 4 deletions.
diff --git a/ros2interface/ros2interface/verb/show.py b/ros2interface/ros2interface/verb/show.py
@@ -12,18 +12,37 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import argparse
+import sys
+
 from ros2interface.api import type_completer
 from ros2interface.verb import VerbExtension
 from rosidl_runtime_py import get_interface_path
 
 
+class ReadStdinPipe(argparse.Action):
+    """Get argument from stdin pipe."""
+
+    def __call__(self, parser, namespace, values, option_string=None):
+        if values == '-':
+            if sys.stdin.isatty():
+                parser.error('expected stdin pipe')
+            values = sys.stdin.readline().strip()
+        if not values:
+            parser.error('the passed value is empty')
+        setattr(namespace, self.dest, values)
+
+
 class ShowVerb(VerbExtension):
     """Output the interface definition."""
 
     def add_arguments(self, parser, cli_name):
         arg = parser.add_argument(
-                'type',
-                help="Show an interface definition (e.g. 'example_interfaces/msg/String')")
+            'type',
+            action=ReadStdinPipe,
+            help="Show an interface definition (e.g. 'example_interfaces/msg/String'). "
+                 "Passing '-' reads the argument from stdin (e.g. "
+                 "'ros2 topic type /chatter | ros2 interface show -').")
         arg.completer = type_completer
 
     def main(self, *, args):

diff --git a/ros2interface/test/test_cli.py b/ros2interface/test/test_cli.py
@@ -77,11 +77,12 @@ def setUpClass(
         proc_output
     ):
         @contextlib.contextmanager
-        def launch_interface_command(self, arguments):
+        def launch_interface_command(self, arguments, prepend_arguments=[], shell=False):
             interface_command_action = ExecuteProcess(
-                cmd=['ros2', 'interface', *arguments],
+                cmd=[*prepend_arguments, 'ros2', 'interface', *arguments],
                 additional_env={'PYTHONUNBUFFERED': '1'},
                 name='ros2interface-cli',
+                shell=shell,
                 output='screen'
             )
             with launch_testing.tools.launch_process(
@@ -335,3 +336,19 @@ def test_show_not_an_interface(self):
             text=interface_command.output,
             strict=True
         )
+
+    def test_show_stdin(self):
+        with self.launch_interface_command(
+            arguments=['show', '-'],
+            prepend_arguments=[sys.executable, '-c', r'"print(\"test_msgs/msg/Nested\")"', '|'],
+            shell=True
+        ) as interface_command:
+            assert interface_command.wait_for_shutdown(timeout=2)
+        assert interface_command.exit_code == launch_testing.asserts.EXIT_OK
+        assert launch_testing.tools.expect_output(
+            expected_lines=[
+                'BasicTypes basic_types_value'
+            ],
+            text=interface_command.output,
+            strict=True
+        )