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.