<s> import dbutils <EOL> import gitutils <EOL> import reviewing . filters <EOL> from operation import ( Operation , OperationResult , OperationFailure , <EOL> OperationError , Optional , typechecker ) <EOL> from extensions . installation import ( installExtension , uninstallExtension , <EOL> reinstallExtension , InstallationError , <EOL> getExtension ) <EOL> from extensions . extension import Extension , ExtensionError <EOL> from extensions . manifest import FilterHookRole <EOL> class ExtensionOperation ( Operation ) : <EOL> def __init__ ( self , perform ) : <EOL> Operation . __init__ ( self , { "<STR_LIT>" : str , <EOL> "<STR_LIT>" : Optional ( str ) , <EOL> "<STR_LIT:version>" : Optional ( str ) , <EOL> "<STR_LIT>" : Optional ( bool ) } ) <EOL> self . perform = perform <EOL> def process ( self , db , user , extension_name , author_name = None , version = None , <EOL> universal = False ) : <EOL> if universal : <EOL> if not user . hasRole ( db , "<STR_LIT>" ) : <EOL> raise OperationFailure ( code = "<STR_LIT>" , <EOL> title = "<STR_LIT>" , <EOL> message = "<STR_LIT>" ) <EOL> user = None <EOL> if version is not None : <EOL> if version == "<STR_LIT>" : <EOL> version = None <EOL> elif version . startswith ( "<STR_LIT>" ) : <EOL> version = version [ <NUM_LIT:8> : ] <EOL> else : <EOL> raise OperationError ( <EOL> "<STR_LIT>" <EOL> % version ) <EOL> try : <EOL> self . perform ( db , user , author_name , extension_name , version ) <EOL> except InstallationError as error : <EOL> raise OperationFailure ( code = "<STR_LIT>" , <EOL> title = error . title , <EOL> message = error . message , <EOL> is_html = error . is_html ) <EOL> return OperationResult ( ) <EOL> class InstallExtension ( ExtensionOperation ) : <EOL> def __init__ ( self ) : <EOL> ExtensionOperation . __init__ ( self , installExtension ) <EOL> class UninstallExtension ( ExtensionOperation ) : <EOL> def __init__ ( self ) : <EOL> ExtensionOperation . __init__ ( self , uninstallExtension ) <EOL> class ReinstallExtension ( ExtensionOperation ) : <EOL> def __init__ ( self ) : <EOL> ExtensionOperation . __init__ ( self , reinstallExtension ) <EOL> class ClearExtensionStorage ( Operation ) : <EOL> def __init__ ( self ) : <EOL> Operation . __init__ ( self , { "<STR_LIT>" : str , <EOL> "<STR_LIT>" : Optional ( str ) } ) <EOL> def process ( self , db , user , extension_name , author_name = None ) : <EOL> extension = getExtension ( author_name , extension_name ) <EOL> extension_id = extension . getExtensionID ( db , create = False ) <EOL> if extension_id is not None : <EOL> cursor = db . cursor ( ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( extension_id , user . id ) ) <EOL> db . commit ( ) <EOL> return OperationResult ( ) <EOL> class AddExtensionHookFilter ( Operation ) : <EOL> def __init__ ( self ) : <EOL> Operation . __init__ ( self , { "<STR_LIT>" : typechecker . User , <EOL> "<STR_LIT>" : typechecker . Extension , <EOL> "<STR_LIT>" : typechecker . Repository , <EOL> "<STR_LIT>" : str , <EOL> "<STR_LIT:path>" : str , <EOL> "<STR_LIT:data>" : Optional ( str ) , <EOL> "<STR_LIT>" : Optional ( int ) } ) <EOL> def process ( self , db , user , subject , extension , repository , <EOL> filterhook_name , path , data = None , replaced_filter_id = None ) : <EOL> if user != subject : <EOL> Operation . requireRole ( db , "<STR_LIT>" , user ) <EOL> path = reviewing . filters . sanitizePath ( path ) <EOL> if "<STR_LIT:*>" in path : <EOL> try : <EOL> reviewing . filters . validatePattern ( path ) <EOL> except reviewing . filters . PatternError as error : <EOL> raise OperationFailure ( <EOL> code = "<STR_LIT>" , <EOL> title = "<STR_LIT>" , <EOL> message = "<STR_LIT>" % error . message ) <EOL> installed_sha1 , _ = extension . getInstalledVersion ( db , subject ) <EOL> if installed_sha1 is False : <EOL> raise OperationFailure ( <EOL> code = "<STR_LIT>" , <EOL> title = "<STR_LIT>" , <EOL> message = ( "<STR_LIT>" <EOL> % extension . getTitle ( db ) ) ) <EOL> manifest = extension . getManifest ( sha1 = installed_sha1 ) <EOL> for role in manifest . roles : <EOL> if isinstance ( role , FilterHookRole ) and role . name == filterhook_name : <EOL> break <EOL> else : <EOL> raise OperationFailure ( <EOL> code = "<STR_LIT>" , <EOL> title = "<STR_LIT>" , <EOL> message = ( "<STR_LIT>" <EOL> % filterhook_name ) ) <EOL> cursor = db . cursor ( ) <EOL> if replaced_filter_id is not None : <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( replaced_filter_id , subject . id ) ) <EOL> if not cursor . fetchone ( ) : <EOL> raise OperationFailure ( <EOL> code = "<STR_LIT>" , <EOL> title = "<STR_LIT>" , <EOL> message = "<STR_LIT>" ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( replaced_filter_id , ) ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( subject . id , extension . getExtensionID ( db ) , repository . id , <EOL> filterhook_name , path , data ) ) <EOL> filter_id , = cursor . fetchone ( ) <EOL> db . commit ( ) <EOL> return OperationResult ( filter_id = filter_id ) <EOL> class DeleteExtensionHookFilter ( Operation ) : <EOL> def __init__ ( self ) : <EOL> Operation . __init__ ( self , { "<STR_LIT>" : typechecker . User , <EOL> "<STR_LIT>" : int } ) <EOL> def process ( self , db , user , subject , filter_id ) : <EOL> if user != subject : <EOL> Operation . requireRole ( db , "<STR_LIT>" , user ) <EOL> cursor = db . cursor ( ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( filter_id , subject . id ) ) <EOL> if not cursor . fetchone ( ) : <EOL> raise OperationFailure ( <EOL> code = "<STR_LIT>" , <EOL> title = "<STR_LIT>" , <EOL> message = "<STR_LIT>" ) <EOL> cursor . execute ( """<STR_LIT>""" , <EOL> ( filter_id , ) ) <EOL> db . commit ( ) <EOL> return OperationResult ( ) </s>
