<s> from __future__ import unicode_literals <EOL> import base64 <EOL> import inspect <EOL> import logging <EOL> import os <EOL> import subprocess <EOL> import sys <EOL> import warnings <EOL> from django . utils import six <EOL> from django . utils . encoding import python_2_unicode_compatible <EOL> from django . utils . six . moves . urllib . error import HTTPError <EOL> from django . utils . six . moves . urllib . parse import urlparse <EOL> from django . utils . six . moves . urllib . request import ( Request as URLRequest , <EOL> urlopen ) <EOL> from django . utils . translation import ugettext_lazy as _ <EOL> import reviewboard . diffviewer . parser as diffparser <EOL> from reviewboard . scmtools . errors import ( AuthenticationError , <EOL> FileNotFoundError , <EOL> SCMError ) <EOL> from reviewboard . ssh import utils as sshutils <EOL> from reviewboard . ssh . errors import SSHAuthenticationError <EOL> class ChangeSet ( object ) : <EOL> def __init__ ( self ) : <EOL> self . changenum = None <EOL> self . summary = "<STR_LIT>" <EOL> self . description = "<STR_LIT>" <EOL> self . testing_done = "<STR_LIT>" <EOL> self . branch = "<STR_LIT>" <EOL> self . bugs_closed = [ ] <EOL> self . files = [ ] <EOL> self . username = "<STR_LIT>" <EOL> self . pending = False <EOL> @ python_2_unicode_compatible <EOL> class Revision ( object ) : <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> def __str__ ( self ) : <EOL> return self . name <EOL> def __eq__ ( self , other ) : <EOL> return self . name == six . text_type ( other ) <EOL> def __ne__ ( self , other ) : <EOL> return self . name != six . text_type ( other ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % self . name <EOL> class Branch ( object ) : <EOL> def __init__ ( self , id , name = None , commit = '<STR_LIT>' , default = False ) : <EOL> assert id <EOL> self . id = id <EOL> self . name = name or self . id <EOL> self . commit = commit <EOL> self . default = default <EOL> def __eq__ ( self , other ) : <EOL> return ( self . id == other . id and <EOL> self . name == other . name and <EOL> self . commit == other . commit and <EOL> self . default == other . default ) <EOL> def __repr__ ( self ) : <EOL> return ( '<STR_LIT>' <EOL> % ( self . id , self . name , self . commit , self . default ) ) <EOL> class Commit ( object ) : <EOL> def __init__ ( self , author_name = '<STR_LIT>' , id = '<STR_LIT>' , date = '<STR_LIT>' , message = '<STR_LIT>' , parent = '<STR_LIT>' , <EOL> diff = None , base_commit_id = None ) : <EOL> self . author_name = author_name <EOL> self . id = id <EOL> self . date = date <EOL> self . message = message <EOL> self . parent = parent <EOL> self . base_commit_id = base_commit_id <EOL> self . diff = diff <EOL> def __eq__ ( self , other ) : <EOL> return ( self . author_name == other . author_name and <EOL> self . id == other . id and <EOL> self . date == other . date and <EOL> self . message == other . message and <EOL> self . parent == other . parent ) <EOL> def __repr__ ( self ) : <EOL> return ( '<STR_LIT>' <EOL> % ( self . id , self . author_name , self . date , self . parent ) ) <EOL> def split_message ( self ) : <EOL> """<STR_LIT>""" <EOL> message = self . message <EOL> parts = message . split ( '<STR_LIT:\n>' , <NUM_LIT:1> ) <EOL> summary = parts [ <NUM_LIT:0> ] <EOL> try : <EOL> message = parts [ <NUM_LIT:1> ] <EOL> except IndexError : <EOL> pass <EOL> return summary , message <EOL> HEAD = Revision ( "<STR_LIT>" ) <EOL> UNKNOWN = Revision ( '<STR_LIT>' ) <EOL> PRE_CREATION = Revision ( "<STR_LIT>" ) <EOL> class SCMTool ( object ) : <EOL> name = None <EOL> supports_pending_changesets = False <EOL> supports_post_commit = False <EOL> supports_raw_file_urls = False <EOL> supports_ticket_auth = False <EOL> field_help_text = { <EOL> '<STR_LIT:path>' : _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> } <EOL> dependencies = { <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : [ ] , <EOL> } <EOL> def __init__ ( self , repository ) : <EOL> self . repository = repository <EOL> def get_file ( self , path , revision = None , base_commit_id = None , <EOL> ** kwargs ) : <EOL> raise NotImplementedError <EOL> def file_exists ( self , path , revision = HEAD , base_commit_id = None , <EOL> ** kwargs ) : <EOL> argspec = inspect . getargspec ( self . get_file ) <EOL> try : <EOL> if argspec . keywords is None : <EOL> warnings . warn ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % self . name , DeprecationWarning ) <EOL> self . get_file ( path , revision ) <EOL> else : <EOL> self . get_file ( path , revision , base_commit_id = base_commit_id ) <EOL> return True <EOL> except FileNotFoundError : <EOL> return False <EOL> def parse_diff_revision ( self , file_str , revision_str , moved = False , <EOL> copied = False , ** kwargs ) : <EOL> raise NotImplementedError <EOL> def get_diffs_use_absolute_paths ( self ) : <EOL> return False <EOL> def get_changeset ( self , changesetid , allow_empty = False ) : <EOL> raise NotImplementedError <EOL> def get_repository_info ( self ) : <EOL> raise NotImplementedError <EOL> def get_branches ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_commits ( self , branch = None , start = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_change ( self , revision ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_fields ( self ) : <EOL> raise NotImplementedError <EOL> def get_parser ( self , data ) : <EOL> return diffparser . DiffParser ( data ) <EOL> def normalize_path_for_display ( self , filename ) : <EOL> return filename <EOL> def normalize_patch ( self , patch , filename , revision ) : <EOL> """<STR_LIT>""" <EOL> return patch <EOL> @ classmethod <EOL> def popen ( cls , command , local_site_name = None ) : <EOL> """<STR_LIT>""" <EOL> env = os . environ . copy ( ) <EOL> if local_site_name : <EOL> env [ b'<STR_LIT>' ] = local_site_name . encode ( '<STR_LIT:utf-8>' ) <EOL> env [ b'<STR_LIT>' ] = ( '<STR_LIT::>' . join ( sys . path ) ) . encode ( '<STR_LIT:utf-8>' ) <EOL> return subprocess . Popen ( command , <EOL> env = env , <EOL> stderr = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> close_fds = ( os . name != '<STR_LIT>' ) ) <EOL> @ classmethod <EOL> def check_repository ( cls , path , username = None , password = None , <EOL> local_site_name = None ) : <EOL> """<STR_LIT>""" <EOL> if sshutils . is_ssh_uri ( path ) : <EOL> username , hostname = SCMTool . get_auth_from_uri ( path , username ) <EOL> logging . debug ( <EOL> "<STR_LIT>" <EOL> % ( cls . __name__ , hostname , username ) ) <EOL> try : <EOL> sshutils . check_host ( hostname , username , password , <EOL> local_site_name ) <EOL> except SSHAuthenticationError as e : <EOL> raise AuthenticationError ( e . allowed_types , six . text_type ( e ) , <EOL> e . user_key ) <EOL> except : <EOL> raise <EOL> @ classmethod <EOL> def get_auth_from_uri ( cls , path , username ) : <EOL> """<STR_LIT>""" <EOL> url = urlparse ( path ) <EOL> if '<STR_LIT:@>' in url [ <NUM_LIT:1> ] : <EOL> netloc_username , hostname = url [ <NUM_LIT:1> ] . split ( '<STR_LIT:@>' , <NUM_LIT:1> ) <EOL> else : <EOL> hostname = url [ <NUM_LIT:1> ] <EOL> netloc_username = None <EOL> if netloc_username : <EOL> return netloc_username , hostname <EOL> else : <EOL> return username , hostname <EOL> @ classmethod <EOL> def accept_certificate ( cls , path , username = None , password = None , <EOL> local_site_name = None , certificate = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> class SCMClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , path , username = None , password = None ) : <EOL> self . path = path <EOL> self . username = username <EOL> self . password = password <EOL> def get_file_http ( self , url , path , revision ) : <EOL> logging . info ( '<STR_LIT>' % url ) <EOL> try : <EOL> request = URLRequest ( url ) <EOL> if self . username : <EOL> auth_string = base64 . b64encode ( '<STR_LIT>' % ( self . username , <EOL> self . password ) ) <EOL> request . add_header ( '<STR_LIT>' , '<STR_LIT>' % auth_string ) <EOL> return urlopen ( request ) . read ( ) <EOL> except HTTPError as e : <EOL> if e . code == <NUM_LIT> : <EOL> logging . error ( '<STR_LIT>' ) <EOL> raise FileNotFoundError ( path , revision ) <EOL> else : <EOL> msg = "<STR_LIT>" % ( e . code , url , e ) <EOL> logging . error ( msg ) <EOL> raise SCMError ( msg ) <EOL> except Exception as e : <EOL> msg = "<STR_LIT>" % ( url , e ) <EOL> logging . error ( msg ) <EOL> raise SCMError ( msg ) </s>
