Best Python code snippet using lisa_python
caserunner.py
Source:caserunner.py
...131 if not self.__pre_check(test):132 return133 self.module_manager.run_module(ModuleType.PRE)134 self.module_manager.run_module(ModuleType.PARALLEL)135 self.__run_case(test)136 self.module_manager.stop_module()137 self.module_manager.run_module(ModuleType.POST)138 def _import_list_case(self, case_tree_node, test_list, log_path=None):139 """140 éå½å¯¼å
¥æµè¯å表ä¸çæµè¯ç¨ä¾141 """142 case_log_path = test_list.test_list_name143 if log_path:144 case_log_path = log_path + "/" + case_log_path145 case_tree_node["list_name"] = test_list.test_list_name146 case_tree_node["test_cases"] = list()147 for testcase in test_list.test_cases:148 if testcase.strip() == "":149 continue150 case_descriptor = dict()151 case_entry = testcase.split(",")152 case_name = case_entry[0]153 case_setting_file = ""154 if len(case_entry) > 1:155 case_setting_file = case_entry[1]156 try:157 # 导å
¥æµè¯ç¨ä¾158 case_descriptor['case'] = self.load_test(case_name)159 case_descriptor['case_name'] = case_name.split(".")[-1]160 case_descriptor['log_path'] = case_log_path161 case_descriptor['setting_file'] = case_setting_file162 # 设置æµè¯ç¨ä¾é
ç½®æ件路å¾163 if test_list.setting.case_setting_path:164 case_descriptor['setting_path'] = test_list.setting.case_setting_path165 else:166 case_descriptor['setting_path'] = CaseRunnerSetting.default_case_setting_path167 case_priority = getattr(case_descriptor['case'], "priority", 999)168 if case_priority not in self.priority_list:169 self.priority_list.append(case_priority)170 except CaseImportError as cie:171 # æµè¯ç¨ä¾å¯¼å
¥å¤±è´¥172 self.logger.error(f"ä¸è½å¯¼å
¥æµè¯ç¨ä¾{case_name}")173 self.logger.exception(cie)174 case_tree_node['test_cases'].append(case_descriptor)175 case_tree_node['sub_list'] = list()176 for sub_list in test_list.sub_list:177 sub_list_dict = dict()178 case_tree_node['sub_list'].append(sub_list_dict)179 self._import_list_case(sub_list_dict, sub_list, log_path=case_log_path)180 def __init_precondition(self, test: TestCaseBase):181 self.pre_conditions.clear()182 self.pre_conditions.append(IsTestCaseType(self.test_list.setting.run_type))183 if any(self.test_list.setting.priority_to_run):184 self.pre_conditions.append(IsTestCasePriority(self.test_list.setting.priority_to_run))185 if any(test.pre_tests):186 self.pre_conditions.append(IsPreCasePassed(self.case_result))187 self.pre_conditions.append(IsHigherPriorityPassed(test.priority, self.case_result))188 def __pre_check(self, test:TestCaseBase):189 for condition in self.pre_conditions:190 if not condition.is_meet(test, self.result_report):191 self.result_report.add(StepResult.INFO, f"{test.__class__.__name__}ä¸è½æ§è¡ï¼")192 return False193 return True194 def __get_case_log(self, path, case_name):195 log_path = os.path.join(self.case_log_folder, path, f"{case_name}.log")196 return logger.register(case_name, filename=log_path, is_test=True)197 def __main_test_thread(self):198 try:199 self.__run_test_list(self.case_tree)200 finally:201 self.status = RunningStatus.Idle202 def __run_test_list(self, testlist):203 self.result_report.add_list(testlist['list_name'])204 for test in testlist['test_cases']:205 test["case"].get_setting(test["setting_path"], test["setting_file"])206 self.result_report.case_logger = self.__get_case_log(test['log_path'], test['case_name'])207 self.case_result[test["case_name"]] = dict()208 self.case_result[test["case_name"]]['priority'] = test["case"].priority209 self.case_result[test["case_name"]]['result'] = False210 self.run_case_lcm(test['case'])211 self.result_report.case_logger = None212 logger.unregister(test['case_name'])213 for list in testlist['sub_list']:214 self.__run_test_list(list)215 self.result_report.end_list()216 def __run_case(self, test: TestCaseBase):217 """218 æµè¯ç¨ä¾æ§è¡çº¿ç¨219 """220 self.result_report.add_test(test.__class__.__name__)221 _continue = True222 try:223 self.result_report.add_step_group("æ¶éæµè¯èµæº")224 test.collect_resource(self.resource_pool)225 except ResourceNotMeetConstraintError as rnce:226 self.result_report.add(StepResult.EXCEPTION, "æµè¯èµæºä¸æ»¡è¶³æ¡ä»¶", str(rnce))227 _continue = False228 except Exception as e:229 self.result_report.add(StepResult.EXCEPTION, "æè·å¼å¸¸ï¼", str(e))230 _continue = False...
Run.py
Source:Run.py
1# -*- coding:utf-8 -*-2#######################################################3# filename:Run.py4# author:Jeff5# date:2016-11-216# function:驱å¨Excelãèæ¬ æµè¯ç¨ä¾7#######################################################8import unittest9from conf.Run_conf import read_config10from multiprocessing import Pool11from src.Public.Server import Server12from src.Public.GetDevices import GetDevices13from src.lib import ExcelRW14from src.Public.Common import public15from src.Public.Global import D,S,L,Data16from src.lib.Driver import Driver17from src.lib.Element import Element18import time,os19# from src.lib.Utils import utils20PATH = lambda p: os.path.abspath(21 os.path.join(os.path.dirname(__file__), p)22)23def run_mode(device,index):24 '''25 runmod éæ©è¿è¡æ¨¡å¼,1:è¿è¡excelç¨ä¾,0:è¿è¡èæ¬ç¨ä¾26 :return:27 '''28 runmod = str(read_config('runmode','mode'))29 if runmod == '1':30 print ('*' * 80)31 L.logger.info(' è¿è¡ Excel æµè¯ç¨ä¾ ')32 xls_file_path = PATH('./TestCase/Excel/TestCase.xlsx')33 print ('*' * 80)34 print (time.ctime(), ' [', __name__, '::', 'ç¨ä¾è·¯å¾: ', xls_file_path)35 xlsEngine = ExcelRW.XlsEngine(xls_file_path)36 xlsEngine.open() # æå¼excel37 global case_sheet138 case_sheet1 = xlsEngine.readsheet(public.case_sheet)39 #驱å¨æµè¯40 runner = unittest.TextTestRunner()41 __Run_Case(runner)42 elif runmod == '0':43 # print '*' * 8044 # print (time.ctime()+' [', __name__, '::'+run_mode.__name__+'] :'+' è¿è¡ Script æµè¯ç¨ä¾ ')45 # print ('*' * 80)46 L.logger.info(' è¿è¡ Script æµè¯ç¨ä¾ ')47 import RunScript48 RunScript.run_pytest()49 elif runmod == '2' :50 # if str(device['platformName']).lower()!='android':51 # raise EnvironmentError,'runmod=2,ä»
æ¯æandroid'52 L.logger.info(' è¿è¡ Yaml æµè¯ç¨ä¾ ')53 from src.Public.integration import RunApp54 r = RunApp(device,index)55 r.case_start()56 else:57 pass58def clean_process():59 '''60 æ¸
ç线ç¨61 :return:62 '''63 from src.Public import CleanProcess64 #ç»ææå¡è¿ç¨65 cp = CleanProcess.Cp()66 cp.clean_process_all()67import RunExcel68def __get_test_suite(case_list):69 '''70 è·åæµè¯å¥71 :param case_list:72 :return:73 '''74 # print time.ctime(), ' [', __name__, '::', __get_test_suite.__name__, '] :', ' è·åsuite '75 print ('*' * 80)76 test_suite = unittest.TestSuite()77 run_excel_case = RunExcel.RunExcelCase(case_list, "function")78 test_suite.addTest(run_excel_case)79 return test_suite80def __Run_Case(runner):81 '''82 è¿è¡æµè¯ç¨ä¾83 :param runner:84 :return:85 '''86 # print time.ctime(), ' [', __name__, '::', __Run_Case.__name__, '] :', ' å¼å§è¿è¡ç¨ä¾éå '87 L.logger.debug(' 循ç¯éåæµè¯ç¨ä¾ ')88 # 循ç¯éåæµè¯ç¨ä¾å表89 for case_list in case_sheet1[1:]:90 #å¤ææ¯å¦æ¯ç¬ææä½,å¦æä¸æ¯å¯¹åºå¹³å°çç¬ææä½å°±è·³è¿å¾ªç¯91 if str(case_list[4]).lower() != str(S.device['platformName']).lower() and len(case_list[4]) !=0:92 continue93 test_suite = __get_test_suite(case_list)94 runner.run(test_suite)95 # æå
¥ç¨ä¾åå设å¤udid96 # Q.sql.insert_per(case_list[0],S.device['udid'],'','','')97 wd=Element(driver)98 # wd.close_app()# éåºapp99 wd.quit() # éåºæå¡100 # çææµè¯æ¥å101 RunExcel.get_html_report()102def Run_one(device,port,index):103 user_datas = [[768486, 768486, 768487], [768488, 768488, 768489], [768490, 768490, 768491]]104 Data.set_data(user_datas[index])105 S.set_device(device)106 from src.lib.Log import LogSignleton107 logsignleton = LogSignleton()108 logger = logsignleton.logger109 L.set_logger(logger)110 # å¯å¨appium æå¡111 A = Server()112 A.start_server(device, port)113 # print ('*' * 80)114 L.logger.debug(' è¿è¡è®¾å¤ device : %s ' % device)115 # å®ä¾åDirver116 Dr = Driver(device, port)117 Dr.init() # åå§ådriver118 global driver119 driver = Dr.getDriver()120 D.set_driver(driver)121 # print ('*' * 80)122 L.logger.debug(' è¿è¡è®¾å¤ driver : %s ' % driver)123 run_mode(device,index) # è¿è¡æ¨¡å¼124if __name__ == '__main__':125 try:126 clean_process()127 G = GetDevices()128 devices = G.get_device()129 count = len(devices)130 ports = G.get_port(count)131 p = Pool(processes=count)132 # å¤çº¿ç¨å¹¶å133 for i in range(count):134 result = p.apply_async(Run_one,(devices[i],ports[i],i,))135 p.close() # å
³éè¿ç¨,ä¸åæ·»å æ°çè¿ç¨136 p.join() # è¿ç¨çå¾
æ§è¡å®æ¯137 # if result.successful():138 # # print ('*' * 80)139 # if count > 1:140 # L.logger.debug( 'å¤è®¾å¤å¹¶åæ§è¡æå')141 # else:142 # L.logger.debug('å设å¤æ§è¡æå')143 # clean_process()144 except Exception as e:145 raise e146 # print ('*' * 80)147 # L.logger.debug( 'ææ代ç æ§è¡å®æ¯!')148 import sys...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!