Base setup for tip_2x package and standard format for wlan-testing

Signed-off-by: shivam <shivam.thakur@candelatech.com>
This commit is contained in:
shivam
2022-07-20 15:13:25 +05:30
committed by haricharan-jaka
parent 98b9bde0ff
commit 3bd86de755
8 changed files with 3474 additions and 0 deletions

79
libs/SetupLibrary.py Normal file
View File

@@ -0,0 +1,79 @@
import logging
import paramiko
from scp import SCPClient
logging.basicConfig(format='%(asctime)s - %(message)s', level=logging.INFO)
class SetupLibrary:
def __init__(self, remote_ip="",
remote_ssh_port=22,
remote_ssh_username="lanforge",
remote_ssh_password="lanforge",
pwd="",
):
self.pwd = pwd
self.remote_ip = remote_ip
self.remote_ssh_username = remote_ssh_username
self.remote_ssh_password = remote_ssh_password
self.remote_ssh_port = remote_ssh_port
def setup_serial_environment(self):
client = self.ssh_cli_connect()
cmd = '[ -f ~/cicd-git/ ] && echo "True" || echo "False"'
stdin, stdout, stderr = client.exec_command(cmd)
output = str(stdout.read())
if output.__contains__("False"):
cmd = 'mkdir ~/cicd-git/'
client.exec_command(cmd)
cmd = '[ -f ~/cicd-git/openwrt_ctl.py ] && echo "True" || echo "False"'
stdin, stdout, stderr = client.exec_command(cmd)
output = str(stdout.read())
if output.__contains__("False"):
print("Copying openwrt_ctl serial control Script...")
with SCPClient(client.get_transport()) as scp:
scp.put(self.pwd + 'openwrt_ctl.py', '~/cicd-git/openwrt_ctl.py') # Copy my_file.txt to the server
cmd = '[ -f ~/cicd-git/openwrt_ctl.py ] && echo "True" || echo "False"'
stdin, stdout, stderr = client.exec_command(cmd)
var = str(stdout.read())
client.close()
def ssh_cli_connect(self):
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
logging.info("Trying SSH Connection to: " + str(self.remote_ip) +
" on port: " + str(self.remote_ssh_port) +
" with username: " + str(self.remote_ssh_username) +
" and password: " + str(self.remote_ssh_password))
client.connect(self.remote_ip, username=self.remote_ssh_username, password=self.remote_ssh_password,
port=self.remote_ssh_port, timeout=10, allow_agent=False, banner_timeout=200)
return client
def check_serial_connection(self, tty="/dev/ttyUSB0"):
client = self.ssh_cli_connect()
cmd = 'ls /dev/tty*'
stdin, stdout, stderr = client.exec_command(cmd)
output = str(stdout.read().decode('utf-8'))
client.close()
available_tty_ports = output.split("\n")
if tty in available_tty_ports:
logging.info("Expected Serial Console Port is available on Remote system: " + tty)
return tty in available_tty_ports
def kill_all_minicom_process(self, tty="/dev/ttyUSB0"):
client = self.ssh_cli_connect()
stdin, stdout, stderr = client.exec_command("fuser -k " + tty)
print(stdout.read())
client.close()
if __name__ == '__main__':
obj = SetupLibrary(remote_ip="192.168.52.89",
remote_ssh_port=22,
pwd="")
obj.setup_serial_environment()
obj.check_serial_connection(tty="/dev/ttyUSB0")
obj.kill_all_minicom_process(tty="/dev/ttyUSB0")

0
libs/__init__.py Normal file
View File

385
libs/openwrt_ctl.py Executable file
View File

@@ -0,0 +1,385 @@
#!/usr/bin/python3
'''
make sure pexpect is installed:
$ sudo yum install python3-pexpect
You might need to install pexpect-serial using pip:
$ pip3 install serial
$ pip3 install pexpect-serial
./openwrt_ctl.py -l stdout -u root -p TIP -s serial --tty ttyUSB0
# Set up reverse ssh tunnel
./openwrt_ctl.py --tty /dev/ttyAP1 --action ssh-tunnel \
--value "ssh -y -y -f -N -T -M -R 9999:localhost:22 lanforge@10.28.3.100" \
--value2 password-for-10.28.3.100 --log stdout --scheme serial --prompt root@Open
'''
import sys
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit()
try:
import importlib
re = importlib.import_module("re")
logging = importlib.import_module("logging")
time = importlib.import_module("time")
sleep = time.sleep
pexpect = importlib.import_module("pexpect")
serial = importlib.import_module("serial")
pexpect_serial = importlib.import_module("pexpect_serial")
SerialSpawn = pexpect_serial.SerialSpawn
pprint = importlib.import_module("pprint")
telnetlib = importlib.import_module("telnetlib")
argparse = importlib.import_module("argparse")
except ImportError as e:
print(e)
sys.exit("Python Import Error: " + str(e))
default_host = "localhost"
default_ports = {
"serial": None,
"ssh": 22,
"telnet": 23
}
NL = "\n"
CR = "\r\n"
Q = '"'
A = "'"
FORMAT = '%(asctime)s %(name)s %(levelname)s: %(message)s'
prompt = "root@OpenWrt:"
def usage():
print("$0 used connect to OpenWrt AP or similar Linux machine:")
print("-d|--dest IP address of the OpenWrt AP, for ssh/telnet scheme")
print("-o|--port IP port of the OpenWrt AP, for ssh/telnet scheme")
print("-t|--tty Serial port, if using serial scheme")
print("-u|--user login name")
print("-p|--pass password")
print("--prompt Prompt to look for when commands are done (default: root@OpenWrt)")
print("-s|--scheme (serial|telnet|ssh): connect via serial, ssh or telnet")
print("-l|--log file log messages here")
print("--action (logread | journalctl | lurk | sysupgrade | download | upload | reboot | cmd | ssh-tunnel")
print("--value (option to help complete the action")
print("--value2 (option to help complete the action, dest filename for download, passwd for ssh-tunnel")
print("-h|--help")
# see https://stackoverflow.com/a/13306095/11014343
class FileAdapter(object):
def __init__(self, logger):
self.logger = logger
def write(self, data):
# NOTE: data can be a partial line, multiple lines
data = data.strip() # ignore leading/trailing whitespace
if data: # non-blank
self.logger.info(data)
def flush(self):
pass # leave it to logging to flush properly
def main():
global prompt
parser = argparse.ArgumentParser(description="OpenWrt AP Control Script")
parser.add_argument("-d", "--dest", type=str, help="address of the cisco controller_tests")
parser.add_argument("-o", "--port", type=int, help="control port on the controller_tests")
parser.add_argument("-u", "--user", type=str, help="credential login/username")
parser.add_argument("-p", "--passwd", type=str, help="credential password")
parser.add_argument("-P", "--prompt", type=str, help="Prompt to look for")
parser.add_argument("-s", "--scheme", type=str, choices=["serial", "ssh", "telnet"],
help="Connect via serial, ssh or telnet")
parser.add_argument("-t", "--tty", type=str, help="tty serial device")
parser.add_argument("-l", "--log", type=str, help="logfile for messages, stdout means output to console")
parser.add_argument("--action", type=str, help="perform action",
choices=["logread", "journalctl", "lurk", "sysupgrade", "sysupgrade-n", "download", "upload",
"reboot", "cmd", "ssh-tunnel"])
parser.add_argument("--value", type=str, help="set value")
parser.add_argument("--value2", type=str, help="set value2")
tty = None
args = None
try:
args = parser.parse_args()
host = args.dest
scheme = args.scheme
port = args.port
# port = (default_ports[scheme], args.port)[args.port != None]
user = args.user
passwd = args.passwd
logfile = args.log
tty = args.tty;
if (args.prompt != None):
prompt = args.prompt
filehandler = None
except Exception as e:
logging.exception(e);
usage()
exit(2);
console_handler = logging.StreamHandler()
formatter = logging.Formatter(FORMAT)
logg = logging.getLogger(__name__)
logg.setLevel(logging.DEBUG)
file_handler = None
if (logfile is not None):
if (logfile != "stdout"):
file_handler = logging.FileHandler(logfile, "w")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
logg.addHandler(file_handler)
logging.basicConfig(format=FORMAT, handlers=[file_handler])
else:
# stdout logging
logging.basicConfig(format=FORMAT, handlers=[console_handler])
CCPROMPT = prompt
ser = None
egg = None # think "eggpect"
try:
if (scheme == "serial"):
# eggspect = pexpect.fdpexpect.fdspan(telcon, logfile=sys.stdout.buffer)
ser = serial.Serial(tty, 115200, timeout=5)
egg = SerialSpawn(ser);
egg.logfile = FileAdapter(logg)
egg.sendline(NL)
has_reset = False
try:
logg.info("prompt: %s user: %s passwd: %s" % (prompt, user, passwd))
while True:
i = egg.expect([prompt, "Please press Enter to activate", "login:", "Password:", "IPQ6018#"],
timeout=3)
logg.info("expect-0: %i" % (i))
if (i == 0):
logg.info("Found prompt, login complete.")
break
if (i == 1):
logg.info("Sending newline")
egg.setdline(NL)
if (i == 2):
logg.info("Sending username: %s" % (user))
egg.sendline(user)
if (i == 3):
logg.info("Sending password: %s" % (passwd))
egg.sendline(passwd)
if (i == 4): # in bootloader
if has_reset:
logg.info("ERROR: Have reset once already, back in bootloader?")
sys.exit(1)
has_reset = True
logg.info("In boot loader, will reset and sleep 30 seconds")
egg.sendline("reset")
time.sleep(30)
egg.sendline(NL)
except Exception as e:
# maybe something like 'logread -f' is running?
# send ctrl-c
egg.send(chr(3))
elif (scheme == "ssh"):
# Not implemented/tested currently. --Ben
if (port is None):
port = 22
cmd = "ssh -p%d %s@%s" % (port, user, host)
logg.info("Spawn: " + cmd + NL)
egg = pexpect.spawn(cmd)
# egg.logfile_read = sys.stdout.buffer
egg.logfile = FileAdapter(logg)
i = egg.expect(["password:", "continue connecting (yes/no)?"], timeout=3)
time.sleep(0.1)
if i == 1:
egg.sendline('yes')
egg.expect('password:')
sleep(0.1)
egg.sendline(passwd)
elif (scheme == "telnet"):
# Not implemented/tested currently. --Ben
if (port is None):
port = 23
cmd = "telnet %s %d" % (host, port)
logg.info("Spawn: " + cmd + NL)
egg = pexpect.spawn(cmd)
egg.logfile = FileAdapter(logg)
time.sleep(0.1)
egg.sendline(' ')
egg.expect('User\:')
egg.sendline(user)
egg.expect('Password\:')
egg.sendline(passwd)
egg.sendline('config paging disable')
else:
usage()
exit(1)
except Exception as e:
logging.exception(e);
command = None
CLOSEDBYREMOTE = "closed by remote host."
CLOSEDCX = "Connection to .* closed."
try:
egg.expect(CCPROMPT)
except Exception as e:
egg.sendline(NL)
TO = 10
wait_forever = False
# Clean pending output
egg.sendline("echo __hello__")
egg.expect("__hello__")
egg.expect(CCPROMPT)
logg.info("Action[%s] Value[%s] Value2[%s]" % (args.action, args.value, args.value2))
if (args.action == "reboot"):
command = "reboot"
TO = 60
if (args.action == "cmd"):
if (args.value is None):
raise Exception("cmd requires value to be set.")
command = "%s" % (args.value)
if (args.action == "logread"):
command = "logread -f"
TO = 1
wait_forever = True
if (args.action == "journalctl"):
command = "journalctl -f"
TO = 1
wait_forever = True
if (args.action == "lurk"):
command = "date"
TO = 1
wait_forever = True
if (args.action == "ssh-tunnel"):
command = "%s" % (args.value)
passwd = "%s" % (args.value2)
logg.info("Command[%s]" % command)
egg.sendline(command);
i = egg.expect(["password:", "Do you want to continue connecting"], timeout=5)
if i == 1:
egg.sendline("y")
egg.expect("password:", timeout=5)
egg.sendline(passwd)
egg.expect(CCPROMPT, timeout=20)
return
if ((args.action == "sysupgrade") or (args.action == "sysupgrade-n")):
command = "scp %s /tmp/new_img.bin" % (args.value)
logg.info("Command[%s]" % command)
egg.sendline(command);
i = egg.expect(["password:", "Do you want to continue connecting"], timeout=5)
if i == 1:
egg.sendline("y")
egg.expect("password:", timeout=5)
egg.sendline("lanforge")
egg.expect(CCPROMPT, timeout=20)
if (args.action == "sysupgrade-n"):
egg.sendline("sysupgrade -n /tmp/new_img.bin")
else:
egg.sendline("sysupgrade /tmp/new_img.bin")
egg.expect("link becomes ready", timeout=100)
return
if (args.action == "download"):
command = "scp %s /tmp/%s" % (args.value, args.value2)
logg.info("Command[%s]" % command)
egg.sendline(command);
i = egg.expect(["password:", "Do you want to continue connecting", "Network unreachable"], timeout=5)
if i == 2:
print("Network unreachable, wait 15 seconds and try again.")
time.sleep(15)
command = "scp %s /tmp/%s" % (args.value, args.value2)
logg.info("Command[%s]" % command)
egg.sendline(command);
i = egg.expect(["password:", "Do you want to continue connecting", "Network unreachable"], timeout=5)
if i == 2:
print("ERROR: Could not connect to LANforge to get download file")
exit(2)
if i == 1:
egg.sendline("y")
egg.expect("password:", timeout=5)
egg.sendline("lanforge")
egg.expect(CCPROMPT, timeout=20)
return
if (args.action == "upload"):
command = "scp %s %s" % (args.value, args.value2)
logg.info("Command[%s]" % command)
egg.sendline(command);
i = egg.expect(["password:", "Do you want to continue connecting", "Network unreachable"], timeout=5)
if i == 2:
print("Network unreachable, wait 15 seconds and try again.")
time.sleep(15)
command = "scp /tmp/%s %s" % (args.value, args.value2)
logg.info("Command[%s]" % command)
egg.sendline(command);
i = egg.expect(["password:", "Do you want to continue connecting", "Network unreachable"], timeout=5)
if i == 2:
print("ERROR: Could not connect to LANforge to put upload file")
exit(2)
if i == 1:
egg.sendline("y")
egg.expect("password:", timeout=5)
egg.sendline("lanforge")
egg.expect(CCPROMPT, timeout=20)
return
if (command is None):
logg.info("No command specified, going to log out.")
else:
logg.info("Command[%s]" % command)
egg.sendline(command);
while True:
try:
i = egg.expect([CCPROMPT, "kmodloader: done loading kernel", "\n"], timeout=TO)
print(egg.before.decode('utf-8', 'ignore'))
if i == 1:
egg.sendline(' ')
egg.expect(CCPROMPT, timeout=20)
print(egg.before.decode('utf-8', 'ignore'))
if i == 2: # new line of text, just print and continue
continue
if not wait_forever:
break
except Exception as e:
# Some commands take a long time (logread -f)
if not wait_forever:
logging.exception(e)
break
# ----- ----- ----- ----- ----- ----- ----- ----- ----- ----- ----- -----
if __name__ == '__main__':
main()
####
####
####

0
libs/tip_1x/__init__.py Normal file
View File

1418
libs/tip_1x/controller.py Normal file

File diff suppressed because it is too large Load Diff

120
libs/tip_2x/__init__.py Normal file
View File

@@ -0,0 +1,120 @@
"""
Telecom Infra Project OpenWifi 2.X (Ucentral libraries for Test Automation)
"""
import importlib
logging = importlib.import_module("logging")
ap_lib = importlib.import_module("ap_lib")
controller = importlib.import_module("controller")
"""
Custom Class Imports needed for OpenWifi 2.X
"""
ConfigureController = controller.ConfigureController
Controller = controller.Controller
FMSUtils = controller.FMSUtils
ProvUtils = controller.ProvUtils
UProfileUtility = controller.UProfileUtility
APLIBS = ap_lib.APLIBS
"""
Setting default Logging
"""
class tip_2x:
"""
Standard OpenWifi wlan-testing specific variables
"""
controller_data = {}
device_under_tests_info = []
"""
OpenWifi 2.x Specific Variables that will be only scoped in tip_2x Library
"""
ow_sec_url = ""
ow_sec_login_username = ""
ow_sec_login_password = ""
controller_library_object = object()
dut_library_object = object()
def __init__(self, controller_data=None,
device_under_tests_info=[], logging_level=logging.INFO):
logging.basicConfig(format='%(asctime)s - %(message)s', level=logging_level)
if controller_data is None:
controller_data = {}
self.controller_data = controller_data
self.device_under_tests_info = device_under_tests_info
self.setup_metadata()
"""
Controller and Access Point specific metadata that is related to OpenWifi 2.x
"""
def setup_metadata(self):
logging.info("setting up the Controller metadata for tip_2x Library: " + str(self.controller_data))
logging.info("setting up the DUT metadata for tip_2x Library: " + str(self.device_under_tests_info))
logging.info("Number of DUT's configured: " + str(len(self.device_under_tests_info)))
self.ow_sec_url = self.controller_data["url"]
self.ow_sec_login_username = self.controller_data["username"]
self.ow_sec_login_password = self.controller_data["password"]
def setup_objects(self):
self.controller_library_object = Controller()
self.dut_library_object = APLIBS()
""" Standard getter methods. Should be available for all type of libraries. Commonly used by wlan-testing"""
def get_dut_library_object(self):
return self.dut_library_object
def get_controller_library_object(self):
return self.controller_library_object
def get_controller_data(self):
return self.controller_data
def get_device_under_tests_info(self):
return self.device_under_tests_info
def get_number_of_dut(self):
return len(self.device_under_tests_info)
def get_dut_logs(self, dut_idx=0):
return self.dut_library_object.get_logs(idx=0)
def get_controller_logs(self):
return ""
if __name__ == '__main__':
basic_1 = {
"target": "tip_2x",
"controller": {
"url": "https://sec-qa01.cicd.lab.wlan.tip.build:16001",
"username": "tip@ucentral.com",
"password": "OpenWifi%123"
},
"device_under_tests": [{
"model": "wallys_dr40x9",
"mode": "wifi5",
"serial": "c44bd1005b30",
"jumphost": True,
"ip": "10.28.3.100",
"username": "lanforge",
"password": "pumpkin77",
"port": 22,
"serial_tty": "/dev/ttyAP8",
"version": "next-latest"
}],
"traffic_generator": {}
}
var = tip_2x(controller_data=basic_1["controller"], device_under_tests_info=basic_1["device_under_tests"])

11
libs/tip_2x/ap_lib.py Normal file
View File

@@ -0,0 +1,11 @@
import importlib
import sys
sys.path.append('..')
SetupLibrary = importlib.import_module("SetupLibrary")
class APLIBS:
def __init__(self):
pass

1461
libs/tip_2x/controller.py Normal file

File diff suppressed because it is too large Load Diff