1#!/usr/bin/env python3 2# coding=utf-8 3 4# 5# Copyright (c) 2020 Huawei Device Co., Ltd. 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17# 18 19import os 20import subprocess 21import sys 22import re 23from xdevice import platform_logger 24from xdevice import EnvironmentManager 25from xdevice import ResultReporter 26from xdevice import ExecInfo 27 28LOG = platform_logger("distribute_utils") 29 30 31def execute_distribute_test_file(py_test_file, result_rootpath): 32 if os.path.exists(py_test_file): 33 environ_dic = dict(os.environ) 34 environ_dic["PYTEST_PYTESTPATH"] = sys.pytest_dir 35 environ_dic["PYTEST_RESULTPATH"] = result_rootpath 36 37 proc = subprocess.Popen([sys.executable, py_test_file], 38 stdout=subprocess.PIPE, 39 stderr=subprocess.PIPE, 40 shell=False, 41 env=environ_dic) 42 43 try: 44 while proc.poll() is None: 45 line = proc.stdout.readline() 46 line = line.strip() 47 if isinstance(line, bytes): 48 line = line.decode('utf-8', 'ignore') 49 if line != "": 50 LOG.info(line) 51 finally: 52 data = proc.stdout.read() 53 if isinstance(data, bytes): 54 data = data.decode('utf-8', 'ignore') 55 if data != "": 56 LOG.info(data.rstrip("\n")) 57 58 data = proc.stderr.read() 59 if isinstance(data, bytes): 60 data = data.decode('utf-8', 'ignore') 61 if data != "": 62 error_message = data.rstrip("\n") 63 LOG.info(error_message) 64 65 proc.stdout.close() 66 proc.stderr.close() 67 return 68 69 70def make_device_info_file(tmp_path): 71 env_manager = EnvironmentManager() 72 device_info_file_path = os.path.join(tmp_path, 73 "device_info_file.txt") 74 with open(device_info_file_path, "w") as file_handle: 75 if env_manager.managers is not None: 76 for device in list(env_manager.managers.values())[1].devices_list: 77 if device.test_device_state.value == "ONLINE": 78 status = device.label if device.label else 'None' 79 LOG.info("%s,%s" % (device.device_sn, status)) 80 file_handle.write("%s,%s,%s,%s\n" % ( 81 device.device_sn, 82 device.label if device.label else 'None', 83 device.host, 84 device.port)) 85 return 86 if env_manager.manager_lite is not None: 87 for device in env_manager.manager_lite.devices_list: 88 if device.test_device_state.value == "ONLINE": 89 status = device.label if device.label else 'None' 90 LOG.info("%s,%s" % (device.device_sn, status)) 91 file_handle.write("%s,%s%s,%s\n" % ( 92 device.device_sn, 93 device.label if device.label else 'None', 94 device.host, 95 device.port)) 96 return 97 98 99def make_reports(result_rootpath, start_time): 100 exec_info = ExecInfo() 101 exec_info.test_type = "systemtest" 102 exec_info.device_name = "ALL" 103 exec_info.host_info = "" 104 exec_info.test_time = start_time 105 exec_info.log_path = os.path.join(result_rootpath, "log") 106 exec_info.platform = "ALL" 107 exec_info.execute_time = "" 108 109 result_report = ResultReporter() 110 result_report.__generate_reports__(report_path=result_rootpath, 111 task_info=exec_info) 112 return True 113 114 115def check_ditributetest_environment(): 116 env_manager = EnvironmentManager() 117 devices_list = list(env_manager.managers.values())[1].devices_list 118 evn_status = True 119 if len(devices_list) == 0: 120 LOG.error("no devices online") 121 return False 122 device_zero = devices_list[0] 123 for device in devices_list: 124 if device != device_zero: 125 device_ip = query_device_ip(device) 126 if device_ip == "": 127 evn_status = False 128 continue 129 if not check_zdn_network(device, device_ip): 130 LOG.error("device_%s ping devices_%s failed" % ( 131 device_zero.device_sn, device.device_sn)) 132 evn_status = False 133 if not evn_status: 134 LOG.error("Environment status = False, test end") 135 return False 136 LOG.info("Environment status == True") 137 return True 138 139 140def check_zdn_network(device, device_ip=""): 141 command = "ping -c 3 " + device_ip 142 output = device.execute_shell_command(command) 143 if "" == output: 144 return False 145 iserror = re.findall(r"error", output) 146 if 0 != len(iserror): 147 return False 148 packet_lose = re.findall(r"\d+%", output) 149 LOG.info("packet lose=%s " % packet_lose[0]) 150 if "0%" == packet_lose[0]: 151 return True 152 return False 153 154 155def query_device_ip(device): 156 output = device.execute_shell_command("getprop ro.hardware") 157 if output == "": 158 return "" 159 160 isemulator = re.findall(r"ranchu", output) 161 output = device.execute_shell_command("ifconfig") 162 if output == "": 163 return "" 164 165 if 0 != len(isemulator): 166 ipaddress = re.findall(r"\b10\.0\.2\.[0-9]{1,3}\b", output) 167 else: 168 ip_template = r"\b192\.168\.(?:[0-9]{1,3}\.)[0-9]{1,3}\b" 169 ipaddress = re.findall(ip_template, output) 170 if len(ipaddress) == 0: 171 LOG.error("get device_%s ip fail,please check the net" 172 % device.device_sn) 173 return "" 174 return ipaddress[0] 175