<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import , division <EOL> import os . path <EOL> import unittest <EOL> from vistrails . core . application import is_running_gui <EOL> from vistrails . core . configuration import get_vistrails_configuration <EOL> import vistrails . core . interpreter . default <EOL> import vistrails . core . db . io <EOL> from vistrails . core . db . io import load_vistrail <EOL> from vistrails . core . db . locator import XMLFileLocator , ZIPFileLocator <EOL> from vistrails . core import debug <EOL> import vistrails . core . interpreter . cached <EOL> from vistrails . core . vistrail . job import Workflow as JobWorkflow <EOL> import vistrails . core . vistrail . pipeline <EOL> from vistrails . core . utils import VistrailsInternalError <EOL> from vistrails . core . vistrail . controller import VistrailController <EOL> import vistrails . core . packagemanager <EOL> import vistrails . core . system <EOL> import vistrails . core . vistrail <EOL> import vistrails . db <EOL> def run_and_get_results ( w_list , parameters = '<STR_LIT>' , <EOL> update_vistrail = True , extra_info = None , <EOL> reason = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> elements = parameters . split ( "<STR_LIT>" ) <EOL> aliases = { } <EOL> params = [ ] <EOL> result = [ ] <EOL> for locator , workflow in w_list : <EOL> ( v , abstractions , thumbnails , mashups ) = load_vistrail ( locator ) <EOL> controller = VistrailController ( v , locator , abstractions , thumbnails , <EOL> mashups , auto_save = update_vistrail ) <EOL> if isinstance ( workflow , basestring ) : <EOL> version = v . get_version_number ( workflow ) <EOL> elif isinstance ( workflow , ( int , long ) ) : <EOL> version = workflow <EOL> elif workflow is None : <EOL> version = controller . get_latest_version_in_graph ( ) <EOL> else : <EOL> msg = "<STR_LIT>" % workflow <EOL> raise VistrailsInternalError ( msg ) <EOL> controller . change_selected_version ( version ) <EOL> for e in elements : <EOL> pos = e . find ( "<STR_LIT:=>" ) <EOL> if pos != - <NUM_LIT:1> : <EOL> key = e [ : pos ] . strip ( ) <EOL> value = e [ pos + <NUM_LIT:1> : ] . strip ( ) <EOL> if controller . current_pipeline . has_alias ( key ) : <EOL> aliases [ key ] = value <EOL> elif '<STR_LIT>' in extra_info : <EOL> for mashuptrail in mashups : <EOL> if mashuptrail . vtVersion == version : <EOL> mashup = mashuptrail . getMashup ( extra_info [ '<STR_LIT>' ] ) <EOL> c = mashup . getAliasByName ( key ) . component <EOL> params . append ( ( c . vttype , c . vtid , value ) ) <EOL> if not update_vistrail : <EOL> conf = get_vistrails_configuration ( ) <EOL> if conf . has ( '<STR_LIT>' ) : <EOL> conf . thumbs . autoSave = False <EOL> jobMonitor = controller . jobMonitor <EOL> current_workflow = jobMonitor . currentWorkflow ( ) <EOL> if not current_workflow : <EOL> for job in jobMonitor . workflows . itervalues ( ) : <EOL> try : <EOL> job_version = int ( job . version ) <EOL> except ValueError : <EOL> try : <EOL> job_version = v . get_version_number ( job . version ) <EOL> except KeyError : <EOL> continue <EOL> if version == job_version : <EOL> current_workflow = job <EOL> jobMonitor . startWorkflow ( job ) <EOL> if not current_workflow : <EOL> current_workflow = JobWorkflow ( version ) <EOL> jobMonitor . startWorkflow ( current_workflow ) <EOL> try : <EOL> ( results , _ ) = controller . execute_current_workflow ( custom_aliases = aliases , <EOL> custom_params = params , <EOL> extra_info = extra_info , <EOL> reason = reason ) <EOL> finally : <EOL> jobMonitor . finishWorkflow ( ) <EOL> new_version = controller . current_version <EOL> if new_version != version : <EOL> debug . log ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> workflow , version , new_version ) ) <EOL> run = results [ <NUM_LIT:0> ] <EOL> run . workflow_info = ( locator . name , new_version ) <EOL> run . pipeline = controller . current_pipeline <EOL> if update_vistrail : <EOL> controller . write_vistrail ( locator ) <EOL> result . append ( run ) <EOL> if current_workflow . jobs : <EOL> if current_workflow . completed ( ) : <EOL> run . job = "<STR_LIT>" <EOL> else : <EOL> run . job = "<STR_LIT>" % current_workflow . id <EOL> for job in current_workflow . jobs . itervalues ( ) : <EOL> if not job . finished : <EOL> run . job += "<STR_LIT>" % ( job . start , job . name , job . description ( ) ) <EOL> print run . job <EOL> return result <EOL> def get_wf_graph ( w_list , output_dir , pdf = False ) : <EOL> """<STR_LIT>""" <EOL> result = [ ] <EOL> if is_running_gui ( ) : <EOL> from vistrails . gui . vistrail_controller import VistrailController as GUIVistrailController <EOL> for locator , workflow in w_list : <EOL> try : <EOL> ( v , abstractions , thumbnails , mashups ) = load_vistrail ( locator ) <EOL> controller = GUIVistrailController ( v , locator , abstractions , <EOL> thumbnails , mashups , <EOL> auto_save = False ) <EOL> controller . current_pipeline_view . set_controller ( controller ) <EOL> version = None <EOL> if isinstance ( workflow , basestring ) : <EOL> version = v . get_version_number ( workflow ) <EOL> elif isinstance ( workflow , ( int , long ) ) : <EOL> version = workflow <EOL> elif workflow is None : <EOL> version = controller . get_latest_version_in_graph ( ) <EOL> else : <EOL> msg = "<STR_LIT>" % workflow <EOL> raise VistrailsInternalError ( msg ) <EOL> controller . change_selected_version ( version ) <EOL> if controller . current_pipeline is not None : <EOL> controller . updatePipelineScene ( ) <EOL> if pdf : <EOL> base_fname = "<STR_LIT>" % ( locator . short_filename , version ) <EOL> filename = os . path . join ( output_dir , base_fname ) <EOL> controller . current_pipeline_scene . saveToPDF ( filename ) <EOL> else : <EOL> base_fname = "<STR_LIT>" % ( locator . short_filename , version ) <EOL> filename = os . path . join ( output_dir , base_fname ) <EOL> controller . current_pipeline_scene . saveToPNG ( filename ) <EOL> result . append ( ( True , "<STR_LIT>" ) ) <EOL> except Exception , e : <EOL> result . append ( ( False , debug . format_exception ( e ) ) ) <EOL> else : <EOL> error_str = "<STR_LIT>" "<STR_LIT>" <EOL> debug . critical ( error_str ) <EOL> result . append ( ( False , error_str ) ) <EOL> return result <EOL> def get_vt_graph ( vt_list , tree_info , pdf = False ) : <EOL> """<STR_LIT>""" <EOL> result = [ ] <EOL> if is_running_gui ( ) : <EOL> from vistrails . gui . vistrail_controller import VistrailController as GUIVistrailController <EOL> for locator in vt_list : <EOL> try : <EOL> ( v , abstractions , thumbnails , mashups ) = load_vistrail ( locator ) <EOL> controller = GUIVistrailController ( v , locator , abstractions , <EOL> thumbnails , mashups ) <EOL> if tree_info is not None : <EOL> from vistrails . gui . version_view import QVersionTreeView <EOL> version_view = QVersionTreeView ( ) <EOL> version_view . scene ( ) . setupScene ( controller ) <EOL> if pdf : <EOL> base_fname = "<STR_LIT>" % locator . short_filename <EOL> filename = os . path . join ( tree_info , base_fname ) <EOL> version_view . scene ( ) . saveToPDF ( filename ) <EOL> else : <EOL> base_fname = "<STR_LIT>" % locator . short_filename <EOL> filename = os . path . join ( tree_info , base_fname ) <EOL> version_view . scene ( ) . saveToPNG ( filename ) <EOL> del version_view <EOL> result . append ( ( True , "<STR_LIT>" ) ) <EOL> except Exception , e : <EOL> result . append ( ( False , debug . format_exception ( e ) ) ) <EOL> else : <EOL> error_str = "<STR_LIT>" "<STR_LIT>" <EOL> debug . critical ( error_str ) <EOL> result . append ( ( False , error_str ) ) <EOL> return result <EOL> def run ( w_list , parameters = '<STR_LIT>' , update_vistrail = True , <EOL> extra_info = None , reason = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> all_errors = [ ] <EOL> results = run_and_get_results ( w_list , parameters , <EOL> update_vistrail , extra_info , reason ) <EOL> for result in results : <EOL> ( objs , errors , executed ) = ( result . objects , <EOL> result . errors , result . executed ) <EOL> for err in sorted ( errors . iteritems ( ) ) : <EOL> all_errors . append ( result . workflow_info + err ) <EOL> return all_errors <EOL> def run_parameter_exploration ( locator , pe_id , extra_info = { } , <EOL> reason = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> if is_running_gui ( ) : <EOL> from vistrails . gui . vistrail_controller import VistrailController as GUIVistrailController <EOL> try : <EOL> ( v , abstractions , thumbnails , mashups ) = load_vistrail ( locator ) <EOL> controller = GUIVistrailController ( v , locator , abstractions , <EOL> thumbnails , mashups ) <EOL> try : <EOL> pe_id = int ( pe_id ) <EOL> pe = controller . vistrail . get_paramexp ( pe_id ) <EOL> except ValueError : <EOL> pe = controller . vistrail . get_named_paramexp ( pe_id ) <EOL> controller . change_selected_version ( pe . action_id ) <EOL> controller . executeParameterExploration ( pe , extra_info = extra_info , <EOL> showProgress = False ) <EOL> except Exception , e : <EOL> return ( locator , pe_id , <EOL> debug . format_exception ( e ) , debug . format_exc ( ) ) <EOL> def run_parameter_explorations ( w_list , extra_info = { } , <EOL> reason = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> all_errors = [ ] <EOL> for locator , pe_id in w_list : <EOL> result = run_parameter_exploration ( locator , pe_id , reason = reason , <EOL> extra_info = extra_info ) <EOL> if result : <EOL> all_errors . append ( result ) <EOL> return all_errors <EOL> def cleanup ( ) : <EOL> vistrails . core . interpreter . cached . CachedInterpreter . cleanup ( ) <EOL> class TestConsoleMode ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> manager = vistrails . core . packagemanager . get_package_manager ( ) <EOL> if manager . has_package ( '<STR_LIT>' ) : <EOL> return <EOL> d = { '<STR_LIT>' : '<STR_LIT>' } <EOL> manager . late_enable_package ( '<STR_LIT>' , d ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> manager = vistrails . core . packagemanager . get_package_manager ( ) <EOL> if manager . has_package ( '<STR_LIT>' ) : <EOL> manager . late_disable_package ( '<STR_LIT>' ) <EOL> def test1 ( self ) : <EOL> from vistrails . core . modules . basic_modules import StandardOutput <EOL> values = [ ] <EOL> def mycompute ( s ) : <EOL> v = s . get_input ( '<STR_LIT:value>' ) <EOL> values . append ( v ) <EOL> orig_compute = StandardOutput . compute <EOL> StandardOutput . compute = mycompute <EOL> try : <EOL> locator = XMLFileLocator ( vistrails . core . system . vistrails_root_directory ( ) + <EOL> '<STR_LIT>' ) <EOL> result = run ( [ ( locator , "<STR_LIT>" ) ] , update_vistrail = False ) <EOL> self . assertEqual ( len ( result ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( values , [ <NUM_LIT:2> ] ) <EOL> finally : <EOL> StandardOutput . compute = orig_compute <EOL> def test_tuple ( self ) : <EOL> from vistrails . core . vistrail . module_param import ModuleParam <EOL> from vistrails . core . vistrail . module_function import ModuleFunction <EOL> from vistrails . core . utils import DummyView <EOL> from vistrails . core . vistrail . module import Module <EOL> import vistrails . db . domain <EOL> id_scope = vistrails . db . domain . IdScope ( ) <EOL> interpreter = vistrails . core . interpreter . default . get_default_interpreter ( ) <EOL> v = DummyView ( ) <EOL> p = vistrails . core . vistrail . pipeline . Pipeline ( ) <EOL> params = [ ModuleParam ( id = id_scope . getNewId ( ModuleParam . vtType ) , <EOL> pos = <NUM_LIT:0> , <EOL> type = '<STR_LIT>' , <EOL> val = '<STR_LIT>' , <EOL> ) , <EOL> ModuleParam ( id = id_scope . getNewId ( ModuleParam . vtType ) , <EOL> pos = <NUM_LIT:1> , <EOL> type = '<STR_LIT>' , <EOL> val = '<STR_LIT>' , <EOL> ) ] <EOL> function = ModuleFunction ( id = id_scope . getNewId ( ModuleFunction . vtType ) , <EOL> name = '<STR_LIT:input>' ) <EOL> function . add_parameters ( params ) <EOL> module = Module ( id = id_scope . getNewId ( Module . vtType ) , <EOL> name = '<STR_LIT>' , <EOL> package = '<STR_LIT>' , <EOL> version = '<STR_LIT>' ) <EOL> module . add_function ( function ) <EOL> p . add_module ( module ) <EOL> interpreter . execute ( p , <EOL> locator = XMLFileLocator ( '<STR_LIT:foo>' ) , <EOL> current_version = <NUM_LIT:1> L , <EOL> view = v ) <EOL> def test_python_source ( self ) : <EOL> locator = XMLFileLocator ( vistrails . core . system . vistrails_root_directory ( ) + <EOL> '<STR_LIT>' ) <EOL> result = run ( [ ( locator , "<STR_LIT>" ) ] , update_vistrail = False ) <EOL> self . assertEqual ( len ( result ) , <NUM_LIT:0> ) <EOL> def test_python_source_2 ( self ) : <EOL> locator = XMLFileLocator ( vistrails . core . system . vistrails_root_directory ( ) + <EOL> '<STR_LIT>' ) <EOL> result = run_and_get_results ( [ ( locator , "<STR_LIT>" ) ] , <EOL> update_vistrail = False ) [ <NUM_LIT:0> ] <EOL> self . assertEquals ( len ( result . executed ) , <NUM_LIT:1> ) <EOL> def test_dynamic_module_error ( self ) : <EOL> locator = XMLFileLocator ( vistrails . core . system . vistrails_root_directory ( ) + <EOL> '<STR_LIT>' ) <EOL> result = run ( [ ( locator , "<STR_LIT:test>" ) ] , update_vistrail = False ) <EOL> self . assertNotEqual ( len ( result ) , <NUM_LIT:0> ) <EOL> def test_change_parameter ( self ) : <EOL> locator = XMLFileLocator ( vistrails . core . system . vistrails_root_directory ( ) + <EOL> '<STR_LIT>' ) <EOL> result = run ( [ ( locator , "<STR_LIT>" ) ] , update_vistrail = False ) <EOL> self . assertEqual ( len ( result ) , <NUM_LIT:0> ) <EOL> result = run ( [ ( locator , "<STR_LIT>" ) ] , update_vistrail = False ) <EOL> self . assertEquals ( len ( result ) , <NUM_LIT:0> ) <EOL> def test_ticket_73 ( self ) : <EOL> locator = XMLFileLocator ( vistrails . core . system . vistrails_root_directory ( ) + <EOL> '<STR_LIT>' ) <EOL> v = locator . load ( ) <EOL> import tempfile <EOL> ( fd , filename ) = tempfile . mkstemp ( ) <EOL> os . close ( fd ) <EOL> locator = XMLFileLocator ( filename ) <EOL> try : <EOL> locator . save ( v ) <EOL> finally : <EOL> os . remove ( filename ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
