Add typing for scrapy/utils/testproc.py.

scrapy/utils/testproc.py

@@ -1,7 +1,13 @@
+from __future__ import annotations
+
 import os
 import sys
+from typing import Iterable, Optional, Tuple, cast
 
-from twisted.internet import defer, protocol
+from twisted.internet.defer import Deferred
+from twisted.internet.error import ProcessTerminated
+from twisted.internet.protocol import ProcessProtocol
+from twisted.python.failure import Failure
 
 
 class ProcessTest:
@@ -9,7 +15,12 @@ class ProcessTest:
     prefix = [sys.executable, "-m", "scrapy.cmdline"]
     cwd = os.getcwd()  # trial chdirs to temp dir
 
-    def execute(self, args, check_code=True, settings=None):
+    def execute(
+        self,
+        args: Iterable[str],
+        check_code: bool = True,
+        settings: Optional[str] = None,
+    ) -> Deferred:
         from twisted.internet import reactor
 
         env = os.environ.copy()
@@ -21,29 +32,31 @@ class ProcessTest:
         reactor.spawnProcess(pp, cmd[0], cmd, env=env, path=self.cwd)
         return pp.deferred
 
-    def _process_finished(self, pp, cmd, check_code):
+    def _process_finished(
+        self, pp: TestProcessProtocol, cmd: str, check_code: bool
+    ) -> Tuple[int, bytes, bytes]:
         if pp.exitcode and check_code:
             msg = f"process {cmd} exit with code {pp.exitcode}"
-            msg += f"\n>>> stdout <<<\n{pp.out}"
+            msg += f"\n>>> stdout <<<\n{pp.out.decode()}"
             msg += "\n"
-            msg += f"\n>>> stderr <<<\n{pp.err}"
+            msg += f"\n>>> stderr <<<\n{pp.err.decode()}"
             raise RuntimeError(msg)
-        return pp.exitcode, pp.out, pp.err
+        return cast(int, pp.exitcode), pp.out, pp.err
 
 
-class TestProcessProtocol(protocol.ProcessProtocol):
-    def __init__(self):
-        self.deferred = defer.Deferred()
-        self.out = b""
-        self.err = b""
-        self.exitcode = None
+class TestProcessProtocol(ProcessProtocol):
+    def __init__(self) -> None:
+        self.deferred: Deferred = Deferred()
+        self.out: bytes = b""
+        self.err: bytes = b""
+        self.exitcode: Optional[int] = None
 
-    def outReceived(self, data):
+    def outReceived(self, data: bytes) -> None:
         self.out += data
 
-    def errReceived(self, data):
+    def errReceived(self, data: bytes) -> None:
         self.err += data
 
-    def processEnded(self, status):
-        self.exitcode = status.value.exitCode
+    def processEnded(self, status: Failure) -> None:
+        self.exitcode = cast(ProcessTerminated, status.value).exitCode
         self.deferred.callback(self)