Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use TeeStdCapture method in Pytest

Best Python code snippet using pytest

test_capture.py

Source:test_capture.py

...107        in_=capture.SysCapture(0) if in_ else None,108        out=capture.SysCapture(1) if out else None,109        err=capture.SysCapture(2) if err else None,110    )111def TeeStdCapture(112    out: bool = True, err: bool = True, in_: bool = True113) -> MultiCapture[str]:114    return capture.MultiCapture(115        in_=capture.SysCapture(0, tee=True) if in_ else None,116        out=capture.SysCapture(1, tee=True) if out else None,117        err=capture.SysCapture(2, tee=True) if err else None,118    )119@pytest.mark.end_to_end120class TestCaptureManager:121    @pytest.mark.parametrize("method", ["no", "sys", "fd"])122    def test_capturing_basic_api(self, method):123        capouter = StdCaptureFD()124        old = sys.stdout, sys.stderr, sys.stdin125        try:126            capman = CaptureManager(method)127            capman.start_capturing()128            capman.suspend()129            outerr = capman.read()130            assert outerr == ("", "")131            capman.suspend()132            outerr = capman.read()133            assert outerr == ("", "")134            print("hello")135            capman.suspend()136            out, err = capman.read()137            if method == "no":138                assert old == (sys.stdout, sys.stderr, sys.stdin)139            else:140                assert not out141            capman.resume()142            print("hello")143            capman.suspend()144            out, err = capman.read()145            if method != "no":146                assert out == "hello\n"147            capman.stop_capturing()148        finally:149            capouter.stop_capturing()150    def test_init_capturing(self):151        capouter = StdCaptureFD()152        try:153            capman = CaptureManager("fd")154            capman.start_capturing()155            pytest.raises(AssertionError, capman.start_capturing)156            capman.stop_capturing()157        finally:158            capouter.stop_capturing()159@pytest.mark.end_to_end160@pytest.mark.parametrize("method", ["fd", "sys"])161def test_capturing_unicode(tmp_path, runner, method):162    obj = "'b\u00f6y'"163    source = f"""164    # taken from issue 227 from nosetests165    def task_unicode():166        import sys167        print(sys.stdout)168        print({obj})169    """170    tmp_path.joinpath("task_unicode.py").write_text(171        textwrap.dedent(source), encoding="utf-8"172    )173    result = runner.invoke(cli, [tmp_path.as_posix(), f"--capture={method}"])174    assert "1  Succeeded" in result.output175    assert result.exit_code == ExitCode.OK176@pytest.mark.end_to_end177@pytest.mark.parametrize("method", ["fd", "sys"])178def test_capturing_bytes_in_utf8_encoding(tmp_path, runner, method):179    source = """180    def task_unicode():181        print('b\\u00f6y')182    """183    tmp_path.joinpath("task_unicode.py").write_text(184        textwrap.dedent(source), encoding="utf-8"185    )186    result = runner.invoke(cli, [tmp_path.as_posix(), f"--capture={method}"])187    assert "1  Succeeded" in result.output188    assert result.exit_code == ExitCode.OK189@pytest.mark.end_to_end190@pytest.mark.xfail(strict=True, reason="pytask cannot capture during collection.")191def test_collect_capturing(tmp_path, runner):192    source = """193    import sys194    print("collect %s failure" % 13)195    sys.stderr.write("collect %s_stderr failure" % 13)196    import xyz42123197    """198    tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))199    result = runner.invoke(cli, [tmp_path.as_posix()])200    for content in [201        "Captured stdout",202        "collect 13 failure",203        "Captured stderr",204        "collect 13_stderr failure",205    ]:206        assert content in result.output207@pytest.mark.end_to_end208def test_capturing_outerr(tmp_path, runner):209    source = """210    import sys211    def task_capturing():212        print(42)213        sys.stderr.write(str(23))214    def task_capturing_error():215        print(1)216        sys.stderr.write(str(2))217        raise ValueError218    """219    tmp_path.joinpath("task_capturing_outerr.py").write_text(textwrap.dedent(source))220    result = runner.invoke(cli, [tmp_path.as_posix()])221    assert "â F" in result.output222    assert "â ." in result.output223    for content in [224        "âââââââââââââââââââââââââââââââââ Failures ââââââââââââââââââââââââââââââââââ",225        "task_capturing_error failed",226        "ValueError",227        "ââââââââââââââââââââââââ Captured stdout during call ââââââââââââââââââââââââ",228        "1",229        "ââââââââââââââââââââââââ Captured stderr during call ââââââââââââââââââââââââ",230        "2",231        "2  Collected tasks",232        "1  Succeeded",233        "1  Failed",234        "âââââââââââ Failed in ",235        "seconds ââââââââ",236    ]:237        assert content in result.output238@pytest.mark.end_to_end239def test_capture_badoutput_issue412(tmp_path, runner):240    source = """241    import os242    def task_func():243        omg = bytearray([1,129,1])244        os.write(1, omg)245        assert 0246    """247    tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))248    result = runner.invoke(cli, [tmp_path.as_posix(), "--capture=fd"])249    for content in [250        "task_func",251        "assert 0",252        "Captured",253        "1  Failed",254    ]:255        assert content in result.output256@pytest.mark.unit257class TestCaptureIO:258    def test_text(self):259        f = capture.CaptureIO()260        f.write("hello")261        s = f.getvalue()262        assert s == "hello"263        f.close()264    def test_unicode_and_str_mixture(self):265        f = capture.CaptureIO()266        f.write("\u00f6")267        pytest.raises(TypeError, f.write, b"hello")268    def test_write_bytes_to_buffer(self):269        """In python3, stdout / stderr are text io wrappers (exposing a buffer270        property of the underlying bytestream).  See issue #1407271        """272        f = capture.CaptureIO()273        f.buffer.write(b"foo\r\n")274        assert f.getvalue() == "foo\r\n"275@pytest.mark.unit276class TestTeeCaptureIO(TestCaptureIO):277    def test_text(self):278        sio = io.StringIO()279        f = capture.TeeCaptureIO(sio)280        f.write("hello")281        s1 = f.getvalue()282        assert s1 == "hello"283        s2 = sio.getvalue()284        assert s2 == s1285        f.close()286        sio.close()287    def test_unicode_and_str_mixture(self):288        sio = io.StringIO()289        f = capture.TeeCaptureIO(sio)290        f.write("\u00f6")291        pytest.raises(TypeError, f.write, b"hello")292@pytest.mark.integration293def test_dontreadfrominput():294    from _pytest.capture import DontReadFromInput295    f = DontReadFromInput()296    assert f.buffer is f297    assert not f.isatty()298    pytest.raises(OSError, f.read)299    pytest.raises(OSError, f.readlines)300    iter_f = iter(f)301    pytest.raises(OSError, next, iter_f)302    pytest.raises(UnsupportedOperation, f.fileno)303    f.close()  # just for completeness304@pytest.mark.unit305def test_captureresult() -> None:306    cr = CaptureResult("out", "err")307    assert len(cr) == 2308    assert cr.out == "out"309    assert cr.err == "err"310    out, err = cr311    assert out == "out"312    assert err == "err"313    assert cr[0] == "out"314    assert cr[1] == "err"315    assert cr == cr316    assert cr == CaptureResult("out", "err")317    assert cr != CaptureResult("wrong", "err")318    assert cr == ("out", "err")319    assert cr != ("out", "wrong")320    assert hash(cr) == hash(CaptureResult("out", "err"))321    assert hash(cr) == hash(("out", "err"))322    assert hash(cr) != hash(("out", "wrong"))323    assert cr < ("z",)324    assert cr < ("z", "b")325    assert cr < ("z", "b", "c")326    assert cr.count("err") == 1327    assert cr.count("wrong") == 0328    assert cr.index("err") == 1329    with pytest.raises(ValueError):330        assert cr.index("wrong") == 0331    assert next(iter(cr)) == "out"332    assert cr._replace(err="replaced") == ("out", "replaced")333@pytest.fixture()334def tmpfile(tmp_path) -> Generator[BinaryIO, None, None]:335    f = tmp_path.joinpath("task_module.py").open("wb+")336    yield f337    if not f.closed:338        f.close()339@contextlib.contextmanager340def lsof_check():341    pid = os.getpid()342    try:343        out = subprocess.check_output(("lsof", "-p", str(pid))).decode()344    except (OSError, subprocess.CalledProcessError, UnicodeDecodeError) as exc:345        # about UnicodeDecodeError, see note on pytester346        pytest.skip(f"could not run 'lsof' ({exc!r})")347    yield348    out2 = subprocess.check_output(("lsof", "-p", str(pid))).decode()349    len1 = len([x for x in out.split("\n") if "REG" in x])350    len2 = len([x for x in out2.split("\n") if "REG" in x])351    assert len2 < len1 + 3, out2352@pytest.mark.unit353class TestFDCapture:354    def test_simple(self, tmpfile):355        fd = tmpfile.fileno()356        cap = capture.FDCapture(fd)357        data = b"hello"358        os.write(fd, data)359        pytest.raises(AssertionError, cap.snap)360        cap.done()361        cap = capture.FDCapture(fd)362        cap.start()363        os.write(fd, data)364        s = cap.snap()365        cap.done()366        assert s == "hello"367    def test_simple_many(self, tmpfile):368        for _ in range(10):369            self.test_simple(tmpfile)370    def test_simple_many_check_open_files(self, tmp_path):371        with lsof_check():372            with tmp_path.joinpath("task_module.py").open("wb+") as tmpfile:373                self.test_simple_many(tmpfile)374    def test_simple_fail_second_start(self, tmpfile):375        fd = tmpfile.fileno()376        cap = capture.FDCapture(fd)377        cap.done()378        pytest.raises(AssertionError, cap.start)379    def test_stderr(self):380        cap = capture.FDCapture(2)381        cap.start()382        print("hello", file=sys.stderr)383        s = cap.snap()384        cap.done()385        assert s == "hello\n"386    def test_stdin(self):387        cap = capture.FDCapture(0)388        cap.start()389        x = os.read(0, 100).strip()390        cap.done()391        assert x == b""392    def test_writeorg(self, tmpfile):393        data1, data2 = b"foo", b"bar"394        cap = capture.FDCapture(tmpfile.fileno())395        cap.start()396        tmpfile.write(data1)397        tmpfile.flush()398        cap.writeorg(data2.decode("ascii"))399        scap = cap.snap()400        cap.done()401        assert scap == data1.decode("ascii")402        with open(tmpfile.name, "rb") as stmp_file:403            stmp = stmp_file.read()404            assert stmp == data2405    def test_simple_resume_suspend(self):406        with saved_fd(1):407            cap = capture.FDCapture(1)408            cap.start()409            data = b"hello"410            os.write(1, data)411            sys.stdout.write("whatever")412            s = cap.snap()413            assert s == "hellowhatever"414            cap.suspend()415            os.write(1, b"world")416            sys.stdout.write("qlwkej")417            assert not cap.snap()418            cap.resume()419            os.write(1, b"but now")420            sys.stdout.write(" yes\n")421            s = cap.snap()422            assert s == "but now yes\n"423            cap.suspend()424            cap.done()425            pytest.raises(AssertionError, cap.suspend)426            assert repr(cap) == (427                "<FDCapture 1 oldfd={} _state='done' tmpfile={!r}>".format(428                    cap.targetfd_save, cap.tmpfile429                )430            )431            # Should not crash with missing "_old".432            assert repr(cap.syscapture) == (433                "<SysCapture stdout _old=<UNSET> _state='done' tmpfile={!r}>".format(434                    cap.syscapture.tmpfile435                )436            )437    def test_capfd_sys_stdout_mode(self, capfd):  # noqa: U100438        assert "b" not in sys.stdout.mode439@contextlib.contextmanager440def saved_fd(fd):441    new_fd = os.dup(fd)442    try:443        yield444    finally:445        os.dup2(new_fd, fd)446        os.close(new_fd)447@pytest.mark.unit448class TestStdCapture:449    captureclass = staticmethod(StdCapture)450    @contextlib.contextmanager451    def getcapture(self, **kw):452        cap = self.__class__.captureclass(**kw)453        cap.start_capturing()454        try:455            yield cap456        finally:457            cap.stop_capturing()458    def test_capturing_done_simple(self):459        with self.getcapture() as cap:460            sys.stdout.write("hello")461            sys.stderr.write("world")462            out, err = cap.readouterr()463        assert out == "hello"464        assert err == "world"465    def test_capturing_reset_simple(self):466        with self.getcapture() as cap:467            print("hello world")468            sys.stderr.write("hello error\n")469            out, err = cap.readouterr()470        assert out == "hello world\n"471        assert err == "hello error\n"472    def test_capturing_readouterr(self):473        with self.getcapture() as cap:474            print("hello world")475            sys.stderr.write("hello error\n")476            out, err = cap.readouterr()477            assert out == "hello world\n"478            assert err == "hello error\n"479            sys.stderr.write("error2")480            out, err = cap.readouterr()481        assert err == "error2"482    def test_capture_results_accessible_by_attribute(self):483        with self.getcapture() as cap:484            sys.stdout.write("hello")485            sys.stderr.write("world")486            capture_result = cap.readouterr()487        assert capture_result.out == "hello"488        assert capture_result.err == "world"489    def test_capturing_readouterr_unicode(self):490        with self.getcapture() as cap:491            print("hxÄÄ")492            out, err = cap.readouterr()493        assert out == "hxÄÄ\n"494    def test_reset_twice_error(self):495        with self.getcapture() as cap:496            print("hello")497            out, err = cap.readouterr()498        pytest.raises(ValueError, cap.stop_capturing)499        assert out == "hello\n"500        assert not err501    def test_capturing_modify_sysouterr_in_between(self):502        oldout = sys.stdout503        olderr = sys.stderr504        with self.getcapture() as cap:505            sys.stdout.write("hello")506            sys.stderr.write("world")507            sys.stdout = capture.CaptureIO()508            sys.stderr = capture.CaptureIO()509            print("not seen")510            sys.stderr.write("not seen\n")511            out, err = cap.readouterr()512        assert out == "hello"513        assert err == "world"514        assert sys.stdout == oldout515        assert sys.stderr == olderr516    def test_capturing_error_recursive(self):517        with self.getcapture() as cap1:518            print("cap1")519            with self.getcapture() as cap2:520                print("cap2")521                out2, err2 = cap2.readouterr()522                out1, err1 = cap1.readouterr()523        assert out1 == "cap1\n"524        assert out2 == "cap2\n"525    def test_just_out_capture(self):526        with self.getcapture(out=True, err=False) as cap:527            sys.stdout.write("hello")528            sys.stderr.write("world")529            out, err = cap.readouterr()530        assert out == "hello"531        assert not err532    def test_just_err_capture(self):533        with self.getcapture(out=False, err=True) as cap:534            sys.stdout.write("hello")535            sys.stderr.write("world")536            out, err = cap.readouterr()537        assert err == "world"538        assert not out539    def test_stdin_restored(self):540        old = sys.stdin541        with self.getcapture(in_=True):542            newstdin = sys.stdin543        assert newstdin != sys.stdin544        assert sys.stdin is old545    def test_stdin_nulled_by_default(self):546        print("XXX this test may well hang instead of crashing")547        print("XXX which indicates an error in the underlying capturing")548        print("XXX mechanisms")549        with self.getcapture():550            pytest.raises(OSError, sys.stdin.read)551@pytest.mark.unit552class TestTeeStdCapture(TestStdCapture):553    captureclass = staticmethod(TeeStdCapture)554    def test_capturing_error_recursive(self):555        r"""For TeeStdCapture since we passthrough stderr/stdout, cap1556        should get all output, while cap2 should only get "cap2\n"."""557        with self.getcapture() as cap1:558            print("cap1")559            with self.getcapture() as cap2:560                print("cap2")561                out2, err2 = cap2.readouterr()562                out1, err1 = cap1.readouterr()563        assert out1 == "cap1\ncap2\n"564        assert out2 == "cap2\n"565@pytest.mark.unit566class TestStdCaptureFD(TestStdCapture):...

Pytest Tutorial

Looking for an in-depth tutorial around pytest? LambdaTest covers the detailed pytest tutorial that has everything related to the pytest, from setting up the pytest framework to automation testing. Delve deeper into pytest testing by exploring advanced use cases like parallel testing, pytest fixtures, parameterization, executing multiple test cases from a single file, and more.

Chapters

What is pytest
Pytest installation: Want to start pytest from scratch? See how to install and configure pytest for Python automation testing.
Run first test with pytest framework: Follow this step-by-step tutorial to write and run your first pytest script.
Parallel testing with pytest: A hands-on guide to parallel testing with pytest to improve the scalability of your test automation.
Generate pytest reports: Reports make it easier to understand the results of pytest-based test runs. Learn how to generate pytest reports.
Pytest Parameterized tests: Create and run your pytest scripts while avoiding code duplication and increasing test coverage with parameterization.
Pytest Fixtures: Check out how to implement pytest fixtures for your end-to-end testing needs.
Execute Multiple Test Cases: Explore different scenarios for running multiple test cases in pytest from a single file.
Stop Test Suite after N Test Failures: See how to stop your test suite after n test failures in pytest using the @pytest.mark.incremental decorator and maxfail command-line option.