<s> from sparts . tasks . twisted_command import CommandTask <EOL> from sparts . tests . base import SingleTaskTestCase <EOL> import threading <EOL> class PollerTests ( SingleTaskTestCase ) : <EOL> TASK = CommandTask <EOL> def test_line_buffered ( self ) : <EOL> result = [ "<STR_LIT>" , None , threading . Event ( ) ] <EOL> def on_stdout ( trans , msg ) : <EOL> result [ <NUM_LIT:0> ] += msg <EOL> def on_exit ( reason ) : <EOL> result [ <NUM_LIT:1> ] = reason <EOL> result [ <NUM_LIT:2> ] . set ( ) <EOL> self . task . run ( [ '<STR_LIT>' , '<STR_LIT>' ] , on_stdout = on_stdout , <EOL> on_exit = on_exit ) <EOL> result [ <NUM_LIT:2> ] . wait ( ) <EOL> self . assertEqual ( result [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> """<STR_LIT>""" </s>
