<s> from __future__ import print_function <EOL> import datetime <EOL> from helpers import unittest <EOL> import luigi <EOL> import luigi . notifications <EOL> from luigi . mock import MockTarget <EOL> from luigi . util import inherits <EOL> luigi . notifications . DEBUG = True <EOL> class A ( luigi . Task ) : <EOL> def output ( self ) : <EOL> return MockTarget ( '<STR_LIT>' ) <EOL> def run ( self ) : <EOL> f = self . output ( ) . open ( '<STR_LIT:w>' ) <EOL> print ( '<STR_LIT>' , file = f ) <EOL> f . close ( ) <EOL> class B ( luigi . Task ) : <EOL> date = luigi . DateParameter ( ) <EOL> def output ( self ) : <EOL> return MockTarget ( self . date . strftime ( '<STR_LIT>' ) ) <EOL> def run ( self ) : <EOL> f = self . output ( ) . open ( '<STR_LIT:w>' ) <EOL> print ( '<STR_LIT>' , file = f ) <EOL> f . close ( ) <EOL> def XMLWrapper ( cls ) : <EOL> @ inherits ( cls ) <EOL> class XMLWrapperCls ( luigi . Task ) : <EOL> def requires ( self ) : <EOL> return self . clone_parent ( ) <EOL> def run ( self ) : <EOL> f = self . input ( ) . open ( '<STR_LIT:r>' ) <EOL> g = self . output ( ) . open ( '<STR_LIT:w>' ) <EOL> print ( '<STR_LIT>' , file = g ) <EOL> for line in f : <EOL> print ( '<STR_LIT>' + line . strip ( ) + '<STR_LIT>' , file = g ) <EOL> g . close ( ) <EOL> return XMLWrapperCls <EOL> class AXML ( XMLWrapper ( A ) ) : <EOL> def output ( self ) : <EOL> return MockTarget ( '<STR_LIT>' ) <EOL> class BXML ( XMLWrapper ( B ) ) : <EOL> def output ( self ) : <EOL> return MockTarget ( self . date . strftime ( '<STR_LIT>' ) ) <EOL> class WrapperTest ( unittest . TestCase ) : <EOL> '''<STR_LIT>''' <EOL> workers = <NUM_LIT:1> <EOL> def setUp ( self ) : <EOL> MockTarget . fs . clear ( ) <EOL> def test_a ( self ) : <EOL> luigi . build ( [ AXML ( ) ] , local_scheduler = True , no_lock = True , workers = self . workers ) <EOL> self . assertEqual ( MockTarget . fs . get_data ( '<STR_LIT>' ) , b'<STR_LIT>' ) <EOL> def test_b ( self ) : <EOL> luigi . build ( [ BXML ( datetime . date ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) ) ] , local_scheduler = True , no_lock = True , workers = self . workers ) <EOL> self . assertEqual ( MockTarget . fs . get_data ( '<STR_LIT>' ) , b'<STR_LIT>' ) <EOL> class WrapperWithMultipleWorkersTest ( WrapperTest ) : <EOL> workers = <NUM_LIT:7> <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> luigi . run ( ) </s>
