TST: Use Popen in contextmanagers (#44611)

Author: mroeschke

pandas/io/clipboard/__init__.py

@@ -111,12 +111,16 @@ def _stringifyText(text) -> str:
 def init_osx_pbcopy_clipboard():
     def copy_osx_pbcopy(text):
         text = _stringifyText(text)  # Converts non-str values to str.
-        p = subprocess.Popen(["pbcopy", "w"], stdin=subprocess.PIPE, close_fds=True)
-        p.communicate(input=text.encode(ENCODING))
+        with subprocess.Popen(
+            ["pbcopy", "w"], stdin=subprocess.PIPE, close_fds=True
+        ) as p:
+            p.communicate(input=text.encode(ENCODING))
 
     def paste_osx_pbcopy():
-        p = subprocess.Popen(["pbpaste", "r"], stdout=subprocess.PIPE, close_fds=True)
-        stdout = p.communicate()[0]
+        with subprocess.Popen(
+            ["pbpaste", "r"], stdout=subprocess.PIPE, close_fds=True
+        ) as p:
+            stdout = p.communicate()[0]
         return stdout.decode(ENCODING)
 
     return copy_osx_pbcopy, paste_osx_pbcopy
@@ -179,22 +183,22 @@ def copy_xclip(text, primary=False):
         selection = DEFAULT_SELECTION
         if primary:
             selection = PRIMARY_SELECTION
-        p = subprocess.Popen(
+        with subprocess.Popen(
             ["xclip", "-selection", selection], stdin=subprocess.PIPE, close_fds=True
-        )
-        p.communicate(input=text.encode(ENCODING))
+        ) as p:
+            p.communicate(input=text.encode(ENCODING))
 
     def paste_xclip(primary=False):
         selection = DEFAULT_SELECTION
         if primary:
             selection = PRIMARY_SELECTION
-        p = subprocess.Popen(
+        with subprocess.Popen(
             ["xclip", "-selection", selection, "-o"],
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             close_fds=True,
-        )
-        stdout = p.communicate()[0]
+        ) as p:
+            stdout = p.communicate()[0]
         # Intentionally ignore extraneous output on stderr when clipboard is empty
         return stdout.decode(ENCODING)
 
@@ -210,19 +214,19 @@ def copy_xsel(text, primary=False):
         selection_flag = DEFAULT_SELECTION
         if primary:
             selection_flag = PRIMARY_SELECTION
-        p = subprocess.Popen(
+        with subprocess.Popen(
             ["xsel", selection_flag, "-i"], stdin=subprocess.PIPE, close_fds=True
-        )
-        p.communicate(input=text.encode(ENCODING))
+        ) as p:
+            p.communicate(input=text.encode(ENCODING))
 
     def paste_xsel(primary=False):
         selection_flag = DEFAULT_SELECTION
         if primary:
             selection_flag = PRIMARY_SELECTION
-        p = subprocess.Popen(
+        with subprocess.Popen(
             ["xsel", selection_flag, "-o"], stdout=subprocess.PIPE, close_fds=True
-        )
-        stdout = p.communicate()[0]
+        ) as p:
+            stdout = p.communicate()[0]
         return stdout.decode(ENCODING)
 
     return copy_xsel, paste_xsel
@@ -231,7 +235,7 @@ def paste_xsel(primary=False):
 def init_klipper_clipboard():
     def copy_klipper(text):
         text = _stringifyText(text)  # Converts non-str values to str.
-        p = subprocess.Popen(
+        with subprocess.Popen(
             [
                 "qdbus",
                 "org.kde.klipper",
@@ -241,16 +245,16 @@ def copy_klipper(text):
             ],
             stdin=subprocess.PIPE,
             close_fds=True,
-        )
-        p.communicate(input=None)
+        ) as p:
+            p.communicate(input=None)
 
     def paste_klipper():
-        p = subprocess.Popen(
+        with subprocess.Popen(
             ["qdbus", "org.kde.klipper", "/klipper", "getClipboardContents"],
             stdout=subprocess.PIPE,
             close_fds=True,
-        )
-        stdout = p.communicate()[0]
+        ) as p:
+            stdout = p.communicate()[0]
 
         # Workaround for https://bugs.kde.org/show_bug.cgi?id=342874
         # TODO: https://github.com/asweigart/pyperclip/issues/43
@@ -483,17 +487,17 @@ def paste_windows():
 def init_wsl_clipboard():
     def copy_wsl(text):
         text = _stringifyText(text)  # Converts non-str values to str.
-        p = subprocess.Popen(["clip.exe"], stdin=subprocess.PIPE, close_fds=True)
-        p.communicate(input=text.encode(ENCODING))
+        with subprocess.Popen(["clip.exe"], stdin=subprocess.PIPE, close_fds=True) as p:
+            p.communicate(input=text.encode(ENCODING))
 
     def paste_wsl():
-        p = subprocess.Popen(
+        with subprocess.Popen(
             ["powershell.exe", "-command", "Get-Clipboard"],
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             close_fds=True,
-        )
-        stdout = p.communicate()[0]
+        ) as p:
+            stdout = p.communicate()[0]
         # WSL appends "\r\n" to the contents.
         return stdout[:-2].decode(ENCODING)
 

pandas/tests/io/conftest.py

@@ -69,27 +69,26 @@ def s3_base(worker_id):
         endpoint_uri = f"http://127.0.0.1:{endpoint_port}/"
 
         # pipe to null to avoid logging in terminal
-        proc = subprocess.Popen(
+        with subprocess.Popen(
             shlex.split(f"moto_server s3 -p {endpoint_port}"),
             stdout=subprocess.DEVNULL,
             stderr=subprocess.DEVNULL,
-        )
-
-        timeout = 5
-        while timeout > 0:
-            try:
-                # OK to go once server is accepting connections
-                r = requests.get(endpoint_uri)
-                if r.ok:
-                    break
-            except Exception:
-                pass
-            timeout -= 0.1
-            time.sleep(0.1)
-        yield endpoint_uri
-
-        proc.terminate()
-        proc.wait()
+        ) as proc:
+
+            timeout = 5
+            while timeout > 0:
+                try:
+                    # OK to go once server is accepting connections
+                    r = requests.get(endpoint_uri)
+                    if r.ok:
+                        break
+                except Exception:
+                    pass
+                timeout -= 0.1
+                time.sleep(0.1)
+            yield endpoint_uri
+
+            proc.terminate()
 
 
 @pytest.fixture()