<s> from twisted . application import service <EOL> from twisted . internet import defer <EOL> from twisted . trial import unittest <EOL> from piped import processing , util <EOL> from piped . processors import tick_processors <EOL> from piped . providers import tick_provider <EOL> class TickProcessorTest ( unittest . TestCase ) : <EOL> timeout = <NUM_LIT:1> <EOL> def setUp ( self ) : <EOL> self . runtime_environment = processing . RuntimeEnvironment ( ) <EOL> self . runtime_environment . application = service . MultiService ( ) <EOL> self . runtime_environment . application . startService ( ) <EOL> self . runtime_environment . resource_manager . register ( '<STR_LIT>' , self ) <EOL> self . dependency_manager = self . runtime_environment . dependency_manager <EOL> self . dependency_manager . configure ( self . runtime_environment ) <EOL> self . provider = tick_provider . TickProvider ( ) <EOL> self . provider . setServiceParent ( self . runtime_environment . application ) <EOL> self . intervals = dict ( ) <EOL> self . runtime_environment . configuration_manager . set ( '<STR_LIT>' , self . intervals ) <EOL> self . start_processor = tick_processors . StartInterval ( '<STR_LIT>' ) <EOL> self . stop_processor = tick_processors . StopInterval ( '<STR_LIT>' ) <EOL> self . ticks = defer . DeferredQueue ( ) <EOL> def tearDown ( self ) : <EOL> self . runtime_environment . application . stopService ( ) <EOL> def add_consumer ( self , resource_dependency ) : <EOL> resource_dependency . on_resource_ready ( self . process ) <EOL> def process ( self , baton ) : <EOL> return self . ticks . put ( baton ) <EOL> @ defer . inlineCallbacks <EOL> def test_starting_an_interval ( self ) : <EOL> self . intervals [ '<STR_LIT>' ] = dict ( interval = <NUM_LIT:0> , auto_start = False , processor = '<STR_LIT>' ) <EOL> self . provider . configure ( self . runtime_environment ) <EOL> self . start_processor . configure ( self . runtime_environment ) <EOL> self . dependency_manager . resolve_initial_states ( ) <EOL> self . assertEquals ( self . ticks . pending , list ( ) ) <EOL> self . start_processor . process ( dict ( ) ) <EOL> yield self . ticks . get ( ) <EOL> self . assertEquals ( self . ticks . pending , list ( ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_stopping_an_interval ( self ) : <EOL> self . intervals [ '<STR_LIT>' ] = dict ( interval = <NUM_LIT:0> , processor = '<STR_LIT>' ) <EOL> self . provider . configure ( self . runtime_environment ) <EOL> self . stop_processor . configure ( self . runtime_environment ) <EOL> self . dependency_manager . resolve_initial_states ( ) <EOL> self . assertEquals ( len ( self . ticks . pending ) , <NUM_LIT:1> ) <EOL> yield self . ticks . get ( ) <EOL> yield self . stop_processor . process ( dict ( ) ) <EOL> yield util . wait ( <NUM_LIT:0> ) <EOL> self . assertEquals ( self . ticks . pending , list ( ) ) <EOL> @ defer . inlineCallbacks <EOL> def test_restarting_an_interval ( self ) : <EOL> self . intervals [ '<STR_LIT>' ] = dict ( interval = <NUM_LIT:0> , auto_start = False , processor = '<STR_LIT>' ) <EOL> self . provider . configure ( self . runtime_environment ) <EOL> self . start_processor . configure ( self . runtime_environment ) <EOL> self . stop_processor . configure ( self . runtime_environment ) <EOL> self . dependency_manager . resolve_initial_states ( ) <EOL> self . assertEquals ( self . ticks . pending , list ( ) ) <EOL> self . start_processor . process ( dict ( ) ) <EOL> yield self . ticks . get ( ) <EOL> self . assertEquals ( self . ticks . pending , list ( ) ) <EOL> yield self . stop_processor . process ( dict ( ) ) <EOL> yield util . wait ( <NUM_LIT:0> ) <EOL> self . assertEquals ( self . ticks . pending , list ( ) ) <EOL> self . start_processor . process ( dict ( ) ) <EOL> yield self . ticks . get ( ) <EOL> self . assertEquals ( self . ticks . pending , list ( ) ) </s>
