<s> from __future__ import absolute_import , unicode_literals <EOL> import logging <EOL> import os <EOL> from datetime import datetime <EOL> from shutil import rmtree <EOL> from tempfile import mkdtemp <EOL> try : <EOL> import pysvn <EOL> from pysvn import ClientError , Revision , opt_revision_kind <EOL> has_svn_backend = True <EOL> except ImportError : <EOL> has_svn_backend = False <EOL> from django . utils import six <EOL> from django . utils . datastructures import SortedDict <EOL> from django . utils . six . moves . urllib . parse import ( urlsplit , urlunsplit , quote ) <EOL> from django . utils . translation import ugettext as _ <EOL> from reviewboard . scmtools . core import HEAD , PRE_CREATION <EOL> from reviewboard . scmtools . errors import FileNotFoundError , SCMError <EOL> from reviewboard . scmtools . svn import base , SVNTool <EOL> class Client ( base . Client ) : <EOL> required_module = '<STR_LIT>' <EOL> def __init__ ( self , config_dir , repopath , username = None , password = None ) : <EOL> super ( Client , self ) . __init__ ( config_dir , repopath , username , password ) <EOL> self . client = pysvn . Client ( config_dir ) <EOL> if username : <EOL> self . client . set_default_username ( six . text_type ( username ) ) <EOL> if password : <EOL> self . client . set_default_password ( password ) <EOL> def set_ssl_server_trust_prompt ( self , cb ) : <EOL> self . client . callback_ssl_server_trust_prompt = cb <EOL> def ssl_trust_prompt ( self , trust_dict ) : <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> return self . callback_ssl_server_trust_prompt ( trust_dict ) <EOL> def _do_on_path ( self , cb , path , revision = HEAD ) : <EOL> if not path : <EOL> raise FileNotFoundError ( path , revision ) <EOL> try : <EOL> normpath = self . normalize_path ( path ) <EOL> if self . client . is_url ( normpath ) : <EOL> pathtuple = urlsplit ( normpath ) <EOL> path = pathtuple [ <NUM_LIT:2> ] <EOL> if isinstance ( path , six . text_type ) : <EOL> path = path . encode ( '<STR_LIT:utf-8>' , '<STR_LIT:ignore>' ) <EOL> normpath = urlunsplit ( ( pathtuple [ <NUM_LIT:0> ] , <EOL> pathtuple [ <NUM_LIT:1> ] , <EOL> quote ( path ) , <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> normrev = self . _normalize_revision ( revision ) <EOL> return cb ( normpath , normrev ) <EOL> except ClientError as e : <EOL> exc = bytes ( e ) . decode ( '<STR_LIT:utf-8>' ) <EOL> if '<STR_LIT>' in exc or '<STR_LIT>' in exc : <EOL> raise FileNotFoundError ( path , revision , detail = exc ) <EOL> elif '<STR_LIT>' in exc : <EOL> raise SCMError ( <EOL> _ ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> % os . path . join ( self . config_dir , '<STR_LIT>' ) ) <EOL> else : <EOL> raise SVNTool . normalize_error ( e ) <EOL> def _get_file_data ( self , normpath , normrev ) : <EOL> data = self . client . cat ( normpath , normrev ) <EOL> keywords = self . client . propget ( "<STR_LIT>" , normpath , normrev , <EOL> recurse = True ) <EOL> if normpath in keywords : <EOL> data = self . collapse_keywords ( data , keywords [ normpath ] ) <EOL> return data <EOL> def get_file ( self , path , revision = HEAD ) : <EOL> """<STR_LIT>""" <EOL> return self . _do_on_path ( self . _get_file_data , path , revision ) <EOL> def _get_file_keywords ( self , normpath , normrev ) : <EOL> keywords = self . client . propget ( "<STR_LIT>" , normpath , normrev , <EOL> recurse = True ) <EOL> return keywords . get ( normpath ) <EOL> def get_keywords ( self , path , revision = HEAD ) : <EOL> """<STR_LIT>""" <EOL> return self . _do_on_path ( self . _get_file_keywords , path , revision ) <EOL> def _normalize_revision ( self , revision ) : <EOL> if revision == HEAD : <EOL> r = Revision ( opt_revision_kind . head ) <EOL> elif revision == PRE_CREATION : <EOL> raise FileNotFoundError ( '<STR_LIT>' , revision ) <EOL> else : <EOL> r = Revision ( opt_revision_kind . number , six . text_type ( revision ) ) <EOL> return r <EOL> @ property <EOL> def repository_info ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> info = self . client . info2 ( self . repopath , recurse = False ) <EOL> except ClientError as e : <EOL> raise SVNTool . normalize_error ( e ) <EOL> return { <EOL> '<STR_LIT>' : info [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . repos_UUID , <EOL> '<STR_LIT>' : info [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . repos_root_URL , <EOL> '<STR_LIT:url>' : info [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . URL <EOL> } <EOL> def accept_ssl_certificate ( self , path , on_failure = None ) : <EOL> """<STR_LIT>""" <EOL> cert = { } <EOL> def ssl_server_trust_prompt ( trust_dict ) : <EOL> cert . update ( trust_dict . copy ( ) ) <EOL> if on_failure : <EOL> return False , <NUM_LIT:0> , False <EOL> else : <EOL> del cert [ '<STR_LIT>' ] <EOL> return True , trust_dict [ '<STR_LIT>' ] , True <EOL> self . client . callback_ssl_server_trust_prompt = ssl_server_trust_prompt <EOL> try : <EOL> info = self . client . info2 ( path , recurse = False ) <EOL> logging . debug ( '<STR_LIT>' % <EOL> ( path , info ) ) <EOL> except ClientError as e : <EOL> if on_failure : <EOL> on_failure ( e , path , cert ) <EOL> return cert <EOL> def get_log ( self , path , start = None , end = None , limit = None , <EOL> discover_changed_paths = False , limit_to_path = False ) : <EOL> """<STR_LIT>""" <EOL> if start is None : <EOL> start = self . LOG_DEFAULT_START <EOL> if end is None : <EOL> end = self . LOG_DEFAULT_END <EOL> commits = self . client . log ( <EOL> self . normalize_path ( path ) , <EOL> limit = limit , <EOL> revision_start = self . _normalize_revision ( start ) , <EOL> revision_end = self . _normalize_revision ( end ) , <EOL> discover_changed_paths = discover_changed_paths , <EOL> strict_node_history = limit_to_path ) <EOL> for commit in commits : <EOL> commit [ '<STR_LIT>' ] = six . text_type ( commit [ '<STR_LIT>' ] . number ) <EOL> if '<STR_LIT:date>' in commit : <EOL> commit [ '<STR_LIT:date>' ] = datetime . utcfromtimestamp ( commit [ '<STR_LIT:date>' ] ) <EOL> return commits <EOL> def list_dir ( self , path ) : <EOL> """<STR_LIT>""" <EOL> result = SortedDict ( ) <EOL> norm_path = self . normalize_path ( path ) <EOL> dirents = self . client . list ( norm_path , recurse = False ) [ <NUM_LIT:1> : ] <EOL> repo_path_len = len ( self . repopath ) <EOL> for dirent , unused in dirents : <EOL> name = dirent [ '<STR_LIT:path>' ] . split ( '<STR_LIT:/>' ) [ - <NUM_LIT:1> ] <EOL> result [ name ] = { <EOL> '<STR_LIT:path>' : dirent [ '<STR_LIT:path>' ] [ repo_path_len : ] , <EOL> '<STR_LIT>' : six . text_type ( dirent [ '<STR_LIT>' ] . number ) , <EOL> } <EOL> return result <EOL> def diff ( self , revision1 , revision2 , path = None ) : <EOL> """<STR_LIT>""" <EOL> if path : <EOL> path = self . normalize_path ( path ) <EOL> else : <EOL> path = self . repopath <EOL> tmpdir = mkdtemp ( prefix = '<STR_LIT>' ) <EOL> try : <EOL> diff = self . client . diff ( <EOL> tmpdir , <EOL> path , <EOL> revision1 = self . _normalize_revision ( revision1 ) , <EOL> revision2 = self . _normalize_revision ( revision2 ) , <EOL> header_encoding = '<STR_LIT>' , <EOL> diff_options = [ '<STR_LIT>' ] ) <EOL> except Exception as e : <EOL> logging . error ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> revision1 , revision2 , path , e , exc_info = <NUM_LIT:1> ) <EOL> raise SCMError ( <EOL> _ ( '<STR_LIT>' ) <EOL> % ( revision1 , revision2 , e ) ) <EOL> finally : <EOL> rmtree ( tmpdir ) <EOL> return diff </s>
