<s> """<STR_LIT>""" <EOL> from __future__ import with_statement <EOL> import sys <EOL> import os <EOL> from tempfile import TemporaryFile <EOL> import py <EOL> import pytest <EOL> from py . io import TextIO <EOL> unicode = py . builtin . text <EOL> patchsysdict = { <NUM_LIT:0> : '<STR_LIT>' , <NUM_LIT:1> : '<STR_LIT>' , <NUM_LIT:2> : '<STR_LIT>' } <EOL> def pytest_addoption ( parser ) : <EOL> group = parser . getgroup ( "<STR_LIT>" ) <EOL> group . _addoption ( <EOL> '<STR_LIT>' , action = "<STR_LIT:store>" , <EOL> default = "<STR_LIT>" if hasattr ( os , "<STR_LIT>" ) else "<STR_LIT>" , <EOL> metavar = "<STR_LIT>" , choices = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> help = "<STR_LIT>" ) <EOL> group . _addoption ( <EOL> '<STR_LIT>' , action = "<STR_LIT>" , const = "<STR_LIT>" , dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> @ pytest . hookimpl ( hookwrapper = True ) <EOL> def pytest_load_initial_conftests ( early_config , parser , args ) : <EOL> _readline_workaround ( ) <EOL> ns = early_config . known_args_namespace <EOL> pluginmanager = early_config . pluginmanager <EOL> capman = CaptureManager ( ns . capture ) <EOL> pluginmanager . register ( capman , "<STR_LIT>" ) <EOL> early_config . add_cleanup ( capman . reset_capturings ) <EOL> def silence_logging_at_shutdown ( ) : <EOL> if "<STR_LIT>" in sys . modules : <EOL> sys . modules [ "<STR_LIT>" ] . raiseExceptions = False <EOL> early_config . add_cleanup ( silence_logging_at_shutdown ) <EOL> capman . init_capturings ( ) <EOL> outcome = yield <EOL> out , err = capman . suspendcapture ( ) <EOL> if outcome . excinfo is not None : <EOL> sys . stdout . write ( out ) <EOL> sys . stderr . write ( err ) <EOL> class CaptureManager : <EOL> def __init__ ( self , method ) : <EOL> self . _method = method <EOL> def _getcapture ( self , method ) : <EOL> if method == "<STR_LIT>" : <EOL> return MultiCapture ( out = True , err = True , Capture = FDCapture ) <EOL> elif method == "<STR_LIT>" : <EOL> return MultiCapture ( out = True , err = True , Capture = SysCapture ) <EOL> elif method == "<STR_LIT>" : <EOL> return MultiCapture ( out = False , err = False , in_ = False ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % method ) <EOL> def init_capturings ( self ) : <EOL> assert not hasattr ( self , "<STR_LIT>" ) <EOL> self . _capturing = self . _getcapture ( self . _method ) <EOL> self . _capturing . start_capturing ( ) <EOL> def reset_capturings ( self ) : <EOL> cap = self . __dict__ . pop ( "<STR_LIT>" , None ) <EOL> if cap is not None : <EOL> cap . pop_outerr_to_orig ( ) <EOL> cap . stop_capturing ( ) <EOL> def resumecapture ( self ) : <EOL> self . _capturing . resume_capturing ( ) <EOL> def suspendcapture ( self , in_ = False ) : <EOL> self . deactivate_funcargs ( ) <EOL> cap = getattr ( self , "<STR_LIT>" , None ) <EOL> if cap is not None : <EOL> try : <EOL> outerr = cap . readouterr ( ) <EOL> finally : <EOL> cap . suspend_capturing ( in_ = in_ ) <EOL> return outerr <EOL> def activate_funcargs ( self , pyfuncitem ) : <EOL> capfuncarg = pyfuncitem . __dict__ . pop ( "<STR_LIT>" , None ) <EOL> if capfuncarg is not None : <EOL> capfuncarg . _start ( ) <EOL> self . _capfuncarg = capfuncarg <EOL> def deactivate_funcargs ( self ) : <EOL> capfuncarg = self . __dict__ . pop ( "<STR_LIT>" , None ) <EOL> if capfuncarg is not None : <EOL> capfuncarg . close ( ) <EOL> @ pytest . hookimpl ( hookwrapper = True ) <EOL> def pytest_make_collect_report ( self , collector ) : <EOL> if isinstance ( collector , pytest . File ) : <EOL> self . resumecapture ( ) <EOL> outcome = yield <EOL> out , err = self . suspendcapture ( ) <EOL> rep = outcome . get_result ( ) <EOL> if out : <EOL> rep . sections . append ( ( "<STR_LIT>" , out ) ) <EOL> if err : <EOL> rep . sections . append ( ( "<STR_LIT>" , err ) ) <EOL> else : <EOL> yield <EOL> @ pytest . hookimpl ( hookwrapper = True ) <EOL> def pytest_runtest_setup ( self , item ) : <EOL> self . resumecapture ( ) <EOL> yield <EOL> self . suspendcapture_item ( item , "<STR_LIT>" ) <EOL> @ pytest . hookimpl ( hookwrapper = True ) <EOL> def pytest_runtest_call ( self , item ) : <EOL> self . resumecapture ( ) <EOL> self . activate_funcargs ( item ) <EOL> yield <EOL> self . suspendcapture_item ( item , "<STR_LIT>" ) <EOL> @ pytest . hookimpl ( hookwrapper = True ) <EOL> def pytest_runtest_teardown ( self , item ) : <EOL> self . resumecapture ( ) <EOL> yield <EOL> self . suspendcapture_item ( item , "<STR_LIT>" ) <EOL> @ pytest . hookimpl ( tryfirst = True ) <EOL> def pytest_keyboard_interrupt ( self , excinfo ) : <EOL> self . reset_capturings ( ) <EOL> @ pytest . hookimpl ( tryfirst = True ) <EOL> def pytest_internalerror ( self , excinfo ) : <EOL> self . reset_capturings ( ) <EOL> def suspendcapture_item ( self , item , when ) : <EOL> out , err = self . suspendcapture ( ) <EOL> item . add_report_section ( when , "<STR_LIT>" , out ) <EOL> item . add_report_section ( when , "<STR_LIT>" , err ) <EOL> error_capsysfderror = "<STR_LIT>" <EOL> @ pytest . fixture <EOL> def capsys ( request ) : <EOL> """<STR_LIT>""" <EOL> if "<STR_LIT>" in request . _funcargs : <EOL> raise request . raiseerror ( error_capsysfderror ) <EOL> request . node . _capfuncarg = c = CaptureFixture ( SysCapture ) <EOL> return c <EOL> @ pytest . fixture <EOL> def capfd ( request ) : <EOL> """<STR_LIT>""" <EOL> if "<STR_LIT>" in request . _funcargs : <EOL> request . raiseerror ( error_capsysfderror ) <EOL> if not hasattr ( os , '<STR_LIT>' ) : <EOL> pytest . skip ( "<STR_LIT>" ) <EOL> request . node . _capfuncarg = c = CaptureFixture ( FDCapture ) <EOL> return c <EOL> class CaptureFixture : <EOL> def __init__ ( self , captureclass ) : <EOL> self . captureclass = captureclass <EOL> def _start ( self ) : <EOL> self . _capture = MultiCapture ( out = True , err = True , in_ = False , <EOL> Capture = self . captureclass ) <EOL> self . _capture . start_capturing ( ) <EOL> def close ( self ) : <EOL> cap = self . __dict__ . pop ( "<STR_LIT>" , None ) <EOL> if cap is not None : <EOL> self . _outerr = cap . pop_outerr_to_orig ( ) <EOL> cap . stop_capturing ( ) <EOL> def readouterr ( self ) : <EOL> try : <EOL> return self . _capture . readouterr ( ) <EOL> except AttributeError : <EOL> return self . _outerr <EOL> def safe_text_dupfile ( f , mode , default_encoding = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> encoding = getattr ( f , "<STR_LIT>" , None ) <EOL> try : <EOL> fd = f . fileno ( ) <EOL> except Exception : <EOL> if "<STR_LIT:b>" not in getattr ( f , "<STR_LIT>" , "<STR_LIT>" ) and hasattr ( f , "<STR_LIT>" ) : <EOL> return f <EOL> else : <EOL> newfd = os . dup ( fd ) <EOL> if "<STR_LIT:b>" not in mode : <EOL> mode += "<STR_LIT:b>" <EOL> f = os . fdopen ( newfd , mode , <NUM_LIT:0> ) <EOL> return EncodedFile ( f , encoding or default_encoding ) <EOL> class EncodedFile ( object ) : <EOL> errors = "<STR_LIT:strict>" <EOL> def __init__ ( self , buffer , encoding ) : <EOL> self . buffer = buffer <EOL> self . encoding = encoding <EOL> def write ( self , obj ) : <EOL> if isinstance ( obj , unicode ) : <EOL> obj = obj . encode ( self . encoding , "<STR_LIT:replace>" ) <EOL> self . buffer . write ( obj ) <EOL> def writelines ( self , linelist ) : <EOL> data = '<STR_LIT>' . join ( linelist ) <EOL> self . write ( data ) <EOL> def __getattr__ ( self , name ) : <EOL> return getattr ( object . __getattribute__ ( self , "<STR_LIT>" ) , name ) <EOL> class MultiCapture ( object ) : <EOL> out = err = in_ = None <EOL> def __init__ ( self , out = True , err = True , in_ = True , Capture = None ) : <EOL> if in_ : <EOL> self . in_ = Capture ( <NUM_LIT:0> ) <EOL> if out : <EOL> self . out = Capture ( <NUM_LIT:1> ) <EOL> if err : <EOL> self . err = Capture ( <NUM_LIT:2> ) <EOL> def start_capturing ( self ) : <EOL> if self . in_ : <EOL> self . in_ . start ( ) <EOL> if self . out : <EOL> self . out . start ( ) <EOL> if self . err : <EOL> self . err . start ( ) <EOL> def pop_outerr_to_orig ( self ) : <EOL> """<STR_LIT>""" <EOL> out , err = self . readouterr ( ) <EOL> if out : <EOL> self . out . writeorg ( out ) <EOL> if err : <EOL> self . err . writeorg ( err ) <EOL> return out , err <EOL> def suspend_capturing ( self , in_ = False ) : <EOL> if self . out : <EOL> self . out . suspend ( ) <EOL> if self . err : <EOL> self . err . suspend ( ) <EOL> if in_ and self . in_ : <EOL> self . in_ . suspend ( ) <EOL> self . _in_suspended = True <EOL> def resume_capturing ( self ) : <EOL> if self . out : <EOL> self . out . resume ( ) <EOL> if self . err : <EOL> self . err . resume ( ) <EOL> if hasattr ( self , "<STR_LIT>" ) : <EOL> self . in_ . resume ( ) <EOL> del self . _in_suspended <EOL> def stop_capturing ( self ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . _reset = True <EOL> if self . out : <EOL> self . out . done ( ) <EOL> if self . err : <EOL> self . err . done ( ) <EOL> if self . in_ : <EOL> self . in_ . done ( ) <EOL> def readouterr ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( self . out . snap ( ) if self . out is not None else "<STR_LIT>" , <EOL> self . err . snap ( ) if self . err is not None else "<STR_LIT>" ) <EOL> class NoCapture : <EOL> __init__ = start = done = suspend = resume = lambda * args : None <EOL> class FDCapture : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , targetfd , tmpfile = None ) : <EOL> self . targetfd = targetfd <EOL> try : <EOL> self . targetfd_save = os . dup ( self . targetfd ) <EOL> except OSError : <EOL> self . start = lambda : None <EOL> self . done = lambda : None <EOL> else : <EOL> if targetfd == <NUM_LIT:0> : <EOL> assert not tmpfile , "<STR_LIT>" <EOL> tmpfile = open ( os . devnull , "<STR_LIT:r>" ) <EOL> self . syscapture = SysCapture ( targetfd ) <EOL> else : <EOL> if tmpfile is None : <EOL> f = TemporaryFile ( ) <EOL> with f : <EOL> tmpfile = safe_text_dupfile ( f , mode = "<STR_LIT>" ) <EOL> if targetfd in patchsysdict : <EOL> self . syscapture = SysCapture ( targetfd , tmpfile ) <EOL> else : <EOL> self . syscapture = NoCapture ( ) <EOL> self . tmpfile = tmpfile <EOL> self . tmpfile_fd = tmpfile . fileno ( ) <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( self . targetfd , self . targetfd_save ) <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> os . fstat ( self . targetfd_save ) <EOL> except ( AttributeError , OSError ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> os . dup2 ( self . tmpfile_fd , self . targetfd ) <EOL> self . syscapture . start ( ) <EOL> def snap ( self ) : <EOL> f = self . tmpfile <EOL> f . seek ( <NUM_LIT:0> ) <EOL> res = f . read ( ) <EOL> if res : <EOL> enc = getattr ( f , "<STR_LIT>" , None ) <EOL> if enc and isinstance ( res , bytes ) : <EOL> res = py . builtin . _totext ( res , enc , "<STR_LIT:replace>" ) <EOL> f . truncate ( <NUM_LIT:0> ) <EOL> f . seek ( <NUM_LIT:0> ) <EOL> return res <EOL> return '<STR_LIT>' <EOL> def done ( self ) : <EOL> """<STR_LIT>""" <EOL> targetfd_save = self . __dict__ . pop ( "<STR_LIT>" ) <EOL> os . dup2 ( targetfd_save , self . targetfd ) <EOL> os . close ( targetfd_save ) <EOL> self . syscapture . done ( ) <EOL> self . tmpfile . close ( ) <EOL> def suspend ( self ) : <EOL> self . syscapture . suspend ( ) <EOL> os . dup2 ( self . targetfd_save , self . targetfd ) <EOL> def resume ( self ) : <EOL> self . syscapture . resume ( ) <EOL> os . dup2 ( self . tmpfile_fd , self . targetfd ) <EOL> def writeorg ( self , data ) : <EOL> """<STR_LIT>""" <EOL> if py . builtin . _istext ( data ) : <EOL> data = data . encode ( "<STR_LIT:utf8>" ) <EOL> os . write ( self . targetfd_save , data ) <EOL> class SysCapture : <EOL> def __init__ ( self , fd , tmpfile = None ) : <EOL> name = patchsysdict [ fd ] <EOL> self . _old = getattr ( sys , name ) <EOL> self . name = name <EOL> if tmpfile is None : <EOL> if name == "<STR_LIT>" : <EOL> tmpfile = DontReadFromInput ( ) <EOL> else : <EOL> tmpfile = TextIO ( ) <EOL> self . tmpfile = tmpfile <EOL> def start ( self ) : <EOL> setattr ( sys , self . name , self . tmpfile ) <EOL> def snap ( self ) : <EOL> f = self . tmpfile <EOL> res = f . getvalue ( ) <EOL> f . truncate ( <NUM_LIT:0> ) <EOL> f . seek ( <NUM_LIT:0> ) <EOL> return res <EOL> def done ( self ) : <EOL> setattr ( sys , self . name , self . _old ) <EOL> del self . _old <EOL> self . tmpfile . close ( ) <EOL> def suspend ( self ) : <EOL> setattr ( sys , self . name , self . _old ) <EOL> def resume ( self ) : <EOL> setattr ( sys , self . name , self . tmpfile ) <EOL> def writeorg ( self , data ) : <EOL> self . _old . write ( data ) <EOL> self . _old . flush ( ) <EOL> class DontReadFromInput : <EOL> """<STR_LIT>""" <EOL> encoding = None <EOL> def read ( self , * args ) : <EOL> raise IOError ( "<STR_LIT>" ) <EOL> readline = read <EOL> readlines = read <EOL> __iter__ = read <EOL> def fileno ( self ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def isatty ( self ) : <EOL> return False <EOL> def close ( self ) : <EOL> pass <EOL> def _readline_workaround ( ) : <EOL> """<STR_LIT>""" <EOL> if not sys . platform . startswith ( '<STR_LIT:win32>' ) : <EOL> return <EOL> try : <EOL> import readline <EOL> except ImportError : <EOL> pass </s>
