Cleanup capture.py and enable all debugging tests.

Author: tobiasraabe

src/_pytask/capture.py

@@ -783,40 +783,40 @@ class CaptureManager:
 
     def __init__(self, method: "_CaptureMethod") -> None:
         self._method = method
-        self._global_capturing: Optional[MultiCapture[str]] = None
+        self._capturing: Optional[MultiCapture[str]] = None
 
     def __repr__(self) -> str:
-        return ("<CaptureManager _method={!r} _global_capturing={!r}>").format(
-            self._method, self._global_capturing
+        return ("<CaptureManager _method={!r} _capturing={!r}>").format(
+            self._method, self._capturing
         )
 
     def is_capturing(self) -> Union[str, bool]:
         return self._method != "no"
 
     def start_capturing(self) -> None:
-        assert self._global_capturing is None
-        self._global_capturing = _get_multicapture(self._method)
-        self._global_capturing.start_capturing()
+        assert self._capturing is None
+        self._capturing = _get_multicapture(self._method)
+        self._capturing.start_capturing()
 
     def stop_capturing(self) -> None:
-        if self._global_capturing is not None:
-            self._global_capturing.pop_outerr_to_orig()
-            self._global_capturing.stop_capturing()
-            self._global_capturing = None
+        if self._capturing is not None:
+            self._capturing.pop_outerr_to_orig()
+            self._capturing.stop_capturing()
+            self._capturing = None
 
     def resume(self) -> None:
         # During teardown of the python process, and on rare occasions, capture
         # attributes can be `None` while trying to resume global capture.
-        if self._global_capturing is not None:
-            self._global_capturing.resume_capturing()
+        if self._capturing is not None:
+            self._capturing.resume_capturing()
 
     def suspend(self, in_: bool = False) -> None:
-        if self._global_capturing is not None:
-            self._global_capturing.suspend_capturing(in_=in_)
+        if self._capturing is not None:
+            self._capturing.suspend_capturing(in_=in_)
 
     def read(self) -> CaptureResult[str]:
-        assert self._global_capturing is not None
-        return self._global_capturing.readouterr()
+        assert self._capturing is not None
+        return self._capturing.readouterr()
 
     # Helper context managers
 

src/_pytask/debugging.py

@@ -207,18 +207,11 @@ def do_continue(self, arg):
                     capman = self._pytask_capman
                     capturing = PytaskPDB._is_capturing(capman)
                     if capturing:
-                        if capturing == "global":
-                            console.rule(
-                                "PDB continue (IO-capturing resumed)",
-                                characters=">",
-                                style=None,
-                            )
-                        else:
-                            console.rule(
-                                f"PDB continue (IO-capturing resumed for {capturing})",
-                                characters=">",
-                                style=None,
-                            )
+                        console.rule(
+                            "PDB continue (IO-capturing resumed)",
+                            characters=">",
+                            style=None,
+                        )
                         assert capman is not None
                         capman.resume()
                     else:
@@ -293,18 +286,12 @@ def _init_pdb(cls, method, *args, **kwargs):
                     console.rule(header, characters=">", style=None)
                 else:
                     capturing = cls._is_capturing(capman)
-                    if capturing == "global":
+                    if capturing:
                         console.rule(
                             f"PDB {method} (IO-capturing turned off)",
                             characters=">",
                             style=None,
                         )
-                    elif capturing:
-                        console.rule(
-                            f"PDB {method} (IO-capturing turned off for {capturing})",
-                            characters=">",
-                            style=None,
-                        )
                     else:
                         console.rule(f"PDB {method}", characters=">", style=None)
 

tests/test_debugging.py

@@ -1,4 +1,5 @@
 import os
+import re
 import sys
 import textwrap
 from contextlib import ExitStack as does_not_raise  # noqa: N813
@@ -289,19 +290,19 @@ def task_1():
     tmp_path.joinpath("task_dummy.py").write_text(textwrap.dedent(source))
 
     child = pexpect.spawn(f"pytask {tmp_path.as_posix()}")
-    child.expect(r"PDB set_trace \(IO-capturing turned off\)")
+    child.expect(["PDB", "set_trace", r"\(IO-capturing", "turned", r"off\)"])
     child.expect("task_1")
     child.expect("x = 3")
     child.expect("Pdb")
     child.sendline("c")
-    child.expect(r"PDB continue \(IO-capturing resumed\)")
-    child.expect(r"PDB set_trace \(IO-capturing turned off\)")
+    child.expect(["PDB", "continue", r"\(IO-capturing", r"resumed\)"])
+    child.expect(["PDB", "set_trace", r"\(IO-capturing", "turned", r"off\)"])
     child.expect("x = 4")
     child.expect("Pdb")
     child.sendline("c")
-    child.expect(r"PDB continue \(IO-capturing resumed\)")
+    child.expect(["PDB", "continue", r"\(IO-capturing", r"resumed\)"])
     child.expect("task_1 failed")
-    rest = child.read().decode("utf8")
+    rest = _escape_ansi(child.read().decode("utf8"))
     assert "Captured stdout during call" in rest
     assert "hello17" in rest  # out is captured
     assert "hello18" in rest  # out is captured
@@ -361,7 +362,7 @@ def task_1():
         env={"PATH": os.environ["PATH"], "PYTHONPATH": f"{tmp_path.as_posix()}"},
     )
 
-    child.expect(r"PDB set_trace \(IO-capturing turned off\)")
+    child.expect(["PDB", "set_trace", r"\(IO-capturing", "turned", r"off\)"])
     child.expect(r"\n\(Pdb")
     child.sendline("debug foo()")
     child.expect("ENTERING RECURSIVE DEBUGGER")
@@ -380,8 +381,8 @@ def task_1():
     assert b"Quitting debugger" not in child.before
 
     child.sendline("c")
-    child.expect(r"PDB continue \(IO-capturing resumed\)")
-    rest = child.read().decode("utf8")
+    child.expect(["PDB", "continue", r"\(IO-capturing", r"resumed\)"])
+    rest = _escape_ansi(child.read().decode("utf8"))
     assert "hello17" in rest  # out is captured
     assert "hello18" in rest  # out is captured
     assert "1 failed" in rest
@@ -405,7 +406,7 @@ def task_1():
     child.expect("Pdb")
     child.sendline("c")
     child.expect(r"PDB continue")
-    child.expect("1 succeeded")
+    child.expect(["1", "succeeded"])
     _flush(child)
 
 
@@ -469,3 +470,9 @@ def test_function():
     rest = child.read().decode("utf8")
     assert "1 passed" in rest
     _flush(child)
+
+
+def _escape_ansi(line):
+    """Escape ANSI sequences produced by rich."""
+    ansi_escape = re.compile(r"(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]")
+    return ansi_escape.sub("", line)