scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-11-01 03:07:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

connectTest.py: refactors test into a class; fixes how port is brought up, now reliably creates port and it get IP address; NOT DONE

Browse Source
This commit is contained in:
Jed Reynolds
2020-06-02 23:57:42 -07:00
parent 61701f2fa6
commit 8fee309d2d
1 changed files with 458 additions and 498 deletions
Download Patch File Download Diff File Expand all files Collapse all files

956
connectTest.py
View File

@@ -1,17 +1,12 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import os
import time
import sys import sys
if 'py-json' not in sys.path: if 'py-json' not in sys.path:
sys.path.append('py-json') sys.path.append('py-json')
import subprocess
import json
import pprint
from LANforge import LFRequest
from LANforge import LFUtils from LANforge import LFUtils
from LANforge.LFUtils import * from LANforge.LFUtils import *
from LANforge.lfcli_base import LFCliBase
import create_genlink as genl import create_genlink as genl
@@ -21,494 +16,459 @@ if sys.version_info[0] != 3:
exit(1) exit(1)
mgrURL = "http://localhost:8080/" mgrURL = "http://localhost:8080/"
staName = "sta0"
def jsonReq(mgrURL, reqURL, data, exitWhenCalled=False): staNameUri = "port/1/1/" + staName
lf_r = LFRequest.LFRequest(mgrURL + reqURL)
lf_r.addPostData(data)
class ConnectTest(LFCliBase):
if exitWhenCalled: def __init__(self, lfhost, lfport):
json_response = lf_r.jsonPost(True) super().__init__(lfhost, lfport, True)
print("jsonReq: debugdie Response: ") super().checkConnect()
LFUtils.debug_printer.pprint(vars(json_response))
print("jsonReq: bye") # compare pre-test values to post-test values
sys.exit(1) @staticmethod
else: def CompareVals(_name, preVal, postVal):
lf_r.jsonPost(exitWhenCalled) print(f"Comparing {_name}")
if postVal > preVal:
def getJsonInfo(mgrURL, reqURL, debug=False): print(" Test Passed")
lf_r = LFRequest.LFRequest(mgrURL + reqURL) else:
json_response = lf_r.getAsJson(debug) print(f" Test Failed: {_name} did not increase after 5 seconds")
return json_response
#print(name) def run(self):
#j_printer = pprint.PrettyPrinter(indent=2) print("See home/lanforge/Documents/connectTestLogs/connectTestLatest for specific values on latest test")
#j_printer.pprint(json_response)
#for record in json_response[key]: eth1IP = super().jsonGet("port/1/1/eth1")
# j_printer.pprint(record) if eth1IP['interface']['ip'] == "0.0.0.0":
print("Warning: Eth1 lacks ip address")
exit(1)
print("Checking for LANforge Client")
response = getJsonInfo(mgrURL, 'port/1/1/wiphy0') # Create stations and turn dhcp on
duration = 0 print("Creating station and turning on dhcp")
while ((response == None) and (duration < 300)):
#print("LANforge Client not found sleeping 5 seconds") response = super().jsonGet(staNameUri)
duration += 2 if response is not None:
time.sleep(2) if response["interface"] is not None:
response = getJsonInfo(mgrURL, 'port/1/1/wiphy0') print("removing old station")
removePort(1, staName, mgrURL)
if duration >= 300: waitUntilPortsDisappear(1, mgrURL, [staName])
print("Could not connect to LANforge Client") time.sleep(1)
sys.exit(1)
url = "cli-json/add_sta"
data = {
print("See home/lanforge/Documents/connectTestLogs/connectTestLatest for specific values on latest test") "shelf": 1,
#Create stations and turn dhcp on "resource": 1,
print("Creating station and turning on dhcp") "radio": "wiphy0",
"sta_name": staName,
url = "port/1/1/sta00000" "ssid": "jedway-wpa2-x2048-5-1",
debugOn = True "key": "jedway-wpa2-x2048-5-1",
response = getJsonInfo(mgrURL, url) "mode": 0,
if (response is not None): "mac": "xx:xx:xx:xx:*:xx",
if (response["interface"] is not None): "flags": (0x400 + 0x20000 + 0x1000000000) # create admin down
print("removing old station") }
LFUtils.removePortByName("1.1.sta00000", mgrURL) super().jsonPost(url, data)
time.sleep(1) time.sleep(0.05)
reqURL = "cli-json/set_port"
url = "cli-json/add_sta" data = {
data = { "shelf": 1,
"shelf":1, "resource": 1,
"resource":1, "port": staName,
"radio":"wiphy0", "current_flags": (0x1 + 0x80000000),
"sta_name":"sta00000", "interest": (0x2 + 0x4000 + 0x800000) # current, dhcp, down,
"ssid":"jedway-wpa2-x2048-5-1", }
"key":"jedway-wpa2-x2048-5-1", super().jsonPost(reqURL, data)
"mode":1, time.sleep(0.5)
"mac":"xx:xx:xx:xx:*:xx", super().jsonPost("cli-json/set_port", portUpRequest(1, staName))
"flags":1024 #0x400 | 1024
} reqURL = "cli-json/nc_show_ports"
print("adding new station") data = {"shelf": 1,
jsonReq(mgrURL, url, data) "resource": 1,
"port": staName,
time.sleep(1) "probe_flags": 1}
super().jsonPost(reqURL, data)
reqURL = "cli-json/set_port" time.sleep(0.5)
data = { waitUntilPortsAdminUp(1, mgrURL, [staName])
"shelf":1,
"resource":1, duration = 0
"port":"sta00000", maxTime = 300
"current_flags": 2147483648, #0x80000000 | 2147483648 ip = "0.0.0.0"
"interest":16386 # 0x4002 | 16386 while (ip == "0.0.0.0") and (duration < maxTime):
} station_info = super().jsonGet(staNameUri + "?fields=port,ip")
print("configuring port") LFUtils.debug_printer.pprint(station_info)
jsonReq(mgrURL, reqURL, data) if (station_info is not None) and ("interface" in station_info) and ("ip" in station_info["interface"]):
ip = station_info["interface"]["ip"]
time.sleep(5) if ip == "0.0.0.0":
duration += 4
eth1IP = getJsonInfo(mgrURL, "port/1/1/eth1") time.sleep(4)
if eth1IP['interface']['ip'] == "0.0.0.0": else:
print("Warning: Eth1 lacks ip address") break
reqURL = "cli-json/nc_show_ports" if duration >= maxTime:
data = { "shelf":1, print(staName+" failed to get an ip. Ending test")
"resource":1, print("Cleaning up...")
"port":"sta0000", removePort(1, staName, mgrURL)
"probe_flags":1 } sys.exit(1)
jsonReq(mgrURL, reqURL, data)
print("Creating endpoints and cross connects")
station_info = getJsonInfo(mgrURL, "port/1/1/sta00000?fields=port,ip") # create cx for tcp and udp
duration = 0 cmd = (
maxTime = 300 f"./lf_firemod.pl --action create_cx --cx_name testTCP --use_ports {staName},eth1 --use_speeds 360000,150000 --endp_type tcp > /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
ip = "0.0.0.0" execWrap(cmd)
while ((ip == "0.0.0.0") and (duration < maxTime)): cmd = (
station_info = getJsonInfo(mgrURL, "port/1/1/sta00000?fields=port,ip") f"./lf_firemod.pl --action create_cx --cx_name testUDP --use_ports {staName},eth1 --use_speeds 360000,150000 --endp_type udp >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
execWrap(cmd)
LFUtils.debug_printer.pprint(station_info) time.sleep(.05)
if ((station_info is not None) and ("interface" in station_info) and ("ip" in station_info["interface"])):
ip = station_info["interface"]["ip"] # create l4 endpoint
duration += 4 url = "cli-json/add_l4_endp"
time.sleep(4) data = {
"alias": "l4Test",
if duration >= maxTime: "shelf": 1,
print("sta00000 failed to get an ip. Ending test") "resource": 1,
print("Cleaning up...") "port": "sta00000",
removePort(1,"sta00000", mgrURL) "type": "l4_generic",
sys.exit(1) "timeout": 1000,
"url_rate": 600,
"url": "dl http://10.40.0.1/ /dev/null"
#create endpoints and cxs }
super().jsonPost(url, data)
time.sleep(.05)
print("Creating endpoints and cross connects")
#create cx for tcp and udp # create cx for l4_endp
cmd = ("./lf_firemod.pl --action create_cx --cx_name testTCP --use_ports sta00000,eth1 --use_speeds 360000,150000 --endp_type tcp > /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") url = "cli-json/add_cx"
execWrap(cmd) data = {
cmd = ("./lf_firemod.pl --action create_cx --cx_name testUDP --use_ports sta00000,eth1 --use_speeds 360000,150000 --endp_type udp >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") "alias": "CX_l4Test",
execWrap(cmd) "test_mgr": "default_tm",
time.sleep(.5) "tx_endp": "l4Test",
"rx_endp": "NA"
#create l4 endpoint }
url = "cli-json/add_l4_endp" super().jsonPost(url, data)
data = { time.sleep(.05)
"alias":"l4Test",
"shelf":1, # create fileio endpoint
"resource":1, url = "cli-json/add_file_endp"
"port":"sta00000", data = {
"type":"l4_generic", "alias": "fioTest",
"timeout":1000, "shelf": 1,
"url_rate":600, "resource": 1,
"url":"dl http://10.40.0.1/ /dev/null" "port": "sta00000",
} "type": "fe_nfs",
"directory": "/mnt/fe-test"
jsonReq(mgrURL, url, data) }
time.sleep(.5) super().jsonPost(url, data)
time.sleep(.05)
#create cx for l4_endp
url = "cli-json/add_cx" # create fileio cx
data = { url = "cli-json/add_cx"
"alias":"CX_l4Test", data = {
"test_mgr":"default_tm", "alias": "CX_fioTest",
"tx_endp":"l4Test", "test_mgr": "default_tm",
"rx_endp":"NA" "tx_endp": "fioTest",
} "rx_endp": "NA"
}
jsonReq(mgrURL, url, data) super().jsonPost(url, data)
time.sleep(.5) time.sleep(.05)
#create fileio endpoint # create generic endpoints
url = "cli-json/add_file_endp" genl.createGenEndp("genTest1", 1, 1, "sta00000", "gen_generic")
data = { genl.createGenEndp("genTest2", 1, 1, "sta00000", "gen_generic")
"alias":"fioTest", genl.setFlags("genTest1", "ClearPortOnStart", 1)
"shelf":1, genl.setFlags("genTest2", "ClearPortOnStart", 1)
"resource":1, genl.setFlags("genTest2", "Unmanaged", 1)
"port":"sta00000", genl.setCmd("genTest1", "lfping -i 0.1 -I sta00000 10.40.0.1")
"type":"fe_nfs", time.sleep(.05)
"directory":"/mnt/fe-test"
} # create generic cx
url = "cli-json/add_cx"
jsonReq(mgrURL,url,data) data = {
time.sleep(.5) "alias": "CX_genTest1",
"test_mgr": "default_tm",
#create fileio cx "tx_endp": "genTest1",
url = "cli-json/add_cx" "rx_endp": "genTest2"
data = { }
"alias":"CX_fioTest", super().jsonPost(url, data)
"test_mgr":"default_tm", time.sleep(.05)
"tx_endp":"fioTest",
"rx_endp":"NA" # create redirects for wanlink
} url = "cli-json/add_rdd"
data = {
jsonReq(mgrURL,url,data) "shelf": 1,
time.sleep(.5) "resource": 1,
"port": "rdd0",
#create generic endpoints "peer_ifname": "rdd1"
genl.createGenEndp("genTest1",1,1,"sta00000","gen_generic") }
genl.createGenEndp("genTest2",1,1,"sta00000","gen_generic") super().jsonPost(url, data)
genl.setFlags("genTest1","ClearPortOnStart",1)
genl.setFlags("genTest2","ClearPortOnStart",1) url = "cli-json/add_rdd"
genl.setFlags("genTest2","Unmanaged",1) data = {
genl.setCmd("genTest1","lfping -i 0.1 -I sta00000 10.40.0.1") "shelf": 1,
time.sleep(.5) "resource": 1,
"port": "rdd1",
#create generic cx "peer_ifname": "rdd0"
url = "cli-json/add_cx" }
data = { super().jsonPost(url, data)
"alias":"CX_genTest1", time.sleep(.05)
"test_mgr":"default_tm",
"tx_endp":"genTest1", # reset redirect ports
"rx_endp":"genTest2" url = "cli-json/reset_port"
} data = {
"shelf": 1,
jsonReq(mgrURL,url,data) "resource": 1,
time.sleep(.5) "port": "rdd0"
}
#create redirects for wanlink super().jsonPost(url, data)
url = "cli-json/add_rdd"
data = { url = "cli-json/reset_port"
"shelf":1, data = {
"resource":1, "shelf": 1,
"port":"rdd0", "resource": 1,
"peer_ifname":"rdd1" "port": "rdd1"
} }
super().jsonPost(url, data)
jsonReq(mgrURL,url,data) time.sleep(.05)
url = "cli-json/add_rdd" # create wanlink endpoints
data = { url = "cli-json/add_wl_endp"
"shelf":1, data = {
"resource":1, "alias": "wlTest1",
"port":"rdd1", "shelf": 1,
"peer_ifname":"rdd0" "resource": 1,
} "port": "rdd0",
"latency": 20,
jsonReq(mgrURL,url,data) "max_rate": 1544000
time.sleep(.5) }
super().jsonPost(url, data)
#reset redirect ports
url = "cli-json/reset_port" url = "cli-json/add_wl_endp"
data = { data = {
"shelf":1, "alias": "wlTest2",
"resource":1, "shelf": 1,
"port":"rdd0" "resource": 1,
} "port": "rdd1",
"latency": 30,
jsonReq(mgrURL,url,data) "max_rate": 1544000
}
url = "cli-json/reset_port" super().jsonPost(url, data)
data = { time.sleep(.05)
"shelf":1,
"resource":1, # create wanlink cx
"port":"rdd1" url = "cli-json/add_cx"
} data = {
"alias": "CX_wlTest1",
jsonReq(mgrURL,url,data) "test_mgr": "default_tm",
time.sleep(.5) "tx_endp": "wlTest1",
"rx_endp": "wlTest2"
}
#create wanlink endpoints super().jsonPost(url, data)
url = "cli-json/add_wl_endp" time.sleep(.5)
data = {
"alias":"wlTest1", cxNames = ["testTCP", "testUDP", "CX_l4Test", "CX_fioTest", "CX_genTest1", "CX_wlTest1"]
"shelf":1,
"resource":1, # get data before running traffic
"port":"rdd0", try:
"latency":20, testTCPA = super().jsonGet("endp/testTCP-A?fields=tx+bytes,rx+bytes")
"max_rate":1544000 testTCPATX = testTCPA['endpoint']['tx bytes']
} testTCPARX = testTCPA['endpoint']['rx bytes']
jsonReq(mgrURL,url,data) testTCPB = super().jsonGet("endp/testTCP-B?fields=tx+bytes,rx+bytes")
testTCPBTX = testTCPB['endpoint']['tx bytes']
url = "cli-json/add_wl_endp" testTCPBRX = testTCPB['endpoint']['rx bytes']
data = {
"alias":"wlTest2", testUDPA = super().jsonGet("endp/testUDP-A?fields=tx+bytes,rx+bytes")
"shelf":1, testUDPATX = testUDPA['endpoint']['tx bytes']
"resource":1, testUDPARX = testUDPA['endpoint']['rx bytes']
"port":"rdd1",
"latency":30, testUDPB = super().jsonGet("endp/testUDP-B?fields=tx+bytes,rx+bytes")
"max_rate":1544000 testUDPBTX = testUDPB['endpoint']['tx bytes']
} testUDPBRX = testUDPB['endpoint']['rx bytes']
jsonReq(mgrURL,url,data) l4Test = super().jsonGet("layer4/l4Test?fields=bytes-rd")
time.sleep(.5) l4TestBR = l4Test['endpoint']['bytes-rd']
#create wanlink cx genTest1 = super().jsonGet("generic/genTest1?fields=last+results")
url = "cli-json/add_cx" genTest1LR = genTest1['endpoint']['last results']
data = {
"alias":"CX_wlTest1", wlTest1 = super().jsonGet("wl_ep/wlTest1")
"test_mgr":"default_tm", wlTest1TXB = wlTest1['endpoint']['tx bytes']
"tx_endp":"wlTest1", wlTest1RXP = wlTest1['endpoint']['rx pkts']
"rx_endp":"wlTest2" wlTest2 = super().jsonGet("wl_ep/wlTest2")
} wlTest2TXB = wlTest2['endpoint']['tx bytes']
wlTest2RXP = wlTest2['endpoint']['rx pkts']
jsonReq(mgrURL,url,data)
time.sleep(.5) except Exception as e:
print("Something went wrong")
cxNames = ["testTCP","testUDP", "CX_l4Test", "CX_fioTest", "CX_genTest1", "CX_wlTest1"] print(e)
print("Cleaning up...")
#get data before running traffic LFUtils.removePort(1, staName, mgrURL)
try: endpNames = ["testTCP-A", "testTCP-B",
testTCPA = getJsonInfo(mgrURL, "endp/testTCP-A?fields=tx+bytes,rx+bytes") "testUDP-A", "testUDP-B",
testTCPATX = testTCPA['endpoint']['tx bytes'] "l4Test", "fioTest",
testTCPARX = testTCPA['endpoint']['rx bytes'] "genTest1", "genTest2",
"wlTest1", "wlTest2"]
testTCPB = getJsonInfo(mgrURL, "endp/testTCP-B?fields=tx+bytes,rx+bytes") removeCX(mgrURL, cxNames)
testTCPBTX = testTCPB['endpoint']['tx bytes'] removeEndps(mgrURL, endpNames)
testTCPBRX = testTCPB['endpoint']['rx bytes'] sys.exit(1)
testUDPA = getJsonInfo(mgrURL, "endp/testUDP-A?fields=tx+bytes,rx+bytes") # start cx traffic
testUDPATX = testUDPA['endpoint']['tx bytes'] print("\nStarting CX Traffic")
testUDPARX = testUDPA['endpoint']['rx bytes'] for name in range(len(cxNames)):
cmd = (
testUDPB = getJsonInfo(mgrURL, "endp/testUDP-B?fields=tx+bytes,rx+bytes") f"./lf_firemod.pl --mgr localhost --quiet yes --action do_cmd --cmd \"set_cx_state default_tm {cxNames[name]} RUNNING\" >> /tmp/connectTest.log")
testUDPBTX = testUDPB['endpoint']['tx bytes'] execWrap(cmd)
testUDPBRX = testUDPB['endpoint']['rx bytes']
# print("Sleeping for 5 seconds")
l4Test = getJsonInfo(mgrURL, "layer4/l4Test?fields=bytes-rd") time.sleep(5)
l4TestBR = l4Test['endpoint']['bytes-rd']
# show tx and rx bytes for ports
genTest1 = getJsonInfo(mgrURL, "generic/genTest1?fields=last+results")
genTest1LR = genTest1['endpoint']['last results'] os.system("echo eth1 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
cmd = (
wlTest1 = getJsonInfo(mgrURL,"wl_ep/wlTest1") "./lf_portmod.pl --quiet 1 --manager localhost --port_name eth1 --show_port \"Txb,Rxb\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
wlTest1TXB = wlTest1['endpoint']['tx bytes'] execWrap(cmd)
wlTest1RXP = wlTest1['endpoint']['rx pkts'] os.system("echo sta00000 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
wlTest2 = getJsonInfo(mgrURL,"wl_ep/wlTest2") cmd = (
wlTest2TXB = wlTest2['endpoint']['tx bytes'] "./lf_portmod.pl --quiet 1 --manager localhost --port_name sta00000 --show_port \"Txb,Rxb\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
wlTest2RXP = wlTest2['endpoint']['rx pkts'] execWrap(cmd)
except Exception as e:
print("Something went wrong") # show tx and rx for endpoints PERL
print(e) os.system("echo TestTCP-A >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
print("Cleaning up...") cmd = (
LFUtils.removePort(1,"sta00000", mgrURL) "./lf_firemod.pl --action show_endp --endp_name testTCP-A --endp_vals \"Tx Bytes,Rx Bytes\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
execWrap(cmd)
endpNames = ["testTCP-A", "testTCP-B", os.system("echo TestTCP-B >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
"testUDP-A", "testUDP-B", cmd = (
"l4Test", "fioTest", "./lf_firemod.pl --action show_endp --endp_name testTCP-B --endp_vals \"Tx Bytes,Rx Bytes\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
"genTest1", "genTest2", execWrap(cmd)
"wlTest1","wlTest2"] os.system("echo TestUDP-A >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
removeCX(mgrURL, cxNames) cmd = (
removeEndps(mgrURL, endpNames) "./lf_firemod.pl --action show_endp --endp_name testUDP-A --endp_vals \"Tx Bytes,Rx Bytes\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
sys.exit(1) execWrap(cmd)
os.system("echo TestUDP-B >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
#start cx traffic cmd = (
print("\nStarting CX Traffic") "./lf_firemod.pl --action show_endp --endp_name testUDP-B --endp_vals \"Tx Bytes,Rx Bytes\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
for name in range(len(cxNames)): execWrap(cmd)
cmd = (f"./lf_firemod.pl --mgr localhost --quiet yes --action do_cmd --cmd \"set_cx_state default_tm {cxNames[name]} RUNNING\" >> /tmp/connectTest.log") os.system("echo l4Test >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
execWrap(cmd) cmd = (
"./lf_firemod.pl --action show_endp --endp_name l4Test --endp_vals Bytes-Read-Total >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
#print("Sleeping for 5 seconds") execWrap(cmd)
time.sleep(5) os.system("echo fioTest >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
cmd = (
#show tx and rx bytes for ports "./lf_firemod.pl --action show_endp --endp_name fioTest --endp_vals \"Bytes Written,Bytes Read\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
execWrap(cmd)
os.system("echo eth1 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") os.system("echo genTest1 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
cmd = ("./lf_portmod.pl --quiet 1 --manager localhost --port_name eth1 --show_port \"Txb,Rxb\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") cmd = (
execWrap(cmd) "./lf_firemod.pl --action show_endp --endp_name genTest1 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
os.system("echo sta00000 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") execWrap(cmd)
cmd = ("./lf_portmod.pl --quiet 1 --manager localhost --port_name sta00000 --show_port \"Txb,Rxb\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") os.system("echo wlTest1 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
execWrap(cmd) cmd = (
"./lf_firemod.pl --action show_endp --endp_name wlTest1 --endp_vals \"Rx Pkts,Tx Bytes,Cur-Backlog,Dump File,Tx3s\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
execWrap(cmd)
#show tx and rx for endpoints PERL os.system("echo wlTest2 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
os.system("echo TestTCP-A >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") cmd = (
cmd = ("./lf_firemod.pl --action show_endp --endp_name testTCP-A --endp_vals \"Tx Bytes,Rx Bytes\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") "./lf_firemod.pl --action show_endp --endp_name wlTest2 --endp_vals \"Rx Pkts,Tx Bytes,Cur-Backlog,Dump File,Tx3s\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
execWrap(cmd) execWrap(cmd)
os.system("echo TestTCP-B >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
cmd = ("./lf_firemod.pl --action show_endp --endp_name testTCP-B --endp_vals \"Tx Bytes,Rx Bytes\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") # stop cx traffic
execWrap(cmd) print("Stopping CX Traffic")
os.system("echo TestUDP-A >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") for name in range(len(cxNames)):
cmd = ("./lf_firemod.pl --action show_endp --endp_name testUDP-A --endp_vals \"Tx Bytes,Rx Bytes\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") cmd = (
execWrap(cmd) f"./lf_firemod.pl --mgr localhost --quiet yes --action do_cmd --cmd \"set_cx_state default_tm {cxNames[name]} STOPPED\" >> /tmp/connectTest.log")
os.system("echo TestUDP-B >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") execWrap(cmd)
cmd = ("./lf_firemod.pl --action show_endp --endp_name testUDP-B --endp_vals \"Tx Bytes,Rx Bytes\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") # print("Sleeping for 15 seconds")
execWrap(cmd) time.sleep(15)
os.system("echo l4Test >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
cmd = ("./lf_firemod.pl --action show_endp --endp_name l4Test --endp_vals Bytes-Read-Total >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") # get data for endpoints JSON
execWrap(cmd) print("Collecting Data")
os.system("echo fioTest >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") try:
cmd = ("./lf_firemod.pl --action show_endp --endp_name fioTest --endp_vals \"Bytes Written,Bytes Read\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") ptestTCPA = super().jsonGet("endp/testTCP-A?fields=tx+bytes,rx+bytes")
execWrap(cmd) ptestTCPATX = ptestTCPA['endpoint']['tx bytes']
os.system("echo genTest1 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") ptestTCPARX = ptestTCPA['endpoint']['rx bytes']
cmd = ("./lf_firemod.pl --action show_endp --endp_name genTest1 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log")
execWrap(cmd) ptestTCPB = super().jsonGet("endp/testTCP-B?fields=tx+bytes,rx+bytes")
os.system("echo wlTest1 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") ptestTCPBTX = ptestTCPB['endpoint']['tx bytes']
cmd = ("./lf_firemod.pl --action show_endp --endp_name wlTest1 --endp_vals \"Rx Pkts,Tx Bytes,Cur-Backlog,Dump File,Tx3s\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") ptestTCPBRX = ptestTCPB['endpoint']['rx bytes']
execWrap(cmd)
os.system("echo wlTest2 >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") ptestUDPA = super().jsonGet("endp/testUDP-A?fields=tx+bytes,rx+bytes")
cmd = ("./lf_firemod.pl --action show_endp --endp_name wlTest2 --endp_vals \"Rx Pkts,Tx Bytes,Cur-Backlog,Dump File,Tx3s\" >> /home/lanforge/Documents/connectTestLogs/connectTestLatest.log") ptestUDPATX = ptestUDPA['endpoint']['tx bytes']
execWrap(cmd) ptestUDPARX = ptestUDPA['endpoint']['rx bytes']
ptestUDPB = super().jsonGet("endp/testUDP-B?fields=tx+bytes,rx+bytes")
ptestUDPBTX = ptestUDPB['endpoint']['tx bytes']
#stop cx traffic ptestUDPBRX = ptestUDPB['endpoint']['rx bytes']
print("Stopping CX Traffic")
for name in range(len(cxNames)): pl4Test = super().jsonGet("layer4/l4Test?fields=bytes-rd")
cmd = (f"./lf_firemod.pl --mgr localhost --quiet yes --action do_cmd --cmd \"set_cx_state default_tm {cxNames[name]} STOPPED\" >> /tmp/connectTest.log") pl4TestBR = pl4Test['endpoint']['bytes-rd']
execWrap(cmd)
#print("Sleeping for 15 seconds") pgenTest1 = super().jsonGet("generic/genTest1?fields=last+results")
time.sleep(15) pgenTest1LR = pgenTest1['endpoint']['last results']
#get data for endpoints JSON pwlTest1 = super().jsonGet("wl_ep/wlTest1")
print("Collecting Data") pwlTest1TXB = pwlTest1['endpoint']['tx bytes']
try: pwlTest1RXP = pwlTest1['endpoint']['rx pkts']
ptestTCPA = getJsonInfo(mgrURL, "endp/testTCP-A?fields=tx+bytes,rx+bytes") pwlTest2 = super().jsonGet("wl_ep/wlTest2")
ptestTCPATX = ptestTCPA['endpoint']['tx bytes'] pwlTest2TXB = pwlTest2['endpoint']['tx bytes']
ptestTCPARX = ptestTCPA['endpoint']['rx bytes'] pwlTest2RXP = pwlTest2['endpoint']['rx pkts']
except Exception as e:
ptestTCPB = getJsonInfo(mgrURL, "endp/testTCP-B?fields=tx+bytes,rx+bytes") print("Something went wrong")
ptestTCPBTX = ptestTCPB['endpoint']['tx bytes'] print(e)
ptestTCPBRX = ptestTCPB['endpoint']['rx bytes'] print("Cleaning up...")
reqURL = "cli-json/rm_vlan"
ptestUDPA = getJsonInfo(mgrURL, "endp/testUDP-A?fields=tx+bytes,rx+bytes") data = {
ptestUDPATX = ptestUDPA['endpoint']['tx bytes'] "shelf": 1,
ptestUDPARX = ptestUDPA['endpoint']['rx bytes'] "resource": 1,
"port": staName
ptestUDPB = getJsonInfo(mgrURL, "endp/testUDP-B?fields=tx+bytes,rx+bytes") }
ptestUDPBTX = ptestUDPB['endpoint']['tx bytes'] super().jsonPost(reqURL, data)
ptestUDPBRX = ptestUDPB['endpoint']['rx bytes']
endpNames = ["testTCP-A", "testTCP-B",
pl4Test = getJsonInfo(mgrURL, "layer4/l4Test?fields=bytes-rd") "testUDP-A", "testUDP-B",
pl4TestBR = pl4Test['endpoint']['bytes-rd'] "l4Test", "fioTest",
"genTest1", "genTest2",
pgenTest1 = getJsonInfo(mgrURL, "generic/genTest1?fields=last+results") "wlTest1", "wlTest2"]
pgenTest1LR = pgenTest1['endpoint']['last results'] removeCX(mgrURL, cxNames)
removeEndps(mgrURL, endpNames)
pwlTest1 = getJsonInfo(mgrURL,"wl_ep/wlTest1") sys.exit(1)
pwlTest1TXB = pwlTest1['endpoint']['tx bytes']
pwlTest1RXP = pwlTest1['endpoint']['rx pkts'] # print("Sleeping for 5 seconds")
pwlTest2 = getJsonInfo(mgrURL,"wl_ep/wlTest2") time.sleep(5)
pwlTest2TXB = pwlTest2['endpoint']['tx bytes']
pwlTest2RXP = pwlTest2['endpoint']['rx pkts'] print("\n")
except Exception as e: self.CompareVals("testTCP-A TX", testTCPATX, ptestTCPATX)
print("Something went wrong") self.CompareVals("testTCP-A RX", testTCPARX, ptestTCPARX)
print(e) self.CompareVals("testTCP-B TX", testTCPBTX, ptestTCPBTX)
print("Cleaning up...") self.CompareVals("testTCP-B RX", testTCPBRX, ptestTCPBRX)
reqURL = "cli-json/rm_vlan" self.CompareVals("testUDP-A TX", testUDPATX, ptestUDPATX)
data = { self.CompareVals("testUDP-A RX", testUDPARX, ptestUDPARX)
"shelf":1, self.CompareVals("testUDP-B TX", testUDPBTX, ptestUDPBTX)
"resource":1, self.CompareVals("testUDP-B RX", testUDPBRX, ptestUDPBRX)
"port":"sta00000" self.CompareVals("l4Test Bytes Read", l4TestBR, pl4TestBR)
} self.CompareVals("genTest1 Last Results", genTest1LR, pgenTest1LR)
self.CompareVals("wlTest1 TX Bytes", wlTest1TXB, pwlTest1TXB)
jsonReq(mgrURL, reqURL, data) self.CompareVals("wlTest1 RX Pkts", wlTest1RXP, pwlTest1RXP)
self.CompareVals("wlTest2 TX Bytes", wlTest2TXB, pwlTest2TXB)
endpNames = ["testTCP-A", "testTCP-B", self.CompareVals("wlTest2 RX Pkts", wlTest2RXP, pwlTest2RXP)
"testUDP-A", "testUDP-B", print("\n")
"l4Test", "fioTest",
"genTest1", "genTest2", # remove all endpoints and cxs
"wlTest1","wlTest2"] print("Cleaning up...")
removeCX(mgrURL, cxNames) LFUtils.removePort(1, "sta00000", mgrURL)
removeEndps(mgrURL, endpNames)
sys.exit(1) endpNames = ["testTCP-A", "testTCP-B",
"testUDP-A", "testUDP-B",
#print("Sleeping for 5 seconds") "l4Test", "fioTest",
time.sleep(5) "genTest1", "genTest2",
"wlTest1", "wlTest2"]
removeCX(mgrURL, cxNames)
removeEndps(mgrURL, endpNames)
#compare pre-test values to post-test values
def compareVals(name, preVal, postVal): # ~class
print(f"Comparing {name}")
if postVal > preVal: def main():
print(" Test Passed") lfjson_host = "localhost"
else: lfjson_port = 8080
print(f" Test Failed: {name} did not increase after 5 seconds") test = ConnectTest(lfjson_host, lfjson_port)
test.run()
print("\n")
compareVals("testTCP-A TX", testTCPATX, ptestTCPATX)
compareVals("testTCP-A RX", testTCPARX, ptestTCPARX) if __name__ == "__main__":
main()
compareVals("testTCP-B TX", testTCPBTX, ptestTCPBTX)
compareVals("testTCP-B RX", testTCPBRX, ptestTCPBRX)
compareVals("testUDP-A TX", testUDPATX, ptestUDPATX)
compareVals("testUDP-A RX", testUDPARX, ptestUDPARX)
compareVals("testUDP-B TX", testUDPBTX, ptestUDPBTX)
compareVals("testUDP-B RX", testUDPBRX, ptestUDPBRX)
compareVals("l4Test Bytes Read", l4TestBR, pl4TestBR)
compareVals("genTest1 Last Results", genTest1LR, pgenTest1LR)
compareVals("wlTest1 TX Bytes", wlTest1TXB, pwlTest1TXB)
compareVals("wlTest1 RX Pkts", wlTest1RXP, pwlTest1RXP)
compareVals("wlTest2 TX Bytes", wlTest2TXB, pwlTest2TXB)
compareVals("wlTest2 RX Pkts", wlTest2RXP, pwlTest2RXP)
print("\n")
#remove all endpoints and cxs
print("Cleaning up...")
LFUtils.removePort(1,"sta00000", mgrURL)
endpNames = ["testTCP-A", "testTCP-B",
"testUDP-A", "testUDP-B",
"l4Test", "fioTest",
"genTest1", "genTest2",
"wlTest1","wlTest2"]
removeCX(mgrURL, cxNames)
removeEndps(mgrURL, endpNames)