#!/usr/bin/python3 # coding=utf-8 linux_path = "/opt/test" windows_path = "c:/opt/test" import sys import os os_platform = (sys.platform).lower() if os_platform == "linux": sys.path.append(linux_path) work_path = "{}/automation/control".format(linux_path) else: sys.path.append(windows_path) work_path = "{}/automation/control".format(windows_path) os.chdir(work_path) import json import shutil import time import copy import configparser import business import argparse try: import xml.etree.cElementTree as ET except ImportError: import xml.etree.ElementTree as ET ''' 命名规范参照：https://www.jianshu.com/p/a793c0d960fe 自动化执行入口， 参数： instructNo：唯一执行标识； executeModule：执行模式：standalone独立执行，独立执行模式无需提交执行测试用例列表及执行参数；fusion控制模式； 测试客户端除移动端为执行ip，其他入参为管理端ip，根据管理端ip和参数类型获取对应的执行ip：获取ip方法已提供，但是需要和类型配合以获取对应的执行ip ''' def executeAutomation(instructNo, runMode, onlyDryrun, kill): """ 程序入口的主函数，只接收fusion的参数执行instructNo，executeModule这个参数需要改成-m的形式传入，目前没有实现（to do） :param instructNo: __，参数形式 :param runMode: runMode 运行模式，非必要参数。runMode=fusion执行fusion模式；runMode=standalone执行独立模式，默认值为fusion :param onlyDryrun: onlyDryrun 是否只运行dryrun，非必要参数。 当为onlyDryrun=yes时，只运行dryrun，onlyDryrun=no时；运行全流程，默认值为no :return: """ #1、执行模式：standlone独立运行模式；fusion，fusion控制模式 #如果standlone模式无需提交执行用来数量和执行结果及捕包数据，非Standalone即为fusion模式 tsglogger = business.tsglogger_fun(instructNo=instructNo) if instructNo is None: tsglogger.info('执行必须参数为空,退出！') sys.exit(0) if runMode == "standalone": tsglogger.info('独立模块执行！') #模拟fusion生成setup的json文件 taskid_timestamp = str(int(time.time())) taskid_timestamp_dir_name = taskid_timestamp + "_taskid_" + "1" project_automation_path = os.path.abspath("/opt/test/automation") project_taskid_timestamp_path = project_automation_path + "/tasks/" + taskid_timestamp_dir_name os.mkdir(project_taskid_timestamp_path) #判断 setup_json_name = "setup_" + taskid_timestamp_dir_name setup_json_name_path = project_taskid_timestamp_path + "/" + setup_json_name + ".json" shutil.copyfile("/opt/test/automation/control/setup_taskid_11223344.json", setup_json_name_path) instructNo = taskid_timestamp + "_taskid_" + "1" #instructNo = "2022032919164900060000313240" instructNo = "2022051721000002800100775113" runMode = "fusion" else: tsglogger.info('fusion模式执行！') """接收fusion唯一参数，并解析json文件""" #根据instructNo唯一标识，重组出setup文件绝对路径 tsglogger.info("instructNo:{}".format(instructNo)) instructNo_setup_json_abs_path = business.get_setup_abs_path(instructNo, tsglogger) tsglogger.info(os.path.dirname(instructNo_setup_json_abs_path)) #判断是否杀掉进程 if kill: if os_platform == "linux": business.kill_all(instructNo_setup_json_abs_path, tsglogger) else: business.kill_all_windows(instructNo_setup_json_abs_path, tsglogger) #获取父进程pid并存文件 ppid = os.getpid() tsglogger.info(f"控制脚本父进程:{ppid}") with open(os.path.dirname(instructNo_setup_json_abs_path) + "/pid.txt", "w") as f: f.write(f"{ppid}") #清空历史垃圾数据 business.clear_historical_reort(instructNo_setup_json_abs_path, tsglogger) #2、重组保存dryrun、report的目录和使用部分文件名 instructNo_setup_json_abs_path_list = instructNo_setup_json_abs_path.split("/") dryrun_dir_path = "/".join(instructNo_setup_json_abs_path_list[:-1]) + "/dryrun" #dryrun目录路径信息 report_dir_path = "/".join(instructNo_setup_json_abs_path_list[:-1]) + "/log" #每个test用例生成的report文件目录 tsglogger.debug(f"dryrun_dir_path:{dryrun_dir_path}") tsglogger.debug(f"report_dir_path:{report_dir_path}") if os.path.exists(report_dir_path): os.system("rm -rf {}".format(report_dir_path)) tsglogger.info(f"删除测试用例旧结果的{report_dir_path}目录") report_name = dryrun_name = instructNo #dryrun和report生成的文件，使用的部分instructNo名信息 # 3、加载json文件 f_dict = business.get_setup_from_fusion(instructNo_setup_json_abs_path, tsglogger) #4、是否版本校验：versionCheck==1，则检验 if int(f_dict["versionCheck"]) == 1: #校验版本信息 #取fusion中版本参数，判断版本是否与毕方版本一致 tsg_version = f_dict["tsgVersion"] bifang_version = "" if (bifang_version != tsg_version) and (bifang_version != ""): tsglogger.info(f"版本信息不一致：bifang_version={bifang_version},tsg_version={tsg_version}") #退出执行 sys.exit(0) else: #不检验版本信息 tsglogger.info("无需校验版本信息！") # 获取执行列表中生成dryrun文件命令，重新组合dryrun命令, 兼容window，基础命令移动到配置文件 config_parse = configparser.ConfigParser() config_parse.read("control_config.ini", encoding="utf-8") if os_platform == "linux": dryrun_command_front_past = config_parse.get("dryrun_front_past_info", "dryrun_command_front_past_linux") else: dryrun_command_front_past = config_parse.get("dryrun_front_past_info", "dryrun_command_front_past_windows") dryrun_command_front_past = "{} ".format(dryrun_command_front_past) # print(dryrun_command_front_past) # dryrun_command_front_past = "cd /opt/test/automation/scripts/gap_tsg_api/ && python3 -m robot.run -d newsreport/220313 --output test.xml --log test.html --report test.html --dry --pythonpath . " for i in range(len(f_dict["commandList"])): f_dict["commandList"][i] = dryrun_command_front_past + f_dict["commandList"][i] dryrun_command_list = f_dict["commandList"] # 5、组合可执行的dryrun命令列表 execute_dryrun_command_list = business.combine_dryrun_commond_from_setup(dryrun_command_list, dryrun_dir_path, dryrun_name, tsglogger) # 执行dryrun命令 dryrun_output_list1 = business.dryrun_commond_concurrency(execute_dryrun_command_list, f_dict, tsglogger) #多线程执行dryrun命令 #6、盘点是否需要提交fusion generate_case_json = business.get_case_name_from_output_xml(dryrun_output_list1, instructNo_setup_json_abs_path, tsglogger) if( runMode == 'fusion'): # 重组case_json文件函数 #robotde dryrun文件执行完后，开始从output文件中取出结果 #将用例列表传给fusion,通过调用fusion的接口回传 business.callback_callback_dryrun_report(generate_case_json, instructNo, tsglogger) #需要接口联调 #判断是否知识提交caselist if onlyDryrun == "yes": #值提交caselist，不执行用例 tsglogger.info("只提交caselist数据，不执行用例") return sys.exit(0) #7、解析执行case参数列表，（1）、写入文件；（2）写入命令参数 #转换caselist的json数据格式转换为python的列表格式，并deepcopy使用 execute_case_list = copy.deepcopy(json.loads(generate_case_json)["caseList"]) #生成robot使用的固定文件配置文件 common_configration_variable_list = business.generate_common_configration_from_fusion_setup(f_dict, instructNo_setup_json_abs_path, tsglogger) #整理robot使用中caselist和设备信息对应关系 execute_case_and_client_list = business.generate_caselist_client_info_list(f_dict, execute_case_list, tsglogger) execute_case_and_client_command_list = business.combine_execute_testcase_commond(execute_case_and_client_list, dryrun_command_list, tsglogger) #8、多线程执行每个设备中，可以执行的case列表 # all_client_task_list --- [[[用例命令],[对应设备信息列表]],[[用例命令],[对应设备信息列表]],[[用例命令],[对应设备信息列表]]] all_client_task_list = [] for i in range(len(execute_case_and_client_command_list)): all_client_task_list.append(business.generate_one_commond_from_execute_list(execute_case_and_client_command_list[i], report_dir_path, report_name)) #9、多线程方式，执行robot命令， business.mult_thread_pool_capture_test_report(all_client_task_list, f_dict, instructNo, runMode, common_configration_variable_list, tsglogger) #合并robot所有的结果文件 combine_report = business.execute_combine_report_of_robot(instructNo_setup_json_abs_path, tsglogger) #判断是否发布结果文件 print("````````````````````````") if runMode == "fusion": business.callbaock_upload_result_file(combine_report, instructNo, tsglogger) if __name__ == "__main__": """ executeModule：执行模式：standalone独立执行，独立执行模式无需提交执行测试用例列表及执行参数；fusion控制模式； standalone, fusion... #exec_mode = “standalone” 变量 """ if 1 == 1 : #当执行该条件时，会从命令行输入参数 try: parser = argparse.ArgumentParser(description="control args") parser.add_argument("-i", "--instructNo", metavar="", help="instructNo 唯一标识，必要参数。", required=True) parser.add_argument("-r", "--runMode", metavar="", help="runMode 运行模式，非必要参数。runMode=fusion执行fusion模式；" "runMode=standalone执行独立模式，默认值为fusion。", default="fusion") parser.add_argument("-o", "--onlyDryrun", metavar="", help="onlyDryrun 是否只运行dryrun，非必要参数。 当为onlyDryrun=yes时，只运行dryrun，" "onlyDryrun=no时；运行全流程，默认值为no。", default="no") parser.add_argument("-k", "--kill", metavar="", help="kill 中断当前唯一标识下所有进程，非必要参数。当有-k时，中断进程。", nargs="?", const="1") args = parser.parse_args() instructNo = args.instructNo runMode = args.runMode onlyDryrun = args.onlyDryrun kill = args.kill except Exception as e: print("执行调试模式") instructNo = 11 runMode = "standaolone" onlyDryrun = "fusion" pass else: #模拟fusion生成setup的json文件 taskid_timestamp = str(int(time.time())) taskid_timestamp_dir_name = taskid_timestamp + "_taskid_" + "1" project_automation_path = os.path.abspath(os.path.dirname(os.getcwd())) project_taskid_timestamp_path = project_automation_path + "/tasks/" + taskid_timestamp_dir_name os.mkdir(project_taskid_timestamp_path) #判断 setup_json_name = "setup_" + taskid_timestamp_dir_name setup_json_name_path = project_taskid_timestamp_path + "/" + setup_json_name + ".json" shutil.copyfile("/opt/test/automation/control/setup_taskid_11223344.json", setup_json_name_path) #/tmp/pycharm_project_327/automation/tasks/taskid_1647148614/setup_taskid_1647148614.json #/tmp/pycharm_project_327/automation/tasks/1647148614_taskid_1/setup_1647148614_taskid_1.json # instructNO = __ #"standaolone" "fusion" instructNo = taskid_timestamp + "_taskid_" + "1" runMode = "standaolone" onlyDryrun = "no" config_abs_path = "/opt/test/automation/control/control_config.ini" executeAutomation(instructNo, runMode, onlyDryrun, kill) print("结束完成") exit(0)