from __future__ import absolute_import, division, print_function # note: py.io capture tests where copied from # pylib 1.4.20.dev2 (rev 13d9af95547e) from __future__ import with_statement import pickle import os import sys from io import UnsupportedOperation import _pytest._code import py import pytest import contextlib from _pytest import capture from _pytest.capture import CaptureManager from _pytest.main import EXIT_NOTESTSCOLLECTED needsosdup = pytest.mark.xfail("not hasattr(os, 'dup')") if sys.version_info >= (3, 0): def tobytes(obj): if isinstance(obj, str): obj = obj.encode('UTF-8') assert isinstance(obj, bytes) return obj def totext(obj): if isinstance(obj, bytes): obj = str(obj, 'UTF-8') assert isinstance(obj, str) return obj else: def tobytes(obj): if isinstance(obj, unicode): obj = obj.encode('UTF-8') assert isinstance(obj, str) return obj def totext(obj): if isinstance(obj, str): obj = unicode(obj, 'UTF-8') assert isinstance(obj, unicode) return obj def oswritebytes(fd, obj): os.write(fd, tobytes(obj)) def StdCaptureFD(out=True, err=True, in_=True): return capture.MultiCapture(out, err, in_, Capture=capture.FDCapture) def StdCapture(out=True, err=True, in_=True): return capture.MultiCapture(out, err, in_, Capture=capture.SysCapture) class TestCaptureManager(object): def test_getmethod_default_no_fd(self, monkeypatch): from _pytest.capture import pytest_addoption from _pytest.config import Parser parser = Parser() pytest_addoption(parser) default = parser._groups[0].options[0].default assert default == "fd" if hasattr(os, "dup") else "sys" parser = Parser() monkeypatch.delattr(os, 'dup', raising=False) pytest_addoption(parser) assert parser._groups[0].options[0].default == "sys" @needsosdup @pytest.mark.parametrize("method", ['no', 'sys', pytest.mark.skipif('not hasattr(os, "dup")', 'fd')]) def test_capturing_basic_api(self, method): capouter = StdCaptureFD() old = sys.stdout, sys.stderr, sys.stdin try: capman = CaptureManager(method) capman.start_global_capturing() outerr = capman.suspend_global_capture() assert outerr == ("", "") outerr = capman.suspend_global_capture() assert outerr == ("", "") print("hello") out, err = capman.suspend_global_capture() if method == "no": assert old == (sys.stdout, sys.stderr, sys.stdin) else: assert not out capman.resume_global_capture() print("hello") out, err = capman.suspend_global_capture() if method != "no": assert out == "hello\n" capman.stop_global_capturing() finally: capouter.stop_capturing() @needsosdup def test_init_capturing(self): capouter = StdCaptureFD() try: capman = CaptureManager("fd") capman.start_global_capturing() pytest.raises(AssertionError, "capman.start_global_capturing()") capman.stop_global_capturing() finally: capouter.stop_capturing() @pytest.mark.parametrize("method", ['fd', 'sys']) def test_capturing_unicode(testdir, method): if hasattr(sys, "pypy_version_info") and sys.pypy_version_info < (2, 2): pytest.xfail("does not work on pypy < 2.2") if sys.version_info >= (3, 0): obj = "'b\u00f6y'" else: obj = "u'\u00f6y'" testdir.makepyfile(""" # coding=utf8 # taken from issue 227 from nosetests def test_unicode(): import sys print (sys.stdout) print (%s) """ % obj) result = testdir.runpytest("--capture=%s" % method) result.stdout.fnmatch_lines([ "*1 passed*" ]) @pytest.mark.parametrize("method", ['fd', 'sys']) def test_capturing_bytes_in_utf8_encoding(testdir, method): testdir.makepyfile(""" def test_unicode(): print ('b\\u00f6y') """) result = testdir.runpytest("--capture=%s" % method) result.stdout.fnmatch_lines([ "*1 passed*" ]) def test_collect_capturing(testdir): p = testdir.makepyfile(""" print ("collect %s failure" % 13) import xyz42123 """) result = testdir.runpytest(p) result.stdout.fnmatch_lines([ "*Captured stdout*", "*collect 13 failure*", ]) class TestPerTestCapturing(object): def test_capture_and_fixtures(self, testdir): p = testdir.makepyfile(""" def setup_module(mod): print ("setup module") def setup_function(function): print ("setup " + function.__name__) def test_func1(): print ("in func1") assert 0 def test_func2(): print ("in func2") assert 0 """) result = testdir.runpytest(p) result.stdout.fnmatch_lines([ "setup module*", "setup test_func1*", "in func1*", "setup test_func2*", "in func2*", ]) @pytest.mark.xfail(reason="unimplemented feature") def test_capture_scope_cache(self, testdir): p = testdir.makepyfile(""" import sys def setup_module(func): print ("module-setup") def setup_function(func): print ("function-setup") def test_func(): print ("in function") assert 0 def teardown_function(func): print ("in teardown") """) result = testdir.runpytest(p) result.stdout.fnmatch_lines([ "*test_func():*", "*Captured stdout during setup*", "module-setup*", "function-setup*", "*Captured stdout*", "in teardown*", ]) def test_no_carry_over(self, testdir): p = testdir.makepyfile(""" def test_func1(): print ("in func1") def test_func2(): print ("in func2") assert 0 """) result = testdir.runpytest(p) s = result.stdout.str() assert "in func1" not in s assert "in func2" in s def test_teardown_capturing(self, testdir): p = testdir.makepyfile(""" def setup_function(function): print ("setup func1") def teardown_function(function): print ("teardown func1") assert 0 def test_func1(): print ("in func1") pass """) result = testdir.runpytest(p) result.stdout.fnmatch_lines([ '*teardown_function*', '*Captured stdout*', "setup func1*", "in func1*", "teardown func1*", # "*1 fixture failure*" ]) def test_teardown_capturing_final(self, testdir): p = testdir.makepyfile(""" def teardown_module(mod): print ("teardown module") assert 0 def test_func(): pass """) result = testdir.runpytest(p) result.stdout.fnmatch_lines([ "*def teardown_module(mod):*", "*Captured stdout*", "*teardown module*", "*1 error*", ]) def test_capturing_outerr(self, testdir): p1 = testdir.makepyfile(""" import sys def test_capturing(): print (42) sys.stderr.write(str(23)) def test_capturing_error(): print (1) sys.stderr.write(str(2)) raise ValueError """) result = testdir.runpytest(p1) result.stdout.fnmatch_lines([ "*test_capturing_outerr.py .F*", "====* FAILURES *====", "____*____", "*test_capturing_outerr.py:8: ValueError", "*--- Captured stdout *call*", "1", "*--- Captured stderr *call*", "2", ]) class TestLoggingInteraction(object): def test_logging_stream_ownership(self, testdir): p = testdir.makepyfile(""" def test_logging(): import logging import pytest stream = capture.CaptureIO() logging.basicConfig(stream=stream) stream.close() # to free memory/release resources """) result = testdir.runpytest_subprocess(p) assert result.stderr.str().find("atexit") == -1 def test_logging_and_immediate_setupteardown(self, testdir): p = testdir.makepyfile(""" import logging def setup_function(function): logging.warn("hello1") def test_logging(): logging.warn("hello2") assert 0 def teardown_function(function): logging.warn("hello3") assert 0 """) for optargs in (('--capture=sys',), ('--capture=fd',)): print(optargs) result = testdir.runpytest_subprocess(p, *optargs) s = result.stdout.str() result.stdout.fnmatch_lines([ "*WARN*hello3", # errors show first! "*WARN*hello1", "*WARN*hello2", ]) # verify proper termination assert "closed" not in s def test_logging_and_crossscope_fixtures(self, testdir): p = testdir.makepyfile(""" import logging def setup_module(function): logging.warn("hello1") def test_logging(): logging.warn("hello2") assert 0 def teardown_module(function): logging.warn("hello3") assert 0 """) for optargs in (('--capture=sys',), ('--capture=fd',)): print(optargs) result = testdir.runpytest_subprocess(p, *optargs) s = result.stdout.str() result.stdout.fnmatch_lines([ "*WARN*hello3", # errors come first "*WARN*hello1", "*WARN*hello2", ]) # verify proper termination assert "closed" not in s def test_conftestlogging_is_shown(self, testdir): testdir.makeconftest(""" import logging logging.basicConfig() logging.warn("hello435") """) # make sure that logging is still captured in tests result = testdir.runpytest_subprocess("-s", "-p", "no:capturelog") assert result.ret == EXIT_NOTESTSCOLLECTED result.stderr.fnmatch_lines([ "WARNING*hello435*", ]) assert 'operation on closed file' not in result.stderr.str() def test_conftestlogging_and_test_logging(self, testdir): testdir.makeconftest(""" import logging logging.basicConfig() """) # make sure that logging is still captured in tests p = testdir.makepyfile(""" def test_hello(): import logging logging.warn("hello433") assert 0 """) result = testdir.runpytest_subprocess(p, "-p", "no:capturelog") assert result.ret != 0 result.stdout.fnmatch_lines([ "WARNING*hello433*", ]) assert 'something' not in result.stderr.str() assert 'operation on closed file' not in result.stderr.str() class TestCaptureFixture(object): @pytest.mark.parametrize("opt", [[], ["-s"]]) def test_std_functional(self, testdir, opt): reprec = testdir.inline_runsource(""" def test_hello(capsys): print (42) out, err = capsys.readouterr() assert out.startswith("42") """, *opt) reprec.assertoutcome(passed=1) def test_capsyscapfd(self, testdir): p = testdir.makepyfile(""" def test_one(capsys, capfd): pass def test_two(capfd, capsys): pass """) result = testdir.runpytest(p) result.stdout.fnmatch_lines([ "*ERROR*setup*test_one*", "E*capfd*capsys*same*time*", "*ERROR*setup*test_two*", "E*capsys*capfd*same*time*", "*2 error*"]) def test_capturing_getfixturevalue(self, testdir): """Test that asking for "capfd" and "capsys" using request.getfixturevalue in the same test is an error. """ testdir.makepyfile(""" def test_one(capsys, request): request.getfixturevalue("capfd") def test_two(capfd, request): request.getfixturevalue("capsys") """) result = testdir.runpytest() result.stdout.fnmatch_lines([ "*test_one*", "*capsys*capfd*same*time*", "*test_two*", "*capfd*capsys*same*time*", "*2 failed in*", ]) def test_capsyscapfdbinary(self, testdir): p = testdir.makepyfile(""" def test_one(capsys, capfdbinary): pass """) result = testdir.runpytest(p) result.stdout.fnmatch_lines([ "*ERROR*setup*test_one*", "E*capfdbinary*capsys*same*time*", "*1 error*"]) @pytest.mark.parametrize("method", ["sys", "fd"]) def test_capture_is_represented_on_failure_issue128(self, testdir, method): p = testdir.makepyfile(""" def test_hello(cap%s): print ("xxx42xxx") assert 0 """ % method) result = testdir.runpytest(p) result.stdout.fnmatch_lines([ "xxx42xxx", ]) @needsosdup def test_stdfd_functional(self, testdir): reprec = testdir.inline_runsource(""" def test_hello(capfd): import os os.write(1, "42".encode('ascii')) out, err = capfd.readouterr() assert out.startswith("42") capfd.close() """) reprec.assertoutcome(passed=1) @needsosdup def test_capfdbinary(self, testdir): reprec = testdir.inline_runsource(""" def test_hello(capfdbinary): import os # some likely un-decodable bytes os.write(1, b'\\xfe\\x98\\x20') out, err = capfdbinary.readouterr() assert out == b'\\xfe\\x98\\x20' assert err == b'' """) reprec.assertoutcome(passed=1) @pytest.mark.skipif( sys.version_info < (3,), reason='only have capsysbinary in python 3', ) def test_capsysbinary(self, testdir): reprec = testdir.inline_runsource(""" def test_hello(capsysbinary): import sys # some likely un-decodable bytes sys.stdout.buffer.write(b'\\xfe\\x98\\x20') out, err = capsysbinary.readouterr() assert out == b'\\xfe\\x98\\x20' assert err == b'' """) reprec.assertoutcome(passed=1) @pytest.mark.skipif( sys.version_info >= (3,), reason='only have capsysbinary in python 3', ) def test_capsysbinary_forbidden_in_python2(self, testdir): testdir.makepyfile(""" def test_hello(capsysbinary): pass """) result = testdir.runpytest() result.stdout.fnmatch_lines([ "*test_hello*", "*capsysbinary is only supported on python 3*", "*1 error in*", ]) def test_partial_setup_failure(self, testdir): p = testdir.makepyfile(""" def test_hello(capsys, missingarg): pass """) result = testdir.runpytest(p) result.stdout.fnmatch_lines([ "*test_partial_setup_failure*", "*1 error*", ]) @needsosdup def test_keyboardinterrupt_disables_capturing(self, testdir): p = testdir.makepyfile(""" def test_hello(capfd): import os os.write(1, str(42).encode('ascii')) raise KeyboardInterrupt() """) result = testdir.runpytest_subprocess(p) result.stdout.fnmatch_lines([ "*KeyboardInterrupt*" ]) assert result.ret == 2 @pytest.mark.issue14 def test_capture_and_logging(self, testdir): p = testdir.makepyfile(""" import logging def test_log(capsys): logging.error('x') """) result = testdir.runpytest_subprocess(p) assert 'closed' not in result.stderr.str() @pytest.mark.parametrize('fixture', ['capsys', 'capfd']) @pytest.mark.parametrize('no_capture', [True, False]) def test_disabled_capture_fixture(self, testdir, fixture, no_capture): testdir.makepyfile(""" def test_disabled({fixture}): print('captured before') with {fixture}.disabled(): print('while capture is disabled') print('captured after') assert {fixture}.readouterr() == ('captured before\\ncaptured after\\n', '') def test_normal(): print('test_normal executed') """.format(fixture=fixture)) args = ('-s',) if no_capture else () result = testdir.runpytest_subprocess(*args) result.stdout.fnmatch_lines(""" *while capture is disabled* """) assert 'captured before' not in result.stdout.str() assert 'captured after' not in result.stdout.str() if no_capture: assert 'test_normal executed' in result.stdout.str() else: assert 'test_normal executed' not in result.stdout.str() @pytest.mark.parametrize('fixture', ['capsys', 'capfd']) def test_fixture_use_by_other_fixtures(self, testdir, fixture): """ Ensure that capsys and capfd can be used by other fixtures during setup and teardown. """ testdir.makepyfile(""" from __future__ import print_function import sys import pytest @pytest.fixture def captured_print({fixture}): print('stdout contents begin') print('stderr contents begin', file=sys.stderr) out, err = {fixture}.readouterr() yield out, err print('stdout contents end') print('stderr contents end', file=sys.stderr) out, err = {fixture}.readouterr() assert out == 'stdout contents end\\n' assert err == 'stderr contents end\\n' def test_captured_print(captured_print): out, err = captured_print assert out == 'stdout contents begin\\n' assert err == 'stderr contents begin\\n' """.format(fixture=fixture)) result = testdir.runpytest_subprocess() result.stdout.fnmatch_lines("*1 passed*") assert 'stdout contents begin' not in result.stdout.str() assert 'stderr contents begin' not in result.stdout.str() def test_setup_failure_does_not_kill_capturing(testdir): sub1 = testdir.mkpydir("sub1") sub1.join("conftest.py").write(_pytest._code.Source(""" def pytest_runtest_setup(item): raise ValueError(42) """)) sub1.join("test_mod.py").write("def test_func1(): pass") result = testdir.runpytest(testdir.tmpdir, '--traceconfig') result.stdout.fnmatch_lines([ "*ValueError(42)*", "*1 error*" ]) def test_fdfuncarg_skips_on_no_osdup(testdir): testdir.makepyfile(""" import os if hasattr(os, 'dup'): del os.dup def test_hello(capfd): pass """) result = testdir.runpytest_subprocess("--capture=no") result.stdout.fnmatch_lines([ "*1 skipped*" ]) def test_capture_conftest_runtest_setup(testdir): testdir.makeconftest(""" def pytest_runtest_setup(): print ("hello19") """) testdir.makepyfile("def test_func(): pass") result = testdir.runpytest() assert result.ret == 0 assert 'hello19' not in result.stdout.str() def test_capture_badoutput_issue412(testdir): testdir.makepyfile(""" import os def test_func(): omg = bytearray([1,129,1]) os.write(1, omg) assert 0 """) result = testdir.runpytest('--cap=fd') result.stdout.fnmatch_lines(''' *def test_func* *assert 0* *Captured* *1 failed* ''') def test_capture_early_option_parsing(testdir): testdir.makeconftest(""" def pytest_runtest_setup(): print ("hello19") """) testdir.makepyfile("def test_func(): pass") result = testdir.runpytest("-vs") assert result.ret == 0 assert 'hello19' in result.stdout.str() def test_capture_binary_output(testdir): testdir.makepyfile(r""" import pytest def test_a(): import sys import subprocess subprocess.call([sys.executable, __file__]) def test_foo(): import os;os.write(1, b'\xc3') if __name__ == '__main__': test_foo() """) result = testdir.runpytest('--assert=plain') result.assert_outcomes(passed=2) def test_error_during_readouterr(testdir): """Make sure we suspend capturing if errors occur during readouterr""" testdir.makepyfile(pytest_xyz=""" from _pytest.capture import FDCapture def bad_snap(self): raise Exception('boom') assert FDCapture.snap FDCapture.snap = bad_snap """) result = testdir.runpytest_subprocess( "-p", "pytest_xyz", "--version", syspathinsert=True ) result.stderr.fnmatch_lines([ "*in bad_snap", " raise Exception('boom')", "Exception: boom", ]) class TestCaptureIO(object): def test_text(self): f = capture.CaptureIO() f.write("hello") s = f.getvalue() assert s == "hello" f.close() def test_unicode_and_str_mixture(self): f = capture.CaptureIO() if sys.version_info >= (3, 0): f.write("\u00f6") pytest.raises(TypeError, "f.write(bytes('hello', 'UTF-8'))") else: f.write(unicode("\u00f6", 'UTF-8')) f.write("hello") # bytes s = f.getvalue() f.close() assert isinstance(s, unicode) @pytest.mark.skipif( sys.version_info[0] == 2, reason='python 3 only behaviour', ) def test_write_bytes_to_buffer(self): """In python3, stdout / stderr are text io wrappers (exposing a buffer property of the underlying bytestream). See issue #1407 """ f = capture.CaptureIO() f.buffer.write(b'foo\r\n') assert f.getvalue() == 'foo\r\n' def test_bytes_io(): f = py.io.BytesIO() f.write(tobytes("hello")) pytest.raises(TypeError, "f.write(totext('hello'))") s = f.getvalue() assert s == tobytes("hello") def test_dontreadfrominput(): from _pytest.capture import DontReadFromInput f = DontReadFromInput() assert not f.isatty() pytest.raises(IOError, f.read) pytest.raises(IOError, f.readlines) iter_f = iter(f) pytest.raises(IOError, next, iter_f) pytest.raises(UnsupportedOperation, f.fileno) f.close() # just for completeness @pytest.mark.skipif('sys.version_info < (3,)', reason='python2 has no buffer') def test_dontreadfrominput_buffer_python3(): from _pytest.capture import DontReadFromInput f = DontReadFromInput() fb = f.buffer assert not fb.isatty() pytest.raises(IOError, fb.read) pytest.raises(IOError, fb.readlines) iter_f = iter(f) pytest.raises(IOError, next, iter_f) pytest.raises(ValueError, fb.fileno) f.close() # just for completeness @pytest.mark.skipif('sys.version_info >= (3,)', reason='python2 has no buffer') def test_dontreadfrominput_buffer_python2(): from _pytest.capture import DontReadFromInput f = DontReadFromInput() with pytest.raises(AttributeError): f.buffer f.close() # just for completeness @pytest.yield_fixture def tmpfile(testdir): f = testdir.makepyfile("").open('wb+') yield f if not f.closed: f.close() @needsosdup def test_dupfile(tmpfile): flist = [] for i in range(5): nf = capture.safe_text_dupfile(tmpfile, "wb") assert nf != tmpfile assert nf.fileno() != tmpfile.fileno() assert nf not in flist print(i, end="", file=nf) flist.append(nf) fname_open = flist[0].name assert fname_open == repr(flist[0].buffer) for i in range(5): f = flist[i] f.close() fname_closed = flist[0].name assert fname_closed == repr(flist[0].buffer) assert fname_closed != fname_open tmpfile.seek(0) s = tmpfile.read() assert "01234" in repr(s) tmpfile.close() assert fname_closed == repr(flist[0].buffer) def test_dupfile_on_bytesio(): io = py.io.BytesIO() f = capture.safe_text_dupfile(io, "wb") f.write("hello") assert io.getvalue() == b"hello" assert 'BytesIO object' in f.name def test_dupfile_on_textio(): io = py.io.TextIO() f = capture.safe_text_dupfile(io, "wb") f.write("hello") assert io.getvalue() == "hello" assert not hasattr(f, 'name') @contextlib.contextmanager def lsof_check(): pid = os.getpid() try: out = py.process.cmdexec("lsof -p %d" % pid) except (py.process.cmdexec.Error, UnicodeDecodeError): # about UnicodeDecodeError, see note on pytester pytest.skip("could not run 'lsof'") yield out2 = py.process.cmdexec("lsof -p %d" % pid) len1 = len([x for x in out.split("\n") if "REG" in x]) len2 = len([x for x in out2.split("\n") if "REG" in x]) assert len2 < len1 + 3, out2 class TestFDCapture(object): pytestmark = needsosdup def test_simple(self, tmpfile): fd = tmpfile.fileno() cap = capture.FDCapture(fd) data = tobytes("hello") os.write(fd, data) s = cap.snap() cap.done() assert not s cap = capture.FDCapture(fd) cap.start() os.write(fd, data) s = cap.snap() cap.done() assert s == "hello" def test_simple_many(self, tmpfile): for i in range(10): self.test_simple(tmpfile) def test_simple_many_check_open_files(self, testdir): with lsof_check(): with testdir.makepyfile("").open('wb+') as tmpfile: self.test_simple_many(tmpfile) def test_simple_fail_second_start(self, tmpfile): fd = tmpfile.fileno() cap = capture.FDCapture(fd) cap.done() pytest.raises(ValueError, cap.start) def test_stderr(self): cap = capture.FDCapture(2) cap.start() print("hello", file=sys.stderr) s = cap.snap() cap.done() assert s == "hello\n" def test_stdin(self, tmpfile): cap = capture.FDCapture(0) cap.start() x = os.read(0, 100).strip() cap.done() assert x == tobytes('') def test_writeorg(self, tmpfile): data1, data2 = tobytes("foo"), tobytes("bar") cap = capture.FDCapture(tmpfile.fileno()) cap.start() tmpfile.write(data1) tmpfile.flush() cap.writeorg(data2) scap = cap.snap() cap.done() assert scap == totext(data1) with open(tmpfile.name, 'rb') as stmp_file: stmp = stmp_file.read() assert stmp == data2 def test_simple_resume_suspend(self, tmpfile): with saved_fd(1): cap = capture.FDCapture(1) cap.start() data = tobytes("hello") os.write(1, data) sys.stdout.write("whatever") s = cap.snap() assert s == "hellowhatever" cap.suspend() os.write(1, tobytes("world")) sys.stdout.write("qlwkej") assert not cap.snap() cap.resume() os.write(1, tobytes("but now")) sys.stdout.write(" yes\n") s = cap.snap() assert s == "but now yes\n" cap.suspend() cap.done() pytest.raises(AttributeError, cap.suspend) @contextlib.contextmanager def saved_fd(fd): new_fd = os.dup(fd) try: yield finally: os.dup2(new_fd, fd) os.close(new_fd) class TestStdCapture(object): captureclass = staticmethod(StdCapture) @contextlib.contextmanager def getcapture(self, **kw): cap = self.__class__.captureclass(**kw) cap.start_capturing() try: yield cap finally: cap.stop_capturing() def test_capturing_done_simple(self): with self.getcapture() as cap: sys.stdout.write("hello") sys.stderr.write("world") out, err = cap.readouterr() assert out == "hello" assert err == "world" def test_capturing_reset_simple(self): with self.getcapture() as cap: print("hello world") sys.stderr.write("hello error\n") out, err = cap.readouterr() assert out == "hello world\n" assert err == "hello error\n" def test_capturing_readouterr(self): with self.getcapture() as cap: print("hello world") sys.stderr.write("hello error\n") out, err = cap.readouterr() assert out == "hello world\n" assert err == "hello error\n" sys.stderr.write("error2") out, err = cap.readouterr() assert err == "error2" def test_capture_results_accessible_by_attribute(self): with self.getcapture() as cap: sys.stdout.write("hello") sys.stderr.write("world") capture_result = cap.readouterr() assert capture_result.out == "hello" assert capture_result.err == "world" def test_capturing_readouterr_unicode(self): with self.getcapture() as cap: print("hx\xc4\x85\xc4\x87") out, err = cap.readouterr() assert out == py.builtin._totext("hx\xc4\x85\xc4\x87\n", "utf8") @pytest.mark.skipif('sys.version_info >= (3,)', reason='text output different for bytes on python3') def test_capturing_readouterr_decode_error_handling(self): with self.getcapture() as cap: # triggered an internal error in pytest print('\xa6') out, err = cap.readouterr() assert out == py.builtin._totext('\ufffd\n', 'unicode-escape') def test_reset_twice_error(self): with self.getcapture() as cap: print("hello") out, err = cap.readouterr() pytest.raises(ValueError, cap.stop_capturing) assert out == "hello\n" assert not err def test_capturing_modify_sysouterr_in_between(self): oldout = sys.stdout olderr = sys.stderr with self.getcapture() as cap: sys.stdout.write("hello") sys.stderr.write("world") sys.stdout = capture.CaptureIO() sys.stderr = capture.CaptureIO() print("not seen") sys.stderr.write("not seen\n") out, err = cap.readouterr() assert out == "hello" assert err == "world" assert sys.stdout == oldout assert sys.stderr == olderr def test_capturing_error_recursive(self): with self.getcapture() as cap1: print("cap1") with self.getcapture() as cap2: print("cap2") out2, err2 = cap2.readouterr() out1, err1 = cap1.readouterr() assert out1 == "cap1\n" assert out2 == "cap2\n" def test_just_out_capture(self): with self.getcapture(out=True, err=False) as cap: sys.stdout.write("hello") sys.stderr.write("world") out, err = cap.readouterr() assert out == "hello" assert not err def test_just_err_capture(self): with self.getcapture(out=False, err=True) as cap: sys.stdout.write("hello") sys.stderr.write("world") out, err = cap.readouterr() assert err == "world" assert not out def test_stdin_restored(self): old = sys.stdin with self.getcapture(in_=True): newstdin = sys.stdin assert newstdin != sys.stdin assert sys.stdin is old def test_stdin_nulled_by_default(self): print("XXX this test may well hang instead of crashing") print("XXX which indicates an error in the underlying capturing") print("XXX mechanisms") with self.getcapture(): pytest.raises(IOError, "sys.stdin.read()") class TestStdCaptureFD(TestStdCapture): pytestmark = needsosdup captureclass = staticmethod(StdCaptureFD) def test_simple_only_fd(self, testdir): testdir.makepyfile(""" import os def test_x(): os.write(1, "hello\\n".encode("ascii")) assert 0 """) result = testdir.runpytest_subprocess() result.stdout.fnmatch_lines(""" *test_x* *assert 0* *Captured stdout* """) def test_intermingling(self): with self.getcapture() as cap: oswritebytes(1, "1") sys.stdout.write(str(2)) sys.stdout.flush() oswritebytes(1, "3") oswritebytes(2, "a") sys.stderr.write("b") sys.stderr.flush() oswritebytes(2, "c") out, err = cap.readouterr() assert out == "123" assert err == "abc" def test_many(self, capfd): with lsof_check(): for i in range(10): cap = StdCaptureFD() cap.stop_capturing() class TestStdCaptureFDinvalidFD(object): pytestmark = needsosdup def test_stdcapture_fd_invalid_fd(self, testdir): testdir.makepyfile(""" import os from _pytest import capture def StdCaptureFD(out=True, err=True, in_=True): return capture.MultiCapture(out, err, in_, Capture=capture.FDCapture) def test_stdout(): os.close(1) cap = StdCaptureFD(out=True, err=False, in_=False) cap.stop_capturing() def test_stderr(): os.close(2) cap = StdCaptureFD(out=False, err=True, in_=False) cap.stop_capturing() def test_stdin(): os.close(0) cap = StdCaptureFD(out=False, err=False, in_=True) cap.stop_capturing() """) result = testdir.runpytest_subprocess("--capture=fd") assert result.ret == 0 assert result.parseoutcomes()['passed'] == 3 def test_capture_not_started_but_reset(): capsys = StdCapture() capsys.stop_capturing() def test_using_capsys_fixture_works_with_sys_stdout_encoding(capsys): test_text = 'test text' print(test_text.encode(sys.stdout.encoding, 'replace')) (out, err) = capsys.readouterr() assert out assert err == '' def test_capsys_results_accessible_by_attribute(capsys): sys.stdout.write("spam") sys.stderr.write("eggs") capture_result = capsys.readouterr() assert capture_result.out == "spam" assert capture_result.err == "eggs" @needsosdup @pytest.mark.parametrize('use', [True, False]) def test_fdcapture_tmpfile_remains_the_same(tmpfile, use): if not use: tmpfile = True cap = StdCaptureFD(out=False, err=tmpfile) try: cap.start_capturing() capfile = cap.err.tmpfile cap.readouterr() finally: cap.stop_capturing() capfile2 = cap.err.tmpfile assert capfile2 == capfile @needsosdup def test_close_and_capture_again(testdir): testdir.makepyfile(""" import os def test_close(): os.close(1) def test_capture_again(): os.write(1, b"hello\\n") assert 0 """) result = testdir.runpytest_subprocess() result.stdout.fnmatch_lines(""" *test_capture_again* *assert 0* *stdout* *hello* """) @pytest.mark.parametrize('method', ['SysCapture', 'FDCapture']) def test_capturing_and_logging_fundamentals(testdir, method): if method == "StdCaptureFD" and not hasattr(os, 'dup'): pytest.skip("need os.dup") # here we check a fundamental feature p = testdir.makepyfile(""" import sys, os import py, logging from _pytest import capture cap = capture.MultiCapture(out=False, in_=False, Capture=capture.%s) cap.start_capturing() logging.warn("hello1") outerr = cap.readouterr() print ("suspend, captured %%s" %%(outerr,)) logging.warn("hello2") cap.pop_outerr_to_orig() logging.warn("hello3") outerr = cap.readouterr() print ("suspend2, captured %%s" %% (outerr,)) """ % (method,)) result = testdir.runpython(p) result.stdout.fnmatch_lines(""" suspend, captured*hello1* suspend2, captured*WARNING:root:hello3* """) result.stderr.fnmatch_lines(""" WARNING:root:hello2 """) assert "atexit" not in result.stderr.str() def test_error_attribute_issue555(testdir): testdir.makepyfile(""" import sys def test_capattr(): assert sys.stdout.errors == "strict" assert sys.stderr.errors == "strict" """) reprec = testdir.inline_run() reprec.assertoutcome(passed=1) @pytest.mark.skipif(not sys.platform.startswith('win') and sys.version_info[:2] >= (3, 6), reason='only py3.6+ on windows') def test_py36_windowsconsoleio_workaround_non_standard_streams(): """ Ensure _py36_windowsconsoleio_workaround function works with objects that do not implement the full ``io``-based stream protocol, for example execnet channels (#2666). """ from _pytest.capture import _py36_windowsconsoleio_workaround class DummyStream(object): def write(self, s): pass stream = DummyStream() _py36_windowsconsoleio_workaround(stream) def test_dontreadfrominput_has_encoding(testdir): testdir.makepyfile(""" import sys def test_capattr(): # should not raise AttributeError assert sys.stdout.encoding assert sys.stderr.encoding """) reprec = testdir.inline_run() reprec.assertoutcome(passed=1) def test_crash_on_closing_tmpfile_py27(testdir): testdir.makepyfile(''' from __future__ import print_function import time import threading import sys def spam(): f = sys.stderr while True: print('.', end='', file=f) def test_silly(): t = threading.Thread(target=spam) t.daemon = True t.start() time.sleep(0.5) ''') result = testdir.runpytest_subprocess() assert result.ret == 0 assert 'IOError' not in result.stdout.str() def test_pickling_and_unpickling_encoded_file(): # See https://bitbucket.org/pytest-dev/pytest/pull-request/194 # pickle.loads() raises infinite recursion if # EncodedFile.__getattr__ is not implemented properly ef = capture.EncodedFile(None, None) ef_as_str = pickle.dumps(ef) pickle.loads(ef_as_str)