Fixing broken tests and linting

Author: fluzzywuzzy

src/pipx/animate.py

@@ -3,7 +3,7 @@
 from contextlib import contextmanager
 from multiprocessing import Queue
 from threading import Event, Thread
-from typing import Generator, List
+from typing import Generator, List, Optional
 
 from pipx.constants import WINDOWS
 from pipx.emojis import EMOJI_SUPPORT
@@ -31,7 +31,9 @@ def _env_supports_animation() -> bool:
 
 
 @contextmanager
-def animate(message: str, do_animation: bool, *, delay: float = 0, stream: Queue = None) -> Generator[None, None, None]:
+def animate(
+    message: str, do_animation: bool, *, delay: float = 0, stream: Optional[Queue] = None
+) -> Generator[None, None, None]:
     if not do_animation or not _env_supports_animation():
         # No animation, just a single print of message
         sys.stderr.write(f"{message}...\n")

src/pipx/util.py

@@ -1,3 +1,4 @@
+import importlib
 import logging
 import os
 import random
@@ -10,6 +11,7 @@
 from dataclasses import dataclass
 from multiprocessing import Queue
 from pathlib import Path
+from subprocess import TimeoutExpired
 from typing import (
     Any,
     Dict,
@@ -166,7 +168,7 @@ def run_subprocess(
     log_stdout: bool = True,
     log_stderr: bool = True,
     run_dir: Optional[str] = None,
-    stream: Queue = None,
+    stream: Optional[Queue] = None,
 ) -> "subprocess.CompletedProcess[str]":
     """Run arbitrary command as subprocess, capturing stderr and stout"""
     env = dict(os.environ)
@@ -193,6 +195,13 @@ def run_subprocess(
         text=True,
         cwd=run_dir,
     ) as process:
+        try:
+            importlib.util.find_spec("msvcrt")
+        except ModuleNotFoundError:
+            _mswindows = False
+        else:
+            _mswindows = True
+
         try:
             stdout, stderr = "", ""
             while True:
@@ -204,10 +213,10 @@ def run_subprocess(
                         stream.put_nowait(out)
                 if not out and not err:
                     break
-        except subprocess.TimeoutExpired as exc:
+        except TimeoutExpired:
             process.kill()
-            if process._mswindows:
-                exc.stdout, exc.stderr = process.communicate()
+            if _mswindows:
+                process.communicate()
             else:
                 process.wait()
             raise
@@ -216,7 +225,8 @@ def run_subprocess(
             raise
         retcode = process.poll()
 
-    completed_process = subprocess.CompletedProcess(process.args, retcode, stdout, stderr)
+    if type(retcode) is int:
+        completed_process = subprocess.CompletedProcess(process.args, retcode, stdout, stderr)
 
     if capture_stdout and log_stdout:
         logger.debug(f"stdout: {completed_process.stdout}".rstrip())

src/pipx/venv.py

@@ -245,7 +245,7 @@ def install_package(
         # check syntax and clean up spec and pip_args
         (package_or_url, pip_args) = parse_specifier_for_install(package_or_url, pip_args)
 
-        pip_out_stream = Queue()
+        pip_out_stream: Queue = Queue()
 
         logger.info("Installing %s", package_descr := full_package_description(package_name, package_or_url))
         with animate(f"installing {package_descr}", self.do_animation, stream=pip_out_stream):
@@ -293,7 +293,7 @@ def install_package(
     def install_unmanaged_packages(self, requirements: List[str], pip_args: List[str]) -> None:
         """Install packages in the venv, but do not record them in the metadata."""
 
-        pip_out_stream = Queue()
+        pip_out_stream: Queue = Queue()
 
         # Note: We want to install everything at once, as that lets
         # pip resolve conflicts correctly.
@@ -312,7 +312,9 @@ def install_unmanaged_packages(self, requirements: List[str], pip_args: List[str
             ]
             # no logging because any errors will be specially logged by
             #   subprocess_post_check_handle_pip_error()
-            pip_process = run_subprocess(cmd, log_stdout=False, log_stderr=False, run_dir=str(self.root), stream=pip_out_stream)
+            pip_process = run_subprocess(
+                cmd, log_stdout=False, log_stderr=False, run_dir=str(self.root), stream=pip_out_stream
+            )
         subprocess_post_check_handle_pip_error(pip_process)
         if pip_process.returncode:
             raise PipxError(f"Error installing {', '.join(requirements)}.")