diff --git a/libs/tip_2x/SetupLibrary.py b/libs/tip_2x/SetupLibrary.py index 97b17648b..e1887aff7 100644 --- a/libs/tip_2x/SetupLibrary.py +++ b/libs/tip_2x/SetupLibrary.py @@ -67,7 +67,7 @@ class SetupLibrary: 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()) + # print(stdout.read()) client.close() diff --git a/libs/tip_2x/ap_lib.py b/libs/tip_2x/ap_lib.py index 0a708c68b..7fdce34f3 100644 --- a/libs/tip_2x/ap_lib.py +++ b/libs/tip_2x/ap_lib.py @@ -1,6 +1,8 @@ import importlib +import json import logging +import allure import paramiko import pytest @@ -49,20 +51,94 @@ class APLIBS: logging.error("Serial port not available. Exiting the Test") pytest.exit("Serial port not available. Please check your serial port connection") - def get_dut_logs(self, idx=0): - pass + def setup_serial_environment(self): + status = [] + for dut in self.device_under_tests_data: + if dut["method"] == "serial": + status.append(self.setup_library_objects[0].setup_serial_environment()) + if False in status: + logging.error("Serial port not available. Exiting the Test") + pytest.exit("Serial port not available. Please check your serial port connection") - def check_wan_connectivity(self, idx=0): - pass + def get_dut_logs(self, idx=0, print_log=True, attach_allure=True): + output = self.run_generic_command(cmd="logread", idx=idx, + print_log=print_log, + attach_allure=attach_allure, + expected_attachment_type=allure.attachment_type.TEXT) + return output - def get_uci_show(self, idx=0): - pass + def check_connectivity(self, idx=0, print_log=True, attach_allure=True): + maverick_status = self.run_generic_command(cmd="/etc/init.d/maverick status", idx=idx, + print_log=print_log, + attach_allure=attach_allure, + expected_attachment_type=allure.attachment_type.TEXT) + if maverick_status.__contains__("running"): + logging.error("Maverick is running!!!") + # Maverick check is happening + # TODO: add the steps to diagnose the maverick reason, check certificates, wan ip address, + # check the partition for certificates and check the /certificates dir. Also check if any of the cert is + # missing, if anything is missing, then report that. if everything looks good, please check md5sum of certs + # also, try to do a reboot in case everything looks good and post reboot, check the maverick status. + # Jitendra + pytest.exit("Maverick is running, Please check the wan connection and certificates") + check_iface = self.run_generic_command(cmd="ifconfig up0v0", idx=idx, + print_log=print_log, + attach_allure=attach_allure, + expected_attachment_type=allure.attachment_type.TEXT) + if check_iface.__contains__("error fetching interface information: Device not found"): + logging.error(check_iface) + pytest.exit("up0v0 interface is not available!!!") - def get_uci_show_ucentral(self, idx=0): - pass + output = self.run_generic_command(cmd="ip neigh show dev up0v0 REACHABLE", idx=idx, + print_log=print_log, + attach_allure=attach_allure, + expected_attachment_type=allure.attachment_type.TEXT) - def check_ubus_call_ucentral_status(self, idx=0): - pass + if output.__contains__("INCOMPLETE") or output.__contains__("FAILED"): + logging.error(output) + pytest.exit("up0v0 interface is failed to have connectivity!!!") + + def get_uci_show(self, param="ucentral", idx=0, print_log=True, attach_allure=True): + output = self.run_generic_command(cmd="uci show " + param, idx=idx, + print_log=print_log, + attach_allure=attach_allure, + expected_attachment_type=allure.attachment_type.TEXT) + + return output + + def ubus_call_ucentral_status(self, idx=0, print_log=True, attach_allure=True): + ret_val = dict.fromkeys(["connected", "latest", "active"]) + print(ret_val) + output = self.run_generic_command(cmd="ubus call ucentral status", idx=idx, + print_log=print_log, + attach_allure=attach_allure, + expected_attachment_type=allure.attachment_type.JSON) + try_again = False + try: + data = dict(json.loads(output.replace("\n\t", "").replace("\n", ""))) + except Exception as e: + logging.error("error in converting the ubus call ucentral status output to json" + output) + try_again = True + if try_again or len(data.keys()) != 3: + output = self.run_generic_command(cmd="ubus call ucentral status", idx=idx, + print_log=print_log, + attach_allure=attach_allure, + expected_attachment_type=allure.attachment_type.JSON) + try: + data = dict(json.loads(output.replace("\n\t", "").replace("\n", ""))) + except Exception as e: + logging.error("error in converting the ubus call ucentral status output to json" + output) + if data.keys().__contains__("connected"): + ret_val["connected"] = True + if data.keys().__contains__("disconnected"): + ret_val["connected"] = False + if data.keys().__contains__("latest"): + ret_val["latest"] = data.get("latest") + if data.keys().__contains__("active"): + ret_val["active"] = data.get("active") + + print(ret_val) + return ret_val def get_latest_config_recieved(self, idx=0): pass @@ -79,31 +155,38 @@ class APLIBS: def verify_certificates(self, idx=0): pass - def run_generic_command(self, cmd="", idx=0): - status = True - for dut in self.device_under_tests_data: - try: - client = self.ssh_cli_connect(ip=dut["host_ip"], - port=dut["host_ssh_port"], - username=dut["host_username"], - password=dut["host_password"], - timeout=10, - allow_agent=False, - banner_timeout=200) - cmd = cmd - if dut["method"] == "serial": - owrt_args = "--prompt root@" + dut["identifier"] + " -s serial --log stdout --user root --passwd openwifi" - cmd = f"cd ~/cicd-git/ && ./openwrt_ctl.py {owrt_args} -t {dut['serial_tty']} --action " \ - f"cmd --value \"{cmd}\" " - stdin, stdout, stderr = client.exec_command(cmd) - output = stdout.read() - status = output.decode('utf-8').splitlines() - print(output) - client.close() - except Exception as e: - print(e) - status = "Error" - return status + def run_generic_command(self, cmd="", idx=0, print_log=True, attach_allure=False, + expected_attachment_type=allure.attachment_type.TEXT): + input_command = cmd + try: + self.setup_library_objects[idx].kill_all_minicom_process( + tty=self.device_under_tests_data[idx]["serial_tty"]) + client = self.ssh_cli_connect(ip=self.device_under_tests_data[idx]["host_ip"], + port=self.device_under_tests_data[idx]["host_ssh_port"], + username=self.device_under_tests_data[idx]["host_username"], + password=self.device_under_tests_data[idx]["host_password"], + timeout=10, + allow_agent=False, + banner_timeout=200) + if self.device_under_tests_data[idx]["method"] == "serial": + owrt_args = "--prompt root@" + self.device_under_tests_data[idx][ + "identifier"] + ":~# " + " -s serial --log stdout --user root --passwd openwifi" + cmd = f"cd ~/cicd-git/ && ./openwrt_ctl.py {owrt_args} -t {self.device_under_tests_data[idx]['serial_tty']} --action " \ + f"cmd --value \"{cmd}\" " + stdin, stdout, stderr = client.exec_command(cmd) + output = stdout.read() + status = output.decode('utf-8').splitlines() + status.pop(0) + final_output = '\n'.join(status) + if print_log: + logging.info("Output for command: " + input_command + "\n" + final_output) + if attach_allure: + allure.attach(name=input_command, body=output, attachment_type=expected_attachment_type) + client.close() + except Exception as e: + logging.error(e) + final_output = "Error" + str(e) + return final_output def get_status(self, idx=0): pass @@ -141,7 +224,7 @@ if __name__ == '__main__': "method": "serial", # serial/ssh/telnet "host_ip": "10.28.3.103", "host_username": "lanforge", - "host_password": "pumpkin77", + "host_password": "pumpkin77", # Endurance@123 "host_ssh_port": 22, "serial_tty": "/dev/ttyAP1", "firmware_version": "next-latest" @@ -149,7 +232,8 @@ if __name__ == '__main__': "traffic_generator": {} } obj = APLIBS(dut_data=basic_1["device_under_tests"]) - obj.run_generic_command("uci show ucentral") + # obj.check_serial_connection() # obj.setup_serial_environment() - # obj.check_serial_connection(tty="/dev/ttyUSB0") - # obj.kill_all_minicom_process(tty="/dev/ttyUSB0") + # obj.run_generic_command("uci show ucentral") + # obj.get_dut_logs() + obj.ubus_call_ucentral_status()