<s> """<STR_LIT>""" <EOL> import re <EOL> from cvs2svn_lib . common import InternalError <EOL> from cvs2svn_lib . common import FatalError <EOL> from cvs2svn_lib . context import Ctx <EOL> from cvs2svn_lib . log import logger <EOL> from cvs2svn_lib . symbol import Trunk <EOL> from cvs2svn_lib . symbol import Branch <EOL> from cvs2svn_lib . symbol import Tag <EOL> from cvs2svn_lib . symbol import ExcludedSymbol <EOL> from cvs2svn_lib . cvs_item import CVSRevision <EOL> from cvs2svn_lib . cvs_item import CVSRevisionModification <EOL> from cvs2svn_lib . cvs_item import CVSRevisionAdd <EOL> from cvs2svn_lib . cvs_item import CVSRevisionChange <EOL> from cvs2svn_lib . cvs_item import CVSRevisionAbsent <EOL> from cvs2svn_lib . cvs_item import CVSRevisionNoop <EOL> from cvs2svn_lib . cvs_item import CVSSymbol <EOL> from cvs2svn_lib . cvs_item import CVSBranch <EOL> from cvs2svn_lib . cvs_item import CVSTag <EOL> from cvs2svn_lib . cvs_item import cvs_revision_type_map <EOL> from cvs2svn_lib . cvs_item import cvs_branch_type_map <EOL> from cvs2svn_lib . cvs_item import cvs_tag_type_map <EOL> class VendorBranchError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class LODItems ( object ) : <EOL> def __init__ ( self , lod , cvs_branch , cvs_revisions , cvs_branches , cvs_tags ) : <EOL> self . lod = lod <EOL> self . cvs_branch = cvs_branch <EOL> self . cvs_revisions = cvs_revisions <EOL> self . cvs_branches = cvs_branches <EOL> self . cvs_tags = cvs_tags <EOL> def is_trivial_import ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( <EOL> len ( self . cvs_revisions ) == <NUM_LIT:1> <EOL> and self . cvs_revisions [ <NUM_LIT:0> ] . ntdbr <EOL> ) <EOL> def is_pure_ntdb ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( <EOL> self . cvs_revisions <EOL> and self . cvs_revisions [ - <NUM_LIT:1> ] . ntdbr <EOL> ) <EOL> def iter_blockers ( self ) : <EOL> if self . is_pure_ntdb ( ) : <EOL> pass <EOL> else : <EOL> non_ntdbr_revision_ids = set ( ) <EOL> for cvs_revision in self . cvs_revisions : <EOL> if not cvs_revision . ntdbr : <EOL> non_ntdbr_revision_ids . add ( cvs_revision . id ) <EOL> for cvs_tag in self . cvs_tags : <EOL> if cvs_tag . source_id in non_ntdbr_revision_ids : <EOL> yield cvs_tag <EOL> for cvs_branch in self . cvs_branches : <EOL> if cvs_branch . source_id in non_ntdbr_revision_ids : <EOL> yield cvs_branch <EOL> class CVSFileItems ( object ) : <EOL> def __init__ ( self , cvs_file , trunk , cvs_items , original_ids = None ) : <EOL> self . cvs_file = cvs_file <EOL> self . trunk = trunk <EOL> self . _cvs_items = { } <EOL> self . root_ids = set ( ) <EOL> for cvs_item in cvs_items : <EOL> self . add ( cvs_item ) <EOL> if isinstance ( cvs_item , CVSRevision ) and cvs_item . prev_id is None : <EOL> self . root_ids . add ( cvs_item . id ) <EOL> if original_ids is not None : <EOL> self . original_ids = original_ids <EOL> else : <EOL> self . original_ids = { } <EOL> for cvs_item in cvs_items : <EOL> if isinstance ( cvs_item , CVSRevision ) : <EOL> self . original_ids [ cvs_item . rev ] = cvs_item . id <EOL> def __getstate__ ( self ) : <EOL> return ( self . cvs_file . id , self . values ( ) , self . original_ids , ) <EOL> def __setstate__ ( self , state ) : <EOL> ( cvs_file_id , cvs_items , original_ids , ) = state <EOL> cvs_file = Ctx ( ) . _cvs_path_db . get_path ( cvs_file_id ) <EOL> CVSFileItems . __init__ ( <EOL> self , cvs_file , cvs_file . project . get_trunk ( ) , cvs_items , <EOL> original_ids = original_ids , <EOL> ) <EOL> def add ( self , cvs_item ) : <EOL> self . _cvs_items [ cvs_item . id ] = cvs_item <EOL> def __getitem__ ( self , id ) : <EOL> """<STR_LIT>""" <EOL> return self . _cvs_items [ id ] <EOL> def get ( self , id , default = None ) : <EOL> return self . _cvs_items . get ( id , default ) <EOL> def __delitem__ ( self , id ) : <EOL> assert id not in self . root_ids <EOL> del self . _cvs_items [ id ] <EOL> def values ( self ) : <EOL> return self . _cvs_items . values ( ) <EOL> def check_link_consistency ( self ) : <EOL> """<STR_LIT>""" <EOL> for cvs_item in self . values ( ) : <EOL> try : <EOL> cvs_item . check_links ( self ) <EOL> except AssertionError : <EOL> logger . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( cvs_item , ) ) <EOL> raise <EOL> def _get_lod ( self , lod , cvs_branch , start_id ) : <EOL> """<STR_LIT>""" <EOL> cvs_revisions = [ ] <EOL> cvs_branches = [ ] <EOL> cvs_tags = [ ] <EOL> def process_subitems ( cvs_item ) : <EOL> """<STR_LIT>""" <EOL> for branch_id in cvs_item . branch_ids [ : ] : <EOL> cvs_branches . append ( self [ branch_id ] ) <EOL> for tag_id in cvs_item . tag_ids : <EOL> cvs_tags . append ( self [ tag_id ] ) <EOL> if cvs_branch is not None : <EOL> process_subitems ( cvs_branch ) <EOL> id = start_id <EOL> while id is not None : <EOL> cvs_rev = self [ id ] <EOL> cvs_revisions . append ( cvs_rev ) <EOL> process_subitems ( cvs_rev ) <EOL> id = cvs_rev . next_id <EOL> return LODItems ( lod , cvs_branch , cvs_revisions , cvs_branches , cvs_tags ) <EOL> def get_lod_items ( self , cvs_branch ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_lod ( cvs_branch . symbol , cvs_branch , cvs_branch . next_id ) <EOL> def iter_root_lods ( self ) : <EOL> """<STR_LIT>""" <EOL> for id in list ( self . root_ids ) : <EOL> cvs_item = self [ id ] <EOL> if isinstance ( cvs_item , CVSRevision ) : <EOL> yield self . _get_lod ( cvs_item . lod , None , id ) <EOL> elif isinstance ( cvs_item , CVSBranch ) : <EOL> yield self . _get_lod ( cvs_item . symbol , cvs_item , cvs_item . next_id ) <EOL> else : <EOL> raise InternalError ( '<STR_LIT>' % ( cvs_item , ) ) <EOL> def _iter_tree ( self , lod , cvs_branch , start_id ) : <EOL> """<STR_LIT>""" <EOL> cvs_revisions = [ ] <EOL> cvs_branches = [ ] <EOL> cvs_tags = [ ] <EOL> def process_subitems ( cvs_item ) : <EOL> """<STR_LIT>""" <EOL> for branch_id in cvs_item . branch_ids [ : ] : <EOL> branch = self [ branch_id ] <EOL> for lod_items in self . _iter_tree ( <EOL> branch . symbol , branch , branch . next_id <EOL> ) : <EOL> yield lod_items <EOL> try : <EOL> cvs_branches . append ( self [ branch_id ] ) <EOL> except KeyError : <EOL> pass <EOL> for tag_id in cvs_item . tag_ids : <EOL> cvs_tags . append ( self [ tag_id ] ) <EOL> if cvs_branch is not None : <EOL> for lod_items in process_subitems ( cvs_branch ) : <EOL> yield lod_items <EOL> id = start_id <EOL> while id is not None : <EOL> cvs_rev = self [ id ] <EOL> cvs_revisions . append ( cvs_rev ) <EOL> for lod_items in process_subitems ( cvs_rev ) : <EOL> yield lod_items <EOL> id = cvs_rev . next_id <EOL> yield LODItems ( lod , cvs_branch , cvs_revisions , cvs_branches , cvs_tags ) <EOL> def iter_lods ( self ) : <EOL> """<STR_LIT>""" <EOL> for id in list ( self . root_ids ) : <EOL> cvs_item = self [ id ] <EOL> if isinstance ( cvs_item , CVSRevision ) : <EOL> lod = cvs_item . lod <EOL> cvs_branch = None <EOL> elif isinstance ( cvs_item , CVSBranch ) : <EOL> lod = cvs_item . symbol <EOL> id = cvs_item . next_id <EOL> cvs_branch = cvs_item <EOL> else : <EOL> raise InternalError ( '<STR_LIT>' % ( cvs_item , ) ) <EOL> for lod_items in self . _iter_tree ( lod , cvs_branch , id ) : <EOL> yield lod_items <EOL> def iter_deltatext_ancestors ( self , cvs_rev ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> if isinstance ( cvs_rev . lod , Trunk ) : <EOL> if cvs_rev . next_id is None : <EOL> return <EOL> else : <EOL> cvs_rev = self [ cvs_rev . next_id ] <EOL> else : <EOL> cvs_rev = self [ cvs_rev . prev_id ] <EOL> yield cvs_rev <EOL> def _sever_branch ( self , lod_items ) : <EOL> """<STR_LIT>""" <EOL> cvs_branch = lod_items . cvs_branch <EOL> if cvs_branch is None : <EOL> return <EOL> assert not cvs_branch . tag_ids <EOL> assert not cvs_branch . branch_ids <EOL> source_rev = self [ cvs_branch . source_id ] <EOL> assert isinstance ( source_rev , CVSRevision ) <EOL> lod_items . cvs_branch = None <EOL> del self [ cvs_branch . id ] <EOL> source_rev . branch_ids . remove ( cvs_branch . id ) <EOL> if lod_items . cvs_revisions : <EOL> first_rev = lod_items . cvs_revisions [ <NUM_LIT:0> ] <EOL> first_rev . first_on_branch_id = None <EOL> source_rev . branch_commit_ids . remove ( first_rev . id ) <EOL> first_rev . prev_id = None <EOL> first_rev . __class__ = cvs_revision_type_map [ <EOL> ( isinstance ( first_rev , CVSRevisionModification ) , False , ) <EOL> ] <EOL> self . root_ids . add ( first_rev . id ) <EOL> def adjust_ntdbrs ( self , ntdbr_cvs_revs ) : <EOL> """<STR_LIT>""" <EOL> for cvs_rev in ntdbr_cvs_revs : <EOL> cvs_rev . ntdbr = True <EOL> rev_1_1 = self [ ntdbr_cvs_revs [ <NUM_LIT:0> ] . prev_id ] <EOL> rev_1_2 = self . get ( rev_1_1 . next_id ) <EOL> if rev_1_2 is not None : <EOL> last_ntdbr = ntdbr_cvs_revs [ - <NUM_LIT:1> ] <EOL> rev_1_2 . ntdbr_prev_id = last_ntdbr . id <EOL> last_ntdbr . ntdbr_next_id = rev_1_2 . id <EOL> rev_1_2 . __class__ = cvs_revision_type_map [ ( <EOL> isinstance ( rev_1_2 , CVSRevisionModification ) , <EOL> isinstance ( last_ntdbr , CVSRevisionModification ) , <EOL> ) ] <EOL> def process_live_ntdb ( self , vendor_lod_items ) : <EOL> """<STR_LIT>""" <EOL> rev_1_1 = self [ vendor_lod_items . cvs_branch . source_id ] <EOL> rev_1_2_id = rev_1_1 . next_id <EOL> if rev_1_2_id is not None : <EOL> raise VendorBranchError ( <EOL> '<STR_LIT>' <EOL> % ( self . cvs_file . rcs_path , <EOL> vendor_lod_items . cvs_branch . branch_number , self [ rev_1_2_id ] . rev , ) <EOL> ) <EOL> ntdbr_cvs_revs = list ( vendor_lod_items . cvs_revisions ) <EOL> if ntdbr_cvs_revs : <EOL> self . adjust_ntdbrs ( ntdbr_cvs_revs ) <EOL> return True <EOL> else : <EOL> return False <EOL> def process_historical_ntdb ( self , vendor_lod_items ) : <EOL> """<STR_LIT>""" <EOL> rev_1_1 = self [ vendor_lod_items . cvs_branch . source_id ] <EOL> rev_1_2_id = rev_1_1 . next_id <EOL> if rev_1_2_id is None : <EOL> rev_1_2_timestamp = None <EOL> else : <EOL> rev_1_2_timestamp = self [ rev_1_2_id ] . timestamp <EOL> ntdbr_cvs_revs = [ ] <EOL> for cvs_rev in vendor_lod_items . cvs_revisions : <EOL> if rev_1_2_timestamp is not None and cvs_rev . timestamp >= rev_1_2_timestamp : <EOL> break <EOL> ntdbr_cvs_revs . append ( cvs_rev ) <EOL> if ntdbr_cvs_revs : <EOL> self . adjust_ntdbrs ( ntdbr_cvs_revs ) <EOL> return True <EOL> else : <EOL> return False <EOL> def imported_remove_1_1 ( self , vendor_lod_items ) : <EOL> """<STR_LIT>""" <EOL> assert vendor_lod_items . cvs_revisions <EOL> cvs_rev = vendor_lod_items . cvs_revisions [ <NUM_LIT:0> ] <EOL> if isinstance ( cvs_rev , CVSRevisionModification ) and not cvs_rev . deltatext_exists : <EOL> cvs_branch = vendor_lod_items . cvs_branch <EOL> rev_1_1 = self [ cvs_branch . source_id ] <EOL> assert isinstance ( rev_1_1 , CVSRevision ) <EOL> logger . debug ( '<STR_LIT>' % ( rev_1_1 , ) ) <EOL> self . _sever_branch ( vendor_lod_items ) <EOL> self . root_ids . remove ( rev_1_1 . id ) <EOL> del self [ rev_1_1 . id ] <EOL> rev_1_2_id = rev_1_1 . next_id <EOL> if rev_1_2_id is not None : <EOL> rev_1_2 = self [ rev_1_2_id ] <EOL> rev_1_2 . prev_id = None <EOL> self . root_ids . add ( rev_1_2 . id ) <EOL> cvs_rev . tag_ids . extend ( rev_1_1 . tag_ids ) <EOL> for id in rev_1_1 . tag_ids : <EOL> cvs_tag = self [ id ] <EOL> cvs_tag . source_lod = cvs_rev . lod <EOL> cvs_tag . source_id = cvs_rev . id <EOL> cvs_rev . branch_ids [ <NUM_LIT:0> : <NUM_LIT:0> ] = rev_1_1 . branch_ids <EOL> for id in rev_1_1 . branch_ids : <EOL> cvs_branch = self [ id ] <EOL> cvs_branch . source_lod = cvs_rev . lod <EOL> cvs_branch . source_id = cvs_rev . id <EOL> cvs_rev . branch_commit_ids [ <NUM_LIT:0> : <NUM_LIT:0> ] = rev_1_1 . branch_commit_ids <EOL> for id in rev_1_1 . branch_commit_ids : <EOL> cvs_rev2 = self [ id ] <EOL> cvs_rev2 . prev_id = cvs_rev . id <EOL> def _is_unneeded_initial_trunk_delete ( self , cvs_item , metadata_db ) : <EOL> if not isinstance ( cvs_item , CVSRevisionNoop ) : <EOL> return False <EOL> if cvs_item . rev != '<STR_LIT>' : <EOL> return False <EOL> if not isinstance ( cvs_item . lod , Trunk ) : <EOL> return False <EOL> if cvs_item . closed_symbols : <EOL> return False <EOL> if cvs_item . ntdbr : <EOL> return False <EOL> log_msg = metadata_db [ cvs_item . metadata_id ] . log_msg <EOL> return bool ( <EOL> re . match ( <EOL> r'<STR_LIT>' , <EOL> log_msg , <EOL> ) <EOL> or re . match ( <EOL> r'<STR_LIT>' , <EOL> log_msg , <EOL> ) <EOL> ) <EOL> def remove_unneeded_initial_trunk_delete ( self , metadata_db ) : <EOL> """<STR_LIT>""" <EOL> for id in self . root_ids : <EOL> cvs_item = self [ id ] <EOL> if self . _is_unneeded_initial_trunk_delete ( cvs_item , metadata_db ) : <EOL> logger . debug ( '<STR_LIT>' % ( cvs_item , ) ) <EOL> for cvs_branch_id in cvs_item . branch_ids [ : ] : <EOL> cvs_branch = self [ cvs_branch_id ] <EOL> self . _sever_branch ( self . get_lod_items ( cvs_branch ) ) <EOL> while cvs_item . tag_ids : <EOL> del self [ cvs_item . tag_ids . pop ( ) ] <EOL> self . root_ids . remove ( cvs_item . id ) <EOL> del self [ cvs_item . id ] <EOL> if cvs_item . next_id is not None : <EOL> cvs_rev_next = self [ cvs_item . next_id ] <EOL> cvs_rev_next . prev_id = None <EOL> self . root_ids . add ( cvs_rev_next . id ) <EOL> return <EOL> def _is_unneeded_initial_branch_delete ( self , lod_items , metadata_db ) : <EOL> """<STR_LIT>""" <EOL> if not lod_items . cvs_revisions : <EOL> return False <EOL> cvs_revision = lod_items . cvs_revisions [ <NUM_LIT:0> ] <EOL> if cvs_revision . ntdbr : <EOL> return False <EOL> if not isinstance ( cvs_revision , CVSRevisionAbsent ) : <EOL> return False <EOL> if cvs_revision . branch_ids : <EOL> return False <EOL> log_msg = metadata_db [ cvs_revision . metadata_id ] . log_msg <EOL> return bool ( re . match ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> log_msg , <EOL> ) ) <EOL> def remove_initial_branch_deletes ( self , metadata_db ) : <EOL> """<STR_LIT>""" <EOL> for lod_items in self . iter_lods ( ) : <EOL> if self . _is_unneeded_initial_branch_delete ( lod_items , metadata_db ) : <EOL> cvs_revision = lod_items . cvs_revisions [ <NUM_LIT:0> ] <EOL> logger . debug ( <EOL> '<STR_LIT>' % ( cvs_revision , ) <EOL> ) <EOL> self . _sever_branch ( lod_items ) <EOL> self . root_ids . remove ( cvs_revision . id ) <EOL> del self [ cvs_revision . id ] <EOL> if cvs_revision . next_id is not None : <EOL> cvs_rev_next = self [ cvs_revision . next_id ] <EOL> cvs_rev_next . prev_id = None <EOL> self . root_ids . add ( cvs_rev_next . id ) <EOL> for tag_id in cvs_revision . tag_ids : <EOL> del self [ tag_id ] <EOL> def _exclude_tag ( self , cvs_tag ) : <EOL> """<STR_LIT>""" <EOL> del self [ cvs_tag . id ] <EOL> self [ cvs_tag . source_id ] . tag_ids . remove ( cvs_tag . id ) <EOL> def _exclude_branch ( self , lod_items ) : <EOL> """<STR_LIT>""" <EOL> if lod_items . cvs_revisions and lod_items . cvs_revisions [ <NUM_LIT:0> ] . ntdbr : <EOL> for cvs_rev in lod_items . cvs_revisions : <EOL> if not cvs_rev . ntdbr : <EOL> break <EOL> else : <EOL> cvs_rev = None <EOL> if cvs_rev : <EOL> last_ntdbr = self [ cvs_rev . prev_id ] <EOL> last_ntdbr . next_id = None <EOL> while True : <EOL> del self [ cvs_rev . id ] <EOL> if cvs_rev . next_id is None : <EOL> break <EOL> cvs_rev = self [ cvs_rev . next_id ] <EOL> return True <EOL> else : <EOL> if lod_items . cvs_branch is not None : <EOL> cvs_branch = lod_items . cvs_branch <EOL> del self [ cvs_branch . id ] <EOL> self [ cvs_branch . source_id ] . branch_ids . remove ( cvs_branch . id ) <EOL> if lod_items . cvs_revisions : <EOL> cvs_rev = lod_items . cvs_revisions [ <NUM_LIT:0> ] <EOL> if cvs_rev . prev_id is None : <EOL> self . root_ids . remove ( cvs_rev . id ) <EOL> else : <EOL> self [ cvs_rev . prev_id ] . branch_commit_ids . remove ( cvs_rev . id ) <EOL> for cvs_rev in lod_items . cvs_revisions : <EOL> del self [ cvs_rev . id ] <EOL> return False <EOL> def graft_ntdbr_to_trunk ( self ) : <EOL> """<STR_LIT>""" <EOL> for lod_items in self . iter_lods ( ) : <EOL> if lod_items . cvs_revisions and lod_items . cvs_revisions [ <NUM_LIT:0> ] . ntdbr : <EOL> assert lod_items . is_pure_ntdb ( ) <EOL> first_rev = lod_items . cvs_revisions [ <NUM_LIT:0> ] <EOL> last_rev = lod_items . cvs_revisions [ - <NUM_LIT:1> ] <EOL> rev_1_1 = self . get ( first_rev . prev_id ) <EOL> rev_1_2 = self . get ( last_rev . ntdbr_next_id ) <EOL> self . _sever_branch ( lod_items ) <EOL> if rev_1_1 is not None : <EOL> rev_1_1 . next_id = first_rev . id <EOL> first_rev . prev_id = rev_1_1 . id <EOL> self . root_ids . remove ( first_rev . id ) <EOL> first_rev . __class__ = cvs_revision_type_map [ ( <EOL> isinstance ( first_rev , CVSRevisionModification ) , <EOL> isinstance ( rev_1_1 , CVSRevisionModification ) , <EOL> ) ] <EOL> if rev_1_2 is not None : <EOL> rev_1_2 . ntdbr_prev_id = None <EOL> last_rev . ntdbr_next_id = None <EOL> if rev_1_2 . prev_id is None : <EOL> self . root_ids . remove ( rev_1_2 . id ) <EOL> rev_1_2 . prev_id = last_rev . id <EOL> last_rev . next_id = rev_1_2 . id <EOL> for cvs_rev in lod_items . cvs_revisions : <EOL> cvs_rev . ntdbr = False <EOL> cvs_rev . lod = self . trunk <EOL> for cvs_branch in lod_items . cvs_branches : <EOL> cvs_branch . source_lod = self . trunk <EOL> for cvs_tag in lod_items . cvs_tags : <EOL> cvs_tag . source_lod = self . trunk <EOL> return <EOL> def exclude_non_trunk ( self ) : <EOL> """<STR_LIT>""" <EOL> ntdbr_excluded = False <EOL> for lod_items in self . iter_lods ( ) : <EOL> for cvs_tag in lod_items . cvs_tags [ : ] : <EOL> self . _exclude_tag ( cvs_tag ) <EOL> lod_items . cvs_tags . remove ( cvs_tag ) <EOL> if not isinstance ( lod_items . lod , Trunk ) : <EOL> assert not lod_items . cvs_branches <EOL> ntdbr_excluded |= self . _exclude_branch ( lod_items ) <EOL> if ntdbr_excluded : <EOL> self . graft_ntdbr_to_trunk ( ) <EOL> def filter_excluded_symbols ( self ) : <EOL> """<STR_LIT>""" <EOL> ntdbr_excluded = False <EOL> for lod_items in self . iter_lods ( ) : <EOL> for cvs_tag in lod_items . cvs_tags [ : ] : <EOL> if isinstance ( cvs_tag . symbol , ExcludedSymbol ) : <EOL> self . _exclude_tag ( cvs_tag ) <EOL> lod_items . cvs_tags . remove ( cvs_tag ) <EOL> if isinstance ( lod_items . lod , ExcludedSymbol ) : <EOL> assert not list ( lod_items . iter_blockers ( ) ) <EOL> ntdbr_excluded |= self . _exclude_branch ( lod_items ) <EOL> if ntdbr_excluded : <EOL> self . graft_ntdbr_to_trunk ( ) <EOL> def _mutate_branch_to_tag ( self , cvs_branch ) : <EOL> """<STR_LIT>""" <EOL> if cvs_branch . next_id is not None : <EOL> raise FatalError ( '<STR_LIT>' ) <EOL> cvs_tag = CVSTag ( <EOL> cvs_branch . id , cvs_branch . cvs_file , cvs_branch . symbol , <EOL> cvs_branch . source_lod , cvs_branch . source_id , <EOL> cvs_branch . revision_reader_token , <EOL> ) <EOL> self . add ( cvs_tag ) <EOL> cvs_revision = self [ cvs_tag . source_id ] <EOL> cvs_revision . branch_ids . remove ( cvs_tag . id ) <EOL> cvs_revision . tag_ids . append ( cvs_tag . id ) <EOL> def _mutate_tag_to_branch ( self , cvs_tag ) : <EOL> """<STR_LIT>""" <EOL> cvs_branch = CVSBranch ( <EOL> cvs_tag . id , cvs_tag . cvs_file , cvs_tag . symbol , <EOL> None , cvs_tag . source_lod , cvs_tag . source_id , None , <EOL> cvs_tag . revision_reader_token , <EOL> ) <EOL> self . add ( cvs_branch ) <EOL> cvs_revision = self [ cvs_branch . source_id ] <EOL> cvs_revision . tag_ids . remove ( cvs_branch . id ) <EOL> cvs_revision . branch_ids . append ( cvs_branch . id ) <EOL> def _mutate_symbol ( self , cvs_symbol ) : <EOL> """<STR_LIT>""" <EOL> symbol = cvs_symbol . symbol <EOL> if isinstance ( cvs_symbol , CVSBranch ) and isinstance ( symbol , Tag ) : <EOL> self . _mutate_branch_to_tag ( cvs_symbol ) <EOL> elif isinstance ( cvs_symbol , CVSTag ) and isinstance ( symbol , Branch ) : <EOL> self . _mutate_tag_to_branch ( cvs_symbol ) <EOL> def mutate_symbols ( self ) : <EOL> """<STR_LIT>""" <EOL> for cvs_item in self . values ( ) : <EOL> if isinstance ( cvs_item , CVSRevision ) : <EOL> pass <EOL> elif isinstance ( cvs_item , CVSSymbol ) : <EOL> self . _mutate_symbol ( cvs_item ) <EOL> else : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> def _adjust_tag_parent ( self , cvs_tag ) : <EOL> """<STR_LIT>""" <EOL> preferred_parent = Ctx ( ) . _symbol_db . get_symbol ( <EOL> cvs_tag . symbol . preferred_parent_id ) <EOL> if cvs_tag . source_lod == preferred_parent : <EOL> return <EOL> source = self [ cvs_tag . source_id ] <EOL> assert isinstance ( source , CVSRevision ) <EOL> if isinstance ( preferred_parent , Trunk ) : <EOL> return <EOL> for branch_id in source . branch_ids : <EOL> if self [ branch_id ] . symbol == preferred_parent : <EOL> break <EOL> else : <EOL> return <EOL> parent = self [ branch_id ] <EOL> assert isinstance ( parent , CVSBranch ) <EOL> logger . debug ( '<STR_LIT>' % ( <EOL> cvs_tag , source , source . lod , parent , ) ) <EOL> source . tag_ids . remove ( cvs_tag . id ) <EOL> parent . tag_ids . append ( cvs_tag . id ) <EOL> cvs_tag . source_lod = parent . symbol <EOL> cvs_tag . source_id = parent . id <EOL> def _adjust_branch_parents ( self , cvs_branch ) : <EOL> """<STR_LIT>""" <EOL> preferred_parent = Ctx ( ) . _symbol_db . get_symbol ( <EOL> cvs_branch . symbol . preferred_parent_id ) <EOL> if cvs_branch . source_lod == preferred_parent : <EOL> return <EOL> source = self [ cvs_branch . source_id ] <EOL> assert isinstance ( source , CVSRevision ) <EOL> if isinstance ( preferred_parent , Trunk ) : <EOL> return <EOL> for branch_id in source . branch_ids : <EOL> possible_parent = self [ branch_id ] <EOL> if possible_parent . symbol == preferred_parent : <EOL> break <EOL> elif possible_parent . symbol == cvs_branch . symbol : <EOL> return <EOL> else : <EOL> raise InternalError ( <EOL> '<STR_LIT>' ) <EOL> parent = possible_parent <EOL> assert isinstance ( parent , CVSBranch ) <EOL> logger . debug ( '<STR_LIT>' % ( <EOL> cvs_branch , source , source . lod , parent , ) ) <EOL> source . branch_ids . remove ( cvs_branch . id ) <EOL> parent . branch_ids . append ( cvs_branch . id ) <EOL> cvs_branch . source_lod = parent . symbol <EOL> cvs_branch . source_id = parent . id <EOL> def adjust_parents ( self ) : <EOL> """<STR_LIT>""" <EOL> for lod_items in self . iter_lods ( ) : <EOL> for cvs_tag in lod_items . cvs_tags : <EOL> self . _adjust_tag_parent ( cvs_tag ) <EOL> for cvs_branch in reversed ( lod_items . cvs_branches ) : <EOL> self . _adjust_branch_parents ( cvs_branch ) <EOL> def _get_revision_source ( self , cvs_symbol ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> cvs_item = self [ cvs_symbol . source_id ] <EOL> if isinstance ( cvs_item , CVSRevision ) : <EOL> return cvs_item <EOL> else : <EOL> cvs_symbol = cvs_item <EOL> def refine_symbols ( self ) : <EOL> """<STR_LIT>""" <EOL> for lod_items in self . iter_lods ( ) : <EOL> for cvs_tag in lod_items . cvs_tags : <EOL> source = self . _get_revision_source ( cvs_tag ) <EOL> cvs_tag . __class__ = cvs_tag_type_map [ <EOL> isinstance ( source , CVSRevisionModification ) <EOL> ] <EOL> for cvs_branch in lod_items . cvs_branches : <EOL> source = self . _get_revision_source ( cvs_branch ) <EOL> cvs_branch . __class__ = cvs_branch_type_map [ <EOL> isinstance ( source , CVSRevisionModification ) <EOL> ] <EOL> def determine_revision_properties ( self , revision_property_setters ) : <EOL> """<STR_LIT>""" <EOL> for lod_items in self . iter_lods ( ) : <EOL> for cvs_rev in lod_items . cvs_revisions : <EOL> cvs_rev . properties = { } <EOL> for revision_property_setter in revision_property_setters : <EOL> revision_property_setter . set_properties ( cvs_rev ) <EOL> for lod_items in self . iter_lods ( ) : <EOL> for cvs_rev in lod_items . cvs_revisions : <EOL> if isinstance ( cvs_rev , CVSRevisionAdd ) : <EOL> cvs_rev . properties_changed = True <EOL> elif isinstance ( cvs_rev , CVSRevisionChange ) : <EOL> prev_properties = self [ <EOL> cvs_rev . get_effective_prev_id ( ) <EOL> ] . get_properties ( ) <EOL> properties = cvs_rev . get_properties ( ) <EOL> cvs_rev . properties_changed = properties != prev_properties <EOL> else : <EOL> cvs_rev . properties_changed = False <EOL> def record_opened_symbols ( self ) : <EOL> """<STR_LIT>""" <EOL> for cvs_item in self . values ( ) : <EOL> if isinstance ( cvs_item , ( CVSRevision , CVSBranch ) ) : <EOL> cvs_item . opened_symbols = [ ] <EOL> for cvs_symbol_opened_id in cvs_item . get_cvs_symbol_ids_opened ( ) : <EOL> cvs_symbol_opened = self [ cvs_symbol_opened_id ] <EOL> cvs_item . opened_symbols . append ( <EOL> ( cvs_symbol_opened . symbol . id , cvs_symbol_opened . id , ) <EOL> ) <EOL> def record_closed_symbols ( self ) : <EOL> """<STR_LIT>""" <EOL> for cvs_item in self . values ( ) : <EOL> if isinstance ( cvs_item , CVSRevision ) : <EOL> cvs_item . closed_symbols = [ ] <EOL> for cvs_item_closed_id in cvs_item . get_ids_closed ( ) : <EOL> cvs_item_closed = self [ cvs_item_closed_id ] <EOL> cvs_item . closed_symbols . extend ( cvs_item_closed . opened_symbols ) </s>
