Merge pull request #4951 from blueyed/fix-pdb-capfix

pdb: handle capturing with fixtures only

Author: nicoddemus

changelog/4951.feature.rst

@@ -0,0 +1 @@
+Output capturing is handled correctly when only capturing via fixtures (capsys, capfs) with ``pdb.set_trace()``.

src/_pytest/capture.py

@@ -107,6 +107,16 @@ def _getcapture(self, method):
             return MultiCapture(out=False, err=False, in_=False)
         raise ValueError("unknown capturing method: %r" % method)  # pragma: no cover
 
+    def is_capturing(self):
+        if self.is_globally_capturing():
+            return "global"
+        capture_fixture = getattr(self._current_item, "_capture_fixture", None)
+        if capture_fixture is not None:
+            return (
+                "fixture %s" % self._current_item._capture_fixture.request.fixturename
+            )
+        return False
+
     # Global capturing control
 
     def is_globally_capturing(self):
@@ -134,6 +144,15 @@ def suspend_global_capture(self, in_=False):
         if cap is not None:
             cap.suspend_capturing(in_=in_)
 
+    def suspend(self, in_=False):
+        # Need to undo local capsys-et-al if it exists before disabling global capture.
+        self.suspend_fixture(self._current_item)
+        self.suspend_global_capture(in_)
+
+    def resume(self):
+        self.resume_global_capture()
+        self.resume_fixture(self._current_item)
+
     def read_global_capture(self):
         return self._global_capturing.readouterr()
 
@@ -168,14 +187,11 @@ def resume_fixture(self, item):
     @contextlib.contextmanager
     def global_and_fixture_disabled(self):
         """Context manager to temporarily disable global and current fixture capturing."""
-        # Need to undo local capsys-et-al if it exists before disabling global capture.
-        self.suspend_fixture(self._current_item)
-        self.suspend_global_capture(in_=False)
+        self.suspend()
         try:
             yield
         finally:
-            self.resume_global_capture()
-            self.resume_fixture(self._current_item)
+            self.resume()
 
     @contextlib.contextmanager
     def item_capture(self, when, item):

src/_pytest/debugging.py

@@ -101,6 +101,12 @@ class pytestPDB(object):
     _saved = []
     _recursive_debug = 0
 
+    @classmethod
+    def _is_capturing(cls, capman):
+        if capman:
+            return capman.is_capturing()
+        return False
+
     @classmethod
     def _init_pdb(cls, *args, **kwargs):
         """ Initialize PDB debugging, dropping any IO capturing. """
@@ -109,18 +115,27 @@ def _init_pdb(cls, *args, **kwargs):
         if cls._pluginmanager is not None:
             capman = cls._pluginmanager.getplugin("capturemanager")
             if capman:
-                capman.suspend_global_capture(in_=True)
+                capman.suspend(in_=True)
             tw = _pytest.config.create_terminal_writer(cls._config)
             tw.line()
             if cls._recursive_debug == 0:
                 # Handle header similar to pdb.set_trace in py37+.
                 header = kwargs.pop("header", None)
                 if header is not None:
                     tw.sep(">", header)
-                elif capman and capman.is_globally_capturing():
-                    tw.sep(">", "PDB set_trace (IO-capturing turned off)")
                 else:
-                    tw.sep(">", "PDB set_trace")
+                    capturing = cls._is_capturing(capman)
+                    if capturing:
+                        if capturing == "global":
+                            tw.sep(">", "PDB set_trace (IO-capturing turned off)")
+                        else:
+                            tw.sep(
+                                ">",
+                                "PDB set_trace (IO-capturing turned off for %s)"
+                                % capturing,
+                            )
+                    else:
+                        tw.sep(">", "PDB set_trace")
 
             class _PdbWrapper(cls._pdb_cls, object):
                 _pytest_capman = capman
@@ -134,15 +149,24 @@ def do_debug(self, arg):
 
                 def do_continue(self, arg):
                     ret = super(_PdbWrapper, self).do_continue(arg)
-                    if self._pytest_capman:
+                    if cls._recursive_debug == 0:
                         tw = _pytest.config.create_terminal_writer(cls._config)
                         tw.line()
-                        if cls._recursive_debug == 0:
-                            if self._pytest_capman.is_globally_capturing():
+
+                        capman = self._pytest_capman
+                        capturing = pytestPDB._is_capturing(capman)
+                        if capturing:
+                            if capturing == "global":
                                 tw.sep(">", "PDB continue (IO-capturing resumed)")
                             else:
-                                tw.sep(">", "PDB continue")
-                            self._pytest_capman.resume_global_capture()
+                                tw.sep(
+                                    ">",
+                                    "PDB continue (IO-capturing resumed for %s)"
+                                    % capturing,
+                                )
+                            capman.resume()
+                        else:
+                            tw.sep(">", "PDB continue")
                     cls._pluginmanager.hook.pytest_leave_pdb(
                         config=cls._config, pdb=self
                     )

testing/test_pdb.py

@@ -576,7 +576,8 @@ def test_1():
         child.sendline("c")
         child.expect("LEAVING RECURSIVE DEBUGGER")
         assert b"PDB continue" not in child.before
-        assert b"print_from_foo" in child.before
+        # No extra newline.
+        assert child.before.endswith(b"c\r\nprint_from_foo\r\n")
         child.sendline("c")
         child.expect(r"PDB continue \(IO-capturing resumed\)")
         rest = child.read().decode("utf8")
@@ -603,6 +604,98 @@ def test_1():
         child.expect("1 passed")
         self.flush(child)
 
+    @pytest.mark.parametrize("capture_arg", ("", "-s", "-p no:capture"))
+    def test_pdb_continue_with_recursive_debug(self, capture_arg, testdir):
+        """Full coverage for do_debug without capturing.
+
+        This is very similar to test_pdb_interaction_continue_recursive in general,
+        but mocks out ``pdb.set_trace`` for providing more coverage.
+        """
+        p1 = testdir.makepyfile(
+            """
+            try:
+                input = raw_input
+            except NameError:
+                pass
+
+            def set_trace():
+                __import__('pdb').set_trace()
+
+            def test_1(monkeypatch):
+                import _pytest.debugging
+
+                class pytestPDBTest(_pytest.debugging.pytestPDB):
+                    @classmethod
+                    def set_trace(cls, *args, **kwargs):
+                        # Init _PdbWrapper to handle capturing.
+                        _pdb = cls._init_pdb(*args, **kwargs)
+
+                        # Mock out pdb.Pdb.do_continue.
+                        import pdb
+                        pdb.Pdb.do_continue = lambda self, arg: None
+
+                        print("=== SET_TRACE ===")
+                        assert input() == "debug set_trace()"
+
+                        # Simulate _PdbWrapper.do_debug
+                        cls._recursive_debug += 1
+                        print("ENTERING RECURSIVE DEBUGGER")
+                        print("=== SET_TRACE_2 ===")
+
+                        assert input() == "c"
+                        _pdb.do_continue("")
+                        print("=== SET_TRACE_3 ===")
+
+                        # Simulate _PdbWrapper.do_debug
+                        print("LEAVING RECURSIVE DEBUGGER")
+                        cls._recursive_debug -= 1
+
+                        print("=== SET_TRACE_4 ===")
+                        assert input() == "c"
+                        _pdb.do_continue("")
+
+                    def do_continue(self, arg):
+                        print("=== do_continue")
+                        # _PdbWrapper.do_continue("")
+
+                monkeypatch.setattr(_pytest.debugging, "pytestPDB", pytestPDBTest)
+
+                import pdb
+                monkeypatch.setattr(pdb, "set_trace", pytestPDBTest.set_trace)
+
+                set_trace()
+        """
+        )
+        child = testdir.spawn_pytest("%s %s" % (p1, capture_arg))
+        child.expect("=== SET_TRACE ===")
+        before = child.before.decode("utf8")
+        if not capture_arg:
+            assert ">>> PDB set_trace (IO-capturing turned off) >>>" in before
+        else:
+            assert ">>> PDB set_trace >>>" in before
+        child.sendline("debug set_trace()")
+        child.expect("=== SET_TRACE_2 ===")
+        before = child.before.decode("utf8")
+        assert "\r\nENTERING RECURSIVE DEBUGGER\r\n" in before
+        child.sendline("c")
+        child.expect("=== SET_TRACE_3 ===")
+
+        # No continue message with recursive debugging.
+        before = child.before.decode("utf8")
+        assert ">>> PDB continue " not in before
+
+        child.sendline("c")
+        child.expect("=== SET_TRACE_4 ===")
+        before = child.before.decode("utf8")
+        assert "\r\nLEAVING RECURSIVE DEBUGGER\r\n" in before
+        child.sendline("c")
+        rest = child.read().decode("utf8")
+        if not capture_arg:
+            assert "> PDB continue (IO-capturing resumed) >" in rest
+        else:
+            assert "> PDB continue >" in rest
+        assert "1 passed in" in rest
+
     def test_pdb_used_outside_test(self, testdir):
         p1 = testdir.makepyfile(
             """
@@ -970,3 +1063,52 @@ def test_2():
     rest = child.read().decode("utf8")
     assert "no tests ran" in rest
     TestPDB.flush(child)
+
+
+@pytest.mark.parametrize("fixture", ("capfd", "capsys"))
+def test_pdb_suspends_fixture_capturing(testdir, fixture):
+    """Using "-s" with pytest should suspend/resume fixture capturing."""
+    p1 = testdir.makepyfile(
+        """
+        def test_inner({fixture}):
+            import sys
+
+            print("out_inner_before")
+            sys.stderr.write("err_inner_before\\n")
+
+            __import__("pdb").set_trace()
+
+            print("out_inner_after")
+            sys.stderr.write("err_inner_after\\n")
+
+            out, err = {fixture}.readouterr()
+            assert out =="out_inner_before\\nout_inner_after\\n"
+            assert err =="err_inner_before\\nerr_inner_after\\n"
+        """.format(
+            fixture=fixture
+        )
+    )
+
+    child = testdir.spawn_pytest(str(p1) + " -s")
+
+    child.expect("Pdb")
+    before = child.before.decode("utf8")
+    assert (
+        "> PDB set_trace (IO-capturing turned off for fixture %s) >" % (fixture)
+        in before
+    )
+
+    # Test that capturing is really suspended.
+    child.sendline("p 40 + 2")
+    child.expect("Pdb")
+    assert "\r\n42\r\n" in child.before.decode("utf8")
+
+    child.sendline("c")
+    rest = child.read().decode("utf8")
+    assert "out_inner" not in rest
+    assert "err_inner" not in rest
+
+    TestPDB.flush(child)
+    assert child.exitstatus == 0
+    assert "= 1 passed in " in rest
+    assert "> PDB continue (IO-capturing resumed for fixture %s) >" % (fixture) in rest