Merge pull request #3830 from nicoddemus/capfd-fixture-capture
Fixtures during teardown can use capsys and capfd to get output from tests
This commit is contained in:
		
						commit
						43657f252f
					
				| 
						 | 
				
			
			@ -0,0 +1 @@
 | 
			
		|||
Fixtures during teardown can again use ``capsys`` and ``cafd`` to inspect output captured during tests.
 | 
			
		||||
| 
						 | 
				
			
			@ -309,6 +309,9 @@ class CaptureFixture(object):
 | 
			
		|||
    def __init__(self, captureclass, request):
 | 
			
		||||
        self.captureclass = captureclass
 | 
			
		||||
        self.request = request
 | 
			
		||||
        self._capture = None
 | 
			
		||||
        self._captured_out = self.captureclass.EMPTY_BUFFER
 | 
			
		||||
        self._captured_err = self.captureclass.EMPTY_BUFFER
 | 
			
		||||
 | 
			
		||||
    def _start(self):
 | 
			
		||||
        self._capture = MultiCapture(
 | 
			
		||||
| 
						 | 
				
			
			@ -317,20 +320,26 @@ class CaptureFixture(object):
 | 
			
		|||
        self._capture.start_capturing()
 | 
			
		||||
 | 
			
		||||
    def close(self):
 | 
			
		||||
        cap = self.__dict__.pop("_capture", None)
 | 
			
		||||
        if cap is not None:
 | 
			
		||||
            self._outerr = cap.pop_outerr_to_orig()
 | 
			
		||||
            cap.stop_capturing()
 | 
			
		||||
        if self._capture is not None:
 | 
			
		||||
            out, err = self._capture.pop_outerr_to_orig()
 | 
			
		||||
            self._captured_out += out
 | 
			
		||||
            self._captured_err += err
 | 
			
		||||
            self._capture.stop_capturing()
 | 
			
		||||
            self._capture = None
 | 
			
		||||
 | 
			
		||||
    def readouterr(self):
 | 
			
		||||
        """Read and return the captured output so far, resetting the internal buffer.
 | 
			
		||||
 | 
			
		||||
        :return: captured content as a namedtuple with  ``out`` and ``err`` string attributes
 | 
			
		||||
        """
 | 
			
		||||
        try:
 | 
			
		||||
            return self._capture.readouterr()
 | 
			
		||||
        except AttributeError:
 | 
			
		||||
            return self._outerr
 | 
			
		||||
        captured_out, captured_err = self._captured_out, self._captured_err
 | 
			
		||||
        if self._capture is not None:
 | 
			
		||||
            out, err = self._capture.readouterr()
 | 
			
		||||
            captured_out += out
 | 
			
		||||
            captured_err += err
 | 
			
		||||
        self._captured_out = self.captureclass.EMPTY_BUFFER
 | 
			
		||||
        self._captured_err = self.captureclass.EMPTY_BUFFER
 | 
			
		||||
        return CaptureResult(captured_out, captured_err)
 | 
			
		||||
 | 
			
		||||
    @contextlib.contextmanager
 | 
			
		||||
    def _suspend(self):
 | 
			
		||||
| 
						 | 
				
			
			@ -463,6 +472,7 @@ class MultiCapture(object):
 | 
			
		|||
 | 
			
		||||
 | 
			
		||||
class NoCapture(object):
 | 
			
		||||
    EMPTY_BUFFER = None
 | 
			
		||||
    __init__ = start = done = suspend = resume = lambda *args: None
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -472,6 +482,8 @@ class FDCaptureBinary(object):
 | 
			
		|||
    snap() produces `bytes`
 | 
			
		||||
    """
 | 
			
		||||
 | 
			
		||||
    EMPTY_BUFFER = bytes()
 | 
			
		||||
 | 
			
		||||
    def __init__(self, targetfd, tmpfile=None):
 | 
			
		||||
        self.targetfd = targetfd
 | 
			
		||||
        try:
 | 
			
		||||
| 
						 | 
				
			
			@ -545,6 +557,8 @@ class FDCapture(FDCaptureBinary):
 | 
			
		|||
    snap() produces text
 | 
			
		||||
    """
 | 
			
		||||
 | 
			
		||||
    EMPTY_BUFFER = str()
 | 
			
		||||
 | 
			
		||||
    def snap(self):
 | 
			
		||||
        res = FDCaptureBinary.snap(self)
 | 
			
		||||
        enc = getattr(self.tmpfile, "encoding", None)
 | 
			
		||||
| 
						 | 
				
			
			@ -554,6 +568,9 @@ class FDCapture(FDCaptureBinary):
 | 
			
		|||
 | 
			
		||||
 | 
			
		||||
class SysCapture(object):
 | 
			
		||||
 | 
			
		||||
    EMPTY_BUFFER = str()
 | 
			
		||||
 | 
			
		||||
    def __init__(self, fd, tmpfile=None):
 | 
			
		||||
        name = patchsysdict[fd]
 | 
			
		||||
        self._old = getattr(sys, name)
 | 
			
		||||
| 
						 | 
				
			
			@ -591,6 +608,8 @@ class SysCapture(object):
 | 
			
		|||
 | 
			
		||||
 | 
			
		||||
class SysCaptureBinary(SysCapture):
 | 
			
		||||
    EMPTY_BUFFER = bytes()
 | 
			
		||||
 | 
			
		||||
    def snap(self):
 | 
			
		||||
        res = self.tmpfile.buffer.getvalue()
 | 
			
		||||
        self.tmpfile.seek(0)
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -647,6 +647,34 @@ class TestCaptureFixture(object):
 | 
			
		|||
        assert "stdout contents begin" not in result.stdout.str()
 | 
			
		||||
        assert "stderr contents begin" not in result.stdout.str()
 | 
			
		||||
 | 
			
		||||
    @pytest.mark.parametrize("cap", ["capsys", "capfd"])
 | 
			
		||||
    def test_fixture_use_by_other_fixtures_teardown(self, testdir, cap):
 | 
			
		||||
        """Ensure we can access setup and teardown buffers from teardown when using capsys/capfd (##3033)"""
 | 
			
		||||
        testdir.makepyfile(
 | 
			
		||||
            """
 | 
			
		||||
            import sys
 | 
			
		||||
            import pytest
 | 
			
		||||
            import os
 | 
			
		||||
 | 
			
		||||
            @pytest.fixture()
 | 
			
		||||
            def fix({cap}):
 | 
			
		||||
                print("setup out")
 | 
			
		||||
                sys.stderr.write("setup err\\n")
 | 
			
		||||
                yield
 | 
			
		||||
                out, err = {cap}.readouterr()
 | 
			
		||||
                assert out == 'setup out\\ncall out\\n'
 | 
			
		||||
                assert err == 'setup err\\ncall err\\n'
 | 
			
		||||
 | 
			
		||||
            def test_a(fix):
 | 
			
		||||
                print("call out")
 | 
			
		||||
                sys.stderr.write("call err\\n")
 | 
			
		||||
        """.format(
 | 
			
		||||
                cap=cap
 | 
			
		||||
            )
 | 
			
		||||
        )
 | 
			
		||||
        reprec = testdir.inline_run()
 | 
			
		||||
        reprec.assertoutcome(passed=1)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def test_setup_failure_does_not_kill_capturing(testdir):
 | 
			
		||||
    sub1 = testdir.mkpydir("sub1")
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
		Loading…
	
		Reference in New Issue