Update the launch code for newer flake8 and mypy. (ros2#719)

Newer flake8 complains if you use parentheses around
assert statements, so fix that.

Newer mypy had a number of minor complaints about our
types, so improve the type checking to account for that.

With both of these in place, the code is now clean under
the versions of flake8 and mypy in Fedora 38 (5.0.3 and 1.4.0,
respectively).

Signed-off-by: Chris Lalancette <[email protected]>

launch/launch/actions/declare_launch_argument.py

@@ -108,7 +108,7 @@ def __init__(
         *,
         default_value: Optional[SomeSubstitutionsType] = None,
         description: Optional[Text] = None,
-        choices: List[Text] = None,
+        choices: Optional[List[Text]] = None,
         **kwargs
     ) -> None:
         """Create a DeclareLaunchArgument action."""
@@ -137,6 +137,7 @@ def __init__(
                         'Provided default_value "{}" is not in provided choices "{}".'.format(
                             default_value, choices))
 
+        self.__description = ''
         if description is None:
             if choices is None:
                 self.__description = 'no description given'

launch/launch/event_handlers/on_process_io.py

@@ -41,9 +41,9 @@ def __init__(
         *,
         target_action:
             Optional[Union[Callable[['Action'], bool], 'Action']] = None,
-        on_stdin: Callable[[ProcessIO], Optional[SomeEntitiesType]] = None,
-        on_stdout: Callable[[ProcessIO], Optional[SomeEntitiesType]] = None,
-        on_stderr: Callable[[ProcessIO], Optional[SomeEntitiesType]] = None,
+        on_stdin: Optional[Callable[[ProcessIO], Optional[SomeEntitiesType]]] = None,
+        on_stdout: Optional[Callable[[ProcessIO], Optional[SomeEntitiesType]]] = None,
+        on_stderr: Optional[Callable[[ProcessIO], Optional[SomeEntitiesType]]] = None,
         **kwargs
     ) -> None:
         """Create an OnProcessIO event handler."""

launch/launch/frontend/parse_substitution.py

@@ -39,7 +39,7 @@ class ExtractSubstitution(Transformer):
     """Extract a substitution."""
 
     def part(self, content):
-        assert(len(content) == 1)
+        assert len(content) == 1
         content = content[0]
         if isinstance(content, Token):
             assert content.type.endswith('_RSTRING')

launch/launch/frontend/parser.py

@@ -139,7 +139,7 @@ def parse_description(self, entity: Entity) -> 'LaunchDescription':
     def get_available_extensions(cls) -> List[Text]:
         """Return the registered extensions."""
         cls.load_parser_implementations()
-        assert(cls.frontend_parsers is not None)
+        assert cls.frontend_parsers is not None
         return cls.frontend_parsers.keys()
 
     @classmethod
@@ -151,7 +151,7 @@ def is_extension_valid(
         warnings.warn(
             'Parser.is_extension_valid is deprecated, use Parser.is_filename_valid instead')
         cls.load_parser_implementations()
-        assert(cls.frontend_parsers is not None)
+        assert cls.frontend_parsers is not None
         return extension in cls.frontend_parsers
 
     @classmethod
@@ -164,7 +164,7 @@ def get_parser_from_extension(
             'Parser.get_parser_from_extension is deprecated, '
             'use Parser.get_parsers_from_filename instead')
         cls.load_parser_implementations()
-        assert(cls.frontend_parsers is not None)
+        assert cls.frontend_parsers is not None
         try:
             return cls.frontend_parsers[extension]
         except KeyError:
@@ -185,7 +185,7 @@ def is_filename_valid(
     ) -> bool:
         """Return `True` if the filename is valid for any parser."""
         cls.load_parser_implementations()
-        assert(cls.frontend_parsers is not None)
+        assert cls.frontend_parsers is not None
         return any(
             parser.may_parse(filename)
             for parser in cls.frontend_parsers.values()
@@ -198,7 +198,7 @@ def get_parsers_from_filename(
     ) -> List[Type['Parser']]:
         """Return a list of parsers which entity loaded with a markup file."""
         cls.load_parser_implementations()
-        assert(cls.frontend_parsers is not None)
+        assert cls.frontend_parsers is not None
         return [
             parser for parser in cls.frontend_parsers.values()
             if parser.may_parse(filename)
@@ -208,7 +208,7 @@ def get_parsers_from_filename(
     def get_file_extensions_from_parsers(cls) -> Set[Type['Parser']]:
         """Return a set of file extensions known to the parser implementations."""
         cls.load_parser_implementations()
-        assert(cls.frontend_parsers is not None)
+        assert cls.frontend_parsers is not None
         return set(itertools.chain.from_iterable(
             parser_extension.get_file_extensions()
             for parser_extension in cls.frontend_parsers.values()
@@ -241,7 +241,7 @@ def load(
         try:
             filename = getattr(fileobj, 'name', '')
             implementations = cls.get_parsers_from_filename(filename)
-            assert(cls.frontend_parsers is not None)
+            assert cls.frontend_parsers is not None
             implementations += [
                 parser for parser in cls.frontend_parsers.values()
                 if parser not in implementations

launch/launch/substitutions/python_expression.py

@@ -71,8 +71,8 @@ def parse(cls, data: Sequence[SomeSubstitutionsType]):
             kwargs['python_modules'] = []
             # Check if we got empty list from XML
             # Ensure that we got a list!
-            assert(not isinstance(data[1], str))
-            assert(not isinstance(data[1], Substitution))
+            assert not isinstance(data[1], str)
+            assert not isinstance(data[1], Substitution)
             # Modules
             modules = list(data[1])
             if len(modules) > 0:

launch/launch/utilities/signal_management.py

@@ -78,25 +78,28 @@ def __init__(
         self.__background_loop: Optional[asyncio.AbstractEventLoop] = None
         self.__handlers: dict = {}
         self.__prev_wakeup_handle: Union[int, socket.socket] = -1
-        self.__wsock = None
-        self.__rsock = None
-        self.__close_sockets = None
+        self.__wsock: Optional[socket.socket] = None
+        self.__rsock: Optional[socket.socket] = None
+        self.__close_sockets: Optional[Callable] = None
 
-    def __enter__(self):
+    def __enter__(self) -> 'AsyncSafeSignalManager':
         pair = socket.socketpair()  # type: Tuple[socket.socket, socket.socket]  # noqa
         with ExitStack() as stack:
             self.__wsock = stack.enter_context(pair[0])
             self.__rsock = stack.enter_context(pair[1])
-            self.__wsock.setblocking(False)
-            self.__rsock.setblocking(False)
+            if self.__wsock is not None:
+                self.__wsock.setblocking(False)
+            if self.__rsock is not None:
+                self.__rsock.setblocking(False)
             self.__close_sockets = stack.pop_all().close
 
         self.__add_signal_readers()
         try:
             self.__install_signal_writers()
         except Exception:
             self.__remove_signal_readers()
-            self.__close_sockets()
+            if self.__close_sockets is not None:
+                self.__close_sockets()
             self.__rsock = None
             self.__wsock = None
             self.__close_sockets = None

launch/test/launch/actions/test_push_and_pop_environment.py

@@ -33,7 +33,7 @@ def test_push_and_pop_environment_constructors():
 @sandbox_environment_variables
 def test_push_and_pop_environment_execute():
     """Test the execute() of the PopEnvironment and PushEnvironment classes."""
-    assert(type(os.environ) == os._Environ)
+    assert type(os.environ) == os._Environ
 
     context = LaunchContext()
 
@@ -89,4 +89,4 @@ def test_push_and_pop_environment_execute():
     assert context.environment['foo'] == 'FOO'
 
     # Pushing and popping the environment should not change the type of os.environ
-    assert(type(os.environ) == os._Environ)
+    assert type(os.environ) == os._Environ

launch/test/launch/frontend/test_parser.py

@@ -48,8 +48,8 @@ def test_invalid_launch_extension():
         }
         with warnings.catch_warnings(record=True) as caught_warnings:
             Parser.load_launch_extensions()
-            assert(caught_warnings)
-            assert('Failed to load the launch' in str(caught_warnings[0]))
+            assert caught_warnings
+            assert 'Failed to load the launch' in str(caught_warnings[0])
 
 
 def test_invalid_parser_implementations():
@@ -61,5 +61,5 @@ def test_invalid_parser_implementations():
 
         with warnings.catch_warnings(record=True) as caught_warnings:
             Parser.load_parser_implementations()
-            assert(caught_warnings)
-            assert('Failed to load the parser' in str(caught_warnings[0]))
+            assert caught_warnings
+            assert 'Failed to load the parser' in str(caught_warnings[0])