KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use get_stream method in Airtest
Best Python code snippet using Airtest
GBFS_streamer.py
Source:GBFS_streamer.py 
...80 feed_name.lower(),81 tstamp)82 df_to_csv(df, file_dir, file_name)83 return True84def get_stream(feed_url, program_id, time_zone, time_adjustment, time_adjust_forward = False):85 print('[%s] - Fetching feed from \'%s\'...' % (datetime.now().time(), feed_url))86 try:87 stream_gbfs_feed_json_to_csv(feed_url = feed_url,88 program_id=program_id,89 feed_name=feed_url.split('/')[-1][:-5],90 time_zone = time_zone,91 time_adjustment = time_adjustment,92 time_adjust_forward=time_adjust_forward)93 except:94 # print('---- Something went wrong, moving on ----')95 pass96def main():97 s = sched.scheduler(time.time, time.sleep)98 def do_something(sc,):99 # International100 # get_stream(time_zone = 'America/Toronto', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'sobi_hamilton', feed_url = 'https://hamilton.socialbicycles.com/opendata/station_status.json')101 # get_stream(time_zone = 'America/Toronto', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'velgo', feed_url = 'http://velogo.ca/opendata/station_status.json') # FEED IS EMPTY102 # get_stream(time_zone = 'Asia/Dubai', time_adjustment = pd.Timedelta('4:00:00'), time_adjust_forward = True, program_id = 'ABU', feed_url = 'https://api-core.bikeshare.ae/gbfs/gbfs/en/station_status.json')103 # get_stream(time_zone = 'Europe/Prague', time_adjustment = pd.Timedelta('1:00:00'), time_adjust_forward = True, program_id = 'velonet_cz', feed_url = 'http://velonet.cz/opendata/station_status.json')104 # get_stream(time_zone = 'Australia/Melbourne', time_adjustment = pd.Timedelta('10:00:00'), time_adjust_forward = True, program_id = 'monash_bike_share', feed_url = 'https://monashbikeshare.com/opendata/station_status.json')105 get_stream(time_zone = 'America/Montreal', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bike_share_toronto', feed_url = 'https://tor.publicbikesystem.net/ube/gbfs/v1/en/station_status')106 # get_stream(time_zone = 'Australia/Perth', time_adjustment = pd.Timedelta('8:00:00'), time_adjust_forward = True, program_id = 'curtin_university', feed_url = 'http://curtinbikeshare.com/opendata/station_status.json')107 # Mountain Time108 # get_stream(time_zone = 'America/Denver', time_adjustment = pd.Timedelta('7:00:00'), time_adjust_forward = False, program_id = 'boise_greenbike', feed_url = 'http://boise.greenbike.com/opendata/station_status.json')109 # get_stream(time_zone = 'America/Denver', time_adjustment = pd.Timedelta('7:00:00'), time_adjust_forward = False, program_id = 'bcycle_boulder', feed_url = 'https://gbfs.bcycle.com/bcycle_boulder/station_status.json')110 # get_stream(time_zone = 'America/Denver', time_adjustment = pd.Timedelta('7:00:00'), time_adjust_forward = False, program_id = 'bcycle_denver', feed_url = 'https://gbfs.bcycle.com/bcycle_denver/station_status.json')111 # get_stream(time_zone = 'America/Denver', time_adjustment = pd.Timedelta('7:00:00'), time_adjust_forward = False, program_id = 'bcycle_greenbikeslc', feed_url = 'https://gbfs.bcycle.com/bcycle_greenbikeslc/station_status.json')112 # get_stream(time_zone = 'America/Denver', time_adjustment = pd.Timedelta('7:00:00'), time_adjust_forward = False, program_id = 'mountain_rides_bike_share', feed_url = 'http://mrbikeshare.org/opendata/station_status.json')113 # Central Time114 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_nashville', feed_url = 'https://gbfs.bcycle.com/bcycle_nashville/station_status.json')115 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_austin', feed_url = 'https://gbfs.bcycle.com/bcycle_austin/station_status.json')116 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bike_chattanooga', feed_url = 'https://gbfs.bikechattanooga.com/gbfs/gbfs/en/station_status.json') # replaced by new feed117 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_bublr', feed_url = 'https://gbfs.bcycle.com/bcycle_bublr/station_status.json')118 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_clarksville', feed_url = 'https://gbfs.bcycle.com/bcycle_clarksville/station_status.json')119 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_dallasfairpark', feed_url = 'https://gbfs.bcycle.com/bcycle_dallasfairpark/station_status.json')120 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_desmoines', feed_url = 'https://gbfs.bcycle.com/bcycle_desmoines/station_status.json')121 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_elpaso', feed_url = 'https://gbfs.bcycle.com/bcycle_elpaso/station_status.json')122 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_fortworth', feed_url = 'https://gbfs.bcycle.com/bcycle_fortworth/station_status.json')123 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_greatrides', feed_url = 'https://gbfs.bcycle.com/bcycle_greatrides/station_status.json') # feed is empty...124 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_heartland', feed_url = 'https://gbfs.bcycle.com/bcycle_heartland/station_status.json')125 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_houston', feed_url = 'https://gbfs.bcycle.com/bcycle_houston/station_status.json')126 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_kc', feed_url = 'https://gbfs.bcycle.com/bcycle_kc/station_status.json')127 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_madison', feed_url = 'https://gbfs.bcycle.com/bcycle_madison/station_status.json')128 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_mcallen', feed_url = 'https://gbfs.bcycle.com/bcycle_mcallen/station_status.json')129 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'niceridemn', feed_url = 'https://api-core.niceridemn.org/gbfs/en/station_status.json')130 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_spokies', feed_url = 'https://gbfs.bcycle.com/bcycle_spokies/station_status.json')131 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_rapidcity', feed_url = 'https://gbfs.bcycle.com/bcycle_rapidcity/station_status.json')132 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'bcycle_sanantonio', feed_url = 'https://gbfs.bcycle.com/bcycle_sanantonio/station_status.json')133 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'topeka_metro_bikes', feed_url = 'http://topekametrobikes.org/opendata/station_status.json')134 get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'divvy', feed_url = 'https://gbfs.divvybikes.com/gbfs/en/station_status.json')135 # Pacific Time136 # get_stream(time_zone = 'America/Los_Angeles', time_adjustment = pd.Timedelta('8:00:00'), time_adjust_forward = False, program_id = 'biketown_pdx', feed_url = 'http://biketownpdx.socialbicycles.com/opendata/station_status.json')137 # get_stream(time_zone = 'America/Los_Angeles', time_adjustment = pd.Timedelta('8:00:00'), time_adjust_forward = False, program_id = 'bishop_ranch', feed_url = 'http://britebikes.socialbicycles.com/opendata/station_status.json')138 # get_stream(time_zone = 'America/Los_Angeles', time_adjustment = pd.Timedelta('8:00:00'), time_adjust_forward = False, program_id = 'breeze_bikeshare', feed_url = 'http://santamonicabikeshare.com/opendata/station_status.json')139 # get_stream(time_zone = 'America/Los_Angeles', time_adjustment = pd.Timedelta('8:00:00'), time_adjust_forward = False, program_id = 'grid_bike_share', feed_url = 'https://grid.socialbicycles.com/opendata/station_status.json')140 # get_stream(time_zone = 'America/Los_Angeles', time_adjustment = pd.Timedelta('8:00:00'), time_adjust_forward = False, program_id = 'bcycle_rtcbikeshare', feed_url = 'https://gbfs.bcycle.com/bcycle_rtcbikeshare/station_status.json')141 # get_stream(time_zone = 'America/Los_Angeles', time_adjustment = pd.Timedelta('8:00:00'), time_adjust_forward = False, program_id = 'bcycle_lametro', feed_url = 'https://gbfs.bcycle.com/bcycle_lametro/station_status.json')142 get_stream(time_zone = 'America/Los_Angeles', time_adjustment = pd.Timedelta('8:00:00'), time_adjust_forward = False, program_id = 'BA', feed_url = 'https://gbfs.fordgobike.com/gbfs/en/station_status.json')143 # Eastern Time144 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_columbike', feed_url = 'https://gbfs.bcycle.com/bcycle_columbike/station_status.json')145 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_pacersbikeshare', feed_url = 'https://gbfs.bcycle.com/bcycle_pacersbikeshare/station_status.json')146 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_arborbike', feed_url = 'https://gbfs.bcycle.com/bcycle_arborbike/station_status.json')147 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_aventura', feed_url = 'https://gbfs.bcycle.com/bcycle_aventura/station_status.json')148 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_battlecreek', feed_url = 'https://gbfs.bcycle.com/bcycle_battlecreek/station_status.json')149 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_broward', feed_url = 'https://gbfs.bcycle.com/bcycle_broward/station_status.json')150 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_charlotte', feed_url = 'https://gbfs.bcycle.com/bcycle_charlotte/station_status.json')151 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_cincyredbike', feed_url = 'https://gbfs.bcycle.com/bcycle_cincyredbike/station_status.json')152 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_clemson', feed_url = 'https://gbfs.bcycle.com/bcycle_clemson/station_status.json')153 get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'coast_bike_share', feed_url = 'http://coast.socialbicycles.com/opendata/station_status.json')154 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_columbiacounty', feed_url = 'https://gbfs.bcycle.com/bcycle_columbiacounty/station_status.json')155 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_greenville', feed_url = 'https://gbfs.bcycle.com/bcycle_greenville/station_status.json')156 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_indego', feed_url = 'https://gbfs.bcycle.com/bcycle_indego/station_status.json')157 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_jacksoncounty', feed_url = 'https://gbfs.bcycle.com/bcycle_jacksoncounty/station_status.json')158 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'juice_bike_share', feed_url = 'https://www.juicebikeshare.com/opendata/station_status.json')159 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_linkdayton', feed_url = 'https://gbfs.bcycle.com/bcycle_linkdayton/station_status.json')160 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'reddy_bikeshare', feed_url = 'https://reddybikeshare.socialbicycles.com/opendata/station_status.json')161 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'relay_bike_share', feed_url = 'https://relaybikeshare.socialbicycles.com/opendata/station_status.json')162 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_catbike', feed_url = 'https://gbfs.bcycle.com/bcycle_catbike/station_status.json')163 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'sobi_long_beach', feed_url = 'http://sobilongbeach.com/opendata/station_status.json')164 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'bcycle_spartanburg', feed_url = 'https://gbfs.bcycle.com/bcycle_spartanburg/station_status.json')165 # get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'university_of_virginia', feed_url = 'http://ubike.virginia.edu/opendata/station_status.json')166 get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'cabi', feed_url = 'https://gbfs.capitalbikeshare.com/gbfs/en/station_status.json')167 get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'NYC', feed_url = 'https://gbfs.citibikenyc.com/gbfs/en/station_status.json')168 get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'cogo', feed_url = 'https://gbfs.cogobikeshare.com/gbfs/en/station_status.json')169 get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'hubway', feed_url = 'https://gbfs.thehubway.com/gbfs/en/station_status.json')170 # New Programs!171 # get_stream(time_zone = 'America/Curacao', time_adjustment = pd.Timedelta('4:00:00'), time_adjust_forward = False, program_id = 'aru', feed_url = 'https://aru.publicbikesystem.net/ube/gbfs/v1/en/station_status.json')172 # get_stream(time_zone = 'America/Fortaleza', time_adjustment = pd.Timedelta('3:00:00'), time_adjust_forward = False, program_id = 'rec', feed_url = 'https://rec.publicbikesystem.net/ube/gbfs/v1/en/station_status.json')173 # get_stream(time_zone = 'Atlantic/Reykjavik', time_adjustment = pd.Timedelta('0:00:00'), time_adjust_forward = False, program_id = 'rey', feed_url = 'https://rey.publicbikesystem.net/ube/gbfs/v1/en/station_status.json')174 get_stream(time_zone = 'America/Mexico_City', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'guad', feed_url = 'https://guad.publicbikesystem.net/ube/gbfs/v1/en/station_status.json')175 # get_stream(time_zone = 'America/Denver', time_adjustment = pd.Timedelta('7:00:00'), time_adjust_forward = False, program_id = 'asp', feed_url = 'https://asp.publicbikesystem.net/ube/gbfs/v1/en/station_status.json')176 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'chat', feed_url = 'https://chat.publicbikesystem.net/ube/gbfs/v1/en/station_status.json')177 # get_stream(time_zone = 'America/Chicago', time_adjustment = pd.Timedelta('6:00:00'), time_adjust_forward = False, program_id = 'lou', feed_url = 'https://lou.publicbikesystem.net/ube/gbfs/v1/en/station_status.json')178 get_stream(time_zone = 'America/New_York', time_adjustment = pd.Timedelta('5:00:00'), time_adjust_forward = False, program_id = 'det', feed_url = 'https://det.publicbikesystem.net/ube/gbfs/v1/en/station_status.json')179 # get_stream(time_zone = 'Pacific/Honolulu', time_adjustment = pd.Timedelta('10:00:00'), time_adjust_forward = False, program_id = 'kona', feed_url = 'https://kona.publicbikesystem.net/ube/gbfs/v1/en/station_status.json')180 # get_stream(time_zone = 'Pacific/Honolulu', time_adjustment = pd.Timedelta('10:00:00'), time_adjust_forward = False, program_id = 'hon', feed_url = 'https://hon.publicbikesystem.net/ube/gbfs/v1/en/station_status.json')181 seconds = 60182 print('\t[%s] - Waiting %s minutes to run again...\n' % (datetime.now().time(), INTERVAL))183 s.enter(INTERVAL*seconds, 1, do_something, (sc,))184 s.enter(0, 1, do_something, (s,))185 s.run()186 return True187if __name__ == '__main__':188 global INTERVAL189 INTERVAL = 5...
test_stringdomain_zmq.py
Source:test_stringdomain_zmq.py
...23 24 def _run_letter1(self):25 messaging_system = ZMQMessagingSystem("localhost", self.registry_port, 9001)26 messager = Messager()27 letter1_stream = messager.get_stream(messaging_system, "letter1", StringMessage)28 configuration_stream = messager.get_stream(messaging_system, "configuration", ConfigurationMessage)29 letter1_metrics_stream = messager.get_stream(messaging_system, "letter1_metrics", Metrics)30 service_runner_factory = ServiceRunnerFactory(messager)31 letter1_generator_service_runner = service_runner_factory.get_service_runner(LetterGeneratorService, {"name": "letter1_generator_service", "endpoint": letter1_stream.get_write_endpoint()}, configuration_stream=configuration_stream, metrics_stream=letter1_metrics_stream)32 letter1_generator_service_runner.start()33 self.letter1_stopped = False34 while not self.letter1_stopped:35 time.sleep(1)36 letter1_generator_service_runner.terminate()37 logging.debug("THREAD _run_letter1 terminates.")38 def _run_letter2(self):39 messaging_system = ZMQMessagingSystem("localhost", self.registry_port, 9002)40 messager = Messager()41 letter2_stream = messager.get_stream(messaging_system, "letter2", StringMessage)42 configuration_stream = messager.get_stream(messaging_system, "configuration", ConfigurationMessage)43 letter2_metrics_stream = messager.get_stream(messaging_system, "letter2_metrics", Metrics)44 service_runner_factory = ServiceRunnerFactory(messager)45 letter2_generator_service_runner = service_runner_factory.get_service_runner(LetterGeneratorService, {"name": "letter2_generator_service", "endpoint": letter2_stream.get_write_endpoint()}, configuration_stream=configuration_stream, metrics_stream=letter2_metrics_stream)46 letter2_generator_service_runner.start()47 self.letter2_stopped = False48 while not self.letter2_stopped:49 time.sleep(1)50 letter2_generator_service_runner.terminate()51 logging.debug("THREAD _run_letter2 terminates.")52 def _run_concatenator(self):53 messaging_system = ZMQMessagingSystem("localhost", self.registry_port, 9003)54 messager = Messager()55 letter1_stream = messager.get_stream(messaging_system, "letter1", StringMessage)56 letter2_stream = messager.get_stream(messaging_system, "letter2", StringMessage)57 configuration_stream = messager.get_stream(messaging_system, "configuration", ConfigurationMessage)58 control_stream = messager.get_stream(messaging_system, "control", StringMessage)59 concatenation_stream = messager.get_stream(messaging_system, "concatenation", StringMessage)60 concatenator_metrics_stream = messager.get_stream(messaging_system, "concatenator_metrics", Metrics)61 service_runner_factory = ServiceRunnerFactory(messager)62 string_concatenator_service_runner = service_runner_factory.get_service_runner(StringConcatenatorService, {"name": "string_concatenator_service", "first_endpoint": letter1_stream.get_read_endpoint(), "second_endpoint": letter2_stream.get_read_endpoint(), "concatenation_endpoint": concatenation_stream.get_write_endpoint(), "control_endpoint": control_stream.get_read_endpoint()}, configuration_stream=configuration_stream, metrics_stream=concatenator_metrics_stream)63 string_concatenator_service_runner.start()64 self.concatenator_stopped = False65 while not self.concatenator_stopped:66 time.sleep(1)67 string_concatenator_service_runner.terminate()68 logging.debug("THREAD _run_concatenator terminates.")69 def _run_writer(self):70 messaging_system = ZMQMessagingSystem("localhost", self.registry_port, 9004)71 messager = Messager()72 concatenation_stream = messager.get_stream(messaging_system, "concatenation", StringMessage)73 control_stream = messager.get_stream(messaging_system, "control", StringMessage)74 configuration_stream = messager.get_stream(messaging_system, "configuration", ConfigurationMessage)75 writer_metrics_stream = messager.get_stream(messaging_system, "writer_metrics", Metrics)76 service_runner_factory = ServiceRunnerFactory(messager)77 string_writer_service_runner = service_runner_factory.get_service_runner(StringWriterService, {"name": "string_writer_service", "string_endpoint": concatenation_stream.get_read_endpoint(), "control_endpoint": control_stream.get_read_endpoint()}, configuration_stream=configuration_stream, metrics_stream=writer_metrics_stream)78 string_writer_service_runner.start()79 self.writer_stopped = False80 while not self.writer_stopped:81 time.sleep(1)82 string_writer_service_runner.terminate()83 logging.debug("THREAD _run_writer terminates.")84 def _run_monitor(self):85 messaging_system = ZMQMessagingSystem("localhost", self.registry_port, 9005)86 messager = Messager()87 letter1_metrics_stream = messager.get_stream(messaging_system, "letter1_metrics", Metrics)88 letter2_metrics_stream = messager.get_stream(messaging_system, "letter2_metrics", Metrics)89 concatenator_metrics_stream = messager.get_stream(messaging_system, "concatenator_metrics", Metrics)90 writer_metrics_stream = messager.get_stream(messaging_system, "writer_metrics", Metrics)91 monitor_service = GenericMonitorService("generic_monitor_service", [letter1_metrics_stream.get_read_endpoint(), letter2_metrics_stream.get_read_endpoint(), concatenator_metrics_stream.get_read_endpoint(), writer_metrics_stream.get_read_endpoint()])92 monitor_service_runner = ServiceRunner(monitor_service, messager)93 monitor_service_runner.start()94 self.monitor_stopped = False95 while not self.monitor_stopped:96 time.sleep(1)97 monitor_service_runner.terminate()98 logging.debug("THREAD _run_monitor terminates.")99 def test(self):100 self.registry_port = 9000101 registry = ZMQMessagingSystemRegistry(self.registry_port)102 registry.start()103 time.sleep(1)104 logging.info("starting services")105 letter1_thread = threading.Thread(target=self._run_letter1)106 letter1_thread.start()107 letter2_thread = threading.Thread(target=self._run_letter2)108 letter2_thread.start()109 concatenator_thread = threading.Thread(target=self._run_concatenator)110 concatenator_thread.start()111 writer_thread = threading.Thread(target=self._run_writer)112 writer_thread.start()113 monitor_thread = threading.Thread(target=self._run_monitor)114 monitor_thread.start()115 time.sleep(5)116 logging.info("Setting up messaging system")117 messaging_system = ZMQMessagingSystem("localhost", self.registry_port, 9006)118 messager = Messager()119 control_stream = messager.get_stream(messaging_system, "control", StringMessage)120 configuration_stream = messager.get_stream(messaging_system, "configuration", ConfigurationMessage)121 logging.info("Registering to publish on 'configuration' topic.")122 configuration_endpoint = configuration_stream.get_write_endpoint()123 time.sleep(5)124 logging.info("Publishing ConfigurationMessage")125 configuration_message = ConfigurationMessage({"letter1_generator_service.upper": True, "string_writer_service.output_path": "output.txt"})126 messager.publish(configuration_endpoint, configuration_message)127 time.sleep(5)128 logging.info("Registering to publish on 'control' topic")129 control_endpoint = control_stream.get_write_endpoint()130 time.sleep(5)131 logging.info("Publishing 'Start' control message")132 messager.publish(control_endpoint, StringMessage("Start"))133 time.sleep(20)134 logging.info("Publishing 'Stop' control message")...
urls.py
Source:urls.py
1from django.conf.urls import url2from . import views3app_name = 'activitystream'4urlpatterns = [5 url('^$', views.sitewide_data, name='sitewide_data'),6 url('^stream/$', views.get_stream, name='get_stream'),7 url('^stream/(?P<models>[\w_:]+)/$', views.get_stream, name='get_stream'),8 url('^stream/(?P<models>[\w_:]+)/(?P<object_id>\d+)/$',9 views.get_stream, name='get_stream'),...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Airtest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
