diff --git a/py-scripts/test_ipv4_connection.py b/py-scripts/test_ipv4_connection.py index 018c0980..72120fa3 100755 --- a/py-scripts/test_ipv4_connection.py +++ b/py-scripts/test_ipv4_connection.py @@ -8,16 +8,15 @@ if sys.version_info[0] != 3: if 'py-json' not in sys.path: sys.path.append('../py-json') - -import argparse +import LANforge from LANforge.lfcli_base import LFCliBase -from LANforge.LFUtils import * +from LANforge import LFUtils import realm import time class IPv4Test(LFCliBase): - def __init__(self, host, port, ssid, security, password, num_stations, prefix="00000", _debug_on=False, + def __init__(self, host, port, ssid, security, password, sta_list=None, num_stations=0, prefix="00000", _debug_on=False, _exit_on_error=False, _exit_on_fail=False): super().__init__(host, port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail) @@ -27,6 +26,7 @@ class IPv4Test(LFCliBase): self.security = security self.password = password self.num_stations = num_stations + self.sta_list = sta_list self.timeout = 120 self.prefix = prefix self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port) @@ -34,59 +34,51 @@ class IPv4Test(LFCliBase): security=self.security, prefix=self.prefix, mode=0, up=True, dhcp=True, debug_=False) - def run_test_full(self, print_pass=False, print_fail=False): - print("Testing all stations for association") - port_list = self.local_realm.station_list() - sta_list = [] - for item in list(port_list): - # print(list(item)) - if "sta" in list(item)[0]: - sta_list.append(self.local_realm.name_to_eid(list(item)[0])[2]) + def build(self): + # Build stations + self.profile.use_wpa2(True, self.ssid, self.password) + self.profile.set_prefix(self.prefix) + print("Creating stations") + self.profile.set_command_flag("add_sta", "create_admin_down", 1) + self.profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=False) - return self._run_test(sta_list, print_pass, print_fail) + def start(self): + # Bring stations up + pass - def run_test_custom(self, range_start="000", range_end="000", sta_list=[], print_pass=False, print_fail=False): - if len(sta_list) == 0: - print("Testing range of stations from sta%s to sta%s" % (range_start, range_end)) - if range_start != "000" and range_end != "000": - found_stations = self.local_realm.find_ports_like("sta[%s..%s]" % (range_start, range_end)) - for sta_name in list(found_stations): - sta_list.append(self.local_realm.name_to_eid(sta_name)[2]) - else: - raise ValueError("range_start and range_end not specified") - else: - print("Testing stations in specified list") - return self._run_test(sta_list, print_pass, print_fail) + def stop(self): + # Bring stations down + pass - def _run_test(self, sta_list, print_pass, print_fail): + def run_test(self, sta_list, print_pass=False, print_fail=False): + associated_map = {} + ip_map = {} for sec in range(self.timeout): - associated_map = [] - ip_map = [] for sta_name in sta_list: - sta_status = super().json_get("port/1/1/" + sta_name) + sta_status = super().json_get("port/1/1/" + sta_name + "?fields=port,alias,ip,ap") # print(sta_status) - if len(sta_status['interface']['ap']) == 17 and sta_status['interface']['ap'][-3] == ':' \ - and sta_status['interface']['ip'] == '0.0.0.0': - + if sta_status is None or sta_status['interface'] is None or sta_status['interface']['ap'] is None: + continue + if len(sta_status['interface']['ap']) == 17 and sta_status['interface']['ap'][-3] == ':': # print("Associated", sta_name, sta_status['interface']['ap'], sta_status['interface']['ip']) - associated_map.append(sta_name) - - elif len(sta_status['interface']['ap']) == 17 and sta_status['interface']['ap'][-3] == ':' \ + associated_map[sta_name] = 1 + if len(sta_status['interface']['ap']) == 17 and sta_status['interface']['ap'][-3] == ':' \ and sta_status['interface']['ip'] != '0.0.0.0': - # print("IP", sta_name, sta_status['interface']['ap'], sta_status['interface']['ip']) - associated_map.append(sta_name) - ip_map.append(sta_name) - + associated_map[sta_name] = 1 + ip_map[sta_name] = 1 time.sleep(1) # print(len(sta_list), sta_list) # print(len(ip_map), ip_map) # print(len(associated_map), associated_map) - if len(sta_list) == len(ip_map) and len(sta_list) == len(associated_map): + if (len(sta_list) == len(ip_map)) and (len(sta_list) == len(associated_map)): self._pass("PASS: All stations associated with IP", print_pass) else: self._fail("FAIL: Not all stations able to associate/get IP", print_fail) + print(sta_list) + print(ip_map) + print(associated_map) return self.passes() @@ -109,34 +101,26 @@ class IPv4Test(LFCliBase): super().json_post(req_url, data) def run(self): - super().clear_test_results() - print("Cleaning up old stations") - self.cleanup() - sta_list = [] - - self.profile.use_wpa2(True, self.ssid, self.password) - self.profile.set_prefix(self.prefix) - print("Creating stations") - self.profile.create(resource=1, radio="wiphy0", num_stations=self.num_stations, debug=False) - - for sta_name in list(self.local_realm.find_ports_like("sta[%s..%s]" % ( - self.prefix, str(self.prefix[:-len(str(self.num_stations))]) + str(self.num_stations - 1)))): - sta_list.append(self.local_realm.name_to_eid(sta_name)[2]) + if len(self.sta_list) == 0: + sta_list = [] + for sta_name in list(self.local_realm.find_ports_like("sta[%s..%s]" % ( + self.prefix, str(self.prefix[:-len(str(self.num_stations))]) + str(self.num_stations - 1)))): + sta_list.append(self.local_realm.name_to_eid(sta_name)[2]) def main(): lfjson_host = "localhost" lfjson_port = 8080 + station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=5, padding_number_=10000) ip_test = IPv4Test(lfjson_host, lfjson_port, ssid="jedway-wpa2-x2048-4-4", password="jedway-wpa2-x2048-4-4", - security="open", num_stations=10) + security="open", sta_list=station_list) ip_test.cleanup() ip_test.timeout = 60 - ip_test.run() - print("Full Test Passed: %s" % ip_test.run_test_full()) - print("Range Test Passed: %s" % ip_test.run_test_custom("00005", "00009")) - print("Custom Test Passed: %s" % ip_test.run_test_custom(sta_list=["sta00001", "sta00003", "sta00009", "sta00002"])) + ip_test.build() + #ip_test.start() + print("Full Test Passed: %s" % ip_test.run_test(ip_test.sta_list)) - ip_test.cleanup() + #ip_test.cleanup() if __name__ == "__main__":