Add type annotations

tests/orchestra/test_connection.py

@@ -54,7 +54,6 @@ def test_connect(self):
         m_ssh_instance.get_transport.return_value = m_transport
         got = connection.connect(
             'jdoe@orchestra.test.newdream.net.invalid',
-            _SSHClient=self.m_ssh,
         )
         self.m_ssh.assert_called_once()
         m_ssh_instance.set_missing_host_key_policy.assert_called_once()
@@ -75,7 +74,6 @@ def test_connect_no_verify_host_keys(self):
         m_ssh_instance.get_transport.return_value = m_transport
         got = connection.connect(
             'jdoe@orchestra.test.newdream.net.invalid',
-            _SSHClient=self.m_ssh,
         )
         self.m_ssh.assert_called_once()
         m_ssh_instance.set_missing_host_key_policy.assert_called_once()
@@ -88,20 +86,18 @@ def test_connect_no_verify_host_keys(self):
         m_transport.set_keepalive.assert_called_once_with(False)
         assert got is m_ssh_instance
 
-    def test_connect_override_hostkeys(self):
+    @patch('teuthology.orchestra.connection.create_key')
+    def test_connect_override_hostkeys(self, m_create_key):
         self.clear_config()
         m_ssh_instance = self.m_ssh.return_value = Mock();
         m_transport = Mock()
         m_ssh_instance.get_transport.return_value = m_transport
         m_host_keys = Mock()
         m_ssh_instance.get_host_keys.return_value = m_host_keys
-        m_create_key = Mock()
         m_create_key.return_value = "frobnitz"
         got = connection.connect(
             'jdoe@orchestra.test.newdream.net.invalid',
             host_key='ssh-rsa testkey',
-            _SSHClient=self.m_ssh,
-            _create_key=m_create_key,
             )
         self.m_ssh.assert_called_once()
         m_ssh_instance.get_host_keys.assert_called_once()

tests/orchestra/test_console.py

@@ -155,7 +155,7 @@ def test_get_console_conserver(self):
             cons = self.klass(self.hostname)
         assert cons.has_conserver is True
         with patch(
-            'teuthology.orchestra.console.pexpect.spawn',
+            'teuthology.orchestra.console.LoggedPexpect',
             autospec=True,
         ) as m_spawn:
             cons._get_console()
@@ -174,7 +174,7 @@ def test_get_console_ipmitool(self):
             cons = self.klass(self.hostname)
         assert cons.has_conserver is True
         with patch(
-            'teuthology.orchestra.console.pexpect.spawn',
+            'teuthology.orchestra.console.LoggedPexpect',
             autospec=True,
         ) as m_spawn:
             cons.has_conserver = False
@@ -193,7 +193,7 @@ def test_get_console_fallback(self):
             cons = self.klass(self.hostname)
         assert cons.has_conserver is True
         with patch(
-            'teuthology.orchestra.console.pexpect.spawn',
+            'teuthology.orchestra.console.LoggedPexpect',
             autospec=True,
         ) as m_spawn:
             cons.has_conserver = True

tests/orchestra/test_remote.py

@@ -151,7 +151,7 @@ def test_host_key(self):
         m_transport.get_remote_server_key.assert_called_once_with()
 
     def test_inventory_info(self):
-        r = remote.Remote('user@host', host_key='host_key')
+        r = remote.Remote('user@host', host_key='host_key', ssh=self.m_ssh)
         r._arch = 'arch'
         r._os = opsys.OS(name='os_name', version='1.2.3', codename='code')
         inv_info = r.inventory_info
@@ -233,7 +233,7 @@ def test_resolve_ip(self, m_sh):
         except Exception as e:
             assert 'Cannot get IPv4 address' in str(e)
         try:
-            ip4 = remote.Remote.resolve_ip(r, 'smithi001', 5)
+            ip4 = remote.Remote.resolve_ip(r, 'smithi001', "5")
         except Exception as e:
             assert 'Unknown IP version' in str(e)
 

tests/provision/cloud/test_openstack.py

@@ -2,7 +2,7 @@
 import yaml
 import os
 
-from teuthology.util.compat import parse_qs
+from urllib.parse import parse_qs
 
 from copy import deepcopy
 from libcloud.compute.providers import get_driver

tests/provision/test_downburst.py

@@ -18,68 +18,62 @@ def setup_method(self):
             description="desc",
         )
 
-    def test_create_if_vm_success(self):
+    @patch('teuthology.lock.query.get_status')
+    @patch('teuthology.provision.downburst.Downburst')
+    def test_create_if_vm_success(self, m_downburst, m_get_status):
         name = self.name
         ctx = self.ctx
         status = self.status
+        m_get_status.return_value = status
 
-        dbrst = provision.downburst.Downburst(
-            name, ctx.os_type, ctx.os_version, status)
+        dbrst = MagicMock()
         dbrst.executable = '/fake/path'
         dbrst.build_config = MagicMock(name='build_config')
         dbrst._run_create = MagicMock(name='_run_create')
         dbrst._run_create.return_value = (0, '', '')
+        dbrst.create.return_value = True
         remove_config = MagicMock(name='remove_config')
         dbrst.remove_config = remove_config
+        m_downburst.return_value = dbrst
 
-        result = provision.create_if_vm(ctx, name, dbrst)
+        result = provision.create_if_vm(ctx, name)
         assert result is True
 
-        dbrst._run_create.assert_called_with()
-        dbrst.build_config.assert_called_with()
-        del dbrst
-        remove_config.assert_called_with()
+        dbrst.create.assert_called_with()
 
-    def test_destroy_if_vm_success(self):
+    @patch('teuthology.lock.query.get_status')
+    @patch('teuthology.provision.downburst.Downburst')
+    def test_destroy_if_vm_success(self, m_downburst, m_get_status):
         name = self.name
-        ctx = self.ctx
         status = self.status
+        m_get_status.return_value = status
 
-        dbrst = provision.downburst.Downburst(
-            name, ctx.os_type, ctx.os_version, status)
+        dbrst = MagicMock()
         dbrst.destroy = MagicMock(name='destroy')
         dbrst.destroy.return_value = True
+        m_downburst.return_value = dbrst
 
-        result = provision.destroy_if_vm(name, user="user@a", _downburst=dbrst)
+        result = provision.destroy_if_vm(name, user="user@a")
         assert result is True
 
         dbrst.destroy.assert_called_with()
 
-    def test_destroy_if_vm_wrong_owner(self):
+    @patch('teuthology.lock.query.get_status')
+    def test_destroy_if_vm_wrong_owner(self, m_get_status):
         name = self.name
-        ctx = self.ctx
         status = self.status
+        m_get_status.return_value = status
 
-        dbrst = provision.downburst.Downburst(
-            name, ctx.os_type, ctx.os_version, status)
-        dbrst.destroy = MagicMock(name='destroy', side_effect=RuntimeError)
-
-        result = provision.destroy_if_vm(name, user='user@b',
-                                         _downburst=dbrst)
+        result = provision.destroy_if_vm(name, user='user@b')
         assert result is False
 
-    def test_destroy_if_vm_wrong_description(self):
+    @patch('teuthology.lock.query.get_status')
+    def test_destroy_if_vm_wrong_description(self, m_get_status):
         name = self.name
-        ctx = self.ctx
         status = self.status
+        m_get_status.return_value = status
 
-        dbrst = provision.downburst.Downburst(
-            name, ctx.os_type, ctx.os_version, status)
-        dbrst.destroy = MagicMock(name='destroy')
-        dbrst.destroy = MagicMock(name='destroy', side_effect=RuntimeError)
-
-        result = provision.destroy_if_vm(name, description='desc_b',
-                                         _downburst=dbrst)
+        result = provision.destroy_if_vm(name, description='desc_b')
         assert result is False
 
     @patch('teuthology.provision.downburst.downburst_executable')

tests/test_results.py

@@ -1,9 +1,8 @@
 import textwrap
 from teuthology.config import config
 from teuthology import results
-from teuthology import report
 
-from unittest.mock import patch, DEFAULT
+from unittest.mock import patch
 
 
 class TestResultsEmail(object):
@@ -141,15 +140,11 @@ def setup_method(self):
         config.results_ui_server = "http://example.com/"
         config.archive_server = "http://qa-proxy.ceph.com/teuthology/"
 
-    def test_build_email_body(self):
+    @patch('teuthology.results.ResultsReporter')
+    def test_build_email_body(self, m_reporter):
         run_name = self.reference['run_name']
-        with patch.multiple(
-                report,
-                ResultsReporter=DEFAULT,
-                ):
-            reporter = report.ResultsReporter()
-            reporter.get_jobs.return_value = self.reference['jobs']
-            (subject, body) = results.build_email_body(
-                run_name, _reporter=reporter)
+        reporter = m_reporter.return_value
+        reporter.get_jobs.return_value = self.reference['jobs']
+        (subject, body) = results.build_email_body(run_name)
         assert subject == self.reference['subject']
         assert body == self.reference['body']

tests/test_timer.py

@@ -56,7 +56,7 @@ def test_write(self):
                 self.timer = timer.Timer(path=_path)
                 assert self.timer.path == _path
                 self.timer.write()
-                _open.assert_called_once_with(_path, 'w')
+                _open.assert_called_once_with(_path, mode='w')
                 _safe_dump.assert_called_once_with(
                     dict(),
                     _open.return_value.__enter__.return_value,
@@ -72,7 +72,7 @@ def test_sync(self):
                 assert self.timer.path == _path
                 assert self.timer.sync is True
                 self.timer.mark()
-                _open.assert_called_once_with(_path, 'w')
+                _open.assert_called_once_with(_path, mode='w')
                 _safe_dump.assert_called_once_with(
                     self.timer.data,
                     _open.return_value.__enter__.return_value,

teuthology/__init__.py

@@ -75,7 +75,7 @@
 log.debug('teuthology version: %s', __version__)
 
 
-def setup_log_file(log_path):
+def setup_log_file(log_path: str) -> None:
     root_logger = logging.getLogger()
     handlers = root_logger.handlers
     for handler in handlers:
@@ -93,23 +93,23 @@ def setup_log_file(log_path):
     root_logger.info('teuthology version: %s', __version__)
 
 
-def install_except_hook():
+def install_except_hook() -> None:
     """
     Install an exception hook that first logs any uncaught exception, then
     raises it.
     """
-    def log_exception(exc_type, exc_value, exc_traceback):
+    def log_exception(exc_type, exc_value, exc_traceback) -> None:
         if not issubclass(exc_type, KeyboardInterrupt):
             log.critical("Uncaught exception", exc_info=(exc_type, exc_value,
                                                          exc_traceback))
         sys.__excepthook__(exc_type, exc_value, exc_traceback)
     sys.excepthook = log_exception
 
 
-def patch_gevent_hub_error_handler():
+def patch_gevent_hub_error_handler() -> None:
     Hub._origin_handle_error = Hub.handle_error
 
-    def custom_handle_error(self, context, type, value, tb):
+    def custom_handle_error(self, context, type, value, tb) -> None:
         if context is None or issubclass(type, Hub.SYSTEM_ERROR):
             self.handle_system_error(type, value)
         elif issubclass(type, Hub.NOT_ERROR):