<s> import os <EOL> from six . moves import cStringIO as StringIO <EOL> from fnmatch import fnmatch <EOL> from . import vcs <EOL> from . log import get_logger <EOL> from . utils import is_blacklisted , rel_path_to_url <EOL> def chunks ( data , n ) : <EOL> for i in range ( <NUM_LIT:0> , len ( data ) - <NUM_LIT:1> , n ) : <EOL> yield data [ i : i + n ] <EOL> class TestTree ( object ) : <EOL> def __init__ ( self , tests_root , url_base ) : <EOL> self . tests_root = tests_root <EOL> self . url_base = url_base <EOL> self . logger = get_logger ( ) <EOL> def current_rev ( self ) : <EOL> pass <EOL> def local_changes ( self ) : <EOL> pass <EOL> def committed_changes ( self , base_rev = None ) : <EOL> pass <EOL> class GitTree ( TestTree ) : <EOL> def __init__ ( self , tests_root , url_base ) : <EOL> TestTree . __init__ ( self , tests_root , url_base ) <EOL> self . git = self . setup_git ( ) <EOL> def setup_git ( self ) : <EOL> assert vcs . is_git_repo ( self . tests_root ) <EOL> return vcs . get_git_func ( self . tests_root ) <EOL> def current_rev ( self ) : <EOL> return self . git ( "<STR_LIT>" , "<STR_LIT>" ) . strip ( ) <EOL> def local_changes ( self , path = None ) : <EOL> cmd = [ "<STR_LIT:status>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> if path is not None : <EOL> cmd . extend ( [ "<STR_LIT>" , path ] ) <EOL> rv = { } <EOL> data = self . git ( * cmd ) <EOL> if data == "<STR_LIT>" : <EOL> return rv <EOL> assert data [ - <NUM_LIT:1> ] == "<STR_LIT>" <EOL> f = StringIO ( data ) <EOL> while f . tell ( ) < len ( data ) : <EOL> staged = f . read ( <NUM_LIT:1> ) <EOL> worktree = f . read ( <NUM_LIT:1> ) <EOL> assert f . read ( <NUM_LIT:1> ) == "<STR_LIT:U+0020>" <EOL> if staged == "<STR_LIT:R>" : <EOL> files = <NUM_LIT:2> <EOL> else : <EOL> files = <NUM_LIT:1> <EOL> filenames = [ ] <EOL> for i in range ( files ) : <EOL> filenames . append ( "<STR_LIT>" ) <EOL> char = f . read ( <NUM_LIT:1> ) <EOL> while char != "<STR_LIT>" : <EOL> filenames [ - <NUM_LIT:1> ] += char <EOL> char = f . read ( <NUM_LIT:1> ) <EOL> if not is_blacklisted ( rel_path_to_url ( filenames [ <NUM_LIT:0> ] , self . url_base ) ) : <EOL> rv . update ( self . local_status ( staged , worktree , filenames ) ) <EOL> return rv <EOL> def committed_changes ( self , base_rev = None ) : <EOL> if base_rev is None : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> return [ ( item , "<STR_LIT>" ) for item in self . paths ( ) ] <EOL> self . logger . debug ( "<STR_LIT>" % ( base_rev , self . current_rev ( ) ) ) <EOL> rv = [ ] <EOL> data = self . git ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , base_rev + "<STR_LIT>" ) <EOL> items = data . split ( "<STR_LIT>" ) <EOL> for status , filename in chunks ( items , <NUM_LIT:2> ) : <EOL> if is_blacklisted ( rel_path_to_url ( filename , self . url_base ) ) : <EOL> continue <EOL> if status == "<STR_LIT:D>" : <EOL> rv . append ( ( filename , "<STR_LIT>" ) ) <EOL> else : <EOL> rv . append ( ( filename , "<STR_LIT>" ) ) <EOL> return rv <EOL> def paths ( self ) : <EOL> data = self . git ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> return [ item for item in data . split ( "<STR_LIT:\n>" ) if not item . endswith ( os . path . sep ) ] <EOL> def local_status ( self , staged , worktree , filenames ) : <EOL> rv = { } <EOL> if ( staged , worktree ) in [ ( "<STR_LIT:D>" , "<STR_LIT:D>" ) , ( "<STR_LIT:A>" , "<STR_LIT>" ) , ( "<STR_LIT>" , "<STR_LIT:D>" ) , ( "<STR_LIT>" , "<STR_LIT:A>" ) , <EOL> ( "<STR_LIT:D>" , "<STR_LIT>" ) , ( "<STR_LIT:A>" , "<STR_LIT:A>" ) , ( "<STR_LIT>" , "<STR_LIT>" ) ] : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> if staged == "<STR_LIT:R>" : <EOL> assert len ( filenames ) == <NUM_LIT:2> <EOL> dest , src = filenames <EOL> rv [ dest ] = "<STR_LIT>" <EOL> rv [ src ] = "<STR_LIT>" <EOL> else : <EOL> assert len ( filenames ) == <NUM_LIT:1> <EOL> filename = filenames [ <NUM_LIT:0> ] <EOL> if staged == "<STR_LIT:D>" or worktree == "<STR_LIT:D>" : <EOL> rv [ filename ] = "<STR_LIT>" <EOL> elif staged == "<STR_LIT:?>" and worktree == "<STR_LIT:?>" : <EOL> if os . path . isdir ( os . path . join ( self . tests_root , filename ) ) : <EOL> rv . update ( self . local_changes ( filename ) ) <EOL> else : <EOL> rv [ filename ] = "<STR_LIT>" <EOL> else : <EOL> rv [ filename ] = "<STR_LIT>" <EOL> return rv <EOL> class NoVCSTree ( TestTree ) : <EOL> """<STR_LIT>""" <EOL> ignore = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> def current_rev ( self ) : <EOL> return None <EOL> def local_changes ( self ) : <EOL> rv = [ ] <EOL> for dir_path , dir_names , filenames in os . walk ( self . tests_root ) : <EOL> for filename in filenames : <EOL> if any ( fnmatch ( filename , pattern ) for pattern in self . ignore ) : <EOL> continue <EOL> rel_path = os . path . relpath ( os . path . join ( dir_path , filename ) , <EOL> self . tests_root ) <EOL> if is_blacklisted ( rel_path_to_url ( rel_path , self . url_base ) ) : <EOL> continue <EOL> rv . append ( ( rel_path , "<STR_LIT>" ) ) <EOL> return dict ( rv ) <EOL> def committed_changes ( self , base_rev = None ) : <EOL> return None </s>
