<s> from collections import namedtuple <EOL> import os <EOL> import shutil <EOL> import subprocess <EOL> import sys <EOL> import tempfile <EOL> import textwrap <EOL> import psutil <EOL> import pytest <EOL> def pytest_runtest_setup ( item ) : <EOL> sudo_marker = item . get_marker ( '<STR_LIT>' ) <EOL> if sudo_marker is not None : <EOL> try : <EOL> with open ( os . devnull , '<STR_LIT:w>' ) as devnull : <EOL> subprocess . check_call ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:true>' ] , stdout = devnull , stderr = devnull ) <EOL> except subprocess . CalledProcessError : <EOL> pytest . skip ( '<STR_LIT>' ) <EOL> PyScriptResult = namedtuple ( <EOL> '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> class PyScript ( object ) : <EOL> def __init__ ( self , code , sudo = False ) : <EOL> self . dirname = os . path . realpath ( <EOL> tempfile . mkdtemp ( prefix = '<STR_LIT>' ) ) <EOL> os . chmod ( self . dirname , <NUM_LIT> ) <EOL> self . basename = '<STR_LIT>' <EOL> self . realpath = os . path . join ( self . dirname , self . basename ) <EOL> with open ( self . realpath , '<STR_LIT:wb>' ) as f : <EOL> f . write ( textwrap . dedent ( code . lstrip ( '<STR_LIT:\n>' ) ) . encode ( '<STR_LIT:utf-8>' ) ) <EOL> self . sudo = sudo <EOL> def run ( self , * args ) : <EOL> subenv = os . environ . copy ( ) <EOL> subenv [ '<STR_LIT>' ] = '<STR_LIT:x>' <EOL> base_command = [ sys . executable , self . realpath ] <EOL> if self . sudo : <EOL> base_command = [ '<STR_LIT>' , '<STR_LIT>' ] + base_command <EOL> proc = subprocess . Popen ( <EOL> base_command + list ( args ) , cwd = self . dirname , env = subenv , <EOL> stdout = subprocess . PIPE , stderr = subprocess . PIPE ) <EOL> stdout , stderr = proc . communicate ( ) <EOL> return PyScriptResult ( stdout , stderr , proc . pid , proc . returncode ) <EOL> def teardown ( self ) : <EOL> procs = [ ] <EOL> for proc in psutil . process_iter ( ) : <EOL> try : <EOL> if ( proc . exe ( ) == sys . executable and <EOL> self . realpath in proc . cmdline ( ) ) : <EOL> proc . terminate ( ) <EOL> procs . append ( proc ) <EOL> except ( psutil . NoSuchProcess , psutil . AccessDenied , OSError ) : <EOL> continue <EOL> if psutil . wait_procs ( procs , timeout = <NUM_LIT:1> ) [ <NUM_LIT:1> ] : <EOL> raise OSError ( '<STR_LIT>' ) <EOL> shutil . rmtree ( self . dirname ) <EOL> @ pytest . fixture ( scope = '<STR_LIT>' ) <EOL> def pyscript ( request ) : <EOL> pfs = [ ] <EOL> def factory ( code ) : <EOL> pf = PyScript ( code , sudo = hasattr ( request . function , '<STR_LIT>' ) ) <EOL> pfs . append ( pf ) <EOL> return pf <EOL> def teardown ( ) : <EOL> for pf in pfs : <EOL> pf . teardown ( ) <EOL> request . addfinalizer ( teardown ) <EOL> return factory </s>
