mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-11-01 19:28:00 +00:00
removing files
This commit is contained in:
@@ -1,234 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import sys
|
||||
import pprint
|
||||
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')
|
||||
from LANforge.lfcli_base import LFCliBase
|
||||
from LANforge.LFUtils import *
|
||||
from LANforge import LFUtils
|
||||
import argparse
|
||||
import realm
|
||||
import time
|
||||
|
||||
class IperfTest(LFCliBase):
|
||||
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=[],_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_list = []
|
||||
self.created_cx = []
|
||||
self.sta_list = sta_list
|
||||
self.dhcp = dhcp
|
||||
self.radio = radio
|
||||
self.security = security
|
||||
self.passwd = passwd
|
||||
self.ssid = ssid
|
||||
self.created_endp = []
|
||||
if macvlan_parent is not None:
|
||||
self.macvlan_parent = macvlan_parent
|
||||
self.port_list = port_list
|
||||
self.mvlan_profile = self.local_realm.new_mvlan_profile()
|
||||
self.mvlan_profile.num_macvlans = int(num_ports)
|
||||
self.mvlan_profile.desired_macvlans = self.port_list
|
||||
self.mvlan_profile.macvlan_parent = self.macvlan_parent
|
||||
self.mvlan_profile.dhcp = dhcp
|
||||
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._local_realm = _local_realm
|
||||
self.name_prefix = "generic"
|
||||
|
||||
|
||||
def build(self):
|
||||
print("Creating MACVLANs")
|
||||
self.mvlan_profile.create(admin_down=False, sleep_time=.5, debug=self.debug)
|
||||
self._pass("PASS: MACVLAN build finished")
|
||||
self.created_ports += self.mvlan_profile.created_macvlans
|
||||
self.generic_endps_profile.create(ports=self.mvlan_profile.created_macvlans, sleep_time=.5)
|
||||
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()
|
||||
#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
|
||||
endp_tpls = []
|
||||
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_name_a,
|
||||
"shelf": shelf,
|
||||
"resource": resource,
|
||||
"port": name,
|
||||
"type": "gen_generic"
|
||||
}
|
||||
if self.debug:
|
||||
pprint(data)
|
||||
|
||||
self.json_post("cli-json/add_gen_endp", data, debug_=self.debug)
|
||||
|
||||
self.local_realm.json_post("/cli-json/nc_show_endpoints", {"endpoint": "all"})
|
||||
time.sleep(0.5)
|
||||
|
||||
for endp_tpl in endp_tpls:
|
||||
gen_name_a = endp_tpl[3]
|
||||
gen_name_b = endp_tpl[4]
|
||||
self.set_flags(gen_name_a, "ClearPortOnStart", 1)
|
||||
time.sleep(0.5)
|
||||
for endp_tpl in endp_tpls:
|
||||
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], name)
|
||||
data_cmd = {
|
||||
"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
|
||||
time.sleep(0.5)
|
||||
post_data = []
|
||||
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_name_a,
|
||||
"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", {
|
||||
"test_mgr": "default_tm",
|
||||
"cross_connect": data["alias"]
|
||||
})
|
||||
time.sleep(0.5)
|
||||
|
||||
def macwlan_cx(self):
|
||||
self.generic_endps_profile.start_cx()
|
||||
time.sleep(10)
|
||||
|
||||
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():
|
||||
|
||||
parser = LFCliBase.create_bare_argparse(
|
||||
prog='create_macvlan.py',
|
||||
# formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
formatter_class=argparse.RawTextHelpFormatter,
|
||||
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('--radio', help='radio EID, e.g: 1.wiphy2')
|
||||
parser.add_argument('--macvlan_parent', help='specifies parent port for macvlan creation', default=None)
|
||||
parser.add_argument('--num_ports', help='number of ports to create', default=1)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
port_list = []
|
||||
station_list = []
|
||||
|
||||
|
||||
num_ports = int(args.num_ports)
|
||||
|
||||
port_list = LFUtils.port_name_series(prefix=args.macvlan_parent + "#", start_id=0,
|
||||
end_id=num_ports - 1, padding_number=100000,
|
||||
radio=args.radio)
|
||||
station_list = LFUtils.port_name_series(prefix="sta" + "#", start_id=0,
|
||||
end_id=num_ports - 1, padding_number=100000,
|
||||
radio=args.radio)
|
||||
|
||||
ip_test = IperfTest(args.mgr, args.mgr_port, ssid=args.ssid,_local_realm = None,
|
||||
passwd=args.passwd,
|
||||
security=args.security, port_list=port_list,sta_list=station_list, _debug_on=args.debug, macvlan_parent=args.macvlan_parent,
|
||||
dhcp=True, num_ports=args.num_ports)
|
||||
ip_test.build()
|
||||
time.sleep(15)
|
||||
num_macvlan = 0
|
||||
all_macvlan_ip = []
|
||||
while True:
|
||||
if(num_macvlan < num_ports):
|
||||
macvlan_ip_list = ip_test.json_get("/port/1/1/eth1#%s?field=ip"% (num_macvlan))
|
||||
get_ip = (macvlan_ip_list['interface']['ip'])
|
||||
num_macvlan = num_macvlan + 1
|
||||
all_macvlan_ip.append(get_ip)
|
||||
else:
|
||||
break
|
||||
time.sleep(5)
|
||||
i=0
|
||||
ip_test.create_gen_for_client(station_list, all_macvlan_ip)
|
||||
time.sleep(10)
|
||||
ip_test.macwlan_cx()
|
||||
ip_test.generic_cx()
|
||||
time.sleep(10)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user