<s> from __future__ import unicode_literals <EOL> import base64 <EOL> import json <EOL> import logging <EOL> import mimetools <EOL> from django . conf . urls import include , patterns , url <EOL> from django . dispatch import receiver <EOL> from django . utils import six <EOL> from django . utils . six . moves . urllib . parse import urlparse <EOL> from django . utils . six . moves . urllib . request import ( Request as BaseURLRequest , <EOL> HTTPBasicAuthHandler , <EOL> urlopen ) <EOL> from django . utils . translation import ugettext_lazy as _ <EOL> from djblets . registries . errors import ItemLookupError <EOL> from djblets . registries . registry import ( ALREADY_REGISTERED , LOAD_ENTRY_POINT , <EOL> NOT_REGISTERED ) <EOL> import reviewboard . hostingsvcs . urls as hostingsvcs_urls <EOL> from reviewboard . registries . registry import EntryPointRegistry <EOL> from reviewboard . signals import initializing <EOL> class URLRequest ( BaseURLRequest ) : <EOL> def __init__ ( self , url , body = '<STR_LIT>' , headers = { } , method = '<STR_LIT:GET>' ) : <EOL> BaseURLRequest . __init__ ( self , url , body , headers ) <EOL> self . method = method <EOL> def get_method ( self ) : <EOL> return self . method <EOL> class HostingServiceClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , hosting_service ) : <EOL> pass <EOL> def http_delete ( self , url , headers = { } , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . http_request ( url , headers = headers , method = '<STR_LIT>' , <EOL> ** kwargs ) <EOL> def http_get ( self , url , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . http_request ( url , method = '<STR_LIT:GET>' , ** kwargs ) <EOL> def http_post ( self , url , body = None , fields = { } , files = { } , content_type = None , <EOL> headers = { } , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> headers = headers . copy ( ) <EOL> if body is None : <EOL> if fields is not None : <EOL> body , content_type = self . _build_form_data ( fields , files ) <EOL> else : <EOL> body = '<STR_LIT>' <EOL> if content_type : <EOL> headers [ '<STR_LIT:Content-Type>' ] = content_type <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT>' % len ( body ) <EOL> return self . http_request ( url , body = body , headers = headers , <EOL> method = '<STR_LIT:POST>' , ** kwargs ) <EOL> def http_request ( self , url , body = None , headers = { } , method = '<STR_LIT:GET>' , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> r = self . _build_request ( url , body , headers , method = method , ** kwargs ) <EOL> u = urlopen ( r ) <EOL> return u . read ( ) , u . headers <EOL> def json_delete ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . _do_json_method ( self . http_delete , * args , ** kwargs ) <EOL> def json_get ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . _do_json_method ( self . http_get , * args , ** kwargs ) <EOL> def json_post ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . _do_json_method ( self . http_post , * args , ** kwargs ) <EOL> def _do_json_method ( self , method , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> data , headers = method ( * args , ** kwargs ) <EOL> if data : <EOL> data = json . loads ( data ) <EOL> return data , headers <EOL> def _build_request ( self , url , body = None , headers = { } , username = None , <EOL> password = None , method = '<STR_LIT:GET>' ) : <EOL> """<STR_LIT>""" <EOL> r = URLRequest ( url , body , headers , method = method ) <EOL> if username is not None and password is not None : <EOL> auth_key = username + '<STR_LIT::>' + password <EOL> r . add_header ( HTTPBasicAuthHandler . auth_header , <EOL> '<STR_LIT>' % <EOL> base64 . b64encode ( auth_key . encode ( '<STR_LIT:utf-8>' ) ) ) <EOL> return r <EOL> def _build_form_data ( self , fields , files ) : <EOL> """<STR_LIT>""" <EOL> BOUNDARY = mimetools . choose_boundary ( ) <EOL> content = "<STR_LIT>" <EOL> for key in fields : <EOL> content += "<STR_LIT>" + BOUNDARY + "<STR_LIT:\r\n>" <EOL> content += "<STR_LIT>" % key <EOL> content += "<STR_LIT:\r\n>" <EOL> content += six . text_type ( fields [ key ] ) + "<STR_LIT:\r\n>" <EOL> for key in files : <EOL> filename = files [ key ] [ '<STR_LIT:filename>' ] <EOL> value = files [ key ] [ '<STR_LIT:content>' ] <EOL> content += "<STR_LIT>" + BOUNDARY + "<STR_LIT:\r\n>" <EOL> content += "<STR_LIT>" % key <EOL> content += "<STR_LIT>" % filename <EOL> content += "<STR_LIT:\r\n>" <EOL> content += value + "<STR_LIT:\r\n>" <EOL> content += "<STR_LIT>" + BOUNDARY + "<STR_LIT>" <EOL> content += "<STR_LIT:\r\n>" <EOL> content_type = "<STR_LIT>" % BOUNDARY <EOL> return content , content_type <EOL> class HostingService ( object ) : <EOL> """<STR_LIT>""" <EOL> name = None <EOL> plans = None <EOL> supports_bug_trackers = False <EOL> supports_post_commit = False <EOL> supports_repositories = False <EOL> supports_ssh_key_association = False <EOL> supports_two_factor_auth = False <EOL> supports_list_remote_repositories = False <EOL> has_repository_hook_instructions = False <EOL> self_hosted = False <EOL> repository_url_patterns = None <EOL> client_class = HostingServiceClient <EOL> needs_authorization = False <EOL> supported_scmtools = [ ] <EOL> form = None <EOL> fields = [ ] <EOL> repository_fields = { } <EOL> bug_tracker_field = None <EOL> def __init__ ( self , account ) : <EOL> assert account <EOL> self . account = account <EOL> self . client = self . client_class ( self ) <EOL> def is_authorized ( self ) : <EOL> """<STR_LIT>""" <EOL> return False <EOL> def get_password ( self ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def is_ssh_key_associated ( self , repository , key ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def associate_ssh_key ( self , repository , key ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def authorize ( self , username , password , hosting_url , local_site_name = None , <EOL> * args , ** kwargs ) : <EOL> raise NotImplementedError <EOL> def check_repository ( self , path , username , password , scmtool_class , <EOL> local_site_name , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return scmtool_class . check_repository ( path , username , password , <EOL> local_site_name ) <EOL> def get_file ( self , repository , path , revision , * args , ** kwargs ) : <EOL> if not self . supports_repositories : <EOL> raise NotImplementedError <EOL> return repository . get_scmtool ( ) . get_file ( path , revision , ** kwargs ) <EOL> def get_file_exists ( self , repository , path , revision , * args , ** kwargs ) : <EOL> if not self . supports_repositories : <EOL> raise NotImplementedError <EOL> return repository . get_scmtool ( ) . file_exists ( path , revision , ** kwargs ) <EOL> def get_branches ( self , repository ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_commits ( self , repository , branch = None , start = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_change ( self , repository , revision ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_remote_repositories ( self , owner = None , owner_type = None , <EOL> filter_type = None , start = None , per_page = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_remote_repository ( self , repository_id ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> @ classmethod <EOL> def get_repository_fields ( cls , username , hosting_url , plan , tool_name , <EOL> field_vars ) : <EOL> if not cls . supports_repositories : <EOL> raise NotImplementedError <EOL> fields = cls . _get_field ( plan , '<STR_LIT>' ) <EOL> new_vars = field_vars . copy ( ) <EOL> new_vars [ '<STR_LIT>' ] = username <EOL> if cls . self_hosted : <EOL> new_vars [ '<STR_LIT>' ] = hosting_url <EOL> new_vars [ '<STR_LIT>' ] = urlparse ( hosting_url ) [ <NUM_LIT:1> ] <EOL> results = { } <EOL> assert tool_name in fields <EOL> for field , value in six . iteritems ( fields [ tool_name ] ) : <EOL> try : <EOL> results [ field ] = value % new_vars <EOL> except KeyError as e : <EOL> logging . error ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( field , six . text_type ( cls . name ) , value , <EOL> new_vars , e ) , <EOL> exc_info = <NUM_LIT:1> ) <EOL> raise KeyError ( <EOL> _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % { <EOL> '<STR_LIT>' : field , <EOL> '<STR_LIT:key>' : e , <EOL> } ) <EOL> return results <EOL> def get_repository_hook_instructions ( self , request , repository ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> @ classmethod <EOL> def get_bug_tracker_requires_username ( cls , plan = None ) : <EOL> if not cls . supports_bug_trackers : <EOL> raise NotImplementedError <EOL> return ( '<STR_LIT>' in <EOL> cls . _get_field ( plan , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> @ classmethod <EOL> def get_bug_tracker_field ( cls , plan , field_vars ) : <EOL> if not cls . supports_bug_trackers : <EOL> raise NotImplementedError <EOL> bug_tracker_field = cls . _get_field ( plan , '<STR_LIT>' ) <EOL> if not bug_tracker_field : <EOL> return '<STR_LIT>' <EOL> try : <EOL> return bug_tracker_field % field_vars <EOL> except KeyError as e : <EOL> logging . error ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( bug_tracker_field , six . text_type ( cls . name ) , <EOL> field_vars , e ) , <EOL> exc_info = <NUM_LIT:1> ) <EOL> raise KeyError ( <EOL> _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % { <EOL> '<STR_LIT>' : bug_tracker_field , <EOL> '<STR_LIT:key>' : e , <EOL> } ) <EOL> @ classmethod <EOL> def _get_field ( cls , plan , name , default = None ) : <EOL> if cls . plans : <EOL> assert plan <EOL> for plan_name , info in cls . plans : <EOL> if plan_name == plan and name in info : <EOL> return info [ name ] <EOL> return getattr ( cls , name , default ) <EOL> _hosting_services = { } <EOL> _hostingsvcs_urlpatterns = { } <EOL> _populated = False <EOL> class HostingServiceRegistry ( EntryPointRegistry ) : <EOL> """<STR_LIT>""" <EOL> entry_point = '<STR_LIT>' <EOL> lookup_attrs = [ '<STR_LIT:id>' ] <EOL> errors = { <EOL> ALREADY_REGISTERED : _ ( <EOL> '<STR_LIT>' <EOL> ) , <EOL> LOAD_ENTRY_POINT : _ ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) , <EOL> NOT_REGISTERED : _ ( <EOL> '<STR_LIT>' <EOL> ) , <EOL> } <EOL> def __init__ ( self ) : <EOL> super ( HostingServiceRegistry , self ) . __init__ ( ) <EOL> self . _url_patterns = { } <EOL> def unregister ( self , service ) : <EOL> """<STR_LIT>""" <EOL> super ( HostingServiceRegistry , self ) . unregister ( service ) <EOL> if service . id in self . _url_patterns : <EOL> cls_urlpatterns = self . _url_patterns [ service . id ] <EOL> hostingsvcs_urls . dynamic_urls . remove_patterns ( cls_urlpatterns ) <EOL> del self . _url_patterns [ service . id ] <EOL> def process_value_from_entry_point ( self , entry_point ) : <EOL> """<STR_LIT>""" <EOL> cls = entry_point . load ( ) <EOL> cls . id = entry_point . name <EOL> return cls <EOL> def register ( self , service ) : <EOL> """<STR_LIT>""" <EOL> super ( HostingServiceRegistry , self ) . register ( service ) <EOL> if service . repository_url_patterns : <EOL> cls_urlpatterns = patterns ( <EOL> '<STR_LIT>' , <EOL> url ( r'<STR_LIT>' + service . id + '<STR_LIT>' , <EOL> include ( service . repository_url_patterns ) ) ) <EOL> self . _url_patterns [ service . id ] = cls_urlpatterns <EOL> hostingsvcs_urls . dynamic_urls . add_patterns ( cls_urlpatterns ) <EOL> _hosting_service_registry = HostingServiceRegistry ( ) <EOL> def get_hosting_services ( ) : <EOL> """<STR_LIT>""" <EOL> return list ( _hosting_service_registry ) <EOL> def get_hosting_service ( name ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return _hosting_service_registry . get ( '<STR_LIT:id>' , name ) <EOL> except ItemLookupError : <EOL> return None <EOL> def register_hosting_service ( name , cls ) : <EOL> """<STR_LIT>""" <EOL> cls . id = name <EOL> _hosting_service_registry . register ( cls ) <EOL> def unregister_hosting_service ( name ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> _hosting_service_registry . unregister_by_attr ( '<STR_LIT:id>' , name ) <EOL> except ItemLookupError as e : <EOL> logging . error ( '<STR_LIT>' <EOL> % name ) <EOL> raise e <EOL> @ receiver ( initializing , dispatch_uid = '<STR_LIT>' ) <EOL> def _on_initializing ( ** kwargs ) : <EOL> _hosting_service_registry . populate ( ) </s>
