<s> """<STR_LIT>""" <EOL> import pytest <EOL> import sys <EOL> import threading <EOL> import collections <EOL> from time import sleep <EOL> from collections import deque <EOL> from circuits . core . manager import TIMEOUT <EOL> from circuits import handler , BaseComponent , Debugger , Manager <EOL> class Watcher ( BaseComponent ) : <EOL> def init ( self ) : <EOL> self . _lock = threading . Lock ( ) <EOL> self . events = deque ( ) <EOL> @ handler ( channel = "<STR_LIT:*>" , priority = <NUM_LIT> ) <EOL> def _on_event ( self , event , * args , ** kwargs ) : <EOL> with self . _lock : <EOL> self . events . append ( event ) <EOL> def clear ( self ) : <EOL> self . events . clear ( ) <EOL> def wait ( self , name , channel = None , timeout = <NUM_LIT> ) : <EOL> for i in range ( int ( timeout / TIMEOUT ) ) : <EOL> with self . _lock : <EOL> for event in self . events : <EOL> if event . name == name and event . waitingHandlers == <NUM_LIT:0> : <EOL> if ( channel is None ) or ( channel in event . channels ) : <EOL> return True <EOL> sleep ( TIMEOUT ) <EOL> else : <EOL> return False <EOL> class Flag ( object ) : <EOL> status = False <EOL> def call_event_from_name ( manager , event , event_name , * channels ) : <EOL> fired = False <EOL> value = None <EOL> for r in manager . waitEvent ( event_name ) : <EOL> if not fired : <EOL> fired = True <EOL> value = manager . fire ( event , * channels ) <EOL> sleep ( <NUM_LIT:0.1> ) <EOL> return value <EOL> def call_event ( manager , event , * channels ) : <EOL> return call_event_from_name ( manager , event , event . name , * channels ) <EOL> class WaitEvent ( object ) : <EOL> def __init__ ( self , manager , name , channel = None , timeout = <NUM_LIT> ) : <EOL> if channel is None : <EOL> channel = getattr ( manager , "<STR_LIT>" , None ) <EOL> self . timeout = timeout <EOL> self . manager = manager <EOL> flag = Flag ( ) <EOL> @ handler ( name , channel = channel ) <EOL> def on_event ( self , * args , ** kwargs ) : <EOL> flag . status = True <EOL> self . handler = self . manager . addHandler ( on_event ) <EOL> self . flag = flag <EOL> def wait ( self ) : <EOL> try : <EOL> for i in range ( int ( self . timeout / TIMEOUT ) ) : <EOL> if self . flag . status : <EOL> return True <EOL> sleep ( TIMEOUT ) <EOL> finally : <EOL> self . manager . removeHandler ( self . handler ) <EOL> def wait_for ( obj , attr , value = True , timeout = <NUM_LIT> ) : <EOL> from circuits . core . manager import TIMEOUT <EOL> for i in range ( int ( timeout / TIMEOUT ) ) : <EOL> if isinstance ( value , collections . Callable ) : <EOL> if value ( obj , attr ) : <EOL> return True <EOL> elif getattr ( obj , attr ) == value : <EOL> return True <EOL> sleep ( TIMEOUT ) <EOL> @ pytest . fixture <EOL> def manager ( request ) : <EOL> manager = Manager ( ) <EOL> def finalizer ( ) : <EOL> manager . stop ( ) <EOL> request . addfinalizer ( finalizer ) <EOL> waiter = WaitEvent ( manager , "<STR_LIT>" ) <EOL> manager . start ( ) <EOL> assert waiter . wait ( ) <EOL> if request . config . option . verbose : <EOL> verbose = True <EOL> else : <EOL> verbose = False <EOL> Debugger ( events = verbose ) . register ( manager ) <EOL> return manager <EOL> @ pytest . fixture <EOL> def watcher ( request , manager ) : <EOL> watcher = Watcher ( ) . register ( manager ) <EOL> def finalizer ( ) : <EOL> waiter = WaitEvent ( manager , "<STR_LIT>" ) <EOL> watcher . unregister ( ) <EOL> waiter . wait ( ) <EOL> request . addfinalizer ( finalizer ) <EOL> return watcher <EOL> def pytest_namespace ( ) : <EOL> return dict ( ( <EOL> ( "<STR_LIT>" , WaitEvent ) , <EOL> ( "<STR_LIT>" , wait_for ) , <EOL> ( "<STR_LIT>" , call_event ) , <EOL> ( "<STR_LIT>" , sys . platform ) , <EOL> ( "<STR_LIT>" , sys . version_info [ : <NUM_LIT:3> ] ) , <EOL> ( "<STR_LIT>" , call_event_from_name ) , <EOL> ) ) </s>
