<s> import os <EOL> import re <EOL> import time <EOL> import urllib <EOL> import urllib2 <EOL> import posixpath <EOL> import string <EOL> from dulwich . objects import Tag <EOL> from dulwich . repo import Repo , NotGitRepository <EOL> from vcs import subprocessio <EOL> from vcs . backends . base import BaseRepository , CollectionGenerator <EOL> from vcs . conf import settings <EOL> from vcs . exceptions import ( <EOL> BranchDoesNotExistError , ChangesetDoesNotExistError , EmptyRepositoryError , <EOL> RepositoryError , TagAlreadyExistError , TagDoesNotExistError <EOL> ) <EOL> from vcs . utils import safe_unicode , makedate , date_fromtimestamp <EOL> from vcs . utils . lazy import LazyProperty <EOL> from vcs . utils . ordered_dict import OrderedDict <EOL> from vcs . utils . paths import abspath , get_user_home <EOL> from vcs . utils . hgcompat import ( <EOL> hg_url , httpbasicauthhandler , httpdigestauthhandler <EOL> ) <EOL> from . changeset import GitChangeset <EOL> from . config import ConfigFile <EOL> from . inmemory import GitInMemoryChangeset <EOL> from . workdir import GitWorkdir <EOL> SHA_PATTERN = re . compile ( r'<STR_LIT>' ) <EOL> class GitRepository ( BaseRepository ) : <EOL> """<STR_LIT>""" <EOL> DEFAULT_BRANCH_NAME = '<STR_LIT>' <EOL> scm = '<STR_LIT>' <EOL> def __init__ ( self , repo_path , create = False , src_url = None , <EOL> update_after_clone = False , bare = False ) : <EOL> self . path = abspath ( repo_path ) <EOL> repo = self . _get_repo ( create , src_url , update_after_clone , bare ) <EOL> self . bare = repo . bare <EOL> @ property <EOL> def _config_files ( self ) : <EOL> return [ <EOL> self . bare and abspath ( self . path , '<STR_LIT>' ) <EOL> or abspath ( self . path , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> abspath ( get_user_home ( ) , '<STR_LIT>' ) , <EOL> ] <EOL> @ property <EOL> def _repo ( self ) : <EOL> return Repo ( self . path ) <EOL> @ property <EOL> def head ( self ) : <EOL> try : <EOL> return self . _repo . head ( ) <EOL> except KeyError : <EOL> return None <EOL> @ LazyProperty <EOL> def revisions ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_all_revisions ( ) <EOL> @ classmethod <EOL> def _run_git_command ( cls , cmd , ** opts ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' in opts : <EOL> _copts = [ ] <EOL> del opts [ '<STR_LIT>' ] <EOL> else : <EOL> _copts = [ '<STR_LIT:-c>' , '<STR_LIT>' , ] <EOL> safe_call = False <EOL> if '<STR_LIT>' in opts : <EOL> del opts [ '<STR_LIT>' ] <EOL> safe_call = True <EOL> _str_cmd = False <EOL> if isinstance ( cmd , basestring ) : <EOL> cmd = [ cmd ] <EOL> _str_cmd = True <EOL> gitenv = os . environ <EOL> if '<STR_LIT>' in gitenv : <EOL> del gitenv [ '<STR_LIT>' ] <EOL> gitenv [ '<STR_LIT>' ] = '<STR_LIT:1>' <EOL> _git_path = settings . GIT_EXECUTABLE_PATH <EOL> cmd = [ _git_path ] + _copts + cmd <EOL> if _str_cmd : <EOL> cmd = '<STR_LIT:U+0020>' . join ( cmd ) <EOL> try : <EOL> _opts = dict ( <EOL> env = gitenv , <EOL> shell = False , <EOL> ) <EOL> _opts . update ( opts ) <EOL> p = subprocessio . SubprocessIOChunker ( cmd , ** _opts ) <EOL> except ( EnvironmentError , OSError ) , err : <EOL> tb_err = ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( cmd , err ) ) <EOL> if safe_call : <EOL> return '<STR_LIT>' , err <EOL> else : <EOL> raise RepositoryError ( tb_err ) <EOL> return '<STR_LIT>' . join ( p . output ) , '<STR_LIT>' . join ( p . error ) <EOL> def run_git_command ( self , cmd ) : <EOL> opts = { } <EOL> if os . path . isdir ( self . path ) : <EOL> opts [ '<STR_LIT>' ] = self . path <EOL> return self . _run_git_command ( cmd , ** opts ) <EOL> @ classmethod <EOL> def _check_url ( cls , url ) : <EOL> """<STR_LIT>""" <EOL> if os . path . isdir ( url ) or url . startswith ( '<STR_LIT>' ) : <EOL> return True <EOL> if ( '<STR_LIT:+>' in url [ : url . find ( '<STR_LIT>' ) ] ) : <EOL> url = url [ url . find ( '<STR_LIT:+>' ) + <NUM_LIT:1> : ] <EOL> handlers = [ ] <EOL> test_uri , authinfo = hg_url ( url ) . authinfo ( ) <EOL> if not test_uri . endswith ( '<STR_LIT>' ) : <EOL> test_uri = test_uri . rstrip ( '<STR_LIT:/>' ) + '<STR_LIT>' <EOL> if authinfo : <EOL> passmgr = urllib2 . HTTPPasswordMgrWithDefaultRealm ( ) <EOL> passmgr . add_password ( * authinfo ) <EOL> handlers . extend ( ( httpbasicauthhandler ( passmgr ) , <EOL> httpdigestauthhandler ( passmgr ) ) ) <EOL> o = urllib2 . build_opener ( * handlers ) <EOL> o . addheaders = [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> q = { "<STR_LIT>" : '<STR_LIT>' } <EOL> qs = '<STR_LIT>' % urllib . urlencode ( q ) <EOL> cu = "<STR_LIT>" % ( test_uri , qs ) <EOL> req = urllib2 . Request ( cu , None , { } ) <EOL> try : <EOL> resp = o . open ( req ) <EOL> return resp . code == <NUM_LIT:200> <EOL> except Exception , e : <EOL> raise urllib2 . URLError ( "<STR_LIT>" % ( url , e ) ) <EOL> def _get_repo ( self , create , src_url = None , update_after_clone = False , <EOL> bare = False ) : <EOL> if create and os . path . exists ( self . path ) : <EOL> raise RepositoryError ( "<STR_LIT>" ) <EOL> if src_url and not create : <EOL> raise RepositoryError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> try : <EOL> if create and src_url : <EOL> GitRepository . _check_url ( src_url ) <EOL> self . clone ( src_url , update_after_clone , bare ) <EOL> return Repo ( self . path ) <EOL> elif create : <EOL> os . mkdir ( self . path ) <EOL> if bare : <EOL> return Repo . init_bare ( self . path ) <EOL> else : <EOL> return Repo . init ( self . path ) <EOL> else : <EOL> return self . _repo <EOL> except ( NotGitRepository , OSError ) , err : <EOL> raise RepositoryError ( err ) <EOL> def _get_all_revisions ( self ) : <EOL> try : <EOL> self . _repo . head ( ) <EOL> except KeyError : <EOL> return [ ] <EOL> rev_filter = settings . GIT_REV_FILTER <EOL> cmd = '<STR_LIT>' % ( rev_filter ) <EOL> try : <EOL> so , se = self . run_git_command ( cmd ) <EOL> except RepositoryError : <EOL> return [ ] <EOL> return so . splitlines ( ) <EOL> def _get_all_revisions2 ( self ) : <EOL> includes = [ x [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] for x in self . _parsed_refs . iteritems ( ) <EOL> if x [ <NUM_LIT:1> ] [ <NUM_LIT:1> ] != '<STR_LIT:T>' ] <EOL> return [ c . commit . id for c in self . _repo . get_walker ( include = includes ) ] <EOL> def _get_revision ( self , revision ) : <EOL> """<STR_LIT>""" <EOL> is_null = lambda o : len ( o ) == revision . count ( '<STR_LIT:0>' ) <EOL> try : <EOL> self . revisions [ <NUM_LIT:0> ] <EOL> except ( KeyError , IndexError ) : <EOL> raise EmptyRepositoryError ( "<STR_LIT>" ) <EOL> if revision in ( None , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , - <NUM_LIT:1> ) : <EOL> return self . revisions [ - <NUM_LIT:1> ] <EOL> is_bstr = isinstance ( revision , ( str , unicode ) ) <EOL> if ( ( is_bstr and revision . isdigit ( ) and len ( revision ) < <NUM_LIT:12> ) <EOL> or isinstance ( revision , int ) or is_null ( revision ) ) : <EOL> try : <EOL> revision = self . revisions [ int ( revision ) ] <EOL> except Exception : <EOL> raise ChangesetDoesNotExistError ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( revision ) ) <EOL> elif is_bstr : <EOL> _ref_revision = self . _parsed_refs . get ( revision ) <EOL> if _ref_revision : <EOL> return _ref_revision [ <NUM_LIT:0> ] <EOL> _tags_shas = self . tags . values ( ) <EOL> if revision in _tags_shas : <EOL> return _tags_shas [ _tags_shas . index ( revision ) ] <EOL> elif not SHA_PATTERN . match ( revision ) or revision not in self . revisions : <EOL> raise ChangesetDoesNotExistError ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( revision ) ) <EOL> if not SHA_PATTERN . match ( str ( revision ) ) : <EOL> raise ChangesetDoesNotExistError ( "<STR_LIT>" <EOL> % revision ) <EOL> return revision <EOL> def _get_archives ( self , archive_name = '<STR_LIT>' ) : <EOL> for i in [ ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] : <EOL> yield { "<STR_LIT:type>" : i [ <NUM_LIT:0> ] , "<STR_LIT>" : i [ <NUM_LIT:1> ] , "<STR_LIT>" : archive_name } <EOL> def _get_url ( self , url ) : <EOL> """<STR_LIT>""" <EOL> url = str ( url ) <EOL> if url != '<STR_LIT:default>' and not '<STR_LIT>' in url : <EOL> url = '<STR_LIT>' . join ( ( '<STR_LIT:file>' , url ) ) <EOL> return url <EOL> def get_hook_location ( self ) : <EOL> """<STR_LIT>""" <EOL> loc = os . path . join ( self . path , '<STR_LIT>' ) <EOL> if not self . bare : <EOL> loc = os . path . join ( self . path , '<STR_LIT>' , '<STR_LIT>' ) <EOL> return loc <EOL> @ LazyProperty <EOL> def name ( self ) : <EOL> return os . path . basename ( self . path ) <EOL> @ LazyProperty <EOL> def last_change ( self ) : <EOL> """<STR_LIT>""" <EOL> return date_fromtimestamp ( self . _get_mtime ( ) , makedate ( ) [ <NUM_LIT:1> ] ) <EOL> def _get_mtime ( self ) : <EOL> try : <EOL> return time . mktime ( self . get_changeset ( ) . date . timetuple ( ) ) <EOL> except RepositoryError : <EOL> idx_loc = '<STR_LIT>' if self . bare else '<STR_LIT>' <EOL> in_path = os . path . join ( self . path , idx_loc , "<STR_LIT:index>" ) <EOL> he_path = os . path . join ( self . path , idx_loc , "<STR_LIT>" ) <EOL> if os . path . exists ( in_path ) : <EOL> return os . stat ( in_path ) . st_mtime <EOL> else : <EOL> return os . stat ( he_path ) . st_mtime <EOL> @ LazyProperty <EOL> def description ( self ) : <EOL> idx_loc = '<STR_LIT>' if self . bare else '<STR_LIT>' <EOL> undefined_description = u'<STR_LIT>' <EOL> description_path = os . path . join ( self . path , idx_loc , '<STR_LIT:description>' ) <EOL> if os . path . isfile ( description_path ) : <EOL> return safe_unicode ( open ( description_path ) . read ( ) ) <EOL> else : <EOL> return undefined_description <EOL> @ LazyProperty <EOL> def contact ( self ) : <EOL> undefined_contact = u'<STR_LIT>' <EOL> return undefined_contact <EOL> @ property <EOL> def branches ( self ) : <EOL> if not self . revisions : <EOL> return { } <EOL> sortkey = lambda ctx : ctx [ <NUM_LIT:0> ] <EOL> _branches = [ ( x [ <NUM_LIT:0> ] , x [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] ) <EOL> for x in self . _parsed_refs . iteritems ( ) if x [ <NUM_LIT:1> ] [ <NUM_LIT:1> ] == '<STR_LIT:H>' ] <EOL> return OrderedDict ( sorted ( _branches , key = sortkey , reverse = False ) ) <EOL> @ LazyProperty <EOL> def tags ( self ) : <EOL> return self . _get_tags ( ) <EOL> def _get_tags ( self ) : <EOL> if not self . revisions : <EOL> return { } <EOL> sortkey = lambda ctx : ctx [ <NUM_LIT:0> ] <EOL> _tags = [ ( x [ <NUM_LIT:0> ] , x [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] ) <EOL> for x in self . _parsed_refs . iteritems ( ) if x [ <NUM_LIT:1> ] [ <NUM_LIT:1> ] == '<STR_LIT:T>' ] <EOL> return OrderedDict ( sorted ( _tags , key = sortkey , reverse = True ) ) <EOL> def tag ( self , name , user , revision = None , message = None , date = None , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if name in self . tags : <EOL> raise TagAlreadyExistError ( "<STR_LIT>" % name ) <EOL> changeset = self . get_changeset ( revision ) <EOL> message = message or "<STR_LIT>" % ( name , <EOL> changeset . raw_id ) <EOL> self . _repo . refs [ "<STR_LIT>" % name ] = changeset . _commit . id <EOL> self . _parsed_refs = self . _get_parsed_refs ( ) <EOL> self . tags = self . _get_tags ( ) <EOL> return changeset <EOL> def remove_tag ( self , name , user , message = None , date = None ) : <EOL> """<STR_LIT>""" <EOL> if name not in self . tags : <EOL> raise TagDoesNotExistError ( "<STR_LIT>" % name ) <EOL> tagpath = posixpath . join ( self . _repo . refs . path , '<STR_LIT>' , '<STR_LIT>' , name ) <EOL> try : <EOL> os . remove ( tagpath ) <EOL> self . _parsed_refs = self . _get_parsed_refs ( ) <EOL> self . tags = self . _get_tags ( ) <EOL> except OSError , e : <EOL> raise RepositoryError ( e . strerror ) <EOL> @ LazyProperty <EOL> def _parsed_refs ( self ) : <EOL> return self . _get_parsed_refs ( ) <EOL> def _get_parsed_refs ( self ) : <EOL> _repo = self . _repo <EOL> refs = _repo . get_refs ( ) <EOL> keys = [ ( '<STR_LIT>' , '<STR_LIT:H>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT:T>' ) ] <EOL> _refs = { } <EOL> for ref , sha in refs . iteritems ( ) : <EOL> for k , type_ in keys : <EOL> if ref . startswith ( k ) : <EOL> _key = ref [ len ( k ) : ] <EOL> if type_ == '<STR_LIT:T>' : <EOL> obj = _repo . get_object ( sha ) <EOL> if isinstance ( obj , Tag ) : <EOL> sha = _repo . get_object ( sha ) . object [ <NUM_LIT:1> ] <EOL> _refs [ _key ] = [ sha , type_ ] <EOL> break <EOL> return _refs <EOL> def _heads ( self , reverse = False ) : <EOL> refs = self . _repo . get_refs ( ) <EOL> heads = { } <EOL> for key , val in refs . items ( ) : <EOL> for ref_key in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> if key . startswith ( ref_key ) : <EOL> n = key [ len ( ref_key ) : ] <EOL> if n not in [ '<STR_LIT>' ] : <EOL> heads [ n ] = val <EOL> return heads if reverse else dict ( ( y , x ) for x , y in heads . iteritems ( ) ) <EOL> def get_changeset ( self , revision = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( revision , GitChangeset ) : <EOL> return revision <EOL> revision = self . _get_revision ( revision ) <EOL> changeset = GitChangeset ( repository = self , revision = revision ) <EOL> return changeset <EOL> def get_changesets ( self , start = None , end = None , start_date = None , <EOL> end_date = None , branch_name = None , reverse = False ) : <EOL> """<STR_LIT>""" <EOL> if branch_name and branch_name not in self . branches : <EOL> raise BranchDoesNotExistError ( "<STR_LIT>" % branch_name ) <EOL> cmd_template = '<STR_LIT>' <EOL> cmd_params = { } <EOL> if start_date : <EOL> cmd_template += '<STR_LIT>' <EOL> cmd_params [ '<STR_LIT>' ] = start_date . strftime ( '<STR_LIT>' ) <EOL> if end_date : <EOL> cmd_template += '<STR_LIT>' <EOL> cmd_params [ '<STR_LIT>' ] = end_date . strftime ( '<STR_LIT>' ) <EOL> if branch_name : <EOL> cmd_template += '<STR_LIT>' <EOL> cmd_params [ '<STR_LIT>' ] = branch_name <EOL> else : <EOL> rev_filter = settings . GIT_REV_FILTER <EOL> cmd_template += '<STR_LIT>' % ( rev_filter ) <EOL> cmd = string . Template ( cmd_template ) . safe_substitute ( ** cmd_params ) <EOL> revs = self . run_git_command ( cmd ) [ <NUM_LIT:0> ] . splitlines ( ) <EOL> start_pos = <NUM_LIT:0> <EOL> end_pos = len ( revs ) <EOL> if start : <EOL> _start = self . _get_revision ( start ) <EOL> try : <EOL> start_pos = revs . index ( _start ) <EOL> except ValueError : <EOL> pass <EOL> if end is not None : <EOL> _end = self . _get_revision ( end ) <EOL> try : <EOL> end_pos = revs . index ( _end ) <EOL> except ValueError : <EOL> pass <EOL> if None not in [ start , end ] and start_pos > end_pos : <EOL> raise RepositoryError ( '<STR_LIT>' ) <EOL> if end_pos is not None : <EOL> end_pos += <NUM_LIT:1> <EOL> revs = revs [ start_pos : end_pos ] <EOL> if reverse : <EOL> revs = reversed ( revs ) <EOL> return CollectionGenerator ( self , revs ) <EOL> def get_diff ( self , rev1 , rev2 , path = None , ignore_whitespace = False , <EOL> context = <NUM_LIT:3> ) : <EOL> """<STR_LIT>""" <EOL> flags = [ '<STR_LIT>' % context , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> if ignore_whitespace : <EOL> flags . append ( '<STR_LIT>' ) <EOL> if hasattr ( rev1 , '<STR_LIT>' ) : <EOL> rev1 = getattr ( rev1 , '<STR_LIT>' ) <EOL> if hasattr ( rev2 , '<STR_LIT>' ) : <EOL> rev2 = getattr ( rev2 , '<STR_LIT>' ) <EOL> if rev1 == self . EMPTY_CHANGESET : <EOL> rev2 = self . get_changeset ( rev2 ) . raw_id <EOL> cmd = '<STR_LIT:U+0020>' . join ( [ '<STR_LIT>' ] + flags + [ rev2 ] ) <EOL> else : <EOL> rev1 = self . get_changeset ( rev1 ) . raw_id <EOL> rev2 = self . get_changeset ( rev2 ) . raw_id <EOL> cmd = '<STR_LIT:U+0020>' . join ( [ '<STR_LIT>' ] + flags + [ rev1 , rev2 ] ) <EOL> if path : <EOL> cmd += '<STR_LIT>' % path <EOL> stdout , stderr = self . run_git_command ( cmd ) <EOL> if rev1 == self . EMPTY_CHANGESET : <EOL> lines = stdout . splitlines ( ) <EOL> x = <NUM_LIT:0> <EOL> for line in lines : <EOL> if line . startswith ( '<STR_LIT>' ) : <EOL> break <EOL> x += <NUM_LIT:1> <EOL> stdout = '<STR_LIT:\n>' . join ( lines [ x : ] ) + '<STR_LIT:\n>' <EOL> return stdout <EOL> @ LazyProperty <EOL> def in_memory_changeset ( self ) : <EOL> """<STR_LIT>""" <EOL> return GitInMemoryChangeset ( self ) <EOL> def clone ( self , url , update_after_clone = True , bare = False ) : <EOL> """<STR_LIT>""" <EOL> url = self . _get_url ( url ) <EOL> cmd = [ '<STR_LIT>' ] <EOL> if bare : <EOL> cmd . append ( '<STR_LIT>' ) <EOL> elif not update_after_clone : <EOL> cmd . append ( '<STR_LIT>' ) <EOL> cmd += [ '<STR_LIT>' , '<STR_LIT>' % url , '<STR_LIT>' % self . path ] <EOL> cmd = '<STR_LIT:U+0020>' . join ( cmd ) <EOL> self . run_git_command ( cmd ) <EOL> def pull ( self , url ) : <EOL> """<STR_LIT>""" <EOL> url = self . _get_url ( url ) <EOL> cmd = [ '<STR_LIT>' ] <EOL> cmd . append ( "<STR_LIT>" ) <EOL> cmd . append ( url ) <EOL> cmd = '<STR_LIT:U+0020>' . join ( cmd ) <EOL> self . run_git_command ( cmd ) <EOL> def fetch ( self , url ) : <EOL> """<STR_LIT>""" <EOL> url = self . _get_url ( url ) <EOL> so , se = self . run_git_command ( '<STR_LIT>' % url ) <EOL> refs = [ ] <EOL> for line in ( x for x in so . splitlines ( ) ) : <EOL> sha , ref = line . split ( '<STR_LIT:\t>' ) <EOL> refs . append ( ref ) <EOL> refs = '<STR_LIT:U+0020>' . join ( ( '<STR_LIT>' % ( r , r ) for r in refs ) ) <EOL> cmd = '''<STR_LIT>''' % ( url , refs ) <EOL> self . run_git_command ( cmd ) <EOL> @ LazyProperty <EOL> def workdir ( self ) : <EOL> """<STR_LIT>""" <EOL> return GitWorkdir ( self ) <EOL> def get_config_value ( self , section , name , config_file = None ) : <EOL> """<STR_LIT>""" <EOL> if config_file is None : <EOL> config_file = [ ] <EOL> elif isinstance ( config_file , basestring ) : <EOL> config_file = [ config_file ] <EOL> def gen_configs ( ) : <EOL> for path in config_file + self . _config_files : <EOL> try : <EOL> yield ConfigFile . from_path ( path ) <EOL> except ( IOError , OSError , ValueError ) : <EOL> continue <EOL> for config in gen_configs ( ) : <EOL> try : <EOL> return config . get ( section , name ) <EOL> except KeyError : <EOL> continue <EOL> return None <EOL> def get_user_name ( self , config_file = None ) : <EOL> """<STR_LIT>""" <EOL> return self . get_config_value ( '<STR_LIT:user>' , '<STR_LIT:name>' , config_file ) <EOL> def get_user_email ( self , config_file = None ) : <EOL> """<STR_LIT>""" <EOL> return self . get_config_value ( '<STR_LIT:user>' , '<STR_LIT:email>' , config_file ) </s>
