<s> from helpers import unittest <EOL> from luigi import six <EOL> from helpers import with_config <EOL> import luigi <EOL> from luigi . db_task_history import DbTaskHistory <EOL> from luigi . task_status import DONE , PENDING , RUNNING <EOL> import luigi . scheduler <EOL> class DummyTask ( luigi . Task ) : <EOL> foo = luigi . Parameter ( default = '<STR_LIT:foo>' ) <EOL> class ParamTask ( luigi . Task ) : <EOL> param1 = luigi . Parameter ( ) <EOL> param2 = luigi . IntParameter ( ) <EOL> class DbTaskHistoryTest ( unittest . TestCase ) : <EOL> @ with_config ( dict ( task_history = dict ( db_connection = '<STR_LIT>' ) ) ) <EOL> def setUp ( self ) : <EOL> self . history = DbTaskHistory ( ) <EOL> def test_task_list ( self ) : <EOL> self . run_task ( DummyTask ( ) ) <EOL> self . run_task ( DummyTask ( foo = '<STR_LIT:bar>' ) ) <EOL> tasks = list ( self . history . find_all_by_name ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( len ( tasks ) , <NUM_LIT:2> ) <EOL> for task in tasks : <EOL> self . assertEqual ( task . name , '<STR_LIT>' ) <EOL> self . assertEqual ( task . host , '<STR_LIT>' ) <EOL> def test_task_events ( self ) : <EOL> self . run_task ( DummyTask ( ) ) <EOL> tasks = list ( self . history . find_all_by_name ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( len ( tasks ) , <NUM_LIT:1> ) <EOL> [ task ] = tasks <EOL> self . assertEqual ( task . name , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( task . events ) , <NUM_LIT:3> ) <EOL> for ( event , name ) in zip ( task . events , [ DONE , RUNNING , PENDING ] ) : <EOL> self . assertEqual ( event . event_name , name ) <EOL> def test_task_by_params ( self ) : <EOL> task1 = ParamTask ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> task2 = ParamTask ( '<STR_LIT:bar>' , '<STR_LIT:foo>' ) <EOL> self . run_task ( task1 ) <EOL> self . run_task ( task2 ) <EOL> task1_record = self . history . find_all_by_parameters ( task_name = '<STR_LIT>' , param1 = '<STR_LIT:foo>' , param2 = '<STR_LIT:bar>' ) <EOL> task2_record = self . history . find_all_by_parameters ( task_name = '<STR_LIT>' , param1 = '<STR_LIT:bar>' , param2 = '<STR_LIT:foo>' ) <EOL> for task , records in zip ( ( task1 , task2 ) , ( task1_record , task2_record ) ) : <EOL> records = list ( records ) <EOL> self . assertEqual ( len ( records ) , <NUM_LIT:1> ) <EOL> [ record ] = records <EOL> self . assertEqual ( task . task_family , record . name ) <EOL> for param_name , param_value in six . iteritems ( task . param_kwargs ) : <EOL> self . assertTrue ( param_name in record . parameters ) <EOL> self . assertEqual ( str ( param_value ) , record . parameters [ param_name ] . value ) <EOL> def test_task_blank_param ( self ) : <EOL> self . run_task ( DummyTask ( foo = "<STR_LIT>" ) ) <EOL> tasks = list ( self . history . find_all_by_name ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( len ( tasks ) , <NUM_LIT:1> ) <EOL> task_record = tasks [ <NUM_LIT:0> ] <EOL> self . assertEqual ( task_record . name , '<STR_LIT>' ) <EOL> self . assertEqual ( task_record . host , '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT:foo>' , task_record . parameters ) <EOL> self . assertEqual ( task_record . parameters [ '<STR_LIT:foo>' ] . value , '<STR_LIT>' ) <EOL> def run_task ( self , task ) : <EOL> task2 = luigi . scheduler . Task ( task . task_id , PENDING , [ ] , family = task . task_family , <EOL> params = task . param_kwargs ) <EOL> self . history . task_scheduled ( task2 ) <EOL> self . history . task_started ( task2 , '<STR_LIT>' ) <EOL> self . history . task_finished ( task2 , successful = True ) <EOL> class MySQLDbTaskHistoryTest ( unittest . TestCase ) : <EOL> @ with_config ( dict ( task_history = dict ( db_connection = '<STR_LIT>' ) ) ) <EOL> def setUp ( self ) : <EOL> try : <EOL> self . history = DbTaskHistory ( ) <EOL> except Exception : <EOL> raise unittest . SkipTest ( '<STR_LIT>' ) <EOL> def test_subsecond_timestamp ( self ) : <EOL> task = DummyTask ( ) <EOL> self . run_task ( task ) <EOL> task_record = six . advance_iterator ( self . history . find_all_by_name ( '<STR_LIT>' ) ) <EOL> print ( task_record . events ) <EOL> self . assertEqual ( task_record . events [ <NUM_LIT:0> ] . event_name , DONE ) <EOL> def test_utc_conversion ( self ) : <EOL> from luigi . server import from_utc <EOL> task = DummyTask ( ) <EOL> self . run_task ( task ) <EOL> task_record = six . advance_iterator ( self . history . find_all_by_name ( '<STR_LIT>' ) ) <EOL> last_event = task_record . events [ <NUM_LIT:0> ] <EOL> try : <EOL> print ( from_utc ( str ( last_event . ts ) ) ) <EOL> except ValueError : <EOL> self . fail ( "<STR_LIT>" . format ( last_event . ts ) ) <EOL> def run_task ( self , task ) : <EOL> task2 = luigi . scheduler . Task ( task . task_id , PENDING , [ ] , <EOL> family = task . task_family , params = task . param_kwargs ) <EOL> self . history . task_scheduled ( task2 ) <EOL> self . history . task_started ( task2 , '<STR_LIT>' ) <EOL> self . history . task_finished ( task2 , successful = True ) </s>
