<s> """<STR_LIT>""" <EOL> from __future__ import with_statement , print_function <EOL> import re <EOL> import os <EOL> import sys <EOL> import cgi <EOL> import netrc <EOL> import socket <EOL> import fnmatch <EOL> import getpass <EOL> import httplib <EOL> import urllib2 <EOL> import urlparse <EOL> import unittest <EOL> from contextlib import closing <EOL> from email import parser as rfc2822_parser <EOL> try : <EOL> import dputhelper <EOL> except ImportError : <EOL> sys . path . insert ( <NUM_LIT:0> , "<STR_LIT>" ) <EOL> import dputhelper <EOL> CHUNK_SIZE = <NUM_LIT:16> * <NUM_LIT> <EOL> def trace ( msg , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if trace . debug : <EOL> print ( "<STR_LIT>" + ( msg % kwargs ) ) <EOL> trace . debug = False <EOL> def log ( msg , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> sys . stdout . flush ( ) <EOL> sys . stderr . write ( "<STR_LIT>" + ( msg % kwargs ) + "<STR_LIT:\n>" ) <EOL> sys . stderr . flush ( ) <EOL> def _resolve_credentials ( fqdn , login ) : <EOL> """<STR_LIT>""" <EOL> result = login <EOL> if "<STR_LIT:$>" in result : <EOL> result = os . path . expandvars ( result ) <EOL> if result . startswith ( "<STR_LIT>" ) : <EOL> result = result . split ( '<STR_LIT::>' , <NUM_LIT:1> ) [ <NUM_LIT:1> ] <EOL> if result : <EOL> result = os . path . abspath ( os . path . expanduser ( result ) ) <EOL> accounts = netrc . netrc ( result or None ) <EOL> account = accounts . authenticators ( fqdn ) <EOL> if not account or not ( account [ <NUM_LIT:0> ] or account [ <NUM_LIT:1> ] ) : <EOL> raise dputhelper . DputUploadFatalException ( "<STR_LIT>" % ( <EOL> fqdn , result or "<STR_LIT:default>" ) ) <EOL> user , pwd = account [ <NUM_LIT:0> ] or account [ <NUM_LIT:1> ] , account [ <NUM_LIT:2> ] or "<STR_LIT>" <EOL> result = "<STR_LIT>" % ( user , pwd ) <EOL> else : <EOL> if result . startswith ( "<STR_LIT>" ) : <EOL> result = os . path . abspath ( os . path . expanduser ( result . split ( '<STR_LIT::>' , <NUM_LIT:1> ) [ <NUM_LIT:1> ] ) ) <EOL> with closing ( open ( result , "<STR_LIT:r>" ) ) as handle : <EOL> result = handle . read ( ) . strip ( ) <EOL> try : <EOL> user , pwd = result . split ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> except ValueError : <EOL> user , pwd = result , "<STR_LIT>" <EOL> trace ( "<STR_LIT>" , user = user , pwd = '<STR_LIT:*>' * len ( pwd ) ) <EOL> return result <EOL> class PromptingPasswordMgr ( urllib2 . HTTPPasswordMgr ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , login ) : <EOL> urllib2 . HTTPPasswordMgr . __init__ ( self ) <EOL> self . login = login <EOL> def find_user_password ( self , realm , authuri ) : <EOL> """<STR_LIT>""" <EOL> authuri = self . reduce_uri ( authuri ) [ <NUM_LIT:0> ] <EOL> authinfo = urllib2 . HTTPPasswordMgr . find_user_password ( self , realm , authuri ) <EOL> if authinfo == ( None , None ) : <EOL> credentials = self . login <EOL> if '<STR_LIT::>' in credentials : <EOL> authinfo = credentials . split ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> else : <EOL> password = getpass . getpass ( "<STR_LIT>" % realm ) <EOL> self . add_password ( realm , authuri , credentials , password ) <EOL> authinfo = credentials , password <EOL> return authinfo <EOL> def _distro2repo ( distro , repo_mappings ) : <EOL> """<STR_LIT>""" <EOL> mappings = [ ( i . split ( '<STR_LIT:=>' , <NUM_LIT:1> ) if '<STR_LIT:=>' in i else ( i , i ) ) for i in repo_mappings . split ( ) ] <EOL> result = distro <EOL> for pattern , target in mappings : <EOL> if fnmatch . fnmatchcase ( distro . lower ( ) , pattern . lower ( ) ) : <EOL> result = target <EOL> break <EOL> trace ( "<STR_LIT>" , distro = distro , repo = result ) <EOL> return result <EOL> def _resolve_incoming ( fqdn , login , incoming , changes = None , cli_params = None , repo_mappings = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> scheme , netloc , path , matrix_params , query , anchor = urlparse . urlparse ( incoming , scheme = "<STR_LIT:http>" , allow_fragments = True ) <EOL> if scheme not in ( "<STR_LIT:http>" , "<STR_LIT>" ) : <EOL> raise dputhelper . DputUploadFatalException ( "<STR_LIT>" % scheme ) <EOL> url = urlparse . urlunparse ( ( scheme , netloc or fqdn , path . rstrip ( '<STR_LIT:/>' ) + '<STR_LIT:/>' , '<STR_LIT>' , query , None ) ) <EOL> url_params = dict ( cgi . parse_qsl ( anchor or '<STR_LIT>' , keep_blank_values = True ) ) <EOL> pkgdata = { } <EOL> if changes : <EOL> try : <EOL> changes . read <EOL> except AttributeError : <EOL> with closing ( open ( changes , "<STR_LIT:r>" ) ) as handle : <EOL> changes = handle . read ( ) <EOL> else : <EOL> changes = changes . read ( ) <EOL> if changes . startswith ( "<STR_LIT>" ) : <EOL> trace ( "<STR_LIT>" ) <EOL> changes = changes . split ( "<STR_LIT>" ) [ <NUM_LIT:1> ] . replace ( '<STR_LIT:\r>' , '<STR_LIT>' ) . split ( '<STR_LIT>' , <NUM_LIT:1> ) [ <NUM_LIT:1> ] <EOL> pkgdata = dict ( [ ( key . lower ( ) . replace ( '<STR_LIT:->' , '<STR_LIT:_>' ) , val . strip ( ) ) <EOL> for key , val in rfc2822_parser . HeaderParser ( ) . parsestr ( changes ) . items ( ) <EOL> ] ) <EOL> if '<STR_LIT>' in pkgdata : <EOL> pkgdata [ '<STR_LIT>' ] = '<STR_LIT>' . join ( pkgdata [ '<STR_LIT>' ] . split ( ) ) <EOL> pkgdata [ "<STR_LIT>" ] = login . split ( '<STR_LIT::>' ) [ <NUM_LIT:0> ] <EOL> if "<STR_LIT:version>" in pkgdata : <EOL> pkgdata [ "<STR_LIT>" ] = re . split ( r"<STR_LIT>" , pkgdata [ "<STR_LIT:version>" ] ) [ <NUM_LIT:0> ] <EOL> pkgdata . update ( dict ( <EOL> fqdn = fqdn , repo = _distro2repo ( pkgdata . get ( "<STR_LIT>" , "<STR_LIT>" ) , repo_mappings ) , <EOL> ) ) <EOL> pkgdata . update ( cli_params or { } ) <EOL> trace ( "<STR_LIT>" , meta = "<STR_LIT>" . join ( [ "<STR_LIT>" % ( key , val ) <EOL> for key , val in sorted ( pkgdata . items ( ) ) <EOL> if '<STR_LIT:\n>' not in val <EOL> ] ) ) <EOL> try : <EOL> try : <EOL> url . format <EOL> except AttributeError : <EOL> url = url % pkgdata <EOL> matrix_params = matrix_params % pkgdata <EOL> else : <EOL> url = url . format ( ** pkgdata ) <EOL> matrix_params = matrix_params . format ( ** pkgdata ) <EOL> matrix_params = matrix_params . replace ( '<STR_LIT:U+0020>' , '<STR_LIT:+>' ) <EOL> except KeyError , exc : <EOL> raise dputhelper . DputUploadFatalException ( "<STR_LIT>" % ( exc , incoming ) ) <EOL> trace ( "<STR_LIT>" , url = url , params = url_params ) <EOL> return url , matrix_params , url_params <EOL> def _url_connection ( url , method , skip_host = False , skip_accept_encoding = False ) : <EOL> """<STR_LIT>""" <EOL> scheme , netloc , path , params , query , _ = urlparse . urlparse ( url ) <EOL> result = conn = ( httplib . HTTPSConnection if scheme == "<STR_LIT>" else httplib . HTTPConnection ) ( netloc ) <EOL> conn . debuglevel = int ( trace . debug ) <EOL> try : <EOL> conn . putrequest ( method , urlparse . urlunparse ( ( None , None , path , params , query , None ) ) , skip_host , skip_accept_encoding ) <EOL> conn . putheader ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> conn . putheader ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> conn = None <EOL> finally : <EOL> if conn : <EOL> conn . close ( ) <EOL> return result <EOL> def _file_url ( filepath , url ) : <EOL> """<STR_LIT>""" <EOL> basename = os . path . basename ( filepath ) <EOL> return urlparse . urljoin ( url . rstrip ( '<STR_LIT:/>' ) + '<STR_LIT:/>' , basename ) <EOL> def _dav_put ( filepath , url , matrix_params , login , progress = None ) : <EOL> """<STR_LIT>""" <EOL> fileurl = _file_url ( filepath , url ) <EOL> if matrix_params : <EOL> fileurl += '<STR_LIT:;>' + matrix_params <EOL> sys . stdout . write ( "<STR_LIT>" % os . path . basename ( filepath ) ) <EOL> sys . stdout . flush ( ) <EOL> size = os . path . getsize ( filepath ) <EOL> with closing ( open ( filepath , '<STR_LIT:r>' ) ) as handle : <EOL> if progress : <EOL> handle = dputhelper . FileWithProgress ( handle , ptype = progress , progressf = sys . stdout , size = size ) <EOL> trace ( "<STR_LIT>" % fileurl ) <EOL> try : <EOL> conn = _url_connection ( fileurl , "<STR_LIT>" ) <EOL> try : <EOL> conn . putheader ( "<STR_LIT>" , '<STR_LIT>' % login . encode ( '<STR_LIT>' ) . replace ( '<STR_LIT:\n>' , '<STR_LIT>' ) . strip ( ) ) <EOL> conn . putheader ( "<STR_LIT>" , str ( size ) ) <EOL> conn . endheaders ( ) <EOL> conn . debuglevel = <NUM_LIT:0> <EOL> while True : <EOL> data = handle . read ( CHUNK_SIZE ) <EOL> if not data : <EOL> break <EOL> conn . send ( data ) <EOL> conn . debuglevel = int ( trace . debug ) <EOL> resp = conn . getresponse ( ) <EOL> if <NUM_LIT:200> <= resp . status <= <NUM_LIT> : <EOL> print ( "<STR_LIT>" ) <EOL> elif resp . status == <NUM_LIT> : <EOL> print ( "<STR_LIT>" ) <EOL> raise urllib2 . URLError ( "<STR_LIT>" <EOL> "<STR_LIT>" % resp . reason ) <EOL> else : <EOL> print ( "<STR_LIT>" ) <EOL> raise urllib2 . URLError ( "<STR_LIT>" % ( resp . status , resp . reason ) ) <EOL> resp . read ( ) <EOL> finally : <EOL> conn . close ( ) <EOL> except httplib . HTTPException , exc : <EOL> raise urllib2 . URLError ( exc ) <EOL> def _check_url ( url , login , allowed , mindepth = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> if mindepth : <EOL> scheme , netloc , path , params , query , fragment = urlparse . urlparse ( url ) <EOL> path = '<STR_LIT:/>' . join ( path . split ( '<STR_LIT:/>' ) [ : mindepth + <NUM_LIT:1> ] ) . rstrip ( '<STR_LIT:/>' ) + '<STR_LIT:/>' <EOL> url = urlparse . urlunparse ( ( scheme , netloc , path , params , query , fragment ) ) <EOL> trace ( "<STR_LIT>" , url = url ) <EOL> try : <EOL> request = urllib2 . Request ( url ) <EOL> request . add_header ( "<STR_LIT>" , '<STR_LIT>' % login . encode ( '<STR_LIT>' ) . replace ( '<STR_LIT:\n>' , '<STR_LIT>' ) . strip ( ) ) <EOL> with closing ( urllib2 . urlopen ( request ) ) as handle : <EOL> handle . read ( ) <EOL> code = handle . code <EOL> if code not in allowed : <EOL> raise urllib2 . HTTPError ( url , code , <EOL> "<STR_LIT>" % ( code , handle . msg ) , <EOL> handle . headers , None ) <EOL> except urllib2 . HTTPError , exc : <EOL> code = exc . code <EOL> if code not in allowed : <EOL> raise <EOL> trace ( "<STR_LIT>" , url = url , code = code ) <EOL> def _get_host_argument ( fqdn ) : <EOL> """<STR_LIT>""" <EOL> import __main__ as dput <EOL> config = dput . config <EOL> result = "<STR_LIT>" <EOL> for section in config . sections ( ) : <EOL> if ( config . has_option ( section , "<STR_LIT>" ) <EOL> and config . get ( section , "<STR_LIT>" ) == fqdn <EOL> and config . has_option ( section , section ) ) : <EOL> result = config . get ( section , section ) <EOL> return result <EOL> def _get_config_data ( fqdn ) : <EOL> """<STR_LIT>""" <EOL> if not upload . extended_info : <EOL> try : <EOL> caller = sys . _getframe ( <NUM_LIT:2> ) <EOL> except AttributeError : <EOL> pass <EOL> else : <EOL> config = caller . f_globals . get ( "<STR_LIT>" ) <EOL> host = caller . f_locals . get ( "<STR_LIT:host>" ) <EOL> del caller <EOL> if config and host : <EOL> upload . extended_info = dict ( config = config , host = host ) <EOL> if upload . extended_info : <EOL> host_config = dict ( upload . extended_info [ "<STR_LIT>" ] . items ( upload . extended_info [ "<STR_LIT:host>" ] ) ) <EOL> host_argument = host_config . get ( upload . extended_info [ "<STR_LIT:host>" ] , "<STR_LIT>" ) <EOL> else : <EOL> host_config = { } <EOL> host_argument = _get_host_argument ( fqdn ) <EOL> log ( "<STR_LIT>" ) <EOL> cli_params = dict ( cgi . parse_qsl ( host_argument . replace ( '<STR_LIT:U+002C>' , '<STR_LIT:;>' ) , keep_blank_values = True ) ) <EOL> return host_config , cli_params <EOL> def upload ( fqdn , login , incoming , files_to_upload , <EOL> debug , dummy , progress = None ) : <EOL> """<STR_LIT>""" <EOL> assert sys . version_info >= ( <NUM_LIT:2> , <NUM_LIT:5> ) , "<STR_LIT>" <EOL> trace . debug = bool ( debug ) <EOL> try : <EOL> host_config , cli_params = _get_config_data ( fqdn ) <EOL> login = _resolve_credentials ( fqdn , login ) <EOL> changes_file = [ i for i in files_to_upload if i . endswith ( "<STR_LIT>" ) ] <EOL> if not changes_file : <EOL> log ( "<STR_LIT>" , n = len ( files_to_upload ) ) <EOL> changes_file = None <EOL> else : <EOL> if len ( changes_file ) > <NUM_LIT:1> : <EOL> log ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> n = len ( files_to_upload ) , changes = "<STR_LIT>" . join ( changes_file ) ) <EOL> changes_file = changes_file [ <NUM_LIT:0> ] <EOL> incoming , matrix_params , repo_params = _resolve_incoming ( fqdn , login , incoming , changes = changes_file , <EOL> cli_params = cli_params , repo_mappings = host_config . get ( "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> log ( "<STR_LIT>" , url = urllib2 . quote ( incoming , safe = "<STR_LIT>" ) ) <EOL> repo_params . update ( cli_params ) <EOL> mindepth = int ( repo_params . get ( "<STR_LIT>" , "<STR_LIT:0>" ) , <NUM_LIT:10> ) <EOL> overwrite = int ( repo_params . get ( "<STR_LIT>" , "<STR_LIT:0>" ) , <NUM_LIT:10> ) <EOL> if "<STR_LIT>" in cli_params : <EOL> import pprint <EOL> print ( "<STR_LIT>" , end = "<STR_LIT>" ) <EOL> pprint . pprint ( dict ( ( k , v ) for k , v in locals ( ) . iteritems ( ) if k in ( <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) ) <EOL> print ( "<STR_LIT>" , end = "<STR_LIT>" ) <EOL> pprint . pprint ( host_config ) <EOL> print ( "<STR_LIT>" , end = "<STR_LIT>" ) <EOL> pprint . pprint ( cli_params ) <EOL> else : <EOL> if not overwrite and changes_file : <EOL> try : <EOL> _check_url ( _file_url ( changes_file , incoming ) , login , [ <NUM_LIT> ] ) <EOL> except urllib2 . HTTPError , exc : <EOL> raise dputhelper . DputUploadFatalException ( "<STR_LIT>" % ( <EOL> _file_url ( changes_file , incoming ) , exc ) ) <EOL> if mindepth : <EOL> try : <EOL> _check_url ( incoming , login , range ( <NUM_LIT:200> , <NUM_LIT> ) , mindepth = mindepth ) <EOL> except urllib2 . HTTPError , exc : <EOL> raise dputhelper . DputUploadFatalException ( "<STR_LIT>" % ( <EOL> exc . filename , exc ) ) <EOL> for filepath in files_to_upload : <EOL> if "<STR_LIT>" in cli_params : <EOL> log ( "<STR_LIT>" , filename = os . path . basename ( filepath ) ) <EOL> else : <EOL> _dav_put ( filepath , incoming , matrix_params , login , progress ) <EOL> except ( dputhelper . DputUploadFatalException , socket . error , urllib2 . URLError , EnvironmentError ) , exc : <EOL> log ( "<STR_LIT>" , exc = exc ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> upload . extended_info = { } <EOL> def py25_format ( template ) : <EOL> """<STR_LIT>""" <EOL> return template if sys . version_info >= ( <NUM_LIT:2> , <NUM_LIT:6> ) else template . replace ( "<STR_LIT:{>" , "<STR_LIT>" ) . replace ( "<STR_LIT:}>" , "<STR_LIT>" ) <EOL> class WebdavTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> DISTRO2REPO_DATA = [ <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ] <EOL> def test_distro2repo ( self ) : <EOL> """<STR_LIT>""" <EOL> cfg = "<STR_LIT>" <EOL> for distro , repo in self . DISTRO2REPO_DATA : <EOL> result = _distro2repo ( distro , cfg ) <EOL> self . assertEquals ( result , repo ) <EOL> def test_resolve_incoming ( self ) : <EOL> """<STR_LIT>""" <EOL> result , _ , params = _resolve_incoming ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEquals ( result , "<STR_LIT>" ) <EOL> self . assertEquals ( params , { } ) <EOL> result , _ , _ = _resolve_incoming ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEquals ( result , "<STR_LIT>" ) <EOL> result , _ , _ = _resolve_incoming ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEquals ( result , "<STR_LIT>" ) <EOL> result , _ , _ = _resolve_incoming ( "<STR_LIT>" , "<STR_LIT>" , py25_format ( "<STR_LIT>" ) ) <EOL> self . assertEquals ( result , "<STR_LIT>" ) <EOL> _ , _ , params = _resolve_incoming ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEquals ( params , dict ( a = "<STR_LIT:1>" , b = "<STR_LIT:c>" ) ) <EOL> result , _ , _ = _resolve_incoming ( "<STR_LIT>" , "<STR_LIT>" , py25_format ( "<STR_LIT>" ) ) <EOL> self . assertEquals ( result , "<STR_LIT>" ) <EOL> result , matrix_params , _ = _resolve_incoming ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEquals ( result , "<STR_LIT>" ) <EOL> self . assertEquals ( matrix_params , "<STR_LIT>" ) <EOL> self . assertRaises ( dputhelper . DputUploadFatalException , _resolve_incoming , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertRaises ( dputhelper . DputUploadFatalException , _resolve_incoming , <EOL> "<STR_LIT>" , "<STR_LIT>" , py25_format ( "<STR_LIT>" ) ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> print ( "<STR_LIT>" ) <EOL> unittest . main ( ) </s>
