From 810377b16e0249c4e3e6227840b4a4b9ad411dba Mon Sep 17 00:00:00 2001 From: SushantBawiskar Date: Sat, 24 Apr 2021 01:07:38 +0530 Subject: [PATCH] - integrated cv_commands.pywith cv_test_manager.py - create_chamberview_dut.py to create a dut - cv_dut_profile.py as library to create a dut - cv test manager can be used to implement chamber view also Signed-off-by: SushantBawiskar --- py-json/cv_commands.py | 104 -------------------- py-json/cv_dut_profile.py | 91 +++++++++++++++++ py-json/cv_test_manager.py | 72 ++++++++++++++ py-scripts/create_chamberview.py | 15 ++- py-scripts/create_chamberview_dut.py | 141 +++++++++++++++++++++++++++ 5 files changed, 316 insertions(+), 107 deletions(-) delete mode 100644 py-json/cv_commands.py create mode 100644 py-json/cv_dut_profile.py create mode 100644 py-scripts/create_chamberview_dut.py diff --git a/py-json/cv_commands.py b/py-json/cv_commands.py deleted file mode 100644 index ac78834b..00000000 --- a/py-json/cv_commands.py +++ /dev/null @@ -1,104 +0,0 @@ -""" -Note: This is a library file used to create a chamber view scenario. - import this file as showed in create_chamberview.py to create a scenario -""" - -import time - -# !/usr/bin/env python3 -# ---- ---- ---- ---- LANforge Base Imports ---- ---- ---- ---- -from LANforge.lfcli_base import LFCliBase - - -class chamberview(LFCliBase): - def __init__(self, - lfclient_host="localhost", - lfclient_port=8080, - ): - super().__init__(_lfjson_host=lfclient_host, - _lfjson_port=lfclient_port) - - #behaves same as chamberview manage scenario - def manage_cv_scenario(self, - scenario_name="Automation", - Resources="1.1", - Profile="STA-AC", - Amount="1", - DUT="DUT", - Dut_Radio="Radio-1" , - Uses1="wiphy0", - Uses2="AUTO", - Traffic="http", - Freq="-1", - VLAN=""): - req_url = "/cli-json/add_text_blob" - - text_blob = "profile_link" + " " + Resources + " " + Profile + " " + Amount + " " + "\'DUT:" + " " + DUT\ - + " " + Dut_Radio + "\' " + Traffic + " " + Uses1 + ","+Uses2 + " " + Freq + " " + VLAN - - data = { - "type": "Network-Connectivity", - "name": scenario_name, - "text": text_blob - } - - rsp = self.json_post(req_url, data) - time.sleep(2) - - - def pass_raw_lines_to_cv(self, - scenario_name="Automation", - Rawline=""): - req_url = "/cli-json/add_text_blob" - data = { - "type": "Network-Connectivity", - "name": scenario_name, - "text": Rawline - } - rsp = self.json_post(req_url, data) - time.sleep(2) - - def show_text_blob(self, config_name, blob_test_name, brief): - req_url = "/cli-json/show_text_blob" - data = {"type": "Plugin-Settings"} - if config_name and blob_test_name: - data["name"] = "%s%s" % (blob_test_name, config_name) # config name - else: - data["name"] = "ALL" - if brief: - data["brief"] = "brief" - return self.json_post(req_url, data) - - #This is for chamber view buttons - def apply_cv_scenario(self, cv_scenario): - cmd = "cv apply '%s'" % cv_scenario #To apply scenario - self.run_cv_cmd(cmd) - print("Applying %s scenario" % cv_scenario) - - def build_cv_scenario(self):#build chamber view scenario - cmd = "cv build" - self.run_cv_cmd(cmd) - print("Building scenario") - - def is_cv_build(self):#check if scenario is build - cmd = "cv is_built" - response = self.run_cv_cmd(cmd) - return self.check_reponse(response) - - def sync_cv(self):#sync - cmd = "cv sync" - self.run_cv_cmd(cmd) - - def run_cv_cmd(self, command):#Send chamber view commands - response_json = [] - req_url = "/gui-json/cmd" - data = { - "cmd": command - } - rsp = self.json_post(req_url, data, debug_=False, response_json_list_=response_json) - return response_json - - def check_reponse(self, response): - d1 = {k: v for e in response for (k, v) in e.items()} - return d1["LAST"]["response"] - diff --git a/py-json/cv_dut_profile.py b/py-json/cv_dut_profile.py new file mode 100644 index 00000000..6edd5072 --- /dev/null +++ b/py-json/cv_dut_profile.py @@ -0,0 +1,91 @@ + +import time + +# !/usr/bin/env python3 +# ---- ---- ---- ---- LANforge Base Imports ---- ---- ---- ---- +from LANforge.lfcli_base import LFCliBase + +class cv_dut(LFCliBase): + def __init__(self, + lfclient_host="localhost", + lfclient_port=8080, + ): + super().__init__(_lfjson_host=lfclient_host, + _lfjson_port=lfclient_port) + + def create_dut(self, + dut_name="DUT", + flags="4098", + ssid1="[BLANK]", + pass1="[BLANK]", + ssid2="[BLANK]", + pass2="[BLANK]", + ssid3="[BLANK]", + pass3="[BLANK]", + bssid1="00:00:00:00:00:00", + bssid2="00:00:00:00:00:00", + bssid3="00:00:00:00:00:00", + mgt_ip="0.0.0.0", + eap_id="[BLANK]"): + response_json = [] + req_url = "/cli-json/add_dut" + data = { + "name": dut_name, + "flags": flags, + "img_file": "NONE", + "sw_version": "[BLANK]", + "hw_version": "[BLANK]", + "model_num": "[BLANK]", + "serial_num": "[BLANK]", + "serial_port": "[BLANK]", + "wan_port": "[BLANK]", + "lan_port": "[BLANK]", + "ssid1": ssid1, + "passwd1": pass1, + "ssid2": ssid2, + "passwd2": pass2, + "ssid3": ssid3, + "passwd3": pass3, + "mgt_ip": mgt_ip, + "api_id": "0", + "flags_mask": "NA", + "antenna_count1": "0", + "antenna_count2": "0", + "antenna_count3": "0", + "bssid1": bssid1, + "bssid2": bssid2, + "bssid3": bssid3, + "top_left_x": "0", + "top_left_y": "0", + "eap_id": eap_id, + } + rsp = self.json_post(req_url, data, debug_=False, response_json_list_=response_json) + return rsp + + + def add_ssid(self, + dut_name="DUT", + ssid_idx=0, + ssid='[BLANK]', + passwd='[BLANK]', + bssid='00:00:00:00:00:00', + ssid_flags=0, + ssid_flags_mask=0xFFFFFFFF): + req_url = "/cli-json/add_dut_ssid" + print( "name:" + dut_name, + "ssid_idx:"+ ssid_idx, + "ssid:"+ ssid, + "passwd:" + passwd, + "bssid:" + bssid, + "ssid_flags:" + str(ssid_flags), + "ssid_flags_mask:"+ str(ssid_flags_mask)) + + self.json_post(req_url, { + "name": dut_name, + "ssid_idx": ssid_idx, + "ssid": ssid, + "passwd": passwd, + "bssid": bssid, + "ssid_flags": ssid_flags, + "ssid_flags_mask": ssid_flags_mask, + }) diff --git a/py-json/cv_test_manager.py b/py-json/cv_test_manager.py index 62378f45..f7500bea 100644 --- a/py-json/cv_test_manager.py +++ b/py-json/cv_test_manager.py @@ -383,3 +383,75 @@ class cv_test(Realm): csvtoinflux.post_to_influx() print("All done posting to influx.\n") + + #************************** chamber view ************************** + def add_text_blob_line(self, + scenario_name="Automation", + Resources="1.1", + Profile="STA-AC", + Amount="1", + DUT="DUT", + Dut_Radio="Radio-1", + Uses1="wiphy0", + Uses2="AUTO", + Traffic="http", + Freq="-1", + VLAN=""): + req_url = "/cli-json/add_text_blob" + + text_blob = "profile_link" + " " + Resources + " " + Profile + " " + Amount + " " + "\'DUT:" + " " + DUT \ + + " " + Dut_Radio + "\' " + Traffic + " " + Uses1 + "," + Uses2 + " " + Freq + " " + VLAN + + data = { + "type": "Network-Connectivity", + "name": scenario_name, + "text": text_blob + } + + rsp = self.json_post(req_url, data) + + def pass_raw_lines_to_cv(self, + scenario_name="Automation", + Rawline=""): + req_url = "/cli-json/add_text_blob" + data = { + "type": "Network-Connectivity", + "name": scenario_name, + "text": Rawline + } + rsp = self.json_post(req_url, data) + time.sleep(2) + + # This is for chamber view buttons + + def apply_cv_scenario(self, cv_scenario): + cmd = "cv apply '%s'" % cv_scenario # To apply scenario + self.run_cv_cmd(cmd) + print("Applying %s scenario" % cv_scenario) + + def build_cv_scenario(self): # build chamber view scenario + cmd = "cv build" + self.run_cv_cmd(cmd) + print("Building scenario") + + def get_cv_build_status(self): # check if scenario is build + cmd = "cv is_built" + response = self.run_cv_cmd(cmd) + return self.check_reponse(response) + + def sync_cv(self): # sync + cmd = "cv sync" + print(self.run_cv_cmd(cmd)) + + def run_cv_cmd(self, command): # Send chamber view commands + response_json = [] + req_url = "/gui-json/cmd" + data = { + "cmd": command + } + rsp = self.json_post(req_url, data, debug_=False, response_json_list_=response_json) + return response_json + + def check_reponse(self, response): + d1 = {k: v for e in response for (k, v) in e.items()} + return d1["LAST"]["response"] diff --git a/py-scripts/create_chamberview.py b/py-scripts/create_chamberview.py index 1187003d..dab33415 100755 --- a/py-scripts/create_chamberview.py +++ b/py-scripts/create_chamberview.py @@ -46,7 +46,7 @@ if sys.version_info[0] != 3: if 'py-json' not in sys.path: sys.path.append(os.path.join(os.path.abspath('..'), 'py-json')) -from cv_commands import chamberview as cv +from cv_test_manager import cv_test as cv class CreateChamberview(cv): def __init__(self, @@ -126,7 +126,7 @@ class CreateChamberview(cv): else: continue - self.manage_cv_scenario(scenario_name, + self.add_text_blob_line(scenario_name, Resource, Profile, Amount, @@ -151,7 +151,16 @@ class CreateChamberview(cv): self.show_text_blob(None, None, False) # Show changes on GUI self.apply_cv_scenario(scenario_name) # Apply scenario self.build_cv_scenario() # build scenario - while self.is_cv_build() == "NO": continue # wait for cv to build + tries = 0 + while (True): + if not self.get_cv_is_built(): + print("Waiting %i/60 for Chamber-View to be built." % (tries)) + tries += 1 + if (tries > 60): + break + time.sleep(1) + else: + break print("completed building %s scenario" %scenario_name) diff --git a/py-scripts/create_chamberview_dut.py b/py-scripts/create_chamberview_dut.py new file mode 100644 index 00000000..f27195df --- /dev/null +++ b/py-scripts/create_chamberview_dut.py @@ -0,0 +1,141 @@ +import sys +import os +import argparse +import time +import re + +if sys.version_info[0] != 3: + print("This script requires Python 3") + exit(1) + +if 'py-json' not in sys.path: + sys.path.append(os.path.join(os.path.abspath('..'), 'py-json')) + +from cv_dut_profile import cv_dut as dut +from cv_test_manager import cv_test as cvtest + +class DUT(dut): + def __init__(self, + lfmgr="localhost", + port="8080", + dut_name="DUT", + ssid=[], + password=[], + bssid=[], + security=[] + ): + super().__init__( + lfclient_host=lfmgr, + lfclient_port=port, + ) + self.cv = cvtest(lfmgr, port) + self.dut_name = dut_name + self.ssid = ssid + self.password = password + self.bssid = bssid + self.security = security + + def setup(self): + self.create_dut(dut_name=self.dut_name) + + def add_ssids(self): + if self.ssid: + + for i in range(len(self.ssid)): + + if " " in self.ssid[i][0]: + self.ssid[i][0] = (re.split(' ', self.ssid[i][0])) + elif "," in self.ssid[i][0]: + self.ssid[i][0] = (re.split(',', self.ssid[i][0])) + elif ", " in self.ssid[i][0]: + self.ssid[i][0] = (re.split(',', self.ssid[i][0])) + elif " ," in self.ssid[i][0]: + self.ssid[i][0] = (re.split(',', self.ssid[i][0])) + else: + print("Wrong arguments entered !") + exit(1) + + ssid_idx = 0 + ssid = "[BLANK]" + passwd = "[BLANK]" + bssid = "00:00:00:00:00:00" + flag = 0x0 + + for j in range(len(self.ssid[i][0])): + self.ssid[i][0][j] = self.ssid[i][0][j].split("=") + for k in range(len(self.ssid[i][0][j])): + name = self.ssid[i][0][j][k] + if str(name) == "SSID" or str(name) == "ssid" or str(name) == "s": + ssid = self.ssid[i][0][j][k + 1] + elif str(name) == "PASSWORD" or str(name) == "password" or str(name) == "pass": + passwd = self.ssid[i][0][j][k + 1] + elif str(name) == "ssid_idx" or str(name) == "no" or str(name) == "N": + ssid_idx = self.ssid[i][0][j][k + 1] + elif str(name) == "security" or str(name) == "sec": + if self.ssid[i][0][j][k + 1]: + all_flags = self.ssid[i][0][j][k + 1].split("|") + for flags in all_flags: + if flags == "WEP" or flags == "wep": + flag += 0x8 + if flags == "WPA" or flags == "wpa": + flag += 0x10 + if flags == "WPA2" or flags == "wpa2": + flag += 0x20 + if flags == "WPA3" or flags == "wpa3": + flag += 0x100 + if flags == "11r": + flag += 0x200 + if flags == "EAP-TTLS": + flag += 0x400 + if flags == "EAP-PEAP": + flag += 0x800 + elif str(name) == "BSSID" or str(name) == "bssid" or str(name) == "B": + bssid = self.ssid[i][0][j][k + 1] + else: + continue + + self.add_ssid(dut_name=self.dut_name, + ssid_idx=ssid_idx, + ssid=ssid, + passwd=passwd, + bssid=bssid, + ssid_flags=flag, + ssid_flags_mask=0xFFFFFFFF + ) + + + +def main(): + parser = argparse.ArgumentParser( + description=""" + ./create_chamberview_dut -m "localhost" -o "8080" -d "dut_name" + -s "ssid_idx=0 ssid=NET1 security=WPA|WEP|11r|EAP-PEAP bssid=78:d2:94:bf:16:41" + -s "ssid_idx=1 ssid=NET1 security=WPA password=test bssid=78:d2:94:bf:16:40" + """) + parser.add_argument("-m", "--lfmgr", type=str, default="localhost", + help="address of the LANforge GUI machine (localhost is default)") + parser.add_argument("-o", "--port", type=str, default="8080", + help="IP Port the LANforge GUI is listening on (8080 is default)") + parser.add_argument("-d", "--dut_name", type=str, default="DUT", + help="set dut name") + parser.add_argument("-s", "--ssid", action='append', nargs=1, + help="SSID", default=[]) + + + args = parser.parse_args() + new_dut = DUT(args.lfmgr, + args.port, + args.dut_name, + args.ssid, + ) + + new_dut.setup() + new_dut.add_ssids() + cv =cvtest(args.lfmgr, + args.port) + cv.show_text_blob(None, None, False) # Show changes on GUI + cv.sync_cv() + time.sleep(2) + +if __name__ == "__main__": + main()