<s> from __future__ import print_function <EOL> import os <EOL> import sys <EOL> from unittest import TestCase <EOL> import time <EOL> import tempfile <EOL> import shutil <EOL> import io <EOL> from git import Repo , Remote , GitCommandError , Git <EOL> from git . compat import string_types <EOL> osp = os . path . dirname <EOL> GIT_REPO = os . environ . get ( "<STR_LIT>" , osp ( osp ( osp ( osp ( __file__ ) ) ) ) ) <EOL> GIT_DAEMON_PORT = os . environ . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> __all__ = ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ) <EOL> def fixture_path ( name ) : <EOL> test_dir = osp ( osp ( __file__ ) ) <EOL> return os . path . join ( test_dir , "<STR_LIT>" , name ) <EOL> def fixture ( name ) : <EOL> return open ( fixture_path ( name ) , '<STR_LIT:rb>' ) . read ( ) <EOL> def absolute_project_path ( ) : <EOL> return os . path . abspath ( os . path . join ( osp ( __file__ ) , "<STR_LIT:..>" , "<STR_LIT:..>" ) ) <EOL> class StringProcessAdapter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , input_string ) : <EOL> self . stdout = io . BytesIO ( input_string ) <EOL> self . stderr = io . BytesIO ( ) <EOL> def wait ( self ) : <EOL> return <NUM_LIT:0> <EOL> poll = wait <EOL> def _mktemp ( * args ) : <EOL> """<STR_LIT>""" <EOL> tdir = tempfile . mktemp ( * args ) <EOL> return tdir <EOL> def _rmtree_onerror ( osremove , fullpath , exec_info ) : <EOL> """<STR_LIT>""" <EOL> if os . name != '<STR_LIT>' or osremove is not os . remove : <EOL> raise <EOL> os . chmod ( fullpath , <NUM_LIT> ) <EOL> os . remove ( fullpath ) <EOL> def with_rw_repo ( working_tree_ref , bare = False ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( working_tree_ref , string_types ) , "<STR_LIT>" <EOL> def argument_passer ( func ) : <EOL> def repo_creator ( self ) : <EOL> prefix = '<STR_LIT>' <EOL> if bare : <EOL> prefix = '<STR_LIT>' <EOL> repo_dir = _mktemp ( "<STR_LIT>" % ( prefix , func . __name__ ) ) <EOL> rw_repo = self . rorepo . clone ( repo_dir , shared = True , bare = bare , n = True ) <EOL> rw_repo . head . commit = rw_repo . commit ( working_tree_ref ) <EOL> if not bare : <EOL> rw_repo . head . reference . checkout ( ) <EOL> prev_cwd = os . getcwd ( ) <EOL> os . chdir ( rw_repo . working_dir ) <EOL> try : <EOL> try : <EOL> return func ( self , rw_repo ) <EOL> except : <EOL> print ( "<STR_LIT>" % repo_dir , file = sys . stderr ) <EOL> repo_dir = None <EOL> raise <EOL> finally : <EOL> os . chdir ( prev_cwd ) <EOL> rw_repo . git . clear_cache ( ) <EOL> if repo_dir is not None : <EOL> shutil . rmtree ( repo_dir , onerror = _rmtree_onerror ) <EOL> repo_creator . __name__ = func . __name__ <EOL> return repo_creator <EOL> return argument_passer <EOL> def with_rw_and_rw_remote_repo ( working_tree_ref ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( working_tree_ref , string_types ) , "<STR_LIT>" <EOL> def argument_passer ( func ) : <EOL> def remote_repo_creator ( self ) : <EOL> remote_repo_dir = _mktemp ( "<STR_LIT>" % func . __name__ ) <EOL> repo_dir = _mktemp ( "<STR_LIT>" ) <EOL> rw_remote_repo = self . rorepo . clone ( remote_repo_dir , shared = True , bare = True ) <EOL> rw_repo = rw_remote_repo . clone ( repo_dir , shared = True , bare = False , n = True ) <EOL> rw_repo . head . commit = working_tree_ref <EOL> rw_repo . head . reference . checkout ( ) <EOL> rw_remote_repo . daemon_export = True <EOL> crw = rw_remote_repo . config_writer ( ) <EOL> section = "<STR_LIT>" <EOL> try : <EOL> crw . add_section ( section ) <EOL> except Exception : <EOL> pass <EOL> crw . set ( section , "<STR_LIT>" , True ) <EOL> crw . release ( ) <EOL> del ( crw ) <EOL> d_remote = Remote . create ( rw_repo , "<STR_LIT>" , remote_repo_dir ) <EOL> d_remote . fetch ( ) <EOL> remote_repo_url = "<STR_LIT>" % ( GIT_DAEMON_PORT , remote_repo_dir ) <EOL> d_remote . config_writer . set ( '<STR_LIT:url>' , remote_repo_url ) <EOL> temp_dir = osp ( _mktemp ( ) ) <EOL> try : <EOL> gd = Git ( ) . daemon ( temp_dir , enable = '<STR_LIT>' , listen = '<STR_LIT:127.0.0.1>' , port = GIT_DAEMON_PORT , <EOL> as_process = True ) <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> except Exception : <EOL> gd = None <EOL> try : <EOL> rw_repo . git . ls_remote ( d_remote ) <EOL> except GitCommandError as e : <EOL> if gd is not None : <EOL> os . kill ( gd . proc . pid , <NUM_LIT:15> ) <EOL> print ( str ( e ) ) <EOL> if os . name == '<STR_LIT>' : <EOL> msg = "<STR_LIT>" <EOL> msg += '<STR_LIT>' % temp_dir <EOL> raise AssertionError ( msg ) <EOL> else : <EOL> msg = '<STR_LIT>' <EOL> msg += '<STR_LIT>' <EOL> msg += '<STR_LIT>' <EOL> msg %= temp_dir <EOL> raise AssertionError ( msg ) <EOL> prev_cwd = os . getcwd ( ) <EOL> os . chdir ( rw_repo . working_dir ) <EOL> try : <EOL> try : <EOL> return func ( self , rw_repo , rw_remote_repo ) <EOL> except : <EOL> print ( "<STR_LIT>" <EOL> % ( repo_dir , remote_repo_dir ) , file = sys . stderr ) <EOL> repo_dir = remote_repo_dir = None <EOL> raise <EOL> finally : <EOL> if gd is not None : <EOL> os . kill ( gd . proc . pid , <NUM_LIT:15> ) <EOL> os . chdir ( prev_cwd ) <EOL> rw_repo . git . clear_cache ( ) <EOL> rw_remote_repo . git . clear_cache ( ) <EOL> if repo_dir : <EOL> shutil . rmtree ( repo_dir , onerror = _rmtree_onerror ) <EOL> if remote_repo_dir : <EOL> shutil . rmtree ( remote_repo_dir , onerror = _rmtree_onerror ) <EOL> if gd is not None : <EOL> gd . proc . wait ( ) <EOL> remote_repo_creator . __name__ = func . __name__ <EOL> return remote_repo_creator <EOL> return argument_passer <EOL> class TestBase ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def _small_repo_url ( self ) : <EOL> """<STR_LIT>""" <EOL> return os . path . join ( self . rorepo . working_tree_dir , '<STR_LIT>' ) <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> """<STR_LIT>""" <EOL> cls . rorepo = Repo ( GIT_REPO ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . rorepo . git . clear_cache ( ) <EOL> cls . rorepo . git = None <EOL> def _make_file ( self , rela_path , data , repo = None ) : <EOL> """<STR_LIT>""" <EOL> repo = repo or self . rorepo <EOL> abs_path = os . path . join ( repo . working_tree_dir , rela_path ) <EOL> fp = open ( abs_path , "<STR_LIT:w>" ) <EOL> fp . write ( data ) <EOL> fp . close ( ) <EOL> return abs_path </s>
