<s> import os <EOL> import jip <EOL> import jip . db <EOL> import datetime <EOL> import pytest <EOL> @ pytest . fixture ( scope = "<STR_LIT>" , <EOL> params = [ "<STR_LIT>" , <EOL> pytest . mark . mysqltest ( pytest . config . option . mysql ) ] ) <EOL> def db ( request ) : <EOL> param = request . param <EOL> def clean ( ) : <EOL> session = jip . db . create_session ( ) <EOL> session . close ( ) <EOL> engine = jip . db . engine <EOL> engine . execute ( jip . db . InputFile . __table__ . delete ( ) ) <EOL> engine . execute ( jip . db . OutputFile . __table__ . delete ( ) ) <EOL> engine . execute ( jip . db . job_dependencies . delete ( ) ) <EOL> engine . execute ( jip . db . job_groups . delete ( ) ) <EOL> engine . execute ( jip . db . job_pipes . delete ( ) ) <EOL> engine . execute ( jip . db . Job . __table__ . delete ( ) ) <EOL> engine . execute ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in param : <EOL> request . addfinalizer ( clean ) <EOL> return param <EOL> def test_command_change_save ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> j = jip . db . Job ( ) <EOL> j . command = "<STR_LIT:A>" <EOL> jip . db . save ( j ) <EOL> assert jip . db . get ( j . id ) . command == "<STR_LIT:A>" <EOL> j . command = "<STR_LIT:B>" <EOL> jip . db . save ( j ) <EOL> assert jip . db . get ( j . id ) . command == "<STR_LIT:B>" <EOL> def test_updating_state ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> j = jip . db . Job ( ) <EOL> jip . db . save ( j ) <EOL> j = jip . db . get ( j . id ) <EOL> assert j is not None <EOL> assert j . create_date is not None <EOL> assert j . start_date is None <EOL> assert j . finish_date is None <EOL> assert j . job_id is None <EOL> assert j . state == jip . db . STATE_HOLD <EOL> assert len ( j . pipe_to ) == <NUM_LIT:0> <EOL> assert len ( j . pipe_from ) == <NUM_LIT:0> <EOL> date = datetime . datetime . now ( ) <EOL> j . job_id = <NUM_LIT:10> <EOL> j . start_date = date <EOL> j . finish_date = date <EOL> j . state = jip . db . STATE_DONE <EOL> jip . db . update_job_states ( j ) <EOL> fresh = jip . db . get ( j . id ) <EOL> assert fresh is not None <EOL> assert fresh . job_id == "<STR_LIT>" <EOL> assert fresh . state == jip . db . STATE_DONE <EOL> assert len ( jip . db . get_all ( ) ) == <NUM_LIT:1> <EOL> def test_update_archived_single ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> j = jip . db . Job ( ) <EOL> jip . db . save ( j ) <EOL> assert not jip . db . get ( j . id ) . archived <EOL> jip . db . update_archived ( j , True ) <EOL> assert jip . db . get ( j . id ) . archived <EOL> def test_update_archived_multiple ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> j1 = jip . db . Job ( ) <EOL> j2 = jip . db . Job ( ) <EOL> jip . db . save ( [ j1 , j2 ] ) <EOL> assert not jip . db . get ( j1 . id ) . archived <EOL> assert not jip . db . get ( j2 . id ) . archived <EOL> jip . db . update_archived ( [ j1 , j2 ] , True ) <EOL> assert jip . db . get ( j1 . id ) . archived <EOL> assert jip . db . get ( j2 . id ) . archived <EOL> def test_update_state_of_non_existing_job ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> j = jip . db . Job ( ) <EOL> j . id = <NUM_LIT:100> <EOL> jip . db . update_job_states ( j ) <EOL> assert len ( jip . db . get_all ( ) ) == <NUM_LIT:0> <EOL> def test_delete_single_job ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> j = jip . db . Job ( ) <EOL> jip . db . save ( j ) <EOL> assert len ( jip . db . get_all ( ) ) == <NUM_LIT:1> <EOL> jip . db . delete ( j ) <EOL> assert len ( jip . db . get_all ( ) ) == <NUM_LIT:0> <EOL> def test_get_state ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> j = jip . db . Job ( ) <EOL> jip . db . save ( j ) <EOL> assert jip . db . get_current_state ( j ) == jip . db . STATE_HOLD <EOL> j . state = jip . db . STATE_DONE <EOL> jip . db . update_job_states ( j ) <EOL> assert jip . db . get_current_state ( j ) == jip . db . STATE_DONE <EOL> def test_delete_job_with_parent_job ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> parent = jip . db . Job ( ) <EOL> child = jip . db . Job ( ) <EOL> child . dependencies . append ( parent ) <EOL> jip . db . save ( [ parent , child ] ) <EOL> assert len ( jip . db . get_all ( ) ) == <NUM_LIT:2> <EOL> jip . db . delete ( child ) <EOL> assert len ( jip . db . get_all ( ) ) == <NUM_LIT:1> <EOL> child = jip . db . get ( child . id ) <EOL> assert child is None <EOL> parent = jip . db . get ( parent . id ) <EOL> assert parent is not None <EOL> assert len ( parent . children ) == <NUM_LIT:0> <EOL> s = jip . db . job_dependencies . select ( ) <EOL> c = jip . db . engine . connect ( ) <EOL> res = c . execute ( s ) <EOL> count = sum ( map ( lambda x : <NUM_LIT:1> , res ) ) <EOL> c . close ( ) <EOL> assert count == <NUM_LIT:0> <EOL> def test_delete_unknown_job ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> job = jip . db . Job ( ) <EOL> job . id = <NUM_LIT:100> <EOL> assert len ( jip . db . get_all ( ) ) == <NUM_LIT:0> <EOL> jip . db . delete ( job ) <EOL> assert len ( jip . db . get_all ( ) ) == <NUM_LIT:0> <EOL> def test_cascaded_save ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> parent = jip . db . Job ( ) <EOL> child = jip . db . Job ( ) <EOL> child . dependencies . append ( parent ) <EOL> jip . db . save ( [ parent ] ) <EOL> assert len ( jip . db . get_all ( ) ) == <NUM_LIT:2> <EOL> def test_ouput_file_update_on_insert ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , cmd = "<STR_LIT>" , outfile = "<STR_LIT>" ) <EOL> jobs = jip . create_jobs ( p ) <EOL> jip . db . save ( jobs ) <EOL> assert len ( jobs [ <NUM_LIT:0> ] . out_files ) == <NUM_LIT:1> <EOL> assert jobs [ <NUM_LIT:0> ] . out_files [ <NUM_LIT:0> ] . job == jobs [ <NUM_LIT:0> ] <EOL> assert jip . db . create_session ( ) . query ( jip . db . OutputFile ) . count ( ) == <NUM_LIT:1> <EOL> def test_input_file_update_on_insert ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , cmd = "<STR_LIT>" , input = "<STR_LIT>" ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> jip . db . save ( jobs ) <EOL> assert len ( jobs [ <NUM_LIT:0> ] . in_files ) == <NUM_LIT:1> <EOL> assert jobs [ <NUM_LIT:0> ] . in_files [ <NUM_LIT:0> ] . job == jobs [ <NUM_LIT:0> ] <EOL> assert jip . db . create_session ( ) . query ( jip . db . InputFile ) . count ( ) == <NUM_LIT:1> <EOL> def test_input_file_delete_on_job_delete ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , cmd = "<STR_LIT>" , input = "<STR_LIT>" ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> jip . db . save ( jobs ) <EOL> assert jip . db . create_session ( ) . query ( jip . db . InputFile ) . count ( ) == <NUM_LIT:1> <EOL> jip . db . delete ( jobs ) <EOL> assert jip . db . create_session ( ) . query ( jip . db . Job ) . count ( ) == <NUM_LIT:0> <EOL> def test_output_file_delete_on_job_delete ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , cmd = "<STR_LIT>" , outfile = "<STR_LIT>" ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> jip . db . save ( jobs ) <EOL> assert jip . db . create_session ( ) . query ( jip . db . OutputFile ) . count ( ) == <NUM_LIT:1> <EOL> jip . db . delete ( jobs ) <EOL> assert jip . db . create_session ( ) . query ( jip . db . Job ) . count ( ) == <NUM_LIT:0> <EOL> assert jip . db . create_session ( ) . query ( jip . db . OutputFile ) . count ( ) == <NUM_LIT:0> <EOL> def test_output_file_query ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , cmd = "<STR_LIT>" , outfile = "<STR_LIT>" ) <EOL> abspath = os . path . join ( os . getcwd ( ) , "<STR_LIT>" ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> jip . db . save ( jobs ) <EOL> job = jip . db . query_by_files ( outputs = abspath ) <EOL> assert job . count ( ) == <NUM_LIT:1> <EOL> assert job . one ( ) == jobs [ <NUM_LIT:0> ] <EOL> def test_file_query_and ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , cmd = "<STR_LIT>" , outfile = "<STR_LIT>" , input = "<STR_LIT>" ) <EOL> abspath = os . path . join ( os . getcwd ( ) , "<STR_LIT>" ) <EOL> in_file = os . path . join ( os . getcwd ( ) , '<STR_LIT>' ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> jip . db . save ( jobs ) <EOL> job = jip . db . query_by_files ( inputs = [ ] , outputs = abspath , and_query = True ) <EOL> assert job . count ( ) == <NUM_LIT:0> <EOL> job = jip . db . query_by_files ( inputs = [ in_file ] , <EOL> outputs = abspath , and_query = True ) <EOL> assert job . count ( ) == <NUM_LIT:1> <EOL> assert job . one ( ) == jobs [ <NUM_LIT:0> ] <EOL> def test_output_file_query_multiple_files ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , cmd = "<STR_LIT>" , outfile = [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> abspath_A = os . path . join ( os . getcwd ( ) , "<STR_LIT>" ) <EOL> abspath_B = os . path . join ( os . getcwd ( ) , "<STR_LIT>" ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> jip . db . save ( jobs ) <EOL> job = jip . db . query_by_files ( outputs = [ abspath_A , abspath_B ] ) <EOL> assert job . count ( ) == <NUM_LIT:2> <EOL> def test_input_file_query ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , cmd = "<STR_LIT>" , input = "<STR_LIT>" ) <EOL> abspath = os . path . join ( os . getcwd ( ) , "<STR_LIT>" ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> jip . db . save ( jobs ) <EOL> job = jip . db . query_by_files ( inputs = abspath ) <EOL> assert job . count ( ) == <NUM_LIT:1> <EOL> assert job . one ( ) == jobs [ <NUM_LIT:0> ] <EOL> def test_iput_file_query_multiple_files ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> p = jip . Pipeline ( ) <EOL> p . run ( '<STR_LIT>' , cmd = "<STR_LIT>" , input = [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> abspath_A = os . path . join ( os . getcwd ( ) , "<STR_LIT>" ) <EOL> abspath_B = os . path . join ( os . getcwd ( ) , "<STR_LIT>" ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> assert len ( jobs ) == <NUM_LIT:2> <EOL> jip . db . save ( jobs ) <EOL> job = jip . db . query_by_files ( inputs = [ abspath_A , abspath_B ] ) <EOL> assert job . count ( ) == <NUM_LIT:2> <EOL> def test_no_duplicated_jobs ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> p = jip . Pipeline ( ) <EOL> a = p . bash ( '<STR_LIT>' , input = '<STR_LIT>' ) <EOL> b = p . bash ( '<STR_LIT>' , input = '<STR_LIT>' ) <EOL> p . context ( locals ( ) ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> assert len ( jobs ) == <NUM_LIT:2> <EOL> assert len ( list ( jip . db . get_all ( ) ) ) == <NUM_LIT:0> <EOL> jip . db . save ( jobs [ <NUM_LIT:0> ] ) <EOL> assert len ( list ( jip . db . get_all ( ) ) ) == <NUM_LIT:1> <EOL> assert jobs [ <NUM_LIT:0> ] . id == <NUM_LIT:1> <EOL> assert jobs [ <NUM_LIT:1> ] . id is None <EOL> def test_no_duplicated_jobs_after_file_query_direct ( db , tmpdir ) : <EOL> if not db . startswith ( '<STR_LIT>' ) : <EOL> db = os . path . join ( str ( tmpdir ) , db ) <EOL> jip . db . init ( db ) <EOL> p = jip . Pipeline ( ) <EOL> a = p . bash ( '<STR_LIT>' , input = '<STR_LIT>' , output = '<STR_LIT>' ) <EOL> p . context ( locals ( ) ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> jip . db . save ( jobs ) <EOL> assert len ( list ( jip . db . get_all ( ) ) ) == <NUM_LIT:1> <EOL> assert jobs [ <NUM_LIT:0> ] . id == <NUM_LIT:1> <EOL> p = jip . Pipeline ( ) <EOL> a = p . bash ( '<STR_LIT>' , input = '<STR_LIT>' , output = '<STR_LIT>' ) <EOL> b = p . bash ( '<STR_LIT>' , input = a ) <EOL> p . context ( locals ( ) ) <EOL> jobs = jip . create_jobs ( p , validate = False ) <EOL> existing = jip . db . query_by_files ( <EOL> outputs = jobs [ <NUM_LIT:1> ] . tool . input . value <EOL> ) <EOL> assert len ( list ( existing ) ) == <NUM_LIT:1> <EOL> old = list ( existing ) [ <NUM_LIT:0> ] <EOL> jobs [ <NUM_LIT:1> ] . dependencies = [ old ] <EOL> jip . db . save ( jobs [ <NUM_LIT:1> ] ) <EOL> assert len ( list ( jip . db . get_all ( ) ) ) == <NUM_LIT:2> <EOL> assert jobs [ <NUM_LIT:0> ] . id is None <EOL> @ pytest . mark . mysqltest <EOL> def test_mysql_init ( mysql ) : <EOL> jip . db . init ( mysql ) <EOL> assert not os . path . exists ( '<STR_LIT>' ) </s>
