Add type hints for twisted.application.twist._options

debugmm · Sep 30, 2020 · 406861a · 406861a
1 parent 83935f3
commit 406861a
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 29 deletions.
diff --git a/mypy.ini b/mypy.ini
@@ -52,9 +52,6 @@ check_untyped_defs = False
 [mypy-twisted.application.test.test_service]
 allow_untyped_defs = True
 
-[mypy-twisted.application.twist._options]
-allow_untyped_defs = True
-
 [mypy-twisted.application.twist._twist]
 allow_untyped_defs = True
 

diff --git a/src/twisted/application/twist/_options.py b/src/twisted/application/twist/_options.py
@@ -8,16 +8,18 @@
 
 from sys import stdout, stderr
 from textwrap import dedent
+from typing import Callable, Iterable, List, Mapping, NoReturn, Optional, Tuple, cast
 
 from twisted.copyright import version
-from twisted.python.usage import Options, UsageError
+from twisted.internet.interfaces import IReactorCore
 from twisted.logger import (
     LogLevel,
     InvalidLogLevelError,
     textFileLogObserver,
     jsonFileLogObserver,
 )
 from twisted.plugin import getPlugins
+from twisted.python.usage import Options, UsageError
 
 from ..reactors import installReactor, NoSuchReactor, getReactorTypes
 from ..runner._exit import exit, ExitStatus
@@ -34,23 +36,23 @@ class TwistOptions(Options):
     defaultReactorName = "default"
     defaultLogLevel = LogLevel.info
 
-    def __init__(self):
+    def __init__(self) -> None:
         Options.__init__(self)
 
         self["reactorName"] = self.defaultReactorName
         self["logLevel"] = self.defaultLogLevel
         self["logFile"] = stdout
 
-    def getSynopsis(self):
+    def getSynopsis(self) -> str:
         return "{} plugin [plugin_options]".format(Options.getSynopsis(self))
 
-    def opt_version(self):
+    def opt_version(self) -> NoReturn:
         """
         Print version and exit.
         """
         exit(ExitStatus.EX_OK, "{}".format(version))
 
-    def opt_reactor(self, name):
+    def opt_reactor(self, name: str) -> None:
         """
         The name of the reactor to use.
         (options: {options})
@@ -69,18 +71,18 @@ def opt_reactor(self, name):
         options=", ".join('"{}"'.format(rt.shortName) for rt in getReactorTypes()),
     )
 
-    def installReactor(self, name):
+    def installReactor(self, name: str) -> IReactorCore:
         """
         Install the reactor.
         """
         if name == self.defaultReactorName:
             from twisted.internet import reactor
 
-            return reactor
+            return cast(IReactorCore, reactor)
         else:
-            return installReactor(name)
+            return cast(IReactorCore, installReactor(name))
 
-    def opt_log_level(self, levelName):
+    def opt_log_level(self, levelName: str) -> None:
         """
         Set default log level.
         (options: {options}; default: "{default}")
@@ -97,7 +99,7 @@ def opt_log_level(self, levelName):
         default=defaultLogLevel.name,
     )
 
-    def opt_log_file(self, fileName):
+    def opt_log_file(self, fileName: str) -> None:
         """
         Log to file. ("-" for stdout, "+" for stderr; default: "-")
         """
@@ -117,7 +119,7 @@ def opt_log_file(self, fileName):
                 "Unable to open log file {!r}: {}".format(fileName, e),
             )
 
-    def opt_log_format(self, format):
+    def opt_log_format(self, format: str) -> None:
         """
         Log file format.
         (options: "text", "json"; default: "text" if the log file is a tty,
@@ -135,7 +137,7 @@ def opt_log_format(self, format):
 
     opt_log_format.__doc__ = dedent(opt_log_format.__doc__ or "")
 
-    def selectDefaultLogObserver(self):
+    def selectDefaultLogObserver(self) -> None:
         """
         Set C{fileLogObserverFactory} to the default appropriate for the
         chosen C{logFile}.
@@ -150,7 +152,7 @@ def selectDefaultLogObserver(self):
                 self["fileLogObserverFactory"] = jsonFileLogObserver
                 self["logFormat"] = "json"
 
-    def parseOptions(self, options=None):
+    def parseOptions(self, options: Optional[List[str]] = None) -> None:
         self.selectDefaultLogObserver()
 
         Options.parseOptions(self, options=options)
@@ -159,31 +161,32 @@ def parseOptions(self, options=None):
             self["reactor"] = self.installReactor(self["reactorName"])
 
     @property
-    def plugins(self):
+    def plugins(self) -> Mapping[str, IServiceMaker]:
         if "plugins" not in self:
             plugins = {}
             for plugin in getPlugins(IServiceMaker):
                 plugins[plugin.tapname] = plugin
             self["plugins"] = plugins
 
-        return self["plugins"]
+        return cast(Mapping[str, IServiceMaker], self["plugins"])
 
     @property
-    def subCommands(self):
+    def subCommands(
+        self,
+    ) -> Iterable[Tuple[str, None, Callable[[IServiceMaker], Options], str]]:
         plugins = self.plugins
         for name in sorted(plugins):
             plugin = plugins[name]
-            yield (
-                plugin.tapname,
-                None,
-                # Avoid resolving the options attribute right away, in case
-                # it's a property with a non-trivial getter (eg, one which
-                # imports modules).
-                lambda plugin=plugin: plugin.options(),
-                plugin.description,
-            )
 
-    def postOptions(self):
+            # Don't pass plugin.options along in order to avoid resolving the
+            # options attribute right away, in case it's a property with a
+            # non-trivial getter (eg, one which imports modules).
+            def options(plugin: IServiceMaker) -> Options:
+                return cast(Options, plugin.options())
+
+            yield (plugin.tapname, None, options, plugin.description)
+
+    def postOptions(self) -> None:
         Options.postOptions(self)
 
         if self.subCommand is None: