Remaining typing tidyup

dev-requirements.txt

@@ -20,5 +20,5 @@ setuptools>56
 # Debuggery
 icecream>=2.1
 # typing
-mypy==0.971
+mypy==1.0.0
 types-PyYAML==6.0.12.4

integration/_explicit.py

@@ -2,7 +2,7 @@
 
 
 @task
-def foo(c):
+def foo(c: object) -> None:
     """
     Frobazz
     """

integration/_support/nested_or_piped.py

@@ -1,11 +1,11 @@
-from invoke import task
+from invoke import Config, task
 
 
 @task
-def calls_foo(c):
+def calls_foo(c: Config) -> None:
     c.run("inv -c nested_or_piped foo")
 
 
 @task
-def foo(c):
+def foo(c: Config) -> None:
     c.run("echo meh")

integration/_support/parsing.py

@@ -2,5 +2,5 @@
 
 
 @task(optional=["meh"])
-def foo(c, meh=False):
+def foo(c: object, meh: bool = False) -> None:
     print(meh)

integration/_support/regression.py

@@ -12,11 +12,11 @@
 
 import sys
 
-from invoke import task
+from invoke import Config, task
 
 
 @task
-def check(c):
+def check(c: Config) -> None:
     count = 0
     failures = []
     for _ in range(0, 1000):

integration/_support/tasks.py

@@ -2,19 +2,19 @@
 Tasks module for use within the integration tests.
 """
 
-from invoke import task
+from invoke import Config, task
 
 
 @task
-def print_foo(c):
+def print_foo(c: object) -> None:
     print("foo")
 
 
 @task
-def print_name(c, name):
+def print_name(c: object, name: object) -> None:
     print(name)
 
 
 @task
-def print_config(c):
+def print_config(c: Config) -> None:
     print(c.foo)

integration/_util.py

@@ -1,20 +1,21 @@
 from contextlib import contextmanager
 from functools import wraps
 from resource import getrusage, RUSAGE_SELF
+from typing import Any, Callable, Iterator, Optional, TypeVar
 import sys
 import time
 
-
 from pytest import skip
 
+_T = TypeVar("_T")
 
-def current_cpu_usage():
+def current_cpu_usage() -> float:
     rusage = getrusage(RUSAGE_SELF)
     return rusage.ru_utime + rusage.ru_stime
 
 
 @contextmanager
-def assert_cpu_usage(lt, verbose=False):
+def assert_cpu_usage(lt: float, verbose: bool = False) -> Iterator[None]:
     """
     Execute wrapped block, asserting CPU utilization was less than ``lt``%.
 
@@ -41,13 +42,14 @@ def assert_cpu_usage(lt, verbose=False):
     assert percentage < lt
 
 
-def only_utf8(f):
+def only_utf8(f: Callable[..., _T]) -> Callable[..., Optional[_T]]:
     """
     Decorator causing tests to skip if local shell pipes aren't UTF-8.
     """
     # TODO: use actual test selection labels or whatever nose has
+    # TODO(PY310): Use ParamSpec.
     @wraps(f)
-    def inner(*args, **kwargs):
+    def inner(*args: Any, **kwargs: Any) -> _T:
         if getattr(sys.stdout, "encoding", None) == "UTF-8":
             return f(*args, **kwargs)
         # TODO: could remove this so they show green, but figure yellow is more

integration/context.py

@@ -4,7 +4,7 @@
 
 class Context_:
     class sudo:
-        def base_case(self):
+        def base_case(self) -> None:
             c = Context()
             # Grab CI-oriented sudo user/pass direct from invocations.ci
             # TODO: might be nice to give Collection a way to get a Config
@@ -15,5 +15,9 @@ def base_case(self):
             # Safety 1: ensure configured user even exists
             assert c.run("id {}".format(user), warn=True)
             # Safety 2: make sure we ARE them (and not eg root already)
-            assert c.run("whoami", hide=True).stdout.strip() == user
-            assert c.sudo("whoami", hide=True).stdout.strip() == "root"
+            result = c.run("whoami", hide=True)
+            assert result is not None
+            assert result.stdout.strip() == user
+            result = c.sudo("whoami", hide=True)
+            assert result is not None
+            assert result.stdout.strip() == "root"

integration/main.py

@@ -9,59 +9,66 @@
 from invoke._version import __version__
 from invoke.terminals import WINDOWS
 
-from _util import only_utf8
+from ._util import only_utf8
 
 
-def _output_eq(cmd, expected):
-    assert run(cmd, hide=True).stdout == expected
+def _output_eq(cmd: str, expected: str) -> None:
+    r = run(cmd, hide=True)
+    assert r is not None
+    assert r.stdout == expected
 
 
 class Main:
-    def setup_method(self):
+    def setup_method(self) -> None:
         self.cwd = os.getcwd()
         # Enter integration/_support as all support files are in there now
         os.chdir(Path(__file__).parent / "_support")
 
-    def teardown_method(self):
+    def teardown_method(self) -> None:
         os.chdir(self.cwd)
 
     class basics:
         @trap
-        def basic_invocation(self):
+        def basic_invocation(self) -> None:
             _output_eq("invoke print-foo", "foo\n")
 
         @trap
-        def version_output(self):
+        def version_output(self) -> None:
             _output_eq("invoke --version", "Invoke {}\n".format(__version__))
 
         @trap
-        def help_output(self):
-            assert "Usage: inv[oke] " in run("invoke --help").stdout
+        def help_output(self) -> None:
+            r = run("invoke --help")
+            assert r is not None
+            assert "Usage: inv[oke] " in r.stdout
 
         @trap
-        def per_task_help(self):
-            assert "Frobazz" in run("invoke -c _explicit foo --help").stdout
+        def per_task_help(self) -> None:
+            r = run("invoke -c _explicit foo --help")
+            assert r is not None
+            assert "Frobazz" in r.stdout
 
         @trap
-        def shorthand_binary_name(self):
+        def shorthand_binary_name(self) -> None:
             _output_eq("inv print-foo", "foo\n")
 
         @trap
-        def explicit_task_module(self):
+        def explicit_task_module(self) -> None:
             _output_eq("inv --collection _explicit foo", "Yup\n")
 
         @trap
-        def invocation_with_args(self):
+        def invocation_with_args(self) -> None:
             _output_eq("inv print-name --name whatevs", "whatevs\n")
 
         @trap
-        def bad_collection_exits_nonzero(self):
+        def bad_collection_exits_nonzero(self) -> None:
             result = run("inv -c nope -l", warn=True)
+            assert result is not None
             assert result.exited == 1
             assert not result.stdout
             assert result.stderr
 
-        def loads_real_user_config(self):
+        def loads_real_user_config(self) -> None:
             path = os.path.expanduser("~/.invoke.yaml")
             try:
                 with open(path, "w") as fd:
@@ -74,75 +81,81 @@ def loads_real_user_config(self):
                     pass
 
         @trap
-        def invocable_via_python_dash_m(self):
+        def invocable_via_python_dash_m(self) -> None:
             _output_eq(
                 "python -m invoke print-name --name mainline", "mainline\n"
             )
 
     class funky_characters_in_stdout:
         @only_utf8
-        def basic_nonstandard_characters(self):
+        def basic_nonstandard_characters(self) -> None:
             # Crummy "doesn't explode with decode errors" test
             cmd = ("type" if WINDOWS else "cat") + " tree.out"
             run(cmd, hide="stderr")
 
         @only_utf8
-        def nonprinting_bytes(self):
+        def nonprinting_bytes(self) -> None:
             # Seriously non-printing characters (i.e. non UTF8) also don't
             # asplode (they would print as escapes normally, but still)
             run("echo '\xff'", hide="stderr")
 
         @only_utf8
-        def nonprinting_bytes_pty(self):
+        def nonprinting_bytes_pty(self) -> None:
             if WINDOWS:
                 return
             # PTY use adds another utf-8 decode spot which can also fail.
             run("echo '\xff'", pty=True, hide="stderr")
 
     class ptys:
-        def complex_nesting_under_ptys_doesnt_break(self):
+        def complex_nesting_under_ptys_doesnt_break(self) -> None:
             if WINDOWS:  # Not sure how to make this work on Windows
                 return
             # GH issue 191
             substr = "      hello\t\t\nworld with spaces"
             cmd = """ eval 'echo "{}" ' """.format(substr)
             expected = "      hello\t\t\r\nworld with spaces\r\n"
-            assert run(cmd, pty=True, hide="both").stdout == expected
+            r = run(cmd, pty=True, hide="both")
+            assert r is not None
+            assert r.stdout == expected
 
-        def pty_puts_both_streams_in_stdout(self):
+        def pty_puts_both_streams_in_stdout(self) -> None:
             if WINDOWS:
                 return
             err_echo = "{} err.py".format(sys.executable)
             command = "echo foo && {} bar".format(err_echo)
             r = run(command, hide="both", pty=True)
+            assert r is not None
             assert r.stdout == "foo\r\nbar\r\n"
             assert r.stderr == ""
 
-        def simple_command_with_pty(self):
+        def simple_command_with_pty(self) -> None:
             """
             Run command under PTY
             """
             # Most Unix systems should have stty, which asplodes when not run
             # under a pty, and prints useful info otherwise
             result = run("stty -a", hide=True, pty=True)
             # PTYs use \r\n, not \n, line separation
+            assert result is not None
             assert "\r\n" in result.stdout
             assert result.pty is True
 
         @pytest.mark.skip(reason="CircleCI env actually does have 0x0 stty")
-        def pty_size_is_realistic(self):
+        def pty_size_is_realistic(self) -> None:
             # When we don't explicitly set pty size, 'stty size' sees it as
             # 0x0.
             # When we do set it, it should be some non 0x0, non 80x24 (the
             # default) value. (yes, this means it fails if you really do have
             # an 80x24 terminal. but who does that?)
-            size = run("stty size", hide=True, pty=True).stdout.strip()
+            r = run("stty size", hide=True, pty=True)
+            assert r is not None
+            size = r.stdout.strip()
             assert size != ""
             assert size != "0 0"
             assert size != "24 80"
 
     class parsing:
-        def false_as_optional_arg_default_value_works_okay(self):
+        def false_as_optional_arg_default_value_works_okay(self) -> None:
             # (Dis)proves #416. When bug present, parser gets very confused,
             # asks "what the hell is 'whee'?". See also a unit test for
             # Task.get_arguments.