<s> from frontera . settings import Settings <EOL> from frontera . contrib . messagebus . zeromq import MessageBus <EOL> from frontera . utils . fingerprint import sha1 <EOL> from random import randint <EOL> from time import sleep <EOL> class MessageBusTester ( object ) : <EOL> def __init__ ( self , settings = Settings ( ) ) : <EOL> settings . set ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> settings . set ( '<STR_LIT>' , True ) <EOL> self . messagebus = MessageBus ( settings ) <EOL> spiderlog = self . messagebus . spider_log ( ) <EOL> self . sw_sl_c = spiderlog . consumer ( partition_id = <NUM_LIT:0> , type = '<STR_LIT>' ) <EOL> scoring_log = self . messagebus . scoring_log ( ) <EOL> self . sw_us_p = scoring_log . producer ( ) <EOL> sleep ( <NUM_LIT:0.1> ) <EOL> self . db_sl_c = spiderlog . consumer ( partition_id = None , type = '<STR_LIT>' ) <EOL> self . db_us_c = scoring_log . consumer ( ) <EOL> spider_feed = self . messagebus . spider_feed ( ) <EOL> self . db_sf_p = spider_feed . producer ( ) <EOL> sleep ( <NUM_LIT:0.1> ) <EOL> self . sp_sl_p = spiderlog . producer ( ) <EOL> self . sp_sf_c = spider_feed . consumer ( <NUM_LIT:0> ) <EOL> sleep ( <NUM_LIT:0.1> ) <EOL> def spider_log_activity ( self , messages ) : <EOL> for i in range ( <NUM_LIT:0> , messages ) : <EOL> if i % <NUM_LIT:2> == <NUM_LIT:0> : <EOL> self . sp_sl_p . send ( sha1 ( str ( randint ( <NUM_LIT:1> , <NUM_LIT:1000> ) ) ) , '<STR_LIT>' + str ( <NUM_LIT:0> ) ) <EOL> else : <EOL> self . sp_sl_p . send ( sha1 ( str ( randint ( <NUM_LIT:1> , <NUM_LIT:1000> ) ) ) , '<STR_LIT>' + str ( <NUM_LIT:0> ) ) <EOL> def spider_feed_activity ( self ) : <EOL> sf_c = <NUM_LIT:0> <EOL> for m in self . sp_sf_c . get_messages ( timeout = <NUM_LIT:1.0> , count = <NUM_LIT> ) : <EOL> sf_c += <NUM_LIT:1> <EOL> return sf_c <EOL> def sw_activity ( self ) : <EOL> c = <NUM_LIT:0> <EOL> for m in self . sw_sl_c . get_messages ( timeout = <NUM_LIT:1.0> , count = <NUM_LIT> ) : <EOL> if m . startswith ( '<STR_LIT>' ) : <EOL> self . sw_us_p . send ( None , '<STR_LIT:message>' + str ( <NUM_LIT:0> ) + "<STR_LIT:U+002C>" + str ( c ) ) <EOL> c += <NUM_LIT:1> <EOL> return c <EOL> def db_activity ( self , messages ) : <EOL> sl_c = <NUM_LIT:0> <EOL> us_c = <NUM_LIT:0> <EOL> for m in self . db_sl_c . get_messages ( timeout = <NUM_LIT:1.0> , count = <NUM_LIT> ) : <EOL> sl_c += <NUM_LIT:1> <EOL> for m in self . db_us_c . get_messages ( timeout = <NUM_LIT:1.0> , count = <NUM_LIT> ) : <EOL> us_c += <NUM_LIT:1> <EOL> for i in range ( <NUM_LIT:0> , messages ) : <EOL> if i % <NUM_LIT:2> == <NUM_LIT:0> : <EOL> self . db_sf_p . send ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> else : <EOL> self . db_sf_p . send ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> return ( sl_c , us_c ) <EOL> class IPv6MessageBusTester ( MessageBusTester ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> settings = Settings ( ) <EOL> settings . set ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> super ( IPv6MessageBusTester , self ) . __init__ ( settings ) <EOL> def test_zmq_message_bus ( ) : <EOL> """<STR_LIT>""" <EOL> tester = MessageBusTester ( ) <EOL> tester . spider_log_activity ( <NUM_LIT:64> ) <EOL> assert tester . sw_activity ( ) == <NUM_LIT:64> <EOL> assert tester . db_activity ( <NUM_LIT> ) == ( <NUM_LIT:64> , <NUM_LIT:32> ) <EOL> assert tester . spider_feed_activity ( ) == <NUM_LIT> </s>
