<s> import pytest <EOL> import sys <EOL> import argparse <EOL> from calvin . actorstore . store import ActorStore <EOL> from calvin . runtime . north . calvin_token import Token <EOL> from calvin . runtime . south . endpoint import Endpoint <EOL> from calvin . runtime . north import metering <EOL> def fwrite ( port , value ) : <EOL> if isinstance ( value , Token ) : <EOL> port . fifo . write ( value ) <EOL> else : <EOL> port . fifo . write ( Token ( value = value ) ) <EOL> def pwrite ( actor , portname , value ) : <EOL> port = actor . inports . get ( portname ) <EOL> if not port : <EOL> raise Exception ( "<STR_LIT>" % ( portname , ) ) <EOL> if isinstance ( value , list ) : <EOL> for v in value : <EOL> fwrite ( port , v ) <EOL> else : <EOL> fwrite ( port , value ) <EOL> def pread ( actor , portname , number = <NUM_LIT:1> ) : <EOL> port = actor . outports . get ( portname , None ) <EOL> assert port <EOL> if number > <NUM_LIT:0> : <EOL> if pavailable ( actor , portname ) < number : <EOL> raise AssertionError ( "<STR_LIT>" % ( pavailable ( actor , portname ) , number ) ) <EOL> else : <EOL> if pavailable ( actor , portname ) > number : <EOL> raise AssertionError ( "<STR_LIT>" % ( pavailable ( actor , portname ) , number ) ) <EOL> values = [ port . fifo . read ( actor . id ) . value for _ in range ( number ) ] <EOL> port . fifo . commit_reads ( actor . id ) , True <EOL> return values <EOL> def pavailable ( actor , portname ) : <EOL> port = actor . outports . get ( portname , None ) <EOL> assert port <EOL> return port . fifo . available_tokens ( actor . id ) <EOL> class DummyInEndpoint ( Endpoint ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , port ) : <EOL> super ( DummyInEndpoint , self ) . __init__ ( port ) <EOL> def is_connected ( self ) : <EOL> return True <EOL> def read_token ( self ) : <EOL> token = self . port . fifo . read ( self . port . id ) <EOL> if token : <EOL> self . port . fifo . commit_reads ( self . port . id , True ) <EOL> return token <EOL> def available_tokens ( self ) : <EOL> tokens = <NUM_LIT:0> <EOL> tokens += self . port . fifo . available_tokens ( self . port . id ) <EOL> return tokens <EOL> def peek_token ( self ) : <EOL> return self . port . fifo . read ( self . port . id ) <EOL> def commit_peek_as_read ( self ) : <EOL> self . port . fifo . commit_reads ( self . port . id ) <EOL> def peek_rewind ( self ) : <EOL> self . port . fifo . rollback_reads ( self . port . id ) <EOL> class FDMock ( object ) : <EOL> def __init__ ( self , fname , mode ) : <EOL> self . fp = open ( fname , mode ) <EOL> if '<STR_LIT:r>' in mode : <EOL> self . buffer = self . fp . read ( ) <EOL> else : <EOL> self . buffer = "<STR_LIT>" <EOL> def close ( self ) : <EOL> self . fp . close ( ) <EOL> def eof ( self ) : <EOL> return len ( self . buffer ) == <NUM_LIT:0> <EOL> def has_data ( self ) : <EOL> return len ( self . buffer ) > <NUM_LIT:0> <EOL> def read ( self ) : <EOL> data = self . buffer <EOL> self . buffer = "<STR_LIT>" <EOL> return data <EOL> def write ( self , data ) : <EOL> self . buffer += data <EOL> self . fp . write ( data ) <EOL> def read_line ( self ) : <EOL> if '<STR_LIT:\n>' in self . buffer : <EOL> line , self . buffer = self . buffer . split ( "<STR_LIT:\n>" , <NUM_LIT:1> ) <EOL> else : <EOL> line = self . buffer <EOL> self . buffer = "<STR_LIT>" <EOL> return line <EOL> def write_line ( self , data ) : <EOL> self . buffer += data + "<STR_LIT:\n>" <EOL> self . fp . write ( data + "<STR_LIT:\n>" ) <EOL> class StdInMock ( FDMock ) : <EOL> def __init__ ( self ) : <EOL> self . buffer = "<STR_LIT>" <EOL> class TimerMock ( object ) : <EOL> def __init__ ( self ) : <EOL> self . _triggered = False <EOL> @ property <EOL> def triggered ( self ) : <EOL> return self . _triggered <EOL> def ack ( self ) : <EOL> assert self . _triggered <EOL> self . _triggered = False <EOL> def cancel ( self ) : <EOL> del self . _triggered <EOL> def trigger ( self ) : <EOL> assert not self . _triggered <EOL> self . _triggered = True <EOL> class CalvinSysTimerMock ( object ) : <EOL> def repeat ( self , delay ) : <EOL> return TimerMock ( ) <EOL> def once ( self , delay ) : <EOL> return TimerMock ( ) <EOL> class CalvinSysFileMock ( object ) : <EOL> def open ( self , fname , mode ) : <EOL> return FDMock ( fname , mode ) <EOL> def open_stdin ( self ) : <EOL> return StdInMock ( ) <EOL> def close ( self , fdmock ) : <EOL> fdmock . close ( ) <EOL> def load_python_requirement ( req ) : <EOL> import importlib <EOL> loaded = importlib . import_module ( "<STR_LIT>" + req ) <EOL> return loaded . register <EOL> requirements = { <EOL> '<STR_LIT>' : CalvinSysFileMock , <EOL> '<STR_LIT>' : CalvinSysTimerMock , <EOL> '<STR_LIT>' : load_python_requirement ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : load_python_requirement ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : load_python_requirement ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : load_python_requirement ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : load_python_requirement ( '<STR_LIT>' ) <EOL> } <EOL> class CalvinSysMock ( dict ) : <EOL> def use_requirement ( self , actor , requirement ) : <EOL> return requirements [ requirement ] ( ) <EOL> class ActorTester ( object ) : <EOL> def __init__ ( self ) : <EOL> self . store = ActorStore ( ) <EOL> self . actors = { } <EOL> self . illegal_actors = { } <EOL> self . components = { } <EOL> self . id = "<STR_LIT>" <EOL> self . metering = metering . set_metering ( metering . Metering ( self ) ) <EOL> def collect_actors ( self , actor ) : <EOL> actors = [ m + '<STR_LIT:.>' + a for m in self . store . modules ( ) for a in self . store . actors ( m ) ] <EOL> if actor : <EOL> actors = [ a for a in actors if actor in a ] <EOL> self . actor_names = actors <EOL> def instantiate_actor ( self , actorclass , actorname ) : <EOL> try : <EOL> actor = actorclass ( actorname , disable_state_checks = True ) <EOL> if not hasattr ( actor , '<STR_LIT>' ) : <EOL> self . actors [ actorname ] = '<STR_LIT>' <EOL> return <EOL> actor . _calvinsys = CalvinSysMock ( ) <EOL> actor . _calvinsys [ '<STR_LIT:file>' ] = CalvinSysFileMock ( ) <EOL> actor . _calvinsys [ '<STR_LIT>' ] = CalvinSysTimerMock ( ) <EOL> actor . init ( * actorclass . test_args , ** actorclass . test_kwargs ) <EOL> actor . setup_complete ( ) <EOL> except Exception as e : <EOL> self . illegal_actors [ actorname ] = "<STR_LIT>" <EOL> sys . stderr . write ( "<STR_LIT>" % ( actorname , e ) ) <EOL> import traceback <EOL> sys . stderr . write ( '<STR_LIT>' . join ( traceback . format_exc ( ) ) ) <EOL> raise e <EOL> for inport in actor . inports . values ( ) : <EOL> inport . endpoint = DummyInEndpoint ( inport ) <EOL> for outport in actor . outports . values ( ) : <EOL> outport . fifo . add_reader ( actor . id ) <EOL> self . actors [ actorname ] = actor <EOL> def instantiate_actors ( self ) : <EOL> for a in self . actor_names : <EOL> found , primitive , actorclass = self . store . lookup ( a ) <EOL> if found and primitive : <EOL> self . instantiate_actor ( actorclass , a ) <EOL> elif found and not primitive : <EOL> self . components [ a ] = "<STR_LIT>" % ( a , ) <EOL> else : <EOL> self . illegal_actors [ a ] = "<STR_LIT>" <EOL> def load_actor ( self , path ) : <EOL> actorclass = self . store . load_from_path ( path ) <EOL> if actorclass : <EOL> self . instantiate_actor ( actorclass , path ) <EOL> else : <EOL> self . illegal_actors [ path ] = "<STR_LIT>" <EOL> def test_actor ( self , actor , aut ) : <EOL> for idx in range ( len ( aut . test_set ) ) : <EOL> test_index = idx + <NUM_LIT:1> <EOL> test = aut . test_set [ idx ] <EOL> setups = test . get ( '<STR_LIT>' , [ ] ) <EOL> inputs = test . get ( '<STR_LIT>' , { } ) <EOL> outputs = test . get ( '<STR_LIT>' , { } ) <EOL> postconds = test . get ( '<STR_LIT>' , [ ] ) <EOL> for f in setups : <EOL> try : <EOL> f ( aut ) <EOL> except Exception as e : <EOL> print "<STR_LIT>" % ( actor , test_index , e . message ) <EOL> raise Exception ( "<STR_LIT>" % ( test_index , ) ) <EOL> for port , values in inputs . iteritems ( ) : <EOL> pwrite ( aut , port , values ) <EOL> aut . fire ( ) <EOL> for port , values in outputs . iteritems ( ) : <EOL> try : <EOL> vals = pread ( aut , port , len ( values ) ) <EOL> assert vals == values , "<STR_LIT>" % ( vals , values ) <EOL> except AssertionError as e : <EOL> print "<STR_LIT>" , str ( e ) <EOL> raise AssertionError ( "<STR_LIT>" % ( test_index , ) ) <EOL> if not all ( f ( aut ) for f in postconds ) : <EOL> raise AssertionError ( "<STR_LIT>" % ( test_index , ) ) <EOL> return True <EOL> def test_actors ( self ) : <EOL> test_pass = [ ] <EOL> test_fail = { } <EOL> no_test = [ ] <EOL> for actor in self . actors : <EOL> aut = self . actors [ actor ] <EOL> if aut == "<STR_LIT>" : <EOL> no_test . append ( actor ) <EOL> continue <EOL> try : <EOL> self . test_actor ( actor , aut ) <EOL> test_pass . append ( actor ) <EOL> except AssertionError as e : <EOL> test_fail [ actor ] = e . message <EOL> except Exception as e : <EOL> self . illegal_actors [ actor ] = str ( e ) <EOL> return { '<STR_LIT>' : test_pass , '<STR_LIT>' : test_fail , '<STR_LIT>' : no_test , <EOL> '<STR_LIT>' : self . illegal_actors , '<STR_LIT>' : self . components } <EOL> def show_result ( header , result ) : <EOL> print header <EOL> for actor in result : <EOL> print "<STR_LIT>" % ( actor , ) <EOL> def show_issue ( header , result ) : <EOL> print header <EOL> for actor , reason in result . iteritems ( ) : <EOL> print "<STR_LIT>" % ( actor , reason ) <EOL> def show_issues ( results ) : <EOL> if results [ '<STR_LIT>' ] : <EOL> show_issue ( "<STR_LIT>" , results [ '<STR_LIT>' ] ) <EOL> if results [ '<STR_LIT>' ] : <EOL> show_issue ( "<STR_LIT>" , results [ '<STR_LIT>' ] ) <EOL> def show_results ( results ) : <EOL> if results [ '<STR_LIT>' ] : <EOL> show_result ( "<STR_LIT>" , results [ '<STR_LIT>' ] ) <EOL> if results [ '<STR_LIT>' ] : <EOL> show_result ( "<STR_LIT>" , results [ '<STR_LIT>' ] ) <EOL> if results [ '<STR_LIT>' ] : <EOL> show_issue ( "<STR_LIT>" , results [ '<STR_LIT>' ] ) <EOL> show_issues ( results ) <EOL> @ pytest . mark . essential <EOL> def test_actors ( actor = "<STR_LIT>" , show = False , path = None ) : <EOL> t = ActorTester ( ) <EOL> if path : <EOL> t . load_actor ( path ) <EOL> else : <EOL> t . collect_actors ( actor ) <EOL> t . instantiate_actors ( ) <EOL> results = t . test_actors ( ) <EOL> if not any ( results . values ( ) ) : <EOL> if actor : <EOL> print "<STR_LIT>" % ( actor , ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> if show : <EOL> return results <EOL> if results [ '<STR_LIT>' ] or results [ '<STR_LIT>' ] : <EOL> show_issues ( results ) <EOL> if results [ '<STR_LIT>' ] : <EOL> raise Exception ( "<STR_LIT>" % ( len ( results [ '<STR_LIT>' ] ) , ) ) <EOL> if results [ '<STR_LIT>' ] : <EOL> raise Exception ( "<STR_LIT>" % ( len ( results [ '<STR_LIT>' ] ) , ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> argparser = argparse . ArgumentParser ( description = "<STR_LIT>" ) <EOL> group = argparser . add_mutually_exclusive_group ( ) <EOL> group . add_argument ( '<STR_LIT>' , '<STR_LIT>' , type = str , <EOL> help = '<STR_LIT>' ) <EOL> group . add_argument ( '<STR_LIT>' , type = str , nargs = '<STR_LIT:?>' , <EOL> help = '<STR_LIT>' ) <EOL> args = argparser . parse_args ( ) <EOL> if args . path : <EOL> show_results ( test_actors ( path = args . path , show = True ) ) <EOL> elif args . filter : <EOL> show_results ( test_actors ( actor = args . filter , show = True ) ) <EOL> else : <EOL> show_results ( test_actors ( show = True ) ) </s>
