Add Netgear Clickthru captive portal and Iperf Task

-modify netgear_Iperf_test.py

py-scripts/netgear_Iperf_test.py

@@ -11,33 +11,24 @@ if 'py-json' not in sys.path:
 from LANforge.lfcli_base import LFCliBase
 from LANforge.LFUtils import *
 from LANforge import LFUtils
-#from LANforge.add_file_endp import *
 import argparse
-#from realm import Realm
 import realm
 import time
-import datetime
-
-#from test_generic import GenTest
-
 
 class IperfTest(LFCliBase):
-    def __init__(self, host, port, _local_realm, ssid, security, passwd,upstream, radio="wiphy0", upstream_port="eth1", macvlan_type = "iperf3_serv", sta_type = "iperf3", num_ports=1, macvlan_parent=None, first_mvlan_ip=None, netmask=None, gateway=None,
+    def __init__(self, host, port, _local_realm, ssid, security, passwd, radio="wiphy0", macvlan_type = "iperf3_serv", sta_type = "iperf3", num_ports=1, macvlan_parent=None,
-                 dhcp=False,port_list=[], sta_list=[], ip_list=None,dest=None, test_duration="5m", interval=1, _debug_on=False,_exit_on_error=False,_exit_on_fail=False):
+                 dhcp=False,port_list=[], sta_list=[],_debug_on=False,_exit_on_error=False,_exit_on_fail=False):
         super().__init__(host, port,_local_realm=realm.Realm(host,port), _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
-        self.port = port
+        #self.port = port
-        self.upstream_port = upstream_port
         self.port_list = []
+        self.created_cx = []
         self.sta_list = sta_list
-        self.netmask = netmask
-        self.gateway = gateway
         self.dhcp = dhcp
         self.radio = radio
         self.security = security
         self.passwd = passwd
-        self.upstream = upstream
         self.ssid = ssid
-        self.test_duration = test_duration
+        self.created_endp = []
         if macvlan_parent is not None:
             self.macvlan_parent = macvlan_parent
             self.port_list = port_list
@@ -46,18 +37,12 @@ class IperfTest(LFCliBase):
         self.mvlan_profile.desired_macvlans = self.port_list
         self.mvlan_profile.macvlan_parent = self.macvlan_parent
         self.mvlan_profile.dhcp = dhcp
-        self.mvlan_profile.netmask = netmask
-        self.mvlan_profile.first_ip_addr = first_mvlan_ip
-        self.mvlan_profile.gateway = gateway
         self.generic_endps_profile = self.local_realm.new_generic_endp_profile()
         self.generic_endps_profile.type = macvlan_type
         self.created_ports = []
         self.station_profile = self.local_realm.new_station_profile()
-        self.generic_endps_for_client = self.local_realm.new_generic_endp_profile()
-        self.generic_endps_for_client.type = sta_type
-        self.generic_endps_for_client.dest = dest
-        self.generic_endps_for_client.interval = interval
         self._local_realm = _local_realm
+        self.name_prefix = "generic"
 
 
     def build(self):
@@ -70,16 +55,45 @@ class IperfTest(LFCliBase):
         self.station_profile.create(radio=self.radio, sta_names_=self.sta_list, debug=self.debug)
         self.station_profile.admin_up()
         #self.generic_endps_for_client.create(ports=self.station_profile.station_names, sleep_time=.5)
-
+    def set_flags(self, endp_name, flag_name, val):
+        data = {
+            "name": endp_name,
+            "flag": flag_name,
+            "val": val
+        }
+        self.json_post("cli-json/set_endp_flag", data, debug_=self.debug)
     def create_gen_for_client(self, gen_sta_list, dest, suppress_related_commands_=None):
         server_ip = 0
-        for gen_endp in gen_sta_list:
+        endp_tpls = []
-            port_info = self.local_realm.name_to_eid(gen_endp)
+        for port_name in gen_sta_list:
+            port_info = self.local_realm.name_to_eid(port_name)
+            if len(port_info) == 2:
+                resource = 1
+                shelf = port_info[0]
+                name = port_info[-1]
+            elif len(port_info) == 3:
+                resource = port_info[0]
+                shelf = port_info[1]
+                name = port_info[-1]
+            else:
+                raise ValueError("Unexpected name for port_name %s" % port_name)
+            gen_name_a = "%s-%s" % (self.name_prefix, name)
+            gen_name_b = "D_%s-%s" % (self.name_prefix, name)
+            endp_tpls.append((shelf, resource, name, gen_name_a, gen_name_b))
+
+        for endp_tpl in endp_tpls:
+            shelf = endp_tpl[0]
+            resource = endp_tpl[1]
+            name = endp_tpl[2]
+            gen_name_a = endp_tpl[3]
+            # gen_name_b  = endp_tpl[3]
+            # (self, alias=None, shelf=1, resource=1, port=None, type=None)
+
             data = {
-                "alias": gen_endp,
+                "alias": gen_name_a,
-                "shelf": 1,
+                "shelf": shelf,
-                "resource": 1,
+                "resource": resource,
-                "port": port_info[-1],
+                "port": name,
                 "type": "gen_generic"
             }
             if self.debug:
@@ -89,42 +103,49 @@ class IperfTest(LFCliBase):
 
         self.local_realm.json_post("/cli-json/nc_show_endpoints", {"endpoint": "all"})
         time.sleep(0.5)
-        for gen_endp in gen_sta_list:
+
-            set_flag = {
+        for endp_tpl in endp_tpls:
-                "name": gen_endp,
+            gen_name_a = endp_tpl[3]
-                "flag": "ClearPortOnStart",
+            gen_name_b = endp_tpl[4]
-                "val": 1
+            self.set_flags(gen_name_a, "ClearPortOnStart", 1)
-            }
-            self.json_post("cli-json/set_endp_flag", set_flag, debug_=self.debug)
         time.sleep(0.5)
-
+        for endp_tpl in endp_tpls:
-        for gen_endp in gen_sta_list:
+            name = endp_tpl[2]
+            gen_name_a = endp_tpl[3]
             self.cmd = "iperf3 --forceflush --format k --precision 4 -c %s -t 60 --tos 0 -b 1K --bind_dev %s -i 1 " \
-                       "--pidfile /tmp/lf_helper_iperf3_test.pid" % (dest[server_ip], gen_endp)
+                       "--pidfile /tmp/lf_helper_iperf3_test.pid" % (dest[server_ip], name)
-
             data_cmd = {
-                "name": gen_endp,
+                "name": gen_name_a,
                 "command": self.cmd
             }
             self.json_post("cli-json/set_gen_cmd", data_cmd, debug_=self.debug)
             server_ip = server_ip + 1
-        """post_data = []
+        time.sleep(0.5)
-        for gen_endp in gen_sta_list:
+        post_data = []
-            cx_name = "CX_generic-%s" % (gen_endp)
+        for endp_tpl in endp_tpls:
+            name = endp_tpl[2]
+            gen_name_a = endp_tpl[3]
+            gen_name_b = endp_tpl[4]
+            cx_name = "CX_%s-%s" % (self.name_prefix, name)
             data = {
                 "alias": cx_name,
                 "test_mgr": "default_tm",
-                "tx_endp": gen_endp,
+                "tx_endp": gen_name_a,
-                "rx_endp": "D_%s" % (gen_endp)
+                "rx_endp": gen_name_b
             }
             post_data.append(data)
+            self.created_cx.append(cx_name)
+            self.created_endp.append(gen_name_a)
+            self.created_endp.append(gen_name_b)
+
+        time.sleep(0.5)
+
         for data in post_data:
             url = "/cli-json/add_cx"
             if self.debug:
                 pprint(data)
             self.local_realm.json_post(url, data, debug_=self.debug, suppress_related_commands_=suppress_related_commands_)
             time.sleep(2)
-
         time.sleep(0.5)
         for data in post_data:
             self.local_realm.json_post("/cli-json/show_cx", {
@@ -132,45 +153,22 @@ class IperfTest(LFCliBase):
                 "cross_connect": data["alias"]
             })
         time.sleep(0.5)
-"""
-    def sta(self):
-        self.station_profile.use_security(self.security, self.ssid, self.passwd)
-        self.station_profile.create(radio=self.radio, sta_names_=self.sta_list, debug=self.debug)
-        self.station_profile.admin_up()
-
-    def start(self):
-
-        self.generic_endps_for_client.create(ports=self.station_profile.station_names, sleep_time=.5)
-
 
     def macwlan_cx(self):
         self.generic_endps_profile.start_cx()
-        #self.generic_endps_for_client.start_cx()
+        time.sleep(10)
-
-    def client_cx(self):
-        """self.generic_endps_for_client.start_cx()
-        time.sleep(20)"""
-        temp_stas = []
-        for station in self.sta_list.copy():
-            temp_stas.append(self.local_realm.name_to_eid(station)[2])
-        if self.debug:
-            pprint.pprint(self.station_profile.station_names)
-        LFUtils.wait_until_ports_admin_up(base_url=self.lfclient_url, port_list=self.station_profile.station_names)
-        if self.local_realm.wait_for_ip(temp_stas):
-            self._pass("All stations got IPs")
-        else:
-            self._fail("Stations failed to get IPs")
-            self.exit_fail()
-        cur_time = datetime.datetime.now()
-        passes = 0
-        expected_passes = 0
-        self.generic_endps_for_client.start_cx()
-        time.sleep(15)
-        end_time = self.local_realm.parse_time(self.test_duration) + cur_time
-        print("Starting Test...")
-
-
 
+    def generic_cx(self):
+        #self.generic_endps_for_client.start_cx(self.created_ports)
+        for cx_name in self.created_cx:
+            self.json_post("/cli-json/set_cx_state", {
+                "test_mgr": "default_tm",
+                "cx_name": cx_name,
+                "cx_state": "RUNNING"
+            }, debug_=self.debug)
+            print(".", end='')
+        print("")
+        time.sleep(10)
 
 def main():
 
@@ -181,30 +179,20 @@ def main():
         epilog='''Creates MACVLAN endpoints.''',
 
         description='''\
+         python netgear_Iperf_test.py --mgr 192.168.200.28 --mgr_port 8080 --macvlan_parent eth1 --num_ports 5 --radio wiphy1 --ssid Captive --passwd [Blank] --security open 
+
 ''')
     parser.add_argument('--num_stations', help='Number of stations to create', default=0)
     parser.add_argument('--ssid', help='SSID for stations to associate to')
     parser.add_argument('--passwd', help='Number of stations to create', default=0)
     parser.add_argument('--security', help='security type to use for ssid { wep | wpa | wpa2 | wpa3 | open }')
-    parser.add_argument('--dest', help='destination IP for command', default="10.40.0.1")
-    parser.add_argument('--test_duration', help='duration of the test eg: 30s, 2m, 4h', default="2m")
-    parser.add_argument('--interval', help='interval to use when running lfping (1s, 1m)', default=1)
     parser.add_argument('--radio', help='radio EID, e.g: 1.wiphy2')
-    parser.add_argument('-u', '--upstream_port',
-                        help='non-station port that generates traffic: <resource>.<port>, e.g: 1.eth1',
-                        default='1.eth1')
     parser.add_argument('--macvlan_parent', help='specifies parent port for macvlan creation', default=None)
-    parser.add_argument('--macvlan_type', help='type of command to run: generic, lfping, iperf3-client, iperf3-server, lfcurl',
-                        default="iperf3")
     parser.add_argument('--num_ports', help='number of ports to create', default=1)
-    parser.add_argument('--first_mvlan_ip', help='specifies first static ip address to be used or dhcp', default=None)
+
-    parser.add_argument('--netmask', help='specifies netmask to be used with static ip addresses', default=None)
-    parser.add_argument('--gateway', help='specifies default gateway to be used with static addressing', default=None)
-    parser.add_argument('--cmd', help='specifies command to be run by generic type endp', default='')
     args = parser.parse_args()
 
     port_list = []
-    ip_list = []
     station_list = []
 
 
@@ -216,20 +204,11 @@ def main():
     station_list = LFUtils.port_name_series(prefix="sta" + "#", start_id=0,
                                          end_id=num_ports - 1, padding_number=100000,
                                          radio=args.radio)
-    """if args.first_mvlan_ip is not None:
-        if args.first_mvlan_ip.lower() == "dhcp":
-            dhcp = True
-        else:
-            dhcp = False
-    else:
-        dhcp = True"""
 
     ip_test = IperfTest(args.mgr, args.mgr_port, ssid=args.ssid,_local_realm = None,
-                        passwd=args.passwd, upstream=args.upstream_port,
+                        passwd=args.passwd,
-                        security=args.security, port_list=port_list, test_duration=args.test_duration, sta_list=station_list, ip_list=ip_list,
+                        security=args.security, port_list=port_list,sta_list=station_list, _debug_on=args.debug, macvlan_parent=args.macvlan_parent,
-                        upstream_port=args.upstream_port, _debug_on=args.debug, macvlan_parent=args.macvlan_parent,
+                        dhcp=True, num_ports=args.num_ports)
-                        first_mvlan_ip=args.first_mvlan_ip, dest=args.dest, netmask=args.netmask, gateway=args.gateway,
-                        dhcp=False, num_ports=args.num_ports, interval=1)
     ip_test.build()
     time.sleep(15)
     num_macvlan = 0
@@ -244,34 +223,11 @@ def main():
             break
     time.sleep(5)
     i=0
-    #ip_test.create_gen_for_client(station_list, all_macvlan_ip)
+    ip_test.create_gen_for_client(station_list, all_macvlan_ip)
-
+    time.sleep(10)
-
-
-
-    """for each in all_macvlan_ip:
-        station_list = LFUtils.portNameSeries(radio=args.radio,
-                                              prefix_="sta%s#" % (i),
-                                              start_id_=0,
-                                              end_id_=1 - 1,
-                                              padding_number_=100)
-        ip_test1 = IperfTest(args.mgr, args.mgr_port, ssid=args.ssid,
-                            passwd=args.passwd, upstream=args.upstream_port,
-                            security=args.security, port_list=port_list, test_duration=args.test_duration, sta_list=station_list, ip_list=ip_list,
-                            upstream_port=args.upstream_port, _debug_on=args.debug, macvlan_parent=args.macvlan_parent,
-                            first_mvlan_ip=args.first_mvlan_ip, dest=each, netmask=args.netmask,
-                            gateway=args.gateway,
-                            dhcp=dhcp, num_ports=args.num_ports, interval=1)
-        ip_test1.sta()
-
-        ip_test1.start()
-        time.sleep(5)
-        i = i + 1
-    for gen_end in all_macvlan_ip:
     ip_test.macwlan_cx()
-        time.sleep(0.5)
+    ip_test.generic_cx()
-        ip_test1.client_cx()
+    time.sleep(10)
-        time.sleep(0.5)"""
 
 
 if __name__ == "__main__":