MyPy 0.650 - new checks

(cherry picked from commit 6077362)

exec_helpers/_ssh_client_base.py

@@ -496,7 +496,7 @@ def reconnect(self):  # type: () -> None
 
             self.__connect()
 
-    def sudo(self, enforce=None):  # type: (typing.Optional[bool]) -> typing.ContextManager
+    def sudo(self, enforce=None):  # type: (typing.Optional[bool]) -> typing.ContextManager[None]
         """Call contextmanager for sudo mode change.
 
         :param enforce: Enforce sudo enabled or disabled. By default: None
@@ -506,7 +506,7 @@ def sudo(self, enforce=None):  # type: (typing.Optional[bool]) -> typing.Context
         """
         return self.__get_sudo(ssh=self, enforce=enforce)
 
-    def keepalive(self, enforce=True):  # type: (bool) -> typing.ContextManager
+    def keepalive(self, enforce=True):  # type: (bool) -> typing.ContextManager[None]
         """Call contextmanager with keepalive mode change.
 
         :param enforce: Enforce keepalive enabled or disabled.
@@ -665,7 +665,7 @@ def poll_pipes():  # type: () -> None
 
         # pylint: disable=assignment-from-no-return
         # noinspection PyNoneFunctionAssignment
-        future = poll_pipes()  # type: concurrent.futures.Future
+        future = poll_pipes()
         # pylint: enable=assignment-from-no-return
 
         concurrent.futures.wait([future], timeout)
@@ -841,14 +841,12 @@ def get_result(remote):  # type: (SSHClientBase) -> exec_result.ExecResult
         errors = {}
         raised_exceptions = {}
 
-        (_, not_done) = concurrent.futures.wait(
-            list(futures.values()), timeout=timeout
-        )  # type: typing.Set[concurrent.futures.Future], typing.Set[concurrent.futures.Future]
+        _, not_done = concurrent.futures.wait(list(futures.values()), timeout=timeout)
 
         for fut in not_done:  # pragma: no cover
             fut.cancel()
 
-        for (remote, future) in futures.items():  # type: SSHClientBase, concurrent.futures.Future
+        for (remote, future) in futures.items():
             try:
                 result = future.result()
                 results[(remote.hostname, remote.port)] = result

exec_helpers/exec_result.py

@@ -227,7 +227,7 @@ def _poll_stream(
 
     def read_stdout(
         self,
-        src=None,  # type: typing.Optional[typing.Iterable]
+        src=None,  # type: typing.Optional[typing.Iterable[bytes]]
         log=None,  # type: typing.Optional[logging.Logger]
         verbose=False,  # type: bool
     ):  # type: (...) -> None
@@ -254,7 +254,7 @@ def read_stdout(
 
     def read_stderr(
         self,
-        src=None,  # type: typing.Optional[typing.Iterable]
+        src=None,  # type: typing.Optional[typing.Iterable[bytes]]
         log=None,  # type: typing.Optional[logging.Logger]
         verbose=False,  # type: bool
     ):  # type: (...) -> None

exec_helpers/metaclasses.py

@@ -33,7 +33,7 @@ class SingletonMeta(abc.ABCMeta):
     Main goals: not need to implement __new__ in singleton classes
     """
 
-    _instances = {}  # type: typing.Dict[typing.Type, typing.Any]
+    _instances = {}  # type: typing.Dict[type, typing.Any]
     _lock = threading.RLock()  # type: threading.RLock
 
     def __call__(cls, *args, **kwargs):  # type: (SingletonMeta, typing.Any, typing.Any) -> typing.Any
@@ -57,7 +57,7 @@ def __init__(
 
     def __new__(  # pylint: disable=arguments-differ
         mcs, name, bases, namespace, **kwargs
-    ):  # type: (str, typing.Tuple[type, ...], typing.Dict[str, typing.Any], typing.Any) -> typing.Type
+    ):  # type: (str, typing.Tuple[type, ...], typing.Dict[str, typing.Any], typing.Any) -> "SingleLock"
         """Create lock property for class instances."""
         namespace["lock"] = property(fget=lambda self: self.__class__.lock)
         return super(SingleLock, mcs).__new__(mcs, name, bases, namespace, **kwargs)  # type: ignore

exec_helpers/ssh_auth.py

@@ -117,7 +117,7 @@ def key_filename(self):  # type: () -> typing.Union[typing.List[str], str, None]
         """
         return copy.deepcopy(self.__key_filename)
 
-    def enter_password(self, tgt):  # type: (typing.IO) -> None
+    def enter_password(self, tgt):  # type: (typing.BinaryIO) -> None
         """Enter password to STDIN.
 
         Note: required for 'sudo' call
@@ -126,7 +126,7 @@ def enter_password(self, tgt):  # type: (typing.IO) -> None
         :type tgt: file
         """
         # noinspection PyTypeChecker
-        tgt.write("{}\n".format(self.__password))
+        tgt.write("{}\n".format(self.__password if self.__password is not None else "").encode("utf-8"))
 
     def connect(
         self,

exec_helpers/subprocess_runner.py

@@ -127,9 +127,9 @@ def close_streams():  # type: () -> None
 
         # pylint: disable=assignment-from-no-return
         # noinspection PyNoneFunctionAssignment
-        stdout_future = poll_stdout()  # type: concurrent.futures.Future
+        stdout_future = poll_stdout()
         # noinspection PyNoneFunctionAssignment
-        stderr_future = poll_stderr()  # type: concurrent.futures.Future
+        stderr_future = poll_stderr()
         # pylint: enable=assignment-from-no-return
 
         concurrent.futures.wait([stdout_future, stderr_future], timeout=timeout)  # Wait real timeout here

test/test_sshauth.py

@@ -106,9 +106,9 @@ def test_001_init_checks(run_parameters):
     int_keys = get_internal_keys(**run_parameters)
 
     assert auth.username == username
-    with contextlib.closing(io.StringIO()) as tgt:
+    with contextlib.closing(io.BytesIO()) as tgt:
         auth.enter_password(tgt)
-        assert tgt.getvalue() == "{}\n".format(run_parameters.get("password", None))
+        assert tgt.getvalue() == "{}\n".format(run_parameters.get("password", "")).encode("utf-8")
 
     key = run_parameters.get("key", None)
     if key is not None:

tox.ini

@@ -140,6 +140,6 @@ commands = pipdeptree
 [testenv:mypy]
 usedevelop = False
 deps =
-    mypy>=0.630
+    mypy>=0.650
     -r{toxinidir}/CI_REQUIREMENTS.txt
 commands = mypy --strict exec_helpers