<s> import os <EOL> from git . objects import Object , Commit <EOL> from git . util import ( <EOL> join_path , <EOL> join_path_native , <EOL> to_native_path_linux , <EOL> assure_directory_exists <EOL> ) <EOL> from gitdb . exc import ( <EOL> BadObject , <EOL> BadName <EOL> ) <EOL> from gitdb . util import ( <EOL> join , <EOL> dirname , <EOL> isdir , <EOL> exists , <EOL> isfile , <EOL> rename , <EOL> hex_to_bin , <EOL> LockedFD <EOL> ) <EOL> from git . compat import ( <EOL> string_types , <EOL> defenc <EOL> ) <EOL> from . log import RefLog <EOL> __all__ = [ "<STR_LIT>" ] <EOL> class SymbolicReference ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( "<STR_LIT>" , "<STR_LIT:path>" ) <EOL> _resolve_ref_on_create = False <EOL> _points_to_commits_only = True <EOL> _common_path_default = "<STR_LIT>" <EOL> _remote_common_path_default = "<STR_LIT>" <EOL> _id_attribute_ = "<STR_LIT:name>" <EOL> def __init__ ( self , repo , path ) : <EOL> self . repo = repo <EOL> self . path = path <EOL> def __str__ ( self ) : <EOL> return self . path <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , self . path ) <EOL> def __eq__ ( self , other ) : <EOL> if hasattr ( other , '<STR_LIT:path>' ) : <EOL> return self . path == other . path <EOL> return False <EOL> def __ne__ ( self , other ) : <EOL> return not ( self == other ) <EOL> def __hash__ ( self ) : <EOL> return hash ( self . path ) <EOL> @ property <EOL> def name ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . path <EOL> @ property <EOL> def abspath ( self ) : <EOL> return join_path_native ( self . repo . git_dir , self . path ) <EOL> @ classmethod <EOL> def _get_packed_refs_path ( cls , repo ) : <EOL> return join ( repo . git_dir , '<STR_LIT>' ) <EOL> @ classmethod <EOL> def _iter_packed_refs ( cls , repo ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> with open ( cls . _get_packed_refs_path ( repo ) , '<STR_LIT>' ) as fp : <EOL> for line in fp : <EOL> line = line . strip ( ) <EOL> if not line : <EOL> continue <EOL> if line . startswith ( '<STR_LIT:#>' ) : <EOL> if line . startswith ( '<STR_LIT>' ) and not line . endswith ( '<STR_LIT>' ) : <EOL> raise TypeError ( "<STR_LIT>" % line ) <EOL> continue <EOL> if line [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> continue <EOL> yield tuple ( line . split ( '<STR_LIT:U+0020>' , <NUM_LIT:1> ) ) <EOL> except ( OSError , IOError ) : <EOL> return <EOL> @ classmethod <EOL> def dereference_recursive ( cls , repo , ref_path ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> hexsha , ref_path = cls . _get_ref_info ( repo , ref_path ) <EOL> if hexsha is not None : <EOL> return hexsha <EOL> @ classmethod <EOL> def _get_ref_info ( cls , repo , ref_path ) : <EOL> """<STR_LIT>""" <EOL> tokens = None <EOL> try : <EOL> fp = open ( join ( repo . git_dir , ref_path ) , '<STR_LIT>' ) <EOL> value = fp . read ( ) . rstrip ( ) <EOL> fp . close ( ) <EOL> tokens = value . split ( ) <EOL> assert ( len ( tokens ) != <NUM_LIT:0> ) <EOL> except ( OSError , IOError ) : <EOL> for sha , path in cls . _iter_packed_refs ( repo ) : <EOL> if path != ref_path : <EOL> continue <EOL> tokens = sha , path <EOL> break <EOL> if tokens is None : <EOL> raise ValueError ( "<STR_LIT>" % ref_path ) <EOL> if tokens [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> return ( None , tokens [ <NUM_LIT:1> ] ) <EOL> if repo . re_hexsha_only . match ( tokens [ <NUM_LIT:0> ] ) : <EOL> return ( tokens [ <NUM_LIT:0> ] , None ) <EOL> raise ValueError ( "<STR_LIT>" % ref_path ) <EOL> def _get_object ( self ) : <EOL> """<STR_LIT>""" <EOL> return Object . new_from_sha ( self . repo , hex_to_bin ( self . dereference_recursive ( self . repo , self . path ) ) ) <EOL> def _get_commit ( self ) : <EOL> """<STR_LIT>""" <EOL> obj = self . _get_object ( ) <EOL> if obj . type == '<STR_LIT>' : <EOL> obj = obj . object <EOL> if obj . type != Commit . type : <EOL> raise TypeError ( "<STR_LIT>" % obj ) <EOL> return obj <EOL> def set_commit ( self , commit , logmsg = None ) : <EOL> """<STR_LIT>""" <EOL> invalid_type = False <EOL> if isinstance ( commit , Object ) : <EOL> invalid_type = commit . type != Commit . type <EOL> elif isinstance ( commit , SymbolicReference ) : <EOL> invalid_type = commit . object . type != Commit . type <EOL> else : <EOL> try : <EOL> invalid_type = self . repo . rev_parse ( commit ) . type != Commit . type <EOL> except ( BadObject , BadName ) : <EOL> raise ValueError ( "<STR_LIT>" % commit ) <EOL> if invalid_type : <EOL> raise ValueError ( "<STR_LIT>" % commit ) <EOL> self . set_object ( commit , logmsg ) <EOL> return self <EOL> def set_object ( self , object , logmsg = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( object , SymbolicReference ) : <EOL> object = object . object <EOL> is_detached = True <EOL> try : <EOL> is_detached = self . is_detached <EOL> except ValueError : <EOL> pass <EOL> if is_detached : <EOL> return self . set_reference ( object , logmsg ) <EOL> return self . _get_reference ( ) . set_object ( object , logmsg ) <EOL> commit = property ( _get_commit , set_commit , doc = "<STR_LIT>" ) <EOL> object = property ( _get_object , set_object , doc = "<STR_LIT>" ) <EOL> def _get_reference ( self ) : <EOL> """<STR_LIT>""" <EOL> sha , target_ref_path = self . _get_ref_info ( self . repo , self . path ) <EOL> if target_ref_path is None : <EOL> raise TypeError ( "<STR_LIT>" % ( self , sha ) ) <EOL> return self . from_path ( self . repo , target_ref_path ) <EOL> def set_reference ( self , ref , logmsg = None ) : <EOL> """<STR_LIT>""" <EOL> write_value = None <EOL> obj = None <EOL> if isinstance ( ref , SymbolicReference ) : <EOL> write_value = "<STR_LIT>" % ref . path <EOL> elif isinstance ( ref , Object ) : <EOL> obj = ref <EOL> write_value = ref . hexsha <EOL> elif isinstance ( ref , string_types ) : <EOL> try : <EOL> obj = self . repo . rev_parse ( ref + "<STR_LIT>" ) <EOL> write_value = obj . hexsha <EOL> except ( BadObject , BadName ) : <EOL> raise ValueError ( "<STR_LIT>" % ref ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % ref ) <EOL> if obj is not None and self . _points_to_commits_only and obj . type != Commit . type : <EOL> raise TypeError ( "<STR_LIT>" % obj ) <EOL> oldbinsha = None <EOL> if logmsg is not None : <EOL> try : <EOL> oldbinsha = self . commit . binsha <EOL> except ValueError : <EOL> oldbinsha = Commit . NULL_BIN_SHA <EOL> fpath = self . abspath <EOL> assure_directory_exists ( fpath , is_file = True ) <EOL> lfd = LockedFD ( fpath ) <EOL> fd = lfd . open ( write = True , stream = True ) <EOL> fd . write ( write_value . encode ( '<STR_LIT:ascii>' ) ) <EOL> lfd . commit ( ) <EOL> if logmsg is not None : <EOL> self . log_append ( oldbinsha , logmsg ) <EOL> return self <EOL> reference = property ( _get_reference , set_reference , doc = "<STR_LIT>" ) <EOL> ref = reference <EOL> def is_valid ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . object <EOL> except ( OSError , ValueError ) : <EOL> return False <EOL> else : <EOL> return True <EOL> @ property <EOL> def is_detached ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . ref <EOL> return False <EOL> except TypeError : <EOL> return True <EOL> def log ( self ) : <EOL> """<STR_LIT>""" <EOL> return RefLog . from_file ( RefLog . path ( self ) ) <EOL> def log_append ( self , oldbinsha , message , newbinsha = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> committer_or_reader = self . commit . committer <EOL> except ValueError : <EOL> committer_or_reader = self . repo . config_reader ( ) <EOL> return RefLog . append_entry ( committer_or_reader , RefLog . path ( self ) , oldbinsha , <EOL> ( newbinsha is None and self . commit . binsha ) or newbinsha , <EOL> message ) <EOL> def log_entry ( self , index ) : <EOL> """<STR_LIT>""" <EOL> return RefLog . entry_at ( RefLog . path ( self ) , index ) <EOL> @ classmethod <EOL> def to_full_path ( cls , path ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( path , SymbolicReference ) : <EOL> path = path . path <EOL> full_ref_path = path <EOL> if not cls . _common_path_default : <EOL> return full_ref_path <EOL> if not path . startswith ( cls . _common_path_default + "<STR_LIT:/>" ) : <EOL> full_ref_path = '<STR_LIT>' % ( cls . _common_path_default , path ) <EOL> return full_ref_path <EOL> @ classmethod <EOL> def delete ( cls , repo , path ) : <EOL> """<STR_LIT>""" <EOL> full_ref_path = cls . to_full_path ( path ) <EOL> abs_path = join ( repo . git_dir , full_ref_path ) <EOL> if exists ( abs_path ) : <EOL> os . remove ( abs_path ) <EOL> else : <EOL> pack_file_path = cls . _get_packed_refs_path ( repo ) <EOL> try : <EOL> reader = open ( pack_file_path , '<STR_LIT:rb>' ) <EOL> except ( OSError , IOError ) : <EOL> pass <EOL> else : <EOL> new_lines = list ( ) <EOL> made_change = False <EOL> dropped_last_line = False <EOL> for line in reader : <EOL> line = line . decode ( defenc ) <EOL> if ( line . startswith ( '<STR_LIT:#>' ) or full_ref_path not in line ) and ( not dropped_last_line or dropped_last_line and not line . startswith ( '<STR_LIT>' ) ) : <EOL> new_lines . append ( line ) <EOL> dropped_last_line = False <EOL> continue <EOL> made_change = True <EOL> dropped_last_line = True <EOL> reader . close ( ) <EOL> if made_change : <EOL> open ( pack_file_path , '<STR_LIT:wb>' ) . writelines ( l . encode ( defenc ) for l in new_lines ) <EOL> reflog_path = RefLog . path ( cls ( repo , full_ref_path ) ) <EOL> if os . path . isfile ( reflog_path ) : <EOL> os . remove ( reflog_path ) <EOL> @ classmethod <EOL> def _create ( cls , repo , path , resolve , reference , force , logmsg = None ) : <EOL> """<STR_LIT>""" <EOL> full_ref_path = cls . to_full_path ( path ) <EOL> abs_ref_path = join ( repo . git_dir , full_ref_path ) <EOL> target = reference <EOL> if resolve : <EOL> target = repo . rev_parse ( str ( reference ) ) <EOL> if not force and isfile ( abs_ref_path ) : <EOL> target_data = str ( target ) <EOL> if isinstance ( target , SymbolicReference ) : <EOL> target_data = target . path <EOL> if not resolve : <EOL> target_data = "<STR_LIT>" + target_data <EOL> existing_data = open ( abs_ref_path , '<STR_LIT:rb>' ) . read ( ) . decode ( defenc ) . strip ( ) <EOL> if existing_data != target_data : <EOL> raise OSError ( "<STR_LIT>" % <EOL> ( full_ref_path , existing_data , target_data ) ) <EOL> ref = cls ( repo , full_ref_path ) <EOL> ref . set_reference ( target , logmsg ) <EOL> return ref <EOL> @ classmethod <EOL> def create ( cls , repo , path , reference = '<STR_LIT>' , force = False , logmsg = None ) : <EOL> """<STR_LIT>""" <EOL> return cls . _create ( repo , path , cls . _resolve_ref_on_create , reference , force , logmsg ) <EOL> def rename ( self , new_path , force = False ) : <EOL> """<STR_LIT>""" <EOL> new_path = self . to_full_path ( new_path ) <EOL> if self . path == new_path : <EOL> return self <EOL> new_abs_path = join ( self . repo . git_dir , new_path ) <EOL> cur_abs_path = join ( self . repo . git_dir , self . path ) <EOL> if isfile ( new_abs_path ) : <EOL> if not force : <EOL> if open ( new_abs_path , '<STR_LIT:rb>' ) . read ( ) . strip ( ) != open ( cur_abs_path , '<STR_LIT:rb>' ) . read ( ) . strip ( ) : <EOL> raise OSError ( "<STR_LIT>" % new_abs_path ) <EOL> os . remove ( new_abs_path ) <EOL> dname = dirname ( new_abs_path ) <EOL> if not isdir ( dname ) : <EOL> os . makedirs ( dname ) <EOL> rename ( cur_abs_path , new_abs_path ) <EOL> self . path = new_path <EOL> return self <EOL> @ classmethod <EOL> def _iter_items ( cls , repo , common_path = None ) : <EOL> if common_path is None : <EOL> common_path = cls . _common_path_default <EOL> rela_paths = set ( ) <EOL> for root , dirs , files in os . walk ( join_path_native ( repo . git_dir , common_path ) ) : <EOL> if '<STR_LIT>' not in root : <EOL> refs_id = [ d for d in dirs if d == '<STR_LIT>' ] <EOL> if refs_id : <EOL> dirs [ <NUM_LIT:0> : ] = [ '<STR_LIT>' ] <EOL> for f in files : <EOL> if f == '<STR_LIT>' : <EOL> continue <EOL> abs_path = to_native_path_linux ( join_path ( root , f ) ) <EOL> rela_paths . add ( abs_path . replace ( to_native_path_linux ( repo . git_dir ) + '<STR_LIT:/>' , "<STR_LIT>" ) ) <EOL> for sha , rela_path in cls . _iter_packed_refs ( repo ) : <EOL> if rela_path . startswith ( common_path ) : <EOL> rela_paths . add ( rela_path ) <EOL> for path in sorted ( rela_paths ) : <EOL> try : <EOL> yield cls . from_path ( repo , path ) <EOL> except ValueError : <EOL> continue <EOL> @ classmethod <EOL> def iter_items ( cls , repo , common_path = None ) : <EOL> """<STR_LIT>""" <EOL> return ( r for r in cls . _iter_items ( repo , common_path ) if r . __class__ == SymbolicReference or not r . is_detached ) <EOL> @ classmethod <EOL> def from_path ( cls , repo , path ) : <EOL> """<STR_LIT>""" <EOL> if not path : <EOL> raise ValueError ( "<STR_LIT>" % path ) <EOL> from . import HEAD , Head , RemoteReference , TagReference , Reference <EOL> for ref_type in ( HEAD , Head , RemoteReference , TagReference , Reference , SymbolicReference ) : <EOL> try : <EOL> instance = ref_type ( repo , path ) <EOL> if instance . __class__ == SymbolicReference and instance . is_detached : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return instance <EOL> except ValueError : <EOL> pass <EOL> raise ValueError ( "<STR_LIT>" % path ) <EOL> def is_remote ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . path . startswith ( self . _remote_common_path_default + "<STR_LIT:/>" ) </s>
