autotest/AutoUpgradeTestCase.py

#!/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)