Merge branch 'master' of github.com:greearb/lanforge-scripts

py-json/LANforge/lfcli_base.py

@@ -209,7 +209,11 @@ class LFCliBase:
             print(self.pass_pref + message)
 
     @staticmethod
-    def create_basic_argparse():
+    def create_basic_argparse(prog=None, formatter_class=None, epilog=None, description=None):
+        if prog is not None or formatter_class is not None or epilog is not None or description is not None:
+            parser = argparse.ArgumentParser(prog=prog, formatter_class=formatter_class, epilog=epilog,
+                                             description=description)
+        else:
             parser = argparse.ArgumentParser()
 
         parser.add_argument('--mgr', help='--mgr <hostname for where LANforge GUI is running>', default='localhost')

py-json/realm.py

@@ -884,6 +884,7 @@ class L3CXProfile(LFCliBase):
                 self.local_realm.rm_cx(cx_name)
 
                 for side in range(len(self.created_cx[cx_name])):
+                    ename = self.created_cx[cx_name][side]
                     print("Cleaning endpoint: %s"%(ename))
                     self.local_realm.rm_endp(self.created_cx[cx_name][side])
 

py-scripts/sta_connect2.py

@@ -22,6 +22,7 @@ from LANforge.lfcli_base import LFCliBase
 from LANforge.LFUtils import *
 import realm
 from realm import Realm
+import pprint
 
 OPEN="open"
 WEP="wep"
@@ -141,7 +142,7 @@ class StaConnect2(LFCliBase):
             if (response is not None) and (response["interface"] is not None):
                 for sta_name in self.station_names:
                     LFUtils.removePort(self.resource, sta_name, self.lfclient_url)
-        LFUtils.wait_until_ports_disappear(self.resource, self.lfclient_url, self.station_names)
+        LFUtils.wait_until_ports_disappear(self.lfclient_url, self.station_names)
 
         # Create stations and turn dhcp on
         self.station_profile = self.localrealm.new_station_profile()
@@ -153,8 +154,8 @@ class StaConnect2(LFCliBase):
         self.station_profile.set_command_flag("add_sta", "create_admin_down", 1)
 
         print("Adding new stations ", end="")
-        self.station_profile.create(resource=self.resource, radio=self.radio, sta_names_=self.station_names, up_=False, debug=self.debug, suppress_related_commands_=True)
-        LFUtils.wait_until_ports_appear(self.resource, self.lfclient_url, self.station_names, debug=self.debug)
+        self.station_profile.create(radio=self.radio, sta_names_=self.station_names, up_=False, debug=self.debug, suppress_related_commands_=True)
+        LFUtils.wait_until_ports_appear(self.lfclient_url, self.station_names, debug=self.debug)
 
         # Create UDP endpoints
         self.l3_udp_profile = self.localrealm.new_l3_cx_profile()
@@ -193,7 +194,7 @@ class StaConnect2(LFCliBase):
                      "port": "ALL",
                      "probe_flags": 1}
             self.json_post("/cli-json/nc_show_ports", data)
-            self.station_profile.admin_up(self.resource)
+            self.station_profile.admin_up()
             LFUtils.waitUntilPortsAdminUp(self.resource, self.lfclient_url, self.station_names)
 
         # station_info = self.jsonGet(self.mgr_url, "%s?fields=port,ip,ap" % (self.getStaUrl()))
@@ -329,14 +330,16 @@ class StaConnect2(LFCliBase):
         if self.cleanup_on_exit:
             for sta_name in self.station_names:
                 LFUtils.removePort(self.resource, sta_name, self.lfclient_url)
-            endp_names = []
-
+            curr_endp_names = []
             removeCX(self.lfclient_url, self.l3_udp_profile.get_cx_names())
             removeCX(self.lfclient_url, self.l3_tcp_profile.get_cx_names())
             for (cx_name, endp_names) in self.l3_udp_profile.created_cx.items():
-                endp_names.append(endp_names[0])
-                endp_names.append(endp_names[1])
-            removeEndps(self.lfclient_url, endp_names)
+                curr_endp_names.append(endp_names[0])
+                curr_endp_names.append(endp_names[1])
+            for (cx_name, endp_names) in self.l3_tcp_profile.created_cx.items():
+                curr_endp_names.append(endp_names[0])
+                curr_endp_names.append(endp_names[1])        
+            removeEndps(self.lfclient_url, curr_endp_names, debug= self.debug)
 
 # ~class
 
@@ -408,6 +411,7 @@ Example:
     if args.dut_security is not None:
         staConnect.dut_security = args.dut_security
 
+   # staConnect.cleanup()
     staConnect.setup()
     staConnect.start()
     print("napping %f sec" % staConnect.runtime_secs)

py-scripts/test_generic.py

@@ -19,9 +19,9 @@ import datetime
 
 
 class GenTest(LFCliBase):
-    def __init__(self, host, port, ssid, security, password, sta_list, name_prefix, resource=1,
+    def __init__(self, host, port, ssid, security, password, sta_list, name_prefix, upstream,
                  number_template="00000", test_duration="5m", type="lfping", dest="127.0.0.1",
-                 interval=1,
+                 interval=1, radio="wiphy0",
                  _debug_on=False,
                  _exit_on_error=False,
                  _exit_on_fail=False):
@@ -29,11 +29,12 @@ class GenTest(LFCliBase):
         self.host = host
         self.port = port
         self.ssid = ssid
+        self.radio = radio
+        self.upstream = upstream
         self.sta_list = sta_list
         self.security = security
         self.password = password
         self.number_template = number_template
-        self.resource = resource
         self.name_prefix = name_prefix
         self.test_duration = test_duration
         self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
@@ -52,10 +53,10 @@ class GenTest(LFCliBase):
         self.cx_profile.interval = interval
 
     def start(self, print_pass=False, print_fail=False):
-        self.station_profile.admin_up(self.resource)
+        self.station_profile.admin_up()
         temp_stas = self.sta_list.copy()
-        temp_stas.append("eth1")
-        if self.local_realm.wait_for_ip(self.resource, temp_stas):
+        temp_stas.append(self.upstream)
+        if self.local_realm.wait_for_ip(temp_stas):
             self._pass("All stations got IPs", print_pass)
         else:
             self._fail("Stations failed to get IPs", print_fail)
@@ -102,30 +103,59 @@ class GenTest(LFCliBase):
         self.station_profile.set_command_flag("add_sta", "create_admin_down", 1)
         self.station_profile.set_command_param("set_port", "report_timer", 1500)
         self.station_profile.set_command_flag("set_port", "rpt_timer", 1)
-        temp_sta_list = []
-        for station in range(len(self.sta_list)):
-            temp_sta_list.append(str(self.resource) + "." + self.sta_list[station])
-        self.station_profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=self.debug)
-        self.cx_profile.create(ports=temp_sta_list, sleep_time=.5)
+        self.station_profile.create(radio=self.radio, sta_names_=self.sta_list, debug=self.debug)
+        self.cx_profile.create(ports=self.station_profile.station_names, sleep_time=.5)
         self._pass("PASS: Station build finished")
 
     def cleanup(self, sta_list):
         self.cx_profile.cleanup()
-        self.station_profile.cleanup(self.resource, sta_list)
-        LFUtils.wait_until_ports_disappear(resource_id=self.resource, base_url=self.lfclient_url, port_list=sta_list,
+        self.station_profile.cleanup(sta_list)
+        LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=sta_list,
                                            debug=self.debug)
 
 
 def main():
-    lfjson_host = "localhost"
     lfjson_port = 8080
-    station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=4, padding_number_=10000)
-    generic_test = GenTest(lfjson_host, lfjson_port, number_template="00", sta_list=station_list,
-                           name_prefix="var_time", type="lfping", dest="10.40.0.1", interval=1,
-                           ssid="jedway-wpa2-x2048-4-4",
-                           password="jedway-wpa2-x2048-4-4",
-                           resource=1,
-                           security="wpa2", test_duration="5m", )
+
+    parser = LFCliBase.create_basic_argparse(
+        prog='test_generic.py',
+        # formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=argparse.RawTextHelpFormatter,
+        epilog='''\
+            Useful Information:
+                1. TBD
+                ''',
+
+        description='''\
+    test_generic.py:
+    --------------------
+    TBD
+
+    Generic command layout:
+    python ./test_generic.py --upstream_port <port> --radio <radio 0> <stations> <ssid> <ssid password> <security type: wpa2, open, wpa3> --debug
+
+    Note:   multiple --radio switches may be entered up to the number of radios available:
+                     --radio <radio 0> <stations> <ssid> <ssid password>  --radio <radio 01> <number of last station> <ssid> <ssid password>
+
+     python3 ./test_generic.py --upstream_port eth1 --radio wiphy0 32 candelaTech-wpa2-x2048-4-1 candelaTech-wpa2-x2048-4-1 wpa2 --radio wiphy1 64 candelaTech-wpa2-x2048-5-3 candelaTech-wpa2-x2048-5-3 wpa2
+
+            ''')
+
+    parser.add_argument('--type', help='--type type of command to run', default="lfping")
+    parser.add_argument('--dest', help='--dest destination for command', default="10.40.0.1")
+    parser.add_argument('--test_duration', help='--test_duration sets the duration of the test', default="5m")
+    parser.add_argument('--interval', help='--interval interval to use when running lfping', default=1)
+
+    args = parser.parse_args()
+
+    station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=1, padding_number_=10000,
+                                          radio=args.radio)
+
+    generic_test = GenTest(args.mgr, lfjson_port, number_template="00", sta_list=station_list,
+                           name_prefix="GT", type=args.type, dest=args.dest, interval=1,
+                           ssid=args.ssid, upstream=args.upstream_port,
+                           password=args.passwd, security=args.security, test_duration=args.test_duration,
+                           _debug_on=args.debug, radio=args.radio)
 
     generic_test.cleanup(station_list)
     generic_test.build()

py-scripts/test_ipv4_l4.py

@@ -13,14 +13,15 @@ import argparse
 from LANforge.lfcli_base import LFCliBase
 from LANforge.LFUtils import *
 from LANforge import LFUtils
+import argparse
 import realm
 import time
 import datetime
 
 
 class IPV4L4(LFCliBase):
-    def __init__(self, host, port, ssid, security, password, url, requests_per_ten, station_list, number_template="00000",
-                 resource=1,
+    def __init__(self, host, port, ssid, security, password, url, requests_per_ten, station_list,
+                 number_template="00000", radio="wiphy0",
                  test_duration="5m",
                  _debug_on=False,
                  _exit_on_error=False,
@@ -28,6 +29,7 @@ class IPV4L4(LFCliBase):
         super().__init__(host, port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
         self.host = host
         self.port = port
+        self.radio = radio
         self.ssid = ssid
         self.security = security
         self.password = password
@@ -35,7 +37,6 @@ class IPV4L4(LFCliBase):
         self.requests_per_ten = requests_per_ten
         self.number_template = number_template
         self.sta_list = station_list
-        self.resource = resource
         self.test_duration = test_duration
 
         self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
@@ -90,18 +91,16 @@ class IPV4L4(LFCliBase):
         self.station_profile.set_command_flag("add_sta", "create_admin_down", 1)
         self.station_profile.set_command_param("set_port", "report_timer", 1500)
         self.station_profile.set_command_flag("set_port", "rpt_timer", 1)
-        self.station_profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=self.debug)
+        self.station_profile.create(radio=self.radio, sta_names_=self.sta_list, debug=self.debug)
         self._pass("PASS: Station build finished")
-        temp_sta_list = []
-        for station in range(len(self.sta_list)):
-            temp_sta_list.append(str(self.resource) + "." + self.sta_list[station])
 
-        self.cx_profile.create(ports=temp_sta_list, sleep_time=.5, debug_=self.debug, suppress_related_commands_=None)
+        self.cx_profile.create(ports=self.station_profile.station_names, sleep_time=.5, debug_=self.debug,
+                               suppress_related_commands_=None)
 
     def start(self, print_pass=False, print_fail=False):
         temp_stas = self.sta_list.copy()
         temp_stas.append("eth1")
-        if self.local_realm.wait_for_ip(self.resource, temp_stas):
+        if self.local_realm.wait_for_ip(temp_stas):
             self._pass("All stations got IPs", print_pass)
         else:
             self._fail("Stations failed to get IPs", print_fail)
@@ -109,7 +108,7 @@ class IPV4L4(LFCliBase):
         cur_time = datetime.datetime.now()
         old_rx_values = self.__get_values()
         end_time = self.local_realm.parse_time(self.test_duration) + cur_time
-        self.station_profile.admin_up(1)
+        self.station_profile.admin_up()
         self.cx_profile.start_cx()
         passes = 0
         expected_passes = 0
@@ -139,24 +138,58 @@ class IPV4L4(LFCliBase):
     def stop(self):
         self.cx_profile.stop_cx()
         for sta_name in self.sta_list:
-            data = LFUtils.portDownRequest(1, sta_name)
+            data = LFUtils.port_down_request(1, self.local_realm.name_to_eid(sta_name)[2])
             url = "cli-json/set_port"
             self.json_post(url, data)
 
     def cleanup(self, sta_list):
         self.cx_profile.cleanup()
-        self.station_profile.cleanup(self.resource, sta_list)
-        LFUtils.wait_until_ports_disappear(resource_id=self.resource, base_url=self.lfclient_url, port_list=sta_list,
+        self.station_profile.cleanup(sta_list)
+        LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=sta_list,
                                            debug=self.debug)
 
 
 def main():
     lfjson_host = "localhost"
     lfjson_port = 8080
-    station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=1, padding_number_=10000)
-    ip_test = IPV4L4(lfjson_host, lfjson_port, ssid="jedway-wpa2-x2048-4-4", password="jedway-wpa2-x2048-4-4",
-                     security="wpa2", station_list=station_list, url="dl http://10.40.0.1 /dev/null", test_duration="5m",
-                     requests_per_ten=600, _debug_on=False)
+
+    parser = LFCliBase.create_basic_argparse(
+        prog='test_generic.py',
+        # formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=argparse.RawTextHelpFormatter,
+        epilog='''\
+            Useful Information:
+                1. TBD
+                ''',
+
+        description='''\
+    test_generic.py:
+    --------------------
+    TBD
+
+    Generic command layout:
+    python ./test_ipv4_l4.py --upstream_port <port> --radio <radio 0> <stations> <ssid> <ssid password> <security type: wpa2, open, wpa3> --debug
+
+    Note:   multiple --radio switches may be entered up to the number of radios available:
+                     --radio <radio 0> <stations> <ssid> <ssid password>  --radio <radio 01> <number of last station> <ssid> <ssid password>
+
+     python3 ./test_ipv4_l4.py --upstream_port eth1 --radio wiphy0 32 candelaTech-wpa2-x2048-4-1 candelaTech-wpa2-x2048-4-1 wpa2 --radio wiphy1 64 candelaTech-wpa2-x2048-5-3 candelaTech-wpa2-x2048-5-3 wpa2
+
+            ''')
+
+    parser.add_argument('--test_duration', help='--test_duration sets the duration of the test', default="5m")
+    parser.add_argument('--requests_per_ten', help='--requests_per_ten number of request per ten minutes', default=600)
+    parser.add_argument('--url', help='--url specifies upload/download, address, and dest', default="dl http://10.40.0.1 /dev/null")
+
+    args = parser.parse_args()
+
+    station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=1, padding_number_=10000,
+                                          radio=args.radio)
+
+    ip_test = IPV4L4(args.mgr, lfjson_port, ssid=args.ssid, password=args.passwd,
+                     security=args.security, station_list=station_list, url=args.url,
+                     test_duration=args.test_duration,
+                     requests_per_ten=args.requests_per_ten, _debug_on=args.debug)
     ip_test.cleanup(station_list)
     ip_test.build()
     if not ip_test.passes():

py-scripts/test_ipv4_l4_ftp.py

@@ -0,0 +1,210 @@
+#!/usr/bin/env python3
+
+import sys
+
+if sys.version_info[0] != 3:
+    print("This script requires Python 3")
+    exit(1)
+
+if 'py-json' not in sys.path:
+    sys.path.append('../py-json')
+
+import argparse
+from LANforge.lfcli_base import LFCliBase
+from LANforge.LFUtils import *
+from LANforge import LFUtils
+import realm
+import time
+import datetime
+
+
+class IPV4L4(LFCliBase):
+    def __init__(self, host, port, ssid, security, password, url, requests_per_ten, station_list, number_template="00000",
+                 upstream_port="eth1",
+                 test_duration="5m",
+                 _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)
+        self.host = host
+        self.port = port
+        self.ssid = ssid
+        self.upstream_port = upstream_port
+        self.security = security
+        self.password = password
+        self.url = url
+        self.requests_per_ten = requests_per_ten
+        self.number_template = number_template
+        self.sta_list = station_list
+        self.test_duration = test_duration
+
+        self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
+        self.station_profile = self.local_realm.new_station_profile()
+        self.cx_profile = self.local_realm.new_l4_cx_profile()
+
+        self.station_profile.lfclient_url = self.lfclient_url
+        self.station_profile.ssid = self.ssid
+        self.station_profile.ssid_pass = self.password,
+        self.station_profile.security = self.security
+        self.station_profile.number_template_ = self.number_template
+        self.station_profile.mode = 0
+
+        self.cx_profile.url = self.url
+        self.cx_profile.requests_per_ten = self.requests_per_ten
+
+        self.port_util = realm.PortUtils(self.local_realm)
+
+    def __compare_vals(self, old_list, new_list):
+        passes = 0
+        expected_passes = 0
+        if len(old_list) == len(new_list):
+            for item, value in old_list.items():
+                expected_passes += 1
+                if new_list[item] > old_list[item]:
+                    passes += 1
+                # print(item, old_list[item], new_list[item], passes, expected_passes)
+
+            if passes == expected_passes:
+                return True
+            else:
+                return False
+        else:
+            return False
+
+    def __get_values(self):
+        time.sleep(1)
+        cx_list = self.json_get("layer4/list?fields=name,bytes-rd", debug_=self.debug)
+        # print("==============\n", cx_list, "\n==============")
+        cx_map = {}
+        for cx_name in cx_list['endpoint']:
+            if cx_name != 'uri' and cx_name != 'handler':
+                for item, value in cx_name.items():
+                    for value_name, value_rx in value.items():
+                        if item in self.cx_profile.created_cx.keys() and value_name == 'bytes-rd':
+                            cx_map[item] = value_rx
+        return cx_map
+
+    def build(self):
+        # Build stations
+        self.station_profile.use_security(self.security, self.ssid, self.password)
+        self.station_profile.set_number_template(self.number_template)
+        print("Creating stations")
+        self.station_profile.set_command_flag("add_sta", "create_admin_down", 1)
+        self.station_profile.set_command_param("set_port", "report_timer", 1500)
+        self.station_profile.set_command_flag("set_port", "rpt_timer", 1)
+        self.station_profile.create(radio="wiphy0", sta_names_=self.sta_list, debug=self.debug)
+        self._pass("PASS: Station build finished")
+
+        self.cx_profile.create(ports=self.station_profile.station_names, sleep_time=.5, debug_=self.debug, suppress_related_commands_=True)
+
+    def start(self, print_pass=False, print_fail=False):
+        self.port_util.set_ftp(port_name=self.upstream_port, resource=1, on=True)
+        temp_stas = self.sta_list.copy()
+        temp_stas.append(self.upstream_port)
+        if self.local_realm.wait_for_ip(temp_stas):
+            self._pass("All stations got IPs", print_pass)
+        else:
+            self._fail("Stations failed to get IPs", print_fail)
+            exit(1)
+        cur_time = datetime.datetime.now()
+        old_rx_values = self.__get_values()
+        end_time = self.local_realm.parse_time(self.test_duration) + cur_time
+        self.station_profile.admin_up()
+        self.cx_profile.start_cx()
+        passes = 0
+        expected_passes = 0
+        print("Starting Test...")
+        while cur_time < end_time:
+            interval_time = cur_time + datetime.timedelta(minutes=1)
+            while cur_time < interval_time:
+                cur_time = datetime.datetime.now()
+                time.sleep(1)
+
+            new_rx_values = self.__get_values()
+            # print(old_rx_values, new_rx_values)
+            # print("\n-----------------------------------")
+            # print(cur_time, end_time, cur_time + datetime.timedelta(minutes=1))
+            # print("-----------------------------------\n")
+            expected_passes += 1
+            if self.__compare_vals(old_rx_values, new_rx_values):
+                passes += 1
+            else:
+                self._fail("FAIL: Not all stations increased traffic", print_fail)
+                break
+            old_rx_values = new_rx_values
+            cur_time = datetime.datetime.now()
+        if passes == expected_passes:
+            self._pass("PASS: All tests passes", print_pass)
+
+    def stop(self):
+        self.port_util.set_ftp(port_name=self.upstream_port, resource=1, on=False)
+        self.cx_profile.stop_cx()
+        for sta_name in self.sta_list:
+            data = LFUtils.portDownRequest(1, self.local_realm.name_to_eid(sta_name)[2])
+            url = "cli-json/set_port"
+            self.json_post(url, data)
+
+    def cleanup(self, sta_list):
+        self.cx_profile.cleanup()
+        self.station_profile.cleanup(sta_list)
+        LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=sta_list,
+                                           debug=self.debug)
+
+
+def main():
+    lfjson_port = 8080
+
+    parser = LFCliBase.create_basic_argparse(
+        prog='test_ipv4_l4_ftp',
+        # formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=argparse.RawTextHelpFormatter,
+        epilog='''\
+            Useful Information:
+                1. TBD
+                ''',
+
+        description='''\
+    test_ipv4_l4_ftp.py:
+    --------------------
+    TBD
+
+    Generic command layout:
+    python ./test_ipv4_l4_ftp.py --upstream_port <port> --radio <radio 0> <stations> <ssid> <ssid password> <security type: wpa2, open, wpa3> --debug
+
+    Note:   multiple --radio switches may be entered up to the number of radios available:
+                     --radio <radio 0> <stations> <ssid> <ssid password>  --radio <radio 01> <number of last station> <ssid> <ssid password>
+
+     python3 ./test_ipv4_l4_ftp.py --upstream_port eth1 --radio wiphy0 32 candelaTech-wpa2-x2048-4-1 candelaTech-wpa2-x2048-4-1 wpa2 --radio wiphy1 64 candelaTech-wpa2-x2048-5-3 candelaTech-wpa2-x2048-5-3 wpa2
+
+            ''')
+
+    parser.add_argument('--test_duration', help='--test_duration sets the duration of the test', default="5m")
+    parser.add_argument('--requests_per_ten', help='--requests_per_ten number of request per ten minutes', default=600)
+    parser.add_argument('--url', help='--url specifies upload/download, address, and dest',
+                        default="dl ftp://10.40.0.1 /dev/null")
+
+    args = parser.parse_args()
+    station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=1, padding_number_=10000,
+                                          radio=args.radio)
+
+    ip_test = IPV4L4(args.mgr, lfjson_port, ssid=args.ssid, password=args.passwd,
+                     security=args.security, station_list=station_list, url=args.url, test_duration=args.test_duration,
+                     requests_per_ten=args.requests_per_ten, _debug_on=args.debug)
+    ip_test.cleanup(station_list)
+    ip_test.build()
+    if not ip_test.passes():
+        print(ip_test.get_fail_message())
+        exit(1)
+    ip_test.start(False, False)
+    ip_test.stop()
+    if not ip_test.passes():
+        print(ip_test.get_fail_message())
+        exit(1)
+    time.sleep(30)
+    ip_test.cleanup(station_list)
+    if ip_test.passes():
+        print("Full test passed, all endpoints had increased bytes-rd throughout test duration")
+
+
+if __name__ == "__main__":
+    main()

py-scripts/test_ipv4_l4_urls_per_ten.py

@@ -20,13 +20,14 @@ import datetime
 
 class IPV4L4(LFCliBase):
     def __init__(self, host, port, ssid, security, password, url, requests_per_ten, station_list,
-                 target_requests_per_ten=60, number_template="00000", resource=1, num_tests=1,
+                 target_requests_per_ten=60, number_template="00000", num_tests=1, radio="wiphy0",
                  _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)
         self.host = host
         self.port = port
+        self.radio = radio
         self.ssid = ssid
         self.security = security
         self.password = password
@@ -34,7 +35,6 @@ class IPV4L4(LFCliBase):
         self.requests_per_ten = requests_per_ten
         self.number_template = number_template
         self.sta_list = station_list
-        self.resource = resource
         self.num_tests = num_tests
         self.target_requests_per_ten = target_requests_per_ten
 
@@ -74,13 +74,10 @@ class IPV4L4(LFCliBase):
         self.station_profile.set_command_flag("add_sta", "create_admin_down", 1)
         self.station_profile.set_command_param("set_port", "report_timer", 1500)
         self.station_profile.set_command_flag("set_port", "rpt_timer", 1)
-        self.station_profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=self.debug)
+        self.station_profile.create(radio=self.radio, sta_names_=self.sta_list, debug=self.debug)
         self._pass("PASS: Station build finished")
-        temp_sta_list = []
-        for station in range(len(self.sta_list)):
-            temp_sta_list.append(str(self.resource) + "." + self.sta_list[station])
 
-        self.cx_profile.create(ports=temp_sta_list, sleep_time=.5, debug_=self.debug, suppress_related_commands_=None)
+        self.cx_profile.create(ports=self.station_profile.station_names, sleep_time=.5, debug_=self.debug, suppress_related_commands_=None)
 
     def start(self, print_pass=False, print_fail=False):
         temp_stas = self.sta_list.copy()
@@ -89,8 +86,8 @@ class IPV4L4(LFCliBase):
         interval_time = cur_time + datetime.timedelta(minutes=10)
         passes = 0
         expected_passes = 0
-        self.station_profile.admin_up(1)
-        if self.local_realm.wait_for_ip(self.resource, temp_stas):
+        self.station_profile.admin_up()
+        if self.local_realm.wait_for_ip(temp_stas):
             self._pass("All stations got IPs", print_pass)
         else:
             self._fail("Stations failed to get IPs", print_fail)
@@ -119,25 +116,59 @@ class IPV4L4(LFCliBase):
     def stop(self):
         self.cx_profile.stop_cx()
         for sta_name in self.sta_list:
-            data = LFUtils.portDownRequest(1, sta_name)
+            data = LFUtils.portDownRequest(1, self.local_realm.name_to_eid(sta_name)[2])
             url = "cli-json/set_port"
             self.json_post(url, data)
 
     def cleanup(self, sta_list):
         self.cx_profile.cleanup()
-        self.station_profile.cleanup(self.resource, sta_list)
-        LFUtils.wait_until_ports_disappear(resource_id=self.resource, base_url=self.lfclient_url, port_list=sta_list,
+        self.station_profile.cleanup(sta_list)
+        LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=sta_list,
                                            debug=self.debug)
 
 
 def main():
-    lfjson_host = "localhost"
     lfjson_port = 8080
-    station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=1, padding_number_=10000)
-    ip_test = IPV4L4(lfjson_host, lfjson_port, ssid="jedway-wpa2-x2048-4-4", password="jedway-wpa2-x2048-4-4",
-                     security="wpa2", station_list=station_list, url="dl http://10.40.0.1 /dev/null", num_tests=1,
-                     target_requests_per_ten=600,
-                     requests_per_ten=600)
+
+    parser = LFCliBase.create_basic_argparse(
+        prog='test_ipv4_l4_urls_per_ten',
+        # formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=argparse.RawTextHelpFormatter,
+        epilog='''\
+            Useful Information:
+                1. TBD
+                ''',
+
+        description='''\
+    test_ipv4_l4_urls_per_ten.py:
+    --------------------
+    TBD
+
+    Generic command layout:
+    python ./test_ipv4_l4_urls_per_ten.py --upstream_port <port> --radio <radio 0> <stations> <ssid> <ssid password> <security type: wpa2, open, wpa3> --debug
+
+    Note:   multiple --radio switches may be entered up to the number of radios available:
+                     --radio <radio 0> <stations> <ssid> <ssid password>  --radio <radio 01> <number of last station> <ssid> <ssid password>
+
+     python3 ./test_ipv4_l4_urls_per_ten.py --upstream_port eth1 --radio wiphy0 32 candelaTech-wpa2-x2048-4-1 candelaTech-wpa2-x2048-4-1 wpa2 --radio wiphy1 64 candelaTech-wpa2-x2048-5-3 candelaTech-wpa2-x2048-5-3 wpa2
+
+            ''')
+
+    parser.add_argument('--requests_per_ten', help='--requests_per_ten number of request per ten minutes', default=600)
+    parser.add_argument('--num_tests', help='--num_tests number of tests to run. Each test runs 10 minutes', default=1)
+    parser.add_argument('--url', help='--url specifies upload/download, address, and dest',
+                        default="dl http://10.40.0.1 /dev/null")
+    parser.add_argument('--target_per_ten', help='--target_per_ten target number of request per ten minutes. test will check for 90% this value',
+                        default=600)
+    args = parser.parse_args()
+
+    station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=1, padding_number_=10000,
+                                          radio=args.radio)
+
+    ip_test = IPV4L4(args.mgr, lfjson_port, ssid=args.ssid, password=args.passwd,
+                     security=args.security, station_list=station_list, url=args.url, num_tests=args.num_tests,
+                     target_requests_per_ten=args.target_per_ten,
+                     requests_per_ten=args.requests_per_ten)
     ip_test.cleanup(station_list)
     ip_test.build()
     ip_test.start()

py-scripts/test_ipv4_variable_time.py

@@ -161,7 +161,7 @@ class IPV4VariableTime(LFCliBase):
 def main():
     lfjson_port = 8080
 
-    parser = argparse.ArgumentParser(
+    parser = LFCliBase.create_basic_argparse(
         prog='test_ipv4_variable_time.py',
         #formatter_class=argparse.RawDescriptionHelpFormatter,
         formatter_class=argparse.RawTextHelpFormatter,
@@ -185,14 +185,9 @@ Note:   multiple --radio switches may be entered up to the number of radios avai
 
         ''')
 
-   
-    parser.add_argument('--mgr', help='--mgr <hostname for where LANforge GUI is running>',default='localhost')
-    parser.add_argument('-u', '--upstream_port', help='--upstream_port <1.eth1, etc>',default='1.eth1')
-    parser.add_argument('--radio', help='--radio <radio EID>',default='wiphy2')
-    parser.add_argument('--ssid', help='--ssid <SSID>',default='jedway-wpa2-160')
-    parser.add_argument('--passwd', help='--passwd <Password>',default='jedway-wpa2-160')
-    parser.add_argument('--security', help='--security <wpa2 | open | wpa3>',default='wpa2')
-    parser.add_argument('--debug', help='--debug:  Enable debugging',default=False, action="store_true")
+    parser.add_argument('--a_min', help='--a_min bps rate minimum for side_a', default=256000)
+    parser.add_argument('--b_min', help='--a_min bps rate minimum for side_a', default=256000)
+    parser.add_argument('--test_duration', help='--test_duration sets the duration of the test', default="5m")
 
     args = parser.parse_args()
 
@@ -204,8 +199,8 @@ Note:   multiple --radio switches may be entered up to the number of radios avai
                                    ssid=args.ssid,
                                    password=args.passwd,
                                    radio=args.radio,
-                                   security=args.security, test_duration="5m", use_ht160=False,
-                                   side_a_min_rate=256000, side_b_min_rate=256000, _debug_on=args.debug)
+                                   security=args.security, test_duration=args.test_duration, use_ht160=False,
+                                   side_a_min_rate=args.a_min, side_b_min_rate=args.b_min, _debug_on=args.debug)
 
     ip_var_test.pre_cleanup()
     ip_var_test.build()