diff --git a/py-scripts/test_l3_unicast_traffic_gen.py b/py-scripts/test_l3_unicast_traffic_gen.py index 11be83c8..1e2e8dd9 100755 --- a/py-scripts/test_l3_unicast_traffic_gen.py +++ b/py-scripts/test_l3_unicast_traffic_gen.py @@ -10,7 +10,6 @@ if sys.version_info[0] != 3: print("This script requires Python 3") exit(1) - sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../"))) lfcli_base = importlib.import_module("py-json.LANforge.lfcli_base") @@ -36,21 +35,21 @@ class L3VariableTimeLongevity(LFCliBase): self.side_b = side_b self.ssid_list = ssid_list self.ssid_password_list = ssid_password_list - self.station_lists = station_lists + self.station_lists = station_lists self.security = security self.number_template = number_template self.resource = resource - self.debug=_debug_on + self.debug = _debug_on self.name_prefix = name_prefix self.test_duration = test_duration self.cx_stations_lists = station_lists - self.radios = radios # from the command line + self.radios = radios # from the command line self.radio_list = radio_name_list - self.number_of_stations_per_radio_list = number_of_stations_per_radio_list + self.number_of_stations_per_radio_list = number_of_stations_per_radio_list self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port) self.cx_profile = self.local_realm.new_l3_cx_profile() self.station_profiles = [] - + index = 0 for radio in radios: self.station_profile = self.local_realm.new_station_profile() @@ -62,7 +61,7 @@ class L3VariableTimeLongevity(LFCliBase): self.station_profile.mode = 0 self.station_profiles.append(self.station_profile) index += 1 - + self.cx_profile.host = self.host self.cx_profile.port = self.port self.cx_profile.name_prefix = self.name_prefix @@ -78,8 +77,8 @@ class L3VariableTimeLongevity(LFCliBase): if cx_name != 'uri' and cx_name != 'handler': for item, value in cx_name.items(): for value_name, value_rx in value.items(): - if value_name == 'rx bytes': - cx_rx_map[item] = value_rx + if value_name == 'rx bytes': + cx_rx_map[item] = value_rx return cx_rx_map def __compare_vals(self, old_list, new_list): @@ -107,16 +106,14 @@ class L3VariableTimeLongevity(LFCliBase): if self.debug: print("Bringing up station {}".format(station_profile)) station_profile.admin_up(self.resource) - if self.local_realm.wait_for_ip(self.resource, station_list,timeout_sec=10*len(station_list)): + if self.local_realm.wait_for_ip(self.resource, station_list, timeout_sec=10 * len(station_list)): if self.debug: print("ip's aquired {}".format(station_list)) else: print("print failed to get IP's: {}".format(station_list)) - if self.local_realm.wait_for_ip(self.resource, station_list,timeout_sec=120): + if self.local_realm.wait_for_ip(self.resource, station_list, timeout_sec=120): print("tried again: print failed to get IP's: {}".format(station_list)) exit(1) - - self.cx_profile.start_cx() @@ -134,7 +131,7 @@ class L3VariableTimeLongevity(LFCliBase): while cur_time < interval_time: cur_time = datetime.datetime.now() time.sleep(1) - + new_rx_values = self.__get_rx_values() filtered_new_rx_values = [] filtered_new_rx_values = new_rx_values @@ -162,8 +159,8 @@ class L3VariableTimeLongevity(LFCliBase): def cleanup(self, resource): resource = 1 data = { - "name":"BLANK", - "action":"overwrite" + "name": "BLANK", + "action": "overwrite" } url = "cli-json/load" self.json_post(url, data) @@ -171,7 +168,7 @@ class L3VariableTimeLongevity(LFCliBase): timeout = 20 done = False while timeout > 0 and not done: - time.sleep( 1) + time.sleep(1) port_r = self.json_get("/port/1/1/list?fields=alias") if self.debug: print("port interfaces {}".format(port_r["interfaces"])) @@ -181,37 +178,37 @@ class L3VariableTimeLongevity(LFCliBase): print("interface {}".format(interface)) else: done = True - break + break timeout -= 1 if timeout <= 0: print("not all station ports removed {}".format(port_r["interfaces"])) - def build(self): + def build(self): # refactor in LFUtils.port_zero_request() resource = 1 - - data ={ - 'shelf':1, - 'resource':1, - 'port':'eth1', - 'ip_addr':'0.0.0.0', - 'netmask':'0.0.0.0', - 'gateway':'0.0.0.0', - 'current_flags':0, - 'interest':402653212 - } + + data = { + 'shelf': 1, + 'resource': 1, + 'port': 'eth1', + 'ip_addr': '0.0.0.0', + 'netmask': '0.0.0.0', + 'gateway': '0.0.0.0', + 'current_flags': 0, + 'interest': 402653212 + } url = "cli-json/set_port" self.json_post(url, data) # refactor into LFUtils - data ={ - "shelf":1, - "resource": resource, - "port":"br0", - "network_devs":"eth1", - "br_flags":1 + data = { + "shelf": 1, + "resource": resource, + "port": "br0", + "network_devs": "eth1", + "br_flags": 1 } url = "cli-json/add_br" self.json_post(url, data) @@ -221,39 +218,44 @@ class L3VariableTimeLongevity(LFCliBase): self.json_post("/cli-json/set_port", data) except: print("LFUtils.port_dhcp_up_request didn't complete ") - print("or the json_post failed either way {} did not set up dhcp so test may not pass data ".format(self.side_b)) + print("or the json_post failed either way {} did not set up dhcp so test may not pass data ".format( + self.side_b)) resource = 1 - index = 0 + index = 0 for station_profile, station_list in zip(self.station_profiles, self.station_lists): station_profile.use_security(station_profile.security, station_profile.ssid, station_profile.ssid_pass) station_profile.set_number_template(station_profile.number_template) if self.debug: - print("radio: {} station_profile: {} Creating stations: {} ".format(self.radio_list[index],station_profile, station_list)) - + print("radio: {} station_profile: {} Creating stations: {} ".format(self.radio_list[index], + station_profile, station_list)) + temp_station_list = [] for station in range(len(station_list)): temp_station_list.append(str(self.resource) + "." + station_list[station]) - station_profile.create(radio=self.radio_list[index], sta_names_=station_list, debug=False ) + station_profile.create(radio=self.radio_list[index], sta_names_=station_list, debug=False) index += 1 - self.cx_profile.create(endp_type=self.endp_type, side_a=temp_station_list, side_b='1.'+self.side_b, sleep_time=.5) + self.cx_profile.create(endp_type=self.endp_type, side_a=temp_station_list, side_b='1.' + self.side_b, + sleep_time=.5) self._pass("PASS: Stations build finished") + def valid_endp_type(endp_type): - valid_endp_type=['lf_udp','lf_udp6','lf_tcp','lf_tcp6'] + valid_endp_type = ['lf_udp', 'lf_udp6', 'lf_tcp', 'lf_tcp6'] if str(endp_type) in valid_endp_type: return endp_type else: - print('invalid endp_type. Valid types lf_udp, lf_udp6, lf_tcp, lf_tcp6') + print('invalid endp_type. Valid types lf_udp, lf_udp6, lf_tcp, lf_tcp6') exit(1) + def main(): lfjson_host = "localhost" lfjson_port = 8080 parser = argparse.ArgumentParser( prog='test_l3_longevity.py', - #formatter_class=argparse.RawDescriptionHelpFormatter, + # formatter_class=argparse.RawDescriptionHelpFormatter, formatter_class=argparse.RawTextHelpFormatter, epilog='''\ Useful Information: @@ -264,7 +266,7 @@ Useful Information: 4. Security is fixed at WPA2 5. Maximum stations per radio is 64 ''', - + description='''\ test_l3_longevity.py: -------------------- @@ -317,36 +319,41 @@ python3 .\\test_l3_longevity.py --test_duration 4m --endp_type lf_tcp --upstream ''') - - parser.add_argument('-d','--test_duration', help='--test_duration example --time 5d (5 days) default: 3m options: number followed by d, h, m or s',default='3m') - parser.add_argument('-t', '--endp_type', help='--endp_type example --endp_type lf_udp, default: lf_udp , options: lf_udp, lf_udp6, lf_tcp, lf_tcp6', - default='lf_udp',type=valid_endp_type) - parser.add_argument('-u', '--upstream_port', help='--upstream_port example: --upstream_port eth1',default='eth1') - parser.add_argument('--debug', help='Enable debugging', default=False, action="store_true") + parser.add_argument('-d', '--test_duration', + help='--test_duration example --time 5d (5 days) default: 3m options: number followed by d, h, m or s', + default='3m') + parser.add_argument('-t', '--endp_type', + help='--endp_type example --endp_type lf_udp, default: lf_udp , options: lf_udp, lf_udp6, lf_tcp, lf_tcp6', + default='lf_udp', type=valid_endp_type) + parser.add_argument('-u', '--upstream_port', help='--upstream_port example: --upstream_port eth1', + default='eth1') + parser.add_argument('--debug', help='Enable debugging', default=False, action="store_true") requiredNamed = parser.add_argument_group('required arguments') - requiredNamed.add_argument('-r','--radio', action='append', nargs=4, metavar=('', '','',''), - help ='--radio ',required=True) + requiredNamed.add_argument('-r', '--radio', action='append', nargs=4, + metavar=('', '', '', ''), + help='--radio ', + required=True) args = parser.parse_args() if args.test_duration: test_duration = args.test_duration if args.endp_type: - endp_type = args.endp_type + endp_type = args.endp_type side_b = args.upstream_port if args.radio: radios = args.radio - + radio_offset = 0 number_of_stations_offset = 1 ssid_offset = 2 ssid_password_offset = 3 MAX_NUMBER_OF_STATIONS = 64 - + radio_name_list = [] number_of_stations_per_radio_list = [] ssid_list = [] @@ -371,36 +378,37 @@ python3 .\\test_l3_longevity.py --test_duration 4m --endp_type lf_tcp --upstream if number_of_stations > MAX_NUMBER_OF_STATIONS: print("number of stations per radio exceeded max of : {}".format(MAX_NUMBER_OF_STATIONS)) quit(1) - station_list = LFUtils.portNameSeries(prefix_="sta", start_id_= 1 + index*1000, end_id_= number_of_stations + index*1000, padding_number_=10000) + station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=1 + index * 1000, + end_id_=number_of_stations + index * 1000, padding_number_=10000) station_lists.append(station_list) index += 1 - ip_var_test = L3VariableTimeLongevity(lfjson_host, - lfjson_port, - number_template="00", - station_lists= station_lists, - name_prefix="var_time", - endp_type=endp_type, - side_b=side_b, - radios=radios, - radio_name_list=radio_name_list, - number_of_stations_per_radio_list=number_of_stations_per_radio_list, - ssid_list=ssid_list, - ssid_password_list=ssid_password_list, - resource=1, - security="wpa2", test_duration=test_duration, - side_a_min_rate=256000, side_b_min_rate=256000, - _debug_on=args.debug) + ip_var_test = L3VariableTimeLongevity(lfjson_host, + lfjson_port, + number_template="00", + station_lists=station_lists, + name_prefix="var_time", + endp_type=endp_type, + side_b=side_b, + radios=radios, + radio_name_list=radio_name_list, + number_of_stations_per_radio_list=number_of_stations_per_radio_list, + ssid_list=ssid_list, + ssid_password_list=ssid_password_list, + resource=1, + security="wpa2", test_duration=test_duration, + side_a_min_rate=256000, side_b_min_rate=256000, + _debug_on=args.debug) ip_var_test.cleanup(station_list) ip_var_test.build() if not ip_var_test.passes(): print(ip_var_test.get_fail_message()) - exit(1) + exit(1) ip_var_test.start(False, False) ip_var_test.stop() if not ip_var_test.passes(): print(ip_var_test.get_fail_message()) - exit(1) + exit(1) time.sleep(30) ip_var_test.cleanup(station_list) if ip_var_test.passes():