<s> from __future__ import absolute_import <EOL> from __future__ import division <EOL> from __future__ import print_function <EOL> from __future__ import unicode_literals <EOL> import os <EOL> from timeit import default_timer as timer <EOL> from goodtables . pipeline import Batch <EOL> from goodtables import exceptions <EOL> from tests import base <EOL> class TestPipeline ( base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestPipeline , self ) . setUp ( ) <EOL> self . batch_csv = os . path . join ( self . data_dir , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . batch_datapackage = os . path . join ( self . data_dir , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_batch_from_csv ( self ) : <EOL> batch = Batch ( self . batch_csv ) <EOL> self . assertEqual ( len ( batch . dataset ) , <NUM_LIT:5> ) <EOL> def test_batch_from_datapackage ( self ) : <EOL> batch = Batch ( self . batch_datapackage , source_type = '<STR_LIT>' ) <EOL> self . assertEqual ( len ( batch . dataset ) , <NUM_LIT:1> ) <EOL> def test_batch_with_pipeline_post_process_handler ( self ) : <EOL> def say_hi ( pipeline ) : <EOL> return '<STR_LIT>' <EOL> batch = Batch ( self . batch_csv , pipeline_post_task = say_hi ) <EOL> rv = batch . run ( ) <EOL> self . assertFalse ( rv ) <EOL> def test_batch_with_batch_post_processor ( self ) : <EOL> def say_hi ( batch ) : <EOL> return '<STR_LIT>' <EOL> batch = Batch ( self . batch_csv , post_task = say_hi ) <EOL> rv = batch . run ( ) <EOL> self . assertFalse ( rv ) <EOL> def test_bad_post_task_raises ( self ) : <EOL> say_hi = '<STR_LIT>' <EOL> self . assertRaises ( exceptions . InvalidHandlerError , Batch , <EOL> self . batch_csv , post_task = say_hi ) <EOL> def test_bad_pipeline_post_task_raises ( self ) : <EOL> say_hi = '<STR_LIT>' <EOL> self . assertRaises ( exceptions . InvalidHandlerError , Batch , <EOL> self . batch_csv , pipeline_post_task = say_hi ) <EOL> def test_batch_with_batch_sleep_time ( self ) : <EOL> def default_time ( ) : <EOL> batch = Batch ( self . batch_csv ) <EOL> start = timer ( ) ; batch . run ( ) ; end = timer ( ) <EOL> return end - start <EOL> def custom_sleep_time ( ) : <EOL> batch = Batch ( self . batch_csv , sleep = <NUM_LIT:3> ) <EOL> start = timer ( ) ; batch . run ( ) ; end = timer ( ) <EOL> return end - start <EOL> self . assertTrue ( default_time ( ) < custom_sleep_time ( ) ) </s>
