mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-11-03 20:27:54 +00:00
196 lines
8.1 KiB
Python
196 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
from LANforge.lfcli_base import LFCliBase
|
|
import port_utils
|
|
from port_utils import PortUtils
|
|
from pprint import pprint
|
|
import pprint
|
|
import time
|
|
|
|
class HTTPProfile(LFCliBase):
|
|
def __init__(self, lfclient_host, lfclient_port, local_realm, debug_=False):
|
|
super().__init__(lfclient_host, lfclient_port, debug_)
|
|
self.lfclient_url = "http://%s:%s" % (lfclient_host, lfclient_port)
|
|
self.debug = debug_
|
|
self.requests_per_ten = 600
|
|
self.local_realm = local_realm
|
|
self.created_cx = {}
|
|
self.created_endp = []
|
|
self.ip_map = {}
|
|
self.direction = "dl"
|
|
self.dest = "/dev/null"
|
|
self.port_util = PortUtils(self.local_realm)
|
|
self.max_speed = 0 #infinity
|
|
|
|
def check_errors(self, debug=False):
|
|
fields_list = ["!conn", "acc.+denied", "bad-proto", "bad-url", "other-err", "total-err", "rslv-p", "rslv-h",
|
|
"timeout", "nf+(4xx)", "http-r", "http-p", "http-t", "login-denied"]
|
|
endp_list = self.json_get("layer4/list?fields=%s" % ','.join(fields_list))
|
|
debug_info = {}
|
|
if endp_list is not None and endp_list['endpoint'] is not None:
|
|
endp_list = endp_list['endpoint']
|
|
expected_passes = len(endp_list)
|
|
passes = len(endp_list)
|
|
for item in range(len(endp_list)):
|
|
for name, info in endp_list[item].items():
|
|
for field in fields_list:
|
|
if info[field.replace("+", " ")] > 0:
|
|
passes -= 1
|
|
debug_info[name] = {field: info[field.replace("+", " ")]}
|
|
if debug:
|
|
print(debug_info)
|
|
if passes == expected_passes:
|
|
return True
|
|
else:
|
|
print(list(debug_info), " Endps in this list showed errors getting to its URL") # %s") % self.url)
|
|
return False
|
|
|
|
def start_cx(self):
|
|
print("Starting CXs...")
|
|
for cx_name in self.created_cx.keys():
|
|
self.json_post("/cli-json/set_cx_state", {
|
|
"test_mgr": "default_tm",
|
|
"cx_name": self.created_cx[cx_name],
|
|
"cx_state": "RUNNING"
|
|
}, debug_=self.debug)
|
|
print(".", end='')
|
|
print("")
|
|
|
|
def stop_cx(self):
|
|
print("Stopping CXs...")
|
|
for cx_name in self.created_cx.keys():
|
|
self.json_post("/cli-json/set_cx_state", {
|
|
"test_mgr": "default_tm",
|
|
"cx_name": self.created_cx[cx_name],
|
|
"cx_state": "STOPPED"
|
|
}, debug_=self.debug)
|
|
print(".", end='')
|
|
print("")
|
|
|
|
def cleanup(self):
|
|
print("Cleaning up cxs and endpoints")
|
|
if len(self.created_cx) != 0:
|
|
for cx_name in self.created_cx.keys():
|
|
req_url = "cli-json/rm_cx"
|
|
data = {
|
|
"test_mgr": "default_tm",
|
|
"cx_name": self.created_cx[cx_name]
|
|
}
|
|
self.json_post(req_url, data)
|
|
# pprint(data)
|
|
req_url = "cli-json/rm_endp"
|
|
data = {
|
|
"endp_name": cx_name
|
|
}
|
|
self.json_post(req_url, data)
|
|
# pprint(data)
|
|
|
|
def map_sta_ips(self, sta_list=[]):
|
|
for sta_eid in sta_list:
|
|
eid = self.local_realm.name_to_eid(sta_eid)
|
|
sta_list = self.json_get("/port/%s/%s/%s?fields=alias,ip" %
|
|
(eid[0], eid[1], eid[2]))
|
|
if sta_list['interface'] is not None:
|
|
self.ip_map[sta_list['interface']['alias']] = sta_list['interface']['ip']
|
|
|
|
def create(self, ports=[], sleep_time=.5, debug_=False, suppress_related_commands_=None, http=False, ftp=False,
|
|
https=False, user=None, passwd=None, source=None, ftp_ip=None, upload_name=None, http_ip=None, https_ip=None):
|
|
cx_post_data = []
|
|
self.map_sta_ips(ports)
|
|
print("Create CXs...")
|
|
|
|
for i in range(len(list(self.ip_map))):
|
|
url = None
|
|
if i != len(list(self.ip_map)) - 1:
|
|
port_name = list(self.ip_map)[i]
|
|
ip_addr = self.ip_map[list(self.ip_map)[i + 1]]
|
|
else:
|
|
port_name = list(self.ip_map)[i]
|
|
ip_addr = self.ip_map[list(self.ip_map)[0]]
|
|
|
|
if (ip_addr is None) or (ip_addr == ""):
|
|
raise ValueError("HTTPProfile::create encountered blank ip/hostname")
|
|
|
|
shelf = self.local_realm.name_to_eid(port_name)[0]
|
|
resource = self.local_realm.name_to_eid(port_name)[1]
|
|
name = self.local_realm.name_to_eid(port_name)[2]
|
|
|
|
if upload_name != None:
|
|
name = upload_name
|
|
|
|
if http:
|
|
if http_ip is not None:
|
|
self.port_util.set_http(port_name=name, resource=resource, on=True)
|
|
url = "%s http://%s %s" % (self.direction, http_ip, self.dest)
|
|
else:
|
|
self.port_util.set_http(port_name=name, resource=resource, on=True)
|
|
url = "%s http://%s/ %s" % (self.direction, ip_addr, self.dest)
|
|
if https:
|
|
if https_ip is not None:
|
|
self.port_util.set_http(port_name=name, resource=resource, on=True)
|
|
url = "%s https://%s %s" % (self.direction, https_ip, self.dest)
|
|
else:
|
|
self.port_util.set_http(port_name=name, resource=resource, on=True)
|
|
url = "%s https://%s/ %s" % (self.direction, ip_addr, self.dest)
|
|
|
|
if ftp:
|
|
self.port_util.set_ftp(port_name=name, resource=resource, on=True)
|
|
if user is not None and passwd is not None and source is not None:
|
|
if ftp_ip is not None:
|
|
ip_addr=ftp_ip
|
|
url = "%s ftp://%s:%s@%s%s %s" % (self.direction, user, passwd, ip_addr, source, self.dest)
|
|
print("###### url:{}".format(url))
|
|
else:
|
|
raise ValueError("user: %s, passwd: %s, and source: %s must all be set" % (user, passwd, source))
|
|
if not http and not ftp and not https:
|
|
raise ValueError("Please specify ftp and/or http")
|
|
|
|
if (url is None) or (url == ""):
|
|
raise ValueError("HTTPProfile::create: url unset")
|
|
|
|
if upload_name ==None:
|
|
endp_data = {
|
|
"alias": name + "_l4",
|
|
"shelf": shelf,
|
|
"resource": resource,
|
|
"port": name,
|
|
"type": "l4_generic",
|
|
"timeout": 10,
|
|
"url_rate": self.requests_per_ten,
|
|
"url": url,
|
|
"proxy_auth_type": 0x200
|
|
}
|
|
else:
|
|
endp_data = {
|
|
"alias": name + "_l4",
|
|
"shelf": shelf,
|
|
"resource": resource,
|
|
"port": ports[0],
|
|
"type": "l4_generic",
|
|
"timeout": 10,
|
|
"url_rate": self.requests_per_ten,
|
|
"url": url,
|
|
"ssl_cert_fname": "ca-bundle.crt",
|
|
"proxy_port": 0,
|
|
"max_speed": self.max_speed,
|
|
"proxy_auth_type": 0x200
|
|
}
|
|
url = "cli-json/add_l4_endp"
|
|
self.local_realm.json_post(url, endp_data, debug_=debug_,
|
|
suppress_related_commands_=suppress_related_commands_)
|
|
time.sleep(sleep_time)
|
|
|
|
endp_data = {
|
|
"alias": "CX_" + name + "_l4",
|
|
"test_mgr": "default_tm",
|
|
"tx_endp": name + "_l4",
|
|
"rx_endp": "NA"
|
|
}
|
|
cx_post_data.append(endp_data)
|
|
self.created_cx[name + "_l4"] = "CX_" + name + "_l4"
|
|
|
|
for cx_data in cx_post_data:
|
|
url = "/cli-json/add_cx"
|
|
self.local_realm.json_post(url, cx_data, debug_=debug_,
|
|
suppress_related_commands_=suppress_related_commands_)
|
|
time.sleep(sleep_time)
|