Avoid importing pipes on Python 3.3+

Author: adamchainz

CONTRIBUTORS

@@ -1,3 +1,4 @@
+Adam Johnson
 Albin Vass
 Alex Grönholm
 Alexander Loechel

docs/changelog/2417.bugfix.rst

@@ -0,0 +1 @@
+Avoid importing ``pipes`` on Python 3.3+ to avoid ``DeprecationWarning`` on Python 3.11 -- by :user:`adamchainz`

src/tox/action.py

@@ -1,7 +1,6 @@
 from __future__ import absolute_import, unicode_literals
 
 import os
-import pipes
 import signal
 import subprocess
 import sys
@@ -18,6 +17,11 @@
 from tox.util.lock import get_unique_file
 from tox.util.stdlib import is_main_thread
 
+try:
+    from shlex import quote as shlex_quote
+except ImportError:
+    from pipes import quote as shlex_quote
+
 
 class Action(object):
     """Action is an effort to group operations with the same goal (within reporting)"""
@@ -89,7 +93,7 @@ def popen(
         """this drives an interaction with a subprocess"""
         cwd = py.path.local() if cwd is None else cwd
         cmd_args = [str(x) for x in self._rewrite_args(cwd, args)]
-        cmd_args_shell = " ".join(pipes.quote(i) for i in cmd_args)
+        cmd_args_shell = " ".join(shlex_quote(i) for i in cmd_args)
         stream_getter = self._get_standard_streams(
             capture_err,
             cmd_args_shell,

src/tox/exception.py

@@ -1,7 +1,11 @@
 import os
-import pipes
 import signal
 
+try:
+    from shlex import quote as shlex_quote
+except ImportError:
+    from pipes import quote as shlex_quote
+
 
 def exit_code_str(exception_name, command, exit_code):
     """String representation for an InvocationError, with exit code
@@ -96,7 +100,7 @@ def __init__(self, config):
         self.config = config
 
     def __str__(self):
-        return " ".join(pipes.quote(i) for i in self.config.requires)
+        return " ".join(shlex_quote(i) for i in self.config.requires)
 
 
 class BadRequirement(Error):

src/tox/venv.py

@@ -1,7 +1,6 @@
 import codecs
 import json
 import os
-import pipes
 import re
 import sys
 from itertools import chain
@@ -19,6 +18,11 @@
 
 from .config import DepConfig
 
+try:
+    from shlex import quote as shlex_quote
+except ImportError:
+    from pipes import quote as shlex_quote
+
 #: maximum parsed shebang interpreter length (see: prepend_shebang_interpreter)
 MAXINTERP = 2048
 
@@ -194,7 +198,7 @@ def getcommandpath(self, name, venv=True, cwd=None):
 
         if path is None:
             raise tox.exception.InvocationError(
-                "could not find executable {}".format(pipes.quote(name)),
+                "could not find executable {}".format(shlex_quote(name)),
             )
 
         return str(path)  # will not be rewritten for reporting
@@ -534,7 +538,7 @@ def test(
                 # happens if the same environment is invoked twice
                 message = "commands[{}] | {}".format(
                     i,
-                    " ".join(pipes.quote(str(x)) for x in argv),
+                    " ".join(shlex_quote(str(x)) for x in argv),
                 )
                 action.setactivity(name, message)
                 # check to see if we need to ignore the return code

tests/unit/session/test_session.py

@@ -1,5 +1,4 @@
 import os
-import pipes
 import sys
 import textwrap
 from threading import Thread
@@ -11,6 +10,11 @@
 from tox.package import resolve_package
 from tox.reporter import Verbosity
 
+try:
+    from shlex import quote as shlex_quote
+except ImportError:
+    from pipes import quote as shlex_quote
+
 
 def test_resolve_pkg_missing_directory(tmpdir, mocksession):
     distshare = tmpdir.join("distshare")
@@ -355,7 +359,7 @@ def test_command_prev_fail_command_skip_post_run(cmd, initproj, mock_venv):
             ___________________________________ summary ___________________________________{}
             ERROR:   py: commands failed
         """.format(
-            pipes.quote(sys.executable),
+            shlex_quote(sys.executable),
             "_" if sys.platform != "win32" else "",
         ),
     )