<s> """<STR_LIT>""" <EOL> from os import linesep <EOL> from os . path import join as opj , exists , normpath , isabs , commonprefix , relpath , realpath , isdir <EOL> from os . path import dirname , basename <EOL> import logging <EOL> import shlex <EOL> from six import string_types <EOL> from functools import wraps <EOL> import git <EOL> from git . exc import GitCommandError , NoSuchPathError , InvalidGitRepositoryError <EOL> from git . objects . blob import Blob <EOL> from . . support . exceptions import CommandError <EOL> from . . support . exceptions import FileNotInRepositoryError <EOL> from . . cmd import Runner <EOL> from . . utils import optional_args , on_windows , getpwd <EOL> from . . utils import swallow_outputs <EOL> lgr = logging . getLogger ( '<STR_LIT>' ) <EOL> default_git_odbt = git . GitCmdObjectDB <EOL> def _normalize_path ( base_dir , path ) : <EOL> """<STR_LIT>""" <EOL> if not path : <EOL> return path <EOL> base_dir = realpath ( base_dir ) <EOL> if isabs ( path ) : <EOL> path = opj ( realpath ( dirname ( path ) ) , basename ( path ) ) <EOL> if commonprefix ( [ path , base_dir ] ) != base_dir : <EOL> raise FileNotInRepositoryError ( msg = "<STR_LIT>" <EOL> % path , filename = path ) <EOL> else : <EOL> pass <EOL> elif commonprefix ( [ realpath ( getpwd ( ) ) , base_dir ] ) == base_dir : <EOL> path = opj ( realpath ( getpwd ( ) ) , path ) <EOL> else : <EOL> return path <EOL> return relpath ( path , start = base_dir ) <EOL> @ optional_args <EOL> def normalize_path ( func ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( func ) <EOL> def newfunc ( self , file_ , * args , ** kwargs ) : <EOL> file_new = _normalize_path ( self . path , file_ ) <EOL> return func ( self , file_new , * args , ** kwargs ) <EOL> return newfunc <EOL> @ optional_args <EOL> def normalize_paths ( func , match_return_type = True , map_filenames_back = False ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( func ) <EOL> def newfunc ( self , files , * args , ** kwargs ) : <EOL> normalize = _normalize_path if kwargs . pop ( '<STR_LIT>' , True ) else lambda rpath , filepath : filepath <EOL> if files : <EOL> if isinstance ( files , string_types ) or not files : <EOL> files_new = [ normalize ( self . path , files ) ] <EOL> single_file = True <EOL> elif isinstance ( files , list ) : <EOL> files_new = [ normalize ( self . path , path ) for path in files ] <EOL> single_file = False <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % <EOL> type ( files ) ) <EOL> else : <EOL> single_file = None <EOL> files_new = [ ] <EOL> if map_filenames_back : <EOL> def remap_filenames ( out ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( out , dict ) : <EOL> assert ( len ( out ) == len ( files_new ) ) <EOL> files_ = [ files ] if single_file else files <EOL> mapped = out . __class__ ( ) <EOL> for fin , fout in zip ( files_ , files_new ) : <EOL> mapped [ fin ] = out [ fout ] <EOL> return mapped <EOL> else : <EOL> return out <EOL> else : <EOL> remap_filenames = lambda x : x <EOL> result = func ( self , files_new , * args , ** kwargs ) <EOL> if single_file is None : <EOL> return result <EOL> elif ( result is None ) or not match_return_type or not single_file : <EOL> return remap_filenames ( result ) <EOL> elif single_file : <EOL> if len ( result ) != <NUM_LIT:1> : <EOL> return remap_filenames ( result ) <EOL> elif isinstance ( result , ( list , tuple ) ) : <EOL> return result [ <NUM_LIT:0> ] <EOL> elif isinstance ( result , dict ) and tuple ( result ) [ <NUM_LIT:0> ] == files_new [ <NUM_LIT:0> ] : <EOL> return tuple ( result . values ( ) ) [ <NUM_LIT:0> ] <EOL> else : <EOL> return remap_filenames ( result ) <EOL> else : <EOL> return RuntimeError ( "<STR_LIT>" ) <EOL> return newfunc <EOL> def _remove_empty_items ( list_ ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( list_ , list ) : <EOL> lgr . warning ( <EOL> "<STR_LIT>" % type ( list_ ) ) <EOL> return list_ <EOL> return [ file_ for file_ in list_ if file_ ] <EOL> def Repo ( * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' not in kwargs : <EOL> kwargs [ '<STR_LIT>' ] = default_git_odbt <EOL> return git . Repo ( * args , ** kwargs ) <EOL> class GitRepo ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = [ '<STR_LIT:path>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> _GIT_COMMON_OPTIONS = [ '<STR_LIT:-c>' , '<STR_LIT>' , '<STR_LIT:-c>' , '<STR_LIT>' ] <EOL> def __init__ ( self , path , url = None , runner = None , create = True ) : <EOL> """<STR_LIT>""" <EOL> self . path = normpath ( path ) <EOL> self . cmd_call_wrapper = runner or Runner ( cwd = self . path ) <EOL> if url is not None : <EOL> try : <EOL> self . cmd_call_wrapper ( git . Repo . clone_from , url , path ) <EOL> except GitCommandError as e : <EOL> lgr . error ( str ( e ) ) <EOL> raise <EOL> except ValueError as e : <EOL> if git . __version__ == '<STR_LIT>' and e . message == "<STR_LIT>" : <EOL> raise GitCommandError ( "<STR_LIT>" , <EOL> <NUM_LIT> , <EOL> stdout = "<STR_LIT>" if exists ( path ) else "<STR_LIT>" <EOL> ) <EOL> raise <EOL> if create and not exists ( opj ( path , '<STR_LIT>' ) ) : <EOL> try : <EOL> self . repo = self . cmd_call_wrapper ( git . Repo . init , path , True , odbt = default_git_odbt ) <EOL> except GitCommandError as e : <EOL> lgr . error ( str ( e ) ) <EOL> raise <EOL> else : <EOL> try : <EOL> self . repo = self . cmd_call_wrapper ( Repo , path ) <EOL> except ( GitCommandError , <EOL> NoSuchPathError , <EOL> InvalidGitRepositoryError ) as e : <EOL> lgr . error ( str ( e ) ) <EOL> raise <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( self . path , type ( self ) ) <EOL> def __eq__ ( self , obj ) : <EOL> """<STR_LIT>""" <EOL> return self . path == obj . path <EOL> @ classmethod <EOL> def get_toppath ( cls , path ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> toppath , err = Runner ( ) . run ( <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> cwd = path , <EOL> log_stdout = True , log_stderr = True , <EOL> expect_fail = True , expect_stderr = True ) <EOL> return toppath . rstrip ( '<STR_LIT>' ) <EOL> except CommandError : <EOL> return None <EOL> @ normalize_paths <EOL> def git_add ( self , files ) : <EOL> """<STR_LIT>""" <EOL> files = _remove_empty_items ( files ) <EOL> if files : <EOL> try : <EOL> self . cmd_call_wrapper ( self . repo . index . add , files , write = True ) <EOL> except OSError as e : <EOL> lgr . error ( "<STR_LIT>" % e ) <EOL> raise <EOL> else : <EOL> lgr . warning ( "<STR_LIT>" ) <EOL> @ normalize_paths ( match_return_type = False ) <EOL> def git_remove ( self , files , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> files = _remove_empty_items ( files ) <EOL> return self . repo . index . remove ( files , working_tree = True , ** kwargs ) <EOL> def precommit ( self ) : <EOL> """<STR_LIT>""" <EOL> self . repo . index . write ( ) <EOL> def git_commit ( self , msg = None , options = None ) : <EOL> """<STR_LIT>""" <EOL> if not msg : <EOL> msg = "<STR_LIT>" <EOL> if options : <EOL> raise NotImplementedError <EOL> lgr . debug ( "<STR_LIT>" % msg ) <EOL> self . cmd_call_wrapper ( self . repo . index . commit , msg ) <EOL> def get_indexed_files ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ x [ <NUM_LIT:0> ] for x in self . cmd_call_wrapper ( <EOL> self . repo . index . entries . keys ) ] <EOL> def git_get_hexsha ( self , branch = None ) : <EOL> """<STR_LIT>""" <EOL> if branch is None : <EOL> return self . repo . active_branch . object . hexsha <EOL> for b in self . repo . branches : <EOL> if b . name == branch : <EOL> return b . object . hexsha <EOL> raise ValueError ( "<STR_LIT>" % branch ) <EOL> def git_get_merge_base ( self , treeishes ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( treeishes , string_types ) : <EOL> treeishes = [ treeishes ] <EOL> if not treeishes : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> elif len ( treeishes ) == <NUM_LIT:1> : <EOL> treeishes = treeishes + [ self . git_get_active_branch ( ) ] <EOL> try : <EOL> bases = self . repo . merge_base ( * treeishes ) <EOL> except GitCommandError as exc : <EOL> if "<STR_LIT>" in str ( exc ) : <EOL> return None <EOL> raise <EOL> if not bases : <EOL> return None <EOL> assert ( len ( bases ) == <NUM_LIT:1> ) <EOL> return bases [ <NUM_LIT:0> ] . hexsha <EOL> def git_get_active_branch ( self ) : <EOL> return self . repo . active_branch . name <EOL> def git_get_branches ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ branch . name for branch in self . repo . branches ] <EOL> def git_get_remote_branches ( self ) : <EOL> """<STR_LIT>""" <EOL> remote_branches = list ( ) <EOL> for remote in self . repo . remotes : <EOL> try : <EOL> for ref in remote . refs : <EOL> remote_branches . append ( ref . name ) <EOL> except AssertionError as e : <EOL> if e . message . endswith ( "<STR_LIT>" ) : <EOL> pass <EOL> else : <EOL> raise e <EOL> return remote_branches <EOL> def git_get_remotes ( self ) : <EOL> return [ remote . name for remote in self . repo . remotes ] <EOL> def git_get_files ( self , branch = None ) : <EOL> """<STR_LIT>""" <EOL> if branch is None : <EOL> return self . get_indexed_files ( ) <EOL> else : <EOL> return [ item . path for item in self . repo . tree ( branch ) . traverse ( ) <EOL> if isinstance ( item , Blob ) ] <EOL> def git_get_file_content ( self , file_ , branch = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> content_str = self . repo . commit ( branch ) . tree [ file_ ] . data_stream . read ( ) <EOL> from six import PY3 <EOL> if PY3 : <EOL> conv_str = u'<STR_LIT>' <EOL> for b in bytes ( content_str ) : <EOL> conv_str += chr ( b ) <EOL> return conv_str . splitlines ( ) <EOL> else : <EOL> return content_str . splitlines ( ) <EOL> @ normalize_paths ( match_return_type = False ) <EOL> def _git_custom_command ( self , files , cmd_str , <EOL> log_stdout = True , log_stderr = True , log_online = False , <EOL> expect_stderr = True , cwd = None , env = None , <EOL> shell = None ) : <EOL> """<STR_LIT>""" <EOL> cmd = shlex . split ( cmd_str + "<STR_LIT:U+0020>" + "<STR_LIT:U+0020>" . join ( files ) , posix = not on_windows ) if isinstance ( cmd_str , string_types ) else cmd_str + files <EOL> assert ( cmd [ <NUM_LIT:0> ] == '<STR_LIT>' ) <EOL> cmd = cmd [ : <NUM_LIT:1> ] + self . _GIT_COMMON_OPTIONS + cmd [ <NUM_LIT:1> : ] <EOL> return self . cmd_call_wrapper . run ( cmd , log_stderr = log_stderr , <EOL> log_stdout = log_stdout , log_online = log_online , <EOL> expect_stderr = expect_stderr , cwd = cwd , <EOL> env = env , shell = shell ) <EOL> def git_remote_add ( self , name , url , options = '<STR_LIT>' ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . _git_custom_command ( '<STR_LIT>' , '<STR_LIT>' % <EOL> ( options , name , url ) ) <EOL> def git_remote_remove ( self , name ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . _git_custom_command ( '<STR_LIT>' , '<STR_LIT>' % name ) <EOL> def git_remote_show ( self , name = '<STR_LIT>' , verbose = False ) : <EOL> """<STR_LIT:U+0020>""" <EOL> v = "<STR_LIT>" if verbose else "<STR_LIT>" <EOL> out , err = self . _git_custom_command ( '<STR_LIT>' , '<STR_LIT>' % <EOL> ( v , name ) ) <EOL> return out . rstrip ( linesep ) . splitlines ( ) <EOL> def git_remote_update ( self , name = '<STR_LIT>' , verbose = False ) : <EOL> """<STR_LIT:U+0020>""" <EOL> v = "<STR_LIT>" if verbose else '<STR_LIT>' <EOL> self . _git_custom_command ( '<STR_LIT>' , '<STR_LIT>' % ( name , v ) , <EOL> expect_stderr = True ) <EOL> def git_fetch ( self , name , options = '<STR_LIT>' ) : <EOL> """<STR_LIT:U+0020>""" <EOL> self . _git_custom_command ( '<STR_LIT>' , '<STR_LIT>' % ( options , name ) , <EOL> expect_stderr = True ) <EOL> def git_get_remote_url ( self , name ) : <EOL> """<STR_LIT>""" <EOL> out , err = self . _git_custom_command ( <EOL> '<STR_LIT>' , '<STR_LIT>' % name ) <EOL> return out . rstrip ( linesep ) <EOL> def git_get_branch_commits ( self , branch , limit = None , stop = None , value = None ) : <EOL> """<STR_LIT>""" <EOL> fvalue = { None : lambda x : x , '<STR_LIT>' : lambda x : x . hexsha } [ value ] <EOL> if not limit : <EOL> def gen ( ) : <EOL> co = self . repo . branches [ branch ] . commit <EOL> yield co <EOL> for co_ in co . traverse ( ) : <EOL> yield co_ <EOL> elif limit == '<STR_LIT>' : <EOL> def gen ( ) : <EOL> co = self . repo . branches [ branch ] . commit <EOL> while co : <EOL> yield co <EOL> co = co . parents [ <NUM_LIT:0> ] if co . parents else None <EOL> else : <EOL> raise ValueError ( limit ) <EOL> for c in gen ( ) : <EOL> if stop and c . hexsha == stop : <EOL> return <EOL> yield fvalue ( c ) <EOL> def git_pull ( self , name = '<STR_LIT>' , options = '<STR_LIT>' ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . _git_custom_command ( '<STR_LIT>' , '<STR_LIT>' % ( options , name ) , <EOL> expect_stderr = True ) <EOL> def git_push ( self , name = '<STR_LIT>' , options = '<STR_LIT>' ) : <EOL> """<STR_LIT:U+0020>""" <EOL> self . _git_custom_command ( '<STR_LIT>' , '<STR_LIT>' % ( options , name ) , <EOL> expect_stderr = True ) <EOL> def git_checkout ( self , name , options = '<STR_LIT>' ) : <EOL> """<STR_LIT:U+0020>""" <EOL> self . _git_custom_command ( '<STR_LIT>' , '<STR_LIT>' % ( options , name ) , <EOL> expect_stderr = True ) <EOL> def git_merge ( self , name , options = [ ] , msg = None , ** kwargs ) : <EOL> if msg : <EOL> options = options + [ "<STR_LIT>" , msg ] <EOL> self . _git_custom_command ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] + options + [ name ] , ** kwargs ) <EOL> def git_remove_branch ( self , branch ) : <EOL> self . _git_custom_command ( '<STR_LIT>' , '<STR_LIT>' % branch ) <EOL> def git_ls_remote ( self , remote , options = None ) : <EOL> self . _git_custom_command ( '<STR_LIT>' , '<STR_LIT>' % <EOL> ( options if options is not None else '<STR_LIT>' , <EOL> remote ) ) <EOL> @ property <EOL> def dirty ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . repo . is_dirty ( untracked_files = True ) <EOL> def gc ( self , allow_background = False , auto = False ) : <EOL> """<STR_LIT>""" <EOL> cmd_options = [ '<STR_LIT>' ] <EOL> if not allow_background : <EOL> cmd_options += [ '<STR_LIT:-c>' , '<STR_LIT>' ] <EOL> cmd_options += [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> if auto : <EOL> cmd_options += [ '<STR_LIT>' ] <EOL> self . _git_custom_command ( '<STR_LIT>' , cmd_options ) </s>
