mypy: cinder/cmd/[api,backup,scheduler,status,volume]

Add type annotations to these cmd files.

Change-Id: I5205d706a5d05e33930353cd5f324980fcb6308a

cinder/cmd/api.py

@@ -27,7 +27,8 @@ eventlet.monkey_patch()
 # https://github.com/eventlet/eventlet/issues/592
 import __original_module_threading as orig_threading  # pylint: disable=E0401
 import threading # noqa
-orig_threading.current_thread.__globals__['_active'] = threading._active
+orig_threading.current_thread.__globals__['_active'] = \
+    threading._active  # type: ignore
 
 from oslo_config import cfg
 from oslo_log import log as logging
@@ -48,7 +49,7 @@ from cinder import version
 CONF = cfg.CONF
 
 
-def main():
+def main() -> None:
     objects.register_all()
     gmr_opts.set_defaults(CONF)
     CONF(sys.argv[1:], project='cinder',

cinder/cmd/backup.py

@@ -32,7 +32,10 @@ eventlet.monkey_patch()
 # https://github.com/eventlet/eventlet/issues/592
 import __original_module_threading as orig_threading  # pylint: disable=E0401
 import threading # noqa
-orig_threading.current_thread.__globals__['_active'] = threading._active
+orig_threading.current_thread.__globals__['_active'] = \
+    threading._active  # type: ignore
+import typing
+from typing import Union
 
 from oslo_concurrency import processutils
 from oslo_config import cfg
@@ -40,6 +43,8 @@ from oslo_log import log as logging
 from oslo_privsep import priv_context
 from oslo_reports import guru_meditation_report as gmr
 from oslo_reports import opts as gmr_opts
+if typing.TYPE_CHECKING:
+    import oslo_service
 
 # Need to register global_opts
 from cinder.common import config  # noqa
@@ -77,7 +82,10 @@ LOG = None
 _EXTRA_DEFAULT_LOG_LEVELS = ['swiftclient=WARN', 'botocore=WARN']
 
 
-def _launch_backup_process(launcher, num_process, _semaphore):
+def _launch_backup_process(launcher: 'oslo_service.ProcessLauncher',
+                           num_process: int,
+                           _semaphore: Union[eventlet.semaphore.Semaphore,
+                                             utils.Semaphore]) -> None:
     try:
         server = service.Service.create(binary='cinder-backup',
                                         coordination=True,
@@ -85,6 +93,7 @@ def _launch_backup_process(launcher, num_process, _semaphore):
                                         process_number=num_process + 1,
                                         semaphore=_semaphore)
     except Exception:
+        assert LOG is not None
         LOG.exception('Backup service %s failed to start.', CONF.host)
         sys.exit(1)
     else:
@@ -95,7 +104,7 @@ def _launch_backup_process(launcher, num_process, _semaphore):
         launcher.launch_service(server)
 
 
-def main():
+def main() -> None:
     objects.register_all()
     gmr_opts.set_defaults(CONF)
     CONF(sys.argv[1:], project='cinder',

cinder/cmd/scheduler.py

@@ -27,7 +27,8 @@ eventlet.monkey_patch()
 # https://github.com/eventlet/eventlet/issues/592
 import __original_module_threading as orig_threading  # pylint: disable=E0401
 import threading # noqa
-orig_threading.current_thread.__globals__['_active'] = threading._active
+orig_threading.current_thread.__globals__['_active'] = \
+    threading._active  # type: ignore
 
 from oslo_config import cfg
 from oslo_log import log as logging
@@ -47,7 +48,7 @@ from cinder import version
 CONF = cfg.CONF
 
 
-def main():
+def main() -> None:
     objects.register_all()
     gmr_opts.set_defaults(CONF)
     CONF(sys.argv[1:], project='cinder',

cinder/cmd/status.py

@@ -15,6 +15,8 @@
 
 """CLI interface for cinder status commands."""
 
+from __future__ import annotations
+
 import os
 import sys
 
@@ -49,10 +51,11 @@ REMOVED_DRVRS = [
 ]
 
 
-def _get_enabled_drivers():
+def _get_enabled_drivers() -> list[str]:
     """Returns a list of volume_driver entries"""
     volume_drivers = []
     if CONF.enabled_backends:
+        backend: str
         for backend in filter(None, CONF.enabled_backends):
             # Each backend group needs to be registered first
             CONF.register_opts(volume_manager.volume_backend_opts,
@@ -70,11 +73,11 @@ class Checks(uc.UpgradeCommands):
         super(Checks, self).__init__(*args, **kwargs)
         self.context = context.get_admin_context()
 
-    def _file_exists(self, path):
+    def _file_exists(self, path: str) -> bool:
         """Helper for mocking check of os.path.exists."""
         return os.path.exists(path)
 
-    def _check_backup_module(self):
+    def _check_backup_module(self) -> uc.Result:
         """Checks for the use of backup driver module paths.
 
         The use of backup modules for setting backup_driver was deprecated and
@@ -96,7 +99,7 @@ class Checks(uc.UpgradeCommands):
 
         return uc.Result(SUCCESS)
 
-    def _check_policy_file(self):
+    def _check_policy_file(self) -> uc.Result:
         """Checks if a policy.json file is present.
 
         With the switch to policy-in-code, policy files should be policy.yaml
@@ -145,7 +148,7 @@ class Checks(uc.UpgradeCommands):
 
         return uc.Result(SUCCESS)
 
-    def _check_periodic_interval(self):
+    def _check_periodic_interval(self) -> uc.Result:
         """Checks for non-default use of periodic_interval.
 
         Some new configuration options have been introduced to supplement
@@ -167,7 +170,7 @@ class Checks(uc.UpgradeCommands):
 
         return uc.Result(SUCCESS)
 
-    def _check_nested_quota(self):
+    def _check_nested_quota(self) -> uc.Result:
         """Checks for the use of the nested quota driver.
 
         The NestedDbQuotaDriver is deprecated in the Train release and is
@@ -185,7 +188,7 @@ class Checks(uc.UpgradeCommands):
                 'and is removed in Wallaby release.')
         return uc.Result(SUCCESS)
 
-    def _check_legacy_windows_config(self):
+    def _check_legacy_windows_config(self) -> uc.Result:
         """Checks to ensure that the Windows driver path is properly updated.
 
         The WindowsDriver was renamed in the Queens release to
@@ -209,7 +212,7 @@ class Checks(uc.UpgradeCommands):
 
         return uc.Result(SUCCESS)
 
-    def _check_removed_drivers(self):
+    def _check_removed_drivers(self) -> uc.Result:
         """Checks to ensure that no removed drivers are configured.
 
         Checks start with drivers removed in the Stein release.
@@ -239,7 +242,7 @@ class Checks(uc.UpgradeCommands):
 
         return uc.Result(SUCCESS)
 
-    def _check_service_uuid(self):
+    def _check_service_uuid(self) -> uc.Result:
         try:
             db.service_get_by_uuid(self.context, None)
         except exception.ServiceNotFound:

cinder/cmd/volume.py

@@ -38,13 +38,17 @@ else:
 # https://github.com/eventlet/eventlet/issues/592
 import __original_module_threading as orig_threading  # pylint: disable=E0401
 import threading # noqa
-orig_threading.current_thread.__globals__['_active'] = threading._active
+orig_threading.current_thread.__globals__['_active'] = \
+    threading._active  # type: ignore
+import typing
 
 from oslo_config import cfg
 from oslo_log import log as logging
 from oslo_privsep import priv_context
 from oslo_reports import guru_meditation_report as gmr
 from oslo_reports import opts as gmr_opts
+if typing.TYPE_CHECKING:
+    import oslo_service
 
 # Need to register global_opts
 from cinder.common import config  # noqa
@@ -87,7 +91,8 @@ LOG = None
 service_started = False
 
 
-def _launch_service(launcher, backend):
+def _launch_service(launcher: 'oslo_service.ProcessLauncher',
+                    backend: str) -> None:
     CONF.register_opt(host_opt, group=backend)
     backend_host = getattr(CONF, backend).backend_host
     host = "%s@%s" % (backend_host or CONF.host, backend)
@@ -102,6 +107,7 @@ def _launch_service(launcher, backend):
                                         coordination=True,
                                         cluster=cluster)
     except Exception:
+        assert LOG is not None
         LOG.exception('Volume service %s failed to start.', host)
     else:
         # Dispose of the whole DB connection pool here before
@@ -112,18 +118,19 @@ def _launch_service(launcher, backend):
         _notify_service_started()
 
 
-def _ensure_service_started():
+def _ensure_service_started() -> None:
     if not service_started:
+        assert LOG is not None
         LOG.error('No volume service(s) started successfully, terminating.')
         sys.exit(1)
 
 
-def _notify_service_started():
+def _notify_service_started() -> None:
     global service_started
     service_started = True
 
 
-def _launch_services_win32():
+def _launch_services_win32() -> None:
     if CONF.backend_name and CONF.backend_name not in CONF.enabled_backends:
         msg = _('The explicitly passed backend name "%(backend_name)s" is not '
                 'among the enabled backends: %(enabled_backends)s.')
@@ -143,6 +150,7 @@ def _launch_services_win32():
         # and constructing the service object within the child process.
         launcher = service.WindowsProcessLauncher()
         py_script_re = re.compile(r'.*\.py\w?$')
+        backend: str
         for backend in filter(None, CONF.enabled_backends):
             cmd = sys.argv + ['--backend_name=%s' % backend]
             # Recent setuptools versions will trim '-script.py' and '.exe'
@@ -157,9 +165,10 @@ def _launch_services_win32():
     launcher.wait()
 
 
-def _launch_services_posix():
+def _launch_services_posix() -> None:
     launcher = service.get_launcher()
 
+    backend: str
     for backend in filter(None, CONF.enabled_backends):
         _launch_service(launcher, backend)
 
@@ -168,7 +177,7 @@ def _launch_services_posix():
     launcher.wait()
 
 
-def main():
+def main() -> None:
     objects.register_all()
     gmr_opts.set_defaults(CONF)
     CONF(sys.argv[1:], project='cinder',

mypy-files.txt

@@ -7,6 +7,11 @@ cinder/backup/drivers/ceph.py
 cinder/common/constants.py
 cinder/context.py
 cinder/coordination.py
+cinder/cmd/api.py
+cinder/cmd/backup.py
+cinder/cmd/scheduler.py
+cinder/cmd/status.py
+cinder/cmd/volume.py
 cinder/i18n.py
 cinder/image/cache.py
 cinder/image/glance.py