<s> """<STR_LIT>""" <EOL> import os <EOL> import jip <EOL> import unittest <EOL> import jip . tools <EOL> class BWAPipelineTest ( unittest . TestCase ) : <EOL> def testPipelineStructure ( self ) : <EOL> tool = jip . find ( "<STR_LIT>" ) <EOL> assert tool is not None <EOL> p = jip . Pipeline ( ) <EOL> p . run ( tool , input = "<STR_LIT>" , reference = "<STR_LIT>" , output = "<STR_LIT>" ) <EOL> p . expand ( validate = False ) <EOL> assert len ( p ) == <NUM_LIT:7> <EOL> assert len ( p . edges ) == <NUM_LIT:6> <EOL> ref = p . get ( "<STR_LIT>" ) <EOL> align = p . get ( "<STR_LIT>" ) <EOL> sam = p . get ( "<STR_LIT>" ) <EOL> bam = p . get ( "<STR_LIT>" ) <EOL> dups = p . get ( "<STR_LIT>" ) <EOL> index = p . get ( "<STR_LIT:index>" ) <EOL> pileup = p . get ( "<STR_LIT>" ) <EOL> assert not ref . has_incoming ( ) <EOL> assert align . has_incoming ( ref ) <EOL> assert sam . has_incoming ( align ) <EOL> assert bam . has_incoming ( sam ) <EOL> assert dups . has_incoming ( bam ) <EOL> assert index . has_incoming ( dups ) <EOL> assert pileup . has_incoming ( index ) <EOL> assert not pileup . has_outgoing ( ) <EOL> def testPipelineStructureMultiplexed ( self ) : <EOL> tool = jip . find ( "<STR_LIT>" ) <EOL> assert tool is not None <EOL> p = jip . Pipeline ( ) <EOL> p . run ( tool , input = [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> reference = "<STR_LIT>" , <EOL> output = "<STR_LIT>" ) <EOL> p . expand ( validate = False ) <EOL> assert len ( p ) == <NUM_LIT> <EOL> assert len ( p . edges ) == <NUM_LIT:12> <EOL> ref = p . get ( "<STR_LIT>" ) <EOL> align = p . get ( "<STR_LIT>" ) <EOL> sam = p . get ( "<STR_LIT>" ) <EOL> bam = p . get ( "<STR_LIT>" ) <EOL> dups = p . get ( "<STR_LIT>" ) <EOL> index = p . get ( "<STR_LIT>" ) <EOL> pileup = p . get ( "<STR_LIT>" ) <EOL> assert not ref . has_incoming ( ) <EOL> assert align . has_incoming ( ref ) <EOL> assert sam . has_incoming ( align ) <EOL> assert bam . has_incoming ( sam ) <EOL> assert dups . has_incoming ( bam ) <EOL> assert index . has_incoming ( dups ) <EOL> assert pileup . has_incoming ( index ) <EOL> assert not pileup . has_outgoing ( ) <EOL> ref = p . get ( "<STR_LIT>" ) <EOL> align = p . get ( "<STR_LIT>" ) <EOL> sam = p . get ( "<STR_LIT>" ) <EOL> bam = p . get ( "<STR_LIT>" ) <EOL> dups = p . get ( "<STR_LIT>" ) <EOL> index = p . get ( "<STR_LIT>" ) <EOL> pileup = p . get ( "<STR_LIT>" ) <EOL> assert not ref . has_incoming ( ) <EOL> assert align . has_incoming ( ref ) <EOL> assert sam . has_incoming ( align ) <EOL> assert bam . has_incoming ( sam ) <EOL> assert dups . has_incoming ( bam ) <EOL> assert index . has_incoming ( dups ) <EOL> assert pileup . has_incoming ( index ) <EOL> assert not pileup . has_outgoing ( ) <EOL> @ jip . tool ( '<STR_LIT>' ) <EOL> class GemIndex ( object ) : <EOL> """<STR_LIT>""" <EOL> def init ( self ) : <EOL> self . add_output ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> def setup ( self ) : <EOL> out = "<STR_LIT>" <EOL> if self . options [ '<STR_LIT>' ] : <EOL> out = "<STR_LIT>" + out <EOL> self . options [ '<STR_LIT>' ] . set ( out ) <EOL> def get_command ( self ) : <EOL> return "<STR_LIT>" "<STR_LIT>" <EOL> @ jip . tool ( '<STR_LIT>' ) <EOL> class GemTranscriptomeIndex ( object ) : <EOL> """<STR_LIT>""" <EOL> def init ( self ) : <EOL> self . add_output ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> self . add_output ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> def get_command ( self ) : <EOL> return '<STR_LIT>' , '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' <EOL> def test_gemtools_index_command_rendering_for_options ( ) : <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , input = "<STR_LIT>" , output_dir = '<STR_LIT:test>' ) <EOL> p . expand ( validate = True ) <EOL> node = p . get ( '<STR_LIT>' ) <EOL> print node . _tool . options <EOL> job = jip . create_jobs ( p ) [ <NUM_LIT:0> ] <EOL> infile = os . path . abspath ( "<STR_LIT>" ) <EOL> base = os . path . dirname ( infile ) <EOL> print "<STR_LIT>" , job . command <EOL> assert job . command == '<STR_LIT>' % ( <EOL> infile , os . path . join ( base , "<STR_LIT>" ) ) <EOL> outfiles = list ( job . get_output_files ( ) ) <EOL> assert len ( outfiles ) == <NUM_LIT:1> <EOL> assert outfiles [ <NUM_LIT:0> ] == os . path . join ( base , "<STR_LIT>" ) <EOL> def test_gemtools_t_index_inputs ( ) : <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , index = "<STR_LIT>" , annotation = '<STR_LIT>' , <EOL> output_dir = '<STR_LIT:test>' ) <EOL> p . expand ( validate = False ) <EOL> job = jip . create_jobs ( p ) [ <NUM_LIT:0> ] <EOL> infile = os . path . abspath ( "<STR_LIT>" ) <EOL> annotation = os . path . abspath ( "<STR_LIT>" ) <EOL> base = os . path . dirname ( infile ) <EOL> print "<STR_LIT>" , job . command <EOL> assert job . command == '<STR_LIT>' % ( <EOL> infile , annotation , os . path . join ( base , "<STR_LIT>" ) ) <EOL> outfiles = list ( job . get_output_files ( ) ) <EOL> assert len ( outfiles ) == <NUM_LIT:2> <EOL> assert outfiles [ <NUM_LIT:0> ] == os . path . join ( base , "<STR_LIT>" ) <EOL> assert outfiles [ <NUM_LIT:1> ] == os . path . join ( base , "<STR_LIT>" ) <EOL> @ jip . tool ( '<STR_LIT>' ) <EOL> class gem ( object ) : <EOL> """<STR_LIT>""" <EOL> def init ( self ) : <EOL> self . add_output ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> self . add_output ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> self . add_output ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> self . add_option ( '<STR_LIT>' , False , long = "<STR_LIT>" , <EOL> hidden = False ) <EOL> def setup ( self ) : <EOL> if len ( self . fastq ) == <NUM_LIT:1> : <EOL> self . options [ '<STR_LIT>' ] . set ( True ) <EOL> def get_command ( self ) : <EOL> return '<STR_LIT>' , '<STR_LIT>' <EOL> @ jip . tool ( '<STR_LIT>' ) <EOL> class flux ( object ) : <EOL> """<STR_LIT>""" <EOL> def init ( self ) : <EOL> self . add_option ( '<STR_LIT:name>' , "<STR_LIT>" ) <EOL> self . add_output ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> def get_command ( self ) : <EOL> return '<STR_LIT>' , '<STR_LIT>' <EOL> @ jip . pipeline ( '<STR_LIT>' ) <EOL> class GrapePipeline ( object ) : <EOL> """<STR_LIT>""" <EOL> def pipeline ( self ) : <EOL> p = jip . Pipeline ( ) <EOL> gem = p . run ( <EOL> '<STR_LIT>' , <EOL> index = self . index , annotation = self . annotation , <EOL> fastq = self . fastq , quality = self . quality , <EOL> output_dir = self . output_dir <EOL> ) <EOL> p . run ( <EOL> '<STR_LIT>' , input = gem . bam , annotation = self . annotation , <EOL> output_dir = self . output_dir <EOL> ) <EOL> p . context ( locals ( ) ) <EOL> return p <EOL> def test_gem_name_option_delegation ( ) : <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , fastq = '<STR_LIT>' , index = '<STR_LIT>' , <EOL> annotation = '<STR_LIT>' ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> ldir = os . getcwd ( ) <EOL> j = os . path . join <EOL> assert len ( jobs ) == <NUM_LIT:2> <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT:index>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == '<STR_LIT>' <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == ldir <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT:name>' ] . get ( ) == '<STR_LIT>' <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:1> ] . configuration [ '<STR_LIT:input>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:1> ] . configuration [ '<STR_LIT:name>' ] . get ( ) == '<STR_LIT>' <EOL> assert jobs [ <NUM_LIT:1> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:1> ] . configuration [ '<STR_LIT>' ] . get ( ) == ldir <EOL> assert jobs [ <NUM_LIT:1> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert len ( jobs [ <NUM_LIT:0> ] . children ) == <NUM_LIT:1> <EOL> assert len ( jobs [ <NUM_LIT:1> ] . dependencies ) == <NUM_LIT:1> <EOL> assert jobs [ <NUM_LIT:0> ] . children [ <NUM_LIT:0> ] == jobs [ <NUM_LIT:1> ] <EOL> def test_gem_name_option_delegation_with_output_dir ( ) : <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , fastq = '<STR_LIT>' , index = '<STR_LIT>' , <EOL> annotation = '<STR_LIT>' , output_dir = "<STR_LIT>" ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> ldir = os . getcwd ( ) <EOL> j = os . path . join <EOL> assert len ( jobs ) == <NUM_LIT:2> <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT:index>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == '<STR_LIT>' <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == "<STR_LIT>" <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT:name>' ] . get ( ) == '<STR_LIT>' <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:1> ] . configuration [ '<STR_LIT:input>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:1> ] . configuration [ '<STR_LIT:name>' ] . get ( ) == '<STR_LIT>' <EOL> assert jobs [ <NUM_LIT:1> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:1> ] . configuration [ '<STR_LIT>' ] . get ( ) == "<STR_LIT>" <EOL> assert jobs [ <NUM_LIT:1> ] . configuration [ '<STR_LIT>' ] . get ( ) == j ( ldir , '<STR_LIT>' ) <EOL> assert len ( jobs [ <NUM_LIT:0> ] . children ) == <NUM_LIT:1> <EOL> assert len ( jobs [ <NUM_LIT:1> ] . dependencies ) == <NUM_LIT:1> <EOL> assert jobs [ <NUM_LIT:0> ] . children [ <NUM_LIT:0> ] == jobs [ <NUM_LIT:1> ] <EOL> def test_multiple_pipelines_with_delegated_outputs ( ) : <EOL> @ jip . tool ( '<STR_LIT>' ) <EOL> class GemIndex ( object ) : <EOL> """<STR_LIT>""" <EOL> def get_command ( self ) : <EOL> return "<STR_LIT>" , "<STR_LIT>" <EOL> @ jip . tool ( '<STR_LIT>' ) <EOL> class gem ( object ) : <EOL> """<STR_LIT>""" <EOL> def get_command ( self ) : <EOL> return '<STR_LIT>' , '<STR_LIT>' <EOL> @ jip . pipeline ( '<STR_LIT>' ) <EOL> class SetupPipeline ( object ) : <EOL> """<STR_LIT>""" <EOL> def init ( self ) : <EOL> self . add_output ( '<STR_LIT:index>' , '<STR_LIT>' ) <EOL> def pipeline ( self ) : <EOL> p = jip . Pipeline ( ) <EOL> index = p . run ( '<STR_LIT>' , <EOL> input = self . input , output = self . index ) <EOL> p . context ( locals ( ) ) <EOL> return p <EOL> @ jip . pipeline ( '<STR_LIT>' ) <EOL> class GrapePipeline ( object ) : <EOL> """<STR_LIT>""" <EOL> def pipeline ( self ) : <EOL> p = jip . Pipeline ( ) <EOL> gem_setup = p . run ( '<STR_LIT>' , input = self . genome ) <EOL> gem = p . run ( '<STR_LIT>' , index = gem_setup . index , <EOL> fastq = self . fastq ) <EOL> p . context ( locals ( ) ) <EOL> return p <EOL> p = jip . Pipeline ( ) <EOL> node = p . run ( '<STR_LIT>' ) <EOL> node . fastq = '<STR_LIT>' <EOL> node . genome = '<STR_LIT>' <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> assert len ( jobs ) == <NUM_LIT:2> <EOL> gem_job = filter ( lambda x : x . name == '<STR_LIT>' , jobs ) [ <NUM_LIT:0> ] <EOL> assert gem_job is not None <EOL> assert gem_job . configuration [ '<STR_LIT:index>' ] . get ( ) . endswith ( '<STR_LIT>' ) <EOL> def test_embedded_pipelines_stage_one ( tmpdir ) : <EOL> tmpdir = str ( tmpdir ) <EOL> jip . scanner . add_module ( '<STR_LIT>' ) <EOL> jip . scanner . scan_modules ( ) <EOL> p = jip . Pipeline ( ) <EOL> p . job ( dir = tmpdir ) . run ( '<STR_LIT>' ) <EOL> jobs = jip . create_jobs ( p ) <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] == '<STR_LIT:test>' <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] == [ <EOL> os . path . join ( tmpdir , '<STR_LIT>' ) <EOL> ] <EOL> assert len ( jobs ) == <NUM_LIT:1> <EOL> def test_embedded_pipelines_stage_two ( tmpdir ) : <EOL> tmpdir = str ( tmpdir ) <EOL> open ( os . path . join ( tmpdir , '<STR_LIT>' ) , '<STR_LIT:a>' ) . close ( ) <EOL> open ( os . path . join ( tmpdir , '<STR_LIT>' ) , '<STR_LIT:a>' ) . close ( ) <EOL> open ( os . path . join ( tmpdir , '<STR_LIT>' ) , '<STR_LIT:a>' ) . close ( ) <EOL> open ( os . path . join ( tmpdir , '<STR_LIT>' ) , '<STR_LIT:a>' ) . close ( ) <EOL> open ( os . path . join ( tmpdir , '<STR_LIT>' ) , '<STR_LIT:a>' ) . close ( ) <EOL> jip . scanner . add_module ( '<STR_LIT>' ) <EOL> jip . scanner . scan_modules ( ) <EOL> p = jip . Pipeline ( ) <EOL> p . job ( dir = tmpdir ) . run ( '<STR_LIT>' ) <EOL> jobs = jip . create_jobs ( p ) <EOL> assert len ( jobs ) == <NUM_LIT:8> <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] == '<STR_LIT:test>' <EOL> print "<STR_LIT>" , jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] . raw ( ) <EOL> print "<STR_LIT>" , tmpdir <EOL> assert jobs [ <NUM_LIT:0> ] . configuration [ '<STR_LIT>' ] == [ <EOL> os . path . join ( tmpdir , '<STR_LIT>' ) , <EOL> os . path . join ( tmpdir , '<STR_LIT>' ) , <EOL> os . path . join ( tmpdir , '<STR_LIT>' ) , <EOL> os . path . join ( tmpdir , '<STR_LIT>' ) , <EOL> os . path . join ( tmpdir , '<STR_LIT>' ) , <EOL> ] <EOL> assert jobs [ <NUM_LIT:1> ] . configuration [ '<STR_LIT:input>' ] == os . path . join ( tmpdir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:2> ] . configuration [ '<STR_LIT:input>' ] == os . path . join ( tmpdir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:3> ] . configuration [ '<STR_LIT:input>' ] == os . path . join ( tmpdir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:4> ] . configuration [ '<STR_LIT:input>' ] == os . path . join ( tmpdir , '<STR_LIT>' ) <EOL> assert jobs [ <NUM_LIT:5> ] . configuration [ '<STR_LIT:input>' ] == os . path . join ( tmpdir , '<STR_LIT>' ) <EOL> print jobs [ <NUM_LIT:6> ] . configuration [ '<STR_LIT:input>' ] . raw ( ) <EOL> assert jobs [ <NUM_LIT:6> ] . configuration [ '<STR_LIT:input>' ] == [ <EOL> os . path . join ( tmpdir , '<STR_LIT>' ) , <EOL> os . path . join ( tmpdir , '<STR_LIT>' ) , <EOL> os . path . join ( tmpdir , '<STR_LIT>' ) , <EOL> os . path . join ( tmpdir , '<STR_LIT>' ) , <EOL> os . path . join ( tmpdir , '<STR_LIT>' ) , <EOL> ] <EOL> def test_dynamic_options ( ) : <EOL> script = '''<STR_LIT>''' <EOL> tool = jip . tools . ScriptTool . from_string ( script ) <EOL> tool . init ( ) <EOL> assert tool is not None <EOL> p = jip . Pipeline ( ) <EOL> node = p . job ( '<STR_LIT:test>' ) . run ( tool , prefix = '<STR_LIT:test>' , count = <NUM_LIT:5> ) <EOL> assert node is not None <EOL> p . expand ( ) <EOL> assert len ( p ) == <NUM_LIT:1> <EOL> node = p . get ( '<STR_LIT:test>' ) <EOL> assert node . prefix == '<STR_LIT:test>' <EOL> cwd = os . getcwd ( ) <EOL> assert node . output == [ os . path . join ( cwd , x ) for x in <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ] <EOL> def test_dynamic_options_multiplex ( ) : <EOL> script = '''<STR_LIT>''' <EOL> tool = jip . tools . ScriptTool . from_string ( script ) <EOL> tool . init ( ) <EOL> assert tool is not None <EOL> p = jip . Pipeline ( ) <EOL> node = p . job ( '<STR_LIT:test>' ) . run ( tool , prefix = '<STR_LIT:test>' , count = [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> assert node is not None <EOL> p . expand ( ) <EOL> cwd = os . getcwd ( ) <EOL> assert len ( p ) == <NUM_LIT:2> <EOL> node_1 = p . get ( '<STR_LIT>' ) <EOL> node_2 = p . get ( '<STR_LIT>' ) <EOL> assert node_1 . output == os . path . join ( cwd , '<STR_LIT>' ) <EOL> assert node_2 . output == [ os . path . join ( cwd , x ) for x in <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ] <EOL> def test_hello_world_py_fun ( tmpdir ) : <EOL> tmpdir = str ( tmpdir ) <EOL> jip . scanner . add_module ( '<STR_LIT>' ) <EOL> jip . scanner . scan_modules ( ) <EOL> p = jip . Pipeline ( ) <EOL> p . job ( dir = tmpdir ) . run ( '<STR_LIT>' ) <EOL> jobs = jip . create_jobs ( p ) <EOL> assert len ( jobs ) == <NUM_LIT:1> <EOL> def test_hello_world_py_cls ( tmpdir ) : <EOL> tmpdir = str ( tmpdir ) <EOL> jip . scanner . add_module ( '<STR_LIT>' ) <EOL> jip . scanner . scan_modules ( ) <EOL> p = jip . Pipeline ( ) <EOL> p . job ( dir = tmpdir ) . run ( '<STR_LIT>' ) <EOL> jobs = jip . create_jobs ( p ) <EOL> assert len ( jobs ) == <NUM_LIT:1> <EOL> def test_file_touch ( ) : <EOL> p = jip . Pipeline ( ) <EOL> node = p . run ( '<STR_LIT>' , p = '<STR_LIT:test>' , c = <NUM_LIT:5> ) <EOL> p . expand ( ) <EOL> cwd = os . getcwd ( ) <EOL> j = os . path . join <EOL> assert node . output == [ <EOL> j ( cwd , '<STR_LIT>' + x ) for x in [ '<STR_LIT:1>' , '<STR_LIT:2>' , '<STR_LIT:3>' , '<STR_LIT:4>' , '<STR_LIT:5>' ] <EOL> ] <EOL> def test_pipeline_to_pipeline_edge_delegation ( ) : <EOL> @ jip . tool ( '<STR_LIT>' ) <EOL> class GemIndex ( object ) : <EOL> """<STR_LIT>""" <EOL> def get_command ( self ) : <EOL> return "<STR_LIT>" , "<STR_LIT>" <EOL> @ jip . tool ( '<STR_LIT>' ) <EOL> class gem ( object ) : <EOL> """<STR_LIT>""" <EOL> def get_command ( self ) : <EOL> return '<STR_LIT>' , '<STR_LIT>' <EOL> @ jip . pipeline ( '<STR_LIT>' ) <EOL> class SetupPipeline ( object ) : <EOL> """<STR_LIT>""" <EOL> def init ( self ) : <EOL> self . add_output ( '<STR_LIT:index>' , '<STR_LIT>' ) <EOL> def pipeline ( self ) : <EOL> p = jip . Pipeline ( ) <EOL> index = p . run ( '<STR_LIT>' , <EOL> input = self . input , output = self . index ) <EOL> p . context ( locals ( ) ) <EOL> return p <EOL> @ jip . pipeline ( '<STR_LIT>' ) <EOL> class GrapePipeline ( object ) : <EOL> """<STR_LIT>""" <EOL> def pipeline ( self ) : <EOL> p = jip . Pipeline ( ) <EOL> gem_setup = p . run ( '<STR_LIT>' , input = self . genome ) <EOL> gem = p . run ( '<STR_LIT>' , index = gem_setup . index , <EOL> fastq = self . fastq ) <EOL> p . context ( locals ( ) ) <EOL> return p <EOL> p = jip . Pipeline ( ) <EOL> node = p . run ( '<STR_LIT>' ) <EOL> node . fastq = '<STR_LIT>' <EOL> node . genome = '<STR_LIT>' <EOL> p . expand ( validate = False ) <EOL> index = p . get ( '<STR_LIT:index>' ) <EOL> gem_node = p . get ( '<STR_LIT>' ) <EOL> cwd = os . getcwd ( ) <EOL> j = os . path . join <EOL> assert index . has_outgoing ( gem_node , link = ( '<STR_LIT>' , '<STR_LIT:index>' ) , <EOL> value = j ( cwd , '<STR_LIT>' ) ) <EOL> def test_subpipe_incoming_edge_resolve ( ) : <EOL> @ jip . pipeline ( ) <EOL> def subedge_pipe ( tool ) : <EOL> """<STR_LIT>""" <EOL> p = jip . Pipeline ( ) <EOL> p . job ( '<STR_LIT>' ) . bash ( '<STR_LIT>' , input = tool . input ) <EOL> return p <EOL> p = jip . Pipeline ( ) <EOL> produce = p . job ( '<STR_LIT>' ) . bash ( '<STR_LIT>' , outfile = '<STR_LIT>' ) <EOL> p . run ( '<STR_LIT>' , input = produce . outfile ) <EOL> p . expand ( validate = False ) <EOL> touch = p . get ( '<STR_LIT>' ) <EOL> ls = p . get ( '<STR_LIT>' ) <EOL> cwd = os . getcwd ( ) <EOL> j = os . path . join <EOL> assert touch . has_outgoing ( ls , link = ( '<STR_LIT>' , '<STR_LIT:input>' ) , <EOL> value = j ( cwd , '<STR_LIT>' ) ) <EOL> def test_subpipe_incoming_edge_resolve_pipe_to_pipe ( ) : <EOL> @ jip . pipeline ( ) <EOL> def subedge_pipe_1 ( tool ) : <EOL> """<STR_LIT>""" <EOL> p = jip . Pipeline ( ) <EOL> p . job ( '<STR_LIT>' ) . bash ( '<STR_LIT>' , input = tool . input , output = tool . output ) <EOL> return p <EOL> @ jip . pipeline ( ) <EOL> def subedge_pipe_2 ( tool ) : <EOL> """<STR_LIT>""" <EOL> p = jip . Pipeline ( ) <EOL> p . job ( '<STR_LIT>' ) . bash ( '<STR_LIT>' , input = tool . input , output = tool . output ) <EOL> return p <EOL> @ jip . pipeline ( ) <EOL> def subedge_pipe_combine ( tool ) : <EOL> """<STR_LIT>""" <EOL> p = jip . Pipeline ( ) <EOL> p1 = p . run ( '<STR_LIT>' , input = tool . input , output = '<STR_LIT>' ) <EOL> p . run ( '<STR_LIT>' , input = p1 , output = tool . output ) <EOL> return p <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , input = '<STR_LIT>' , output = "<STR_LIT>" ) <EOL> p . expand ( validate = False ) <EOL> assert len ( p ) == <NUM_LIT:2> <EOL> p1 = p . get ( '<STR_LIT>' ) <EOL> p2 = p . get ( '<STR_LIT>' ) <EOL> cwd = os . getcwd ( ) <EOL> j = os . path . join <EOL> assert p1 . has_outgoing ( p2 ) <EOL> assert p1 . has_outgoing ( p2 , value = j ( cwd , '<STR_LIT>' ) ) <EOL> assert p1 . has_outgoing ( p2 , link = ( '<STR_LIT>' , '<STR_LIT:input>' ) , <EOL> value = j ( cwd , '<STR_LIT>' ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
