#!/usr/bin/env python #coding:utf-8 import requests,json,os,sys,threading,time import logging import logging.handlers import re #打印公司logo f = open('logo.txt', 'r') file_contents = f.read() print (file_contents) f.close() #日志文件 LOG_FILE = 'autotest.log' handler = logging.handlers.RotatingFileHandler(LOG_FILE, maxBytes = 1024*1024, backupCount = 5) # 实例化handler fmt = '%(asctime)s -- %(name)s - %(message)s' # log格式formatter formatter = logging.Formatter(fmt) # 实例化formatter handler.setFormatter(formatter) # 为handler添加formatter logger = logging.getLogger('autotest') # 获取名为tst的logger logger.addHandler(handler) # 为logger添加handler logger.setLevel(logging.DEBUG) # 为设置logger等级 #被测油罐车ID g_oilTankID="83e2e01d8edf74443acd1b7ef5a11f2f" #测试库 版本信息 合规版本1 合规版本2 非合规版本 controller_id=["2acbaf521456186f2ada879f1fcb88fa","ce71527072e1ff8b953b0ade6c9631a3","28350b002504be7c1e9ad0f62c49162c"] controller_version=["2.1.3.9.1.1.0.20220225","2.1.2.9.1.1.0.20220225","2.1.3.9.1.1.0.20220220"] collector_id=["ab9e4bc574b07bce07bccae1dc572f39","9f85cb17fd98a00c751f56d9035c28b6","114e402040d06a06dc1b21bdfe9f5254"] collector_version=["2.0.0.6.1.0.0.20220226","2.0.0.6.1.0.0.20220225","2.1.3.9.1.1.0.20220221"] #controller_old_v='' #collector_old_v='' #controller_current_v='' #collector_current_v='' def get_controller(): #判断设备是否在线 trynum=0 get_error = 1 while True: trynum=trynum+1 if trynum > 5: break url="http://dev.jlwlwkj.com:6002/alarm-app/sendInstructions/readControllerVersion?carId=%s"%g_oilTankID headers = {"Authorization":"bearer "+ token} res = requests.post(url=url,headers=headers) if 200 != res.status_code: logger.error('get controller error') time.sleep(5) get_error = 1 continue res_json = json.loads(res.text) if 0 == res_json['code']: print("oilTank offline try to connect") logger.error('oiltank offline') time.sleep(5) continue else: get_error = 0 time.sleep(10) return get_error def get_collector(): #判断设备是否在线 trynum=0 get_error = 1 while True: trynum=trynum+1 if trynum > 5: break url="http://dev.jlwlwkj.com:6002/alarm-app/sendInstructions/readCollectorVersion?carId=%s"%g_oilTankID headers = {"Authorization":"bearer "+ token} res = requests.post(url=url,headers=headers) if 200 != res.status_code: logger.error('get collector error') time.sleep(5) get_error = 1 continue res_json = json.loads(res.text) if 0 == res_json['code']: print("oilTank offline try to connect") logger.error('oiltank offline') time.sleep(5) continue else: get_error = 0 time.sleep(10) return get_error def get_version(): #读取 系统库中 controller和collector的版本号 url="http://dev.jlwlwkj.com:6002/base-info-app/carInfo/readVersion?carId=%s"%g_oilTankID headers = {"Authorization":"bearer "+ token} res = requests.get(url=url,headers=headers) res_json = json.loads(res.text) controller_v="version_error" collector_v="version_error" for version in res_json['data']: if "COLLECTOR_VERSION" == version['name']: collector_v = version['value'] else: controller_v = version['value'] return controller_v,collector_v #查看当前升级进度 upgrade_finished = 0 def get_process(): url="http://dev.jlwlwkj.com:6002/base-info-app/upgradePack/upgradeProgress?carId=%s"%g_oilTankID headers = {"Authorization":"bearer "+ token} res = requests.get(url=url,headers=headers) res_json = json.loads(res.text) done_process = "100.00%" global upgrade_finished upgrade_finished = 1 #判断升级是否取消 name=res_json['data'] print name if name.find(u"取消") != -1: print "upgrade cancel restart again!" upgrade_finished = 1 logger.error('upgrade cancel') return if done_process in res_json['data']: print "upgrade finished!" upgrade_finished = 1 else: print "upgrade processing" print res_json['data'] upgrade_finished = 0 #定时获取token线程,每隔半小时获取一次token,防止token值发生变化 token='' def get_token_timer(): url="http://dev.jlwlwkj.com:6002/authority-app/logIn" headers={'Content-Type':'application/json;charset=UTF-8'} request_param={ "account": "sys", "password": "123456", "type": 0 } response=requests.post(url,data=json.dumps(request_param), headers=headers) global token token = response.json()["data"]["access_token"] print("...Getting token...") print token global timer timer = threading.Timer(1800, get_token_timer) timer.start() #发送升级指令 def send_upgrade_command(oilTankID, upgradePackId): url="http://dev.jlwlwkj.com:6002/base-info-app/upgradePack/upgrade" headers={'Content-Type':'application/json;charset=UTF-8','Authorization':'bearer '+ token} request_param = { "carIds":[oilTankID], "upgradePackId":upgradePackId } response=requests.post(url,data=json.dumps(request_param), headers=headers) send_result = 0 global upgrade_finished if 200 == response.status_code: print "发送升级指令成功,开始升级" send_result = 0 upgrade_finished = 0 else: print "发送升级指令失败,等待下一次发送升级" logger.error('send upgrade command error') upgrade_finished = 1 send_result = 1 return send_result def main(): #启用定时获取token线程 get_token_timer() time.sleep(5) controller_old_v='' collector_old_v='' controller_current_v='' collector_current_v='' global upgrade_finished #case1 控制板不同版本升级测试 print("\033[0;32;40mRunning Test_case_1 ------------------- different legal controller version upgrade\033[0m") controller_upgrade_times = 0 while True: controller_upgrade_times = controller_upgrade_times+1 if controller_upgrade_times > 50: break; #获取当前控制板版本号 if 0 == get_controller(): controller_old_v, collector_old_v = get_version() #版本比对,选择一个不同的合规版本进行升级 目前库中存在两个合规版本 if controller_old_v != controller_version[0]: t_controllerID=controller_id[0] else: t_controllerID=controller_id[1] #发送升级指令成功开始升级,记录当前升级前版本和升级后版本 if 0 != send_upgrade_command(g_oilTankID,t_controllerID): logger.error('send upgrade command error') #查看升级进度 while 0 == upgrade_finished: time.sleep(20) get_process() time.sleep(20) #获取升级后的版本号 if 0 == get_controller(): controller_current_v, collector_current_v = get_version() #记录日志做下一轮升级 logger.info('controller upgrade before: %s'%controller_old_v) logger.info('controller upgrade after: %s'%controller_current_v) time.sleep(5) #case2 采集板不同版本升级测试 print("\033[0;32;40mRunning Test_case_2 ------------------- different legal collector version upgrade\033[0m") collector_upgrade_times = 0 upgrade_finished = 1 while True: collector_upgrade_times = collector_upgrade_times+1 if collector_upgrade_times > 50: break; #获取当前采集板版本号 if 0 == get_collector(): controller_old_v, collector_old_v = get_version() #版本比对,选择一个不同的合规版本进行升级 目前库中存在两个合规版本 if collector_old_v != collector_version[0]: t_collectorID=collector_id[0] else: t_collectorID=collector_id[1] #发送升级指令成功开始升级,记录当前升级前版本和升级后版本 if 0 != send_upgrade_command(g_oilTankID,t_collectorID): logger.error('send collector upgrade command error') #查看升级进度 while 0 == upgrade_finished: time.sleep(20) get_process() time.sleep(20) #获取升级后的版本号 if 0 != get_collector(): controller_current_v, collector_current_v = get_version() #记录日志做下一轮升级 logger.info('collector upgrade before: %s'%collector_old_v) logger.info('collector upgrade after: %s'%collector_current_v) time.sleep(5) #测试结束取消定时获取token任务 timer.cancel() if __name__ == '__main__': main() sys.exit(0)