Improve TeePopen typing and structure

tests/ci/tee_popen.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 
+from io import TextIOWrapper
 from subprocess import Popen, PIPE, STDOUT
 from threading import Thread
 from time import sleep
@@ -14,15 +15,23 @@ import sys
 # it finishes. stderr and stdout will be redirected both to specified file and
 # stdout.
 class TeePopen:
-    # pylint: disable=W0102
-    def __init__(self, command, log_file, env=os.environ.copy(), timeout=None):
+    def __init__(
+        self,
+        command: str,
+        log_file: str,
+        env: Optional[dict] = None,
+        timeout: Optional[int] = None,
+    ):
         self.command = command
-        self.log_file = log_file
-        self.env = env
+        self._log_file_name = log_file
+        self._log_file = None  # type: Optional[TextIOWrapper]
+        self.env = env or os.environ.copy()
         self._process = None  # type: Optional[Popen]
         self.timeout = timeout
 
-    def _check_timeout(self):
+    def _check_timeout(self) -> None:
+        if self.timeout is None:
+            return
         sleep(self.timeout)
         while self.process.poll() is None:
             logging.warning(
@@ -33,7 +42,7 @@ class TeePopen:
             os.killpg(self.process.pid, 9)
             sleep(10)
 
-    def __enter__(self):
+    def __enter__(self) -> "TeePopen":
         self.process = Popen(
             self.command,
             shell=True,
@@ -44,25 +53,21 @@ class TeePopen:
             stdout=PIPE,
             bufsize=1,
         )
-        self.log_file = open(self.log_file, "w", encoding="utf-8")
         if self.timeout is not None and self.timeout > 0:
             t = Thread(target=self._check_timeout)
             t.daemon = True  # does not block the program from exit
             t.start()
         return self
 
-    def __exit__(self, t, value, traceback):
-        for line in self.process.stdout:  # type: ignore
-            sys.stdout.write(line)
-            self.log_file.write(line)
-
-        self.process.wait()
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.wait()
         self.log_file.close()
 
     def wait(self):
-        for line in self.process.stdout:  # type: ignore
-            sys.stdout.write(line)
-            self.log_file.write(line)
+        if self.process.stdout is not None:
+            for line in self.process.stdout:
+                sys.stdout.write(line)
+                self.log_file.write(line)
 
         return self.process.wait()
 
@@ -75,3 +80,9 @@ class TeePopen:
     @process.setter
     def process(self, process: Popen) -> None:
         self._process = process
+
+    @property
+    def log_file(self) -> TextIOWrapper:
+        if self._log_file is None:
+            self._log_file = open(self._log_file_name, "w", encoding="utf-8")
+        return self._log_file