xref: /developtools/integration_verification/DeployDevice/src/func/liteOsUpgrade/liteOsUpgrade_L3_app.py

#-*- coding:utf-8 -*-
import uuid
import sys
import subprocess
import os
import serial

from core.base import BaseApp, dec_stepmsg
from util.file_locker import FileLock
from util.log_info import logger
from util.time_info import get_now_time_str_info, get_now_time_info, Timeout, timeout
from aw.Telnet.TelnetClient import TelConnect
from aw.Common.Constant import CONSTANT
from aw.Download.Download import *
from aw.Common.Common import getHostIp
from aw.ExtractFile.ExtractFile import *
from aw.poweronoff.serial_power_on_off import serialPowerOnOff

lock_suffix = CONSTANT.File.LOCK_SUFFIX  #通过文件锁实现并发下载
suc_file = CONSTANT.File.SUC_FILE        #通过本文件是区分版本是否成功下载
failed_file = CONSTANT.File.FAILED_FILE  #通过本文件是标记文件下载失败
READ_MAXTIMEOUT = 600
READ_TIMEOUT = 30
READ_MINITIMEOUT = 5
uboot_finish = 'hisilicon #'
cmd_finish = ' #'

class liteOsUpgrade_L3(BaseApp):
    '''
    @author: w00278233
    '''

    def __init__(self, param_file):
        super().__init__(param_file)
        self.param_List = ["upgrade_upgradeLocation"]

    @dec_stepmsg("hongmeng L3 flash")
    def excute(self):
        '''
        #===================================================================================
        #   @Method:        excute(self)
        #   @Precondition:  none
        #   @Func:          升级执行入口
        #   @PostStatus:    none
        #   @eg:            excute()
        #   @return:        True or Flase
        #===================================================================================
        '''
        step_index = self.params_dict.get("step_list").index("liteOsUpgrade_L3_app")

        # 执行下载
        try:
            if not self.download():
                CONSTANT.ENVERRMESSAGE = "image download fail"
                logger.printLog(CONSTANT.ENVERRMESSAGE)
                return False
        except Exception as e:
            raise e

        # 执行升级
        try:
            status = self.upgrade()
            logger.info(status)
            logger.info(type(status))
            logger.info(f'升级状态:{str(status)}')
            if not status:
                CONSTANT.ENVERRMESSAGE = "board upgrade fail"
                logger.printLog(CONSTANT.ENVERRMESSAGE)
                return False
            return True
        except Exception as e:
            raise e

    @dec_stepmsg("download")
    @timeout(1800)
    def download(self):
        '''
        #===================================================================================
        #   @Method:        download(self)
        #   @Precondition:  none
        #   @Func:          构建下载到本地的路径，执行相应包的下载
        #   @PostStatus:    none
        #   @eg:            download()
        #   @return:        True or Flase
        #===================================================================================
        '''
        global version_savepath, version_name
        dir_path = CONSTANT.Path.getDirPath()
        if self.params_dict.get("pbiid"):
            version_path = self.params_dict.get("pbiid")
            version_name = str(uuid.uuid5(uuid.NAMESPACE_URL, str(self.params_dict.get("pbiid")) + "FASTBOOT"))
            version_savepath = os.path.join(dir_path, self.params_dict.get("flash_type"), version_name)
        else:
            version_path = self.params_dict.get("upgrade_upgradeLocation")
            version_name = str(uuid.uuid5(uuid.NAMESPACE_URL, (self.params_dict.get("upgrade_upgradeLocation"))))
            version_savepath = os.path.join(dir_path, version_name)
            logger.printLog(version_savepath)

        if self.params_dict.get("isDownload") == "True":
            logger.printLog("不需要做下载，直接返回")
            return True

        #执行img下载
        import hashlib
        save_file_str = version_path.replace("/", "").replace("\\", "")
        save_file_name = hashlib.sha1(save_file_str.encode("utf-8")).hexdigest()
        logger.info("download hash string:%s, hash value:%s" % (save_file_str, save_file_name))
        save_path_file = os.path.join(dir_path, "record", "%s%s" % (save_file_name, ".txt"))
        if not self.excutedown(version_path, os.path.join(version_savepath, "img"), save_path_file, False):
            logger.error("download img fail")
            return False

        #保存本地版本路径给devicetest去版本路径下取用例
        saveVersion(save_path_file, os.path.join(version_savepath, "img"))

        # 执行升级脚本下载
        if self.params_dict.get("upgrade_script"):
            suc_mark = os.path.join(version_savepath, "scriptfile", suc_file)
            if not self.excutedown(self.params_dict.get("upgrade_script"),
                                   os.path.join(version_savepath, "scriptfile"), suc_mark, True):
                logger.error("download upgrade script fail")
                return False
            with open(suc_mark, "a+") as fp:
                fp.write("")

        return True

    def excutedown(self, source_path, download_dir, suc_mark, is_file):
        '''
        #===================================================================================
        #   @Method:        excutedown(source_path, download_dir, is_file)
        #   @Precondition:  none
        #   @Func:          执行下载动作
        #   @PostStatus:    none
        #   @Param:         source_path：资源文件路径
        #                   download_dir:文件下载到本地的文件夹路径
        #                   is_file:是否是文件
        #
        #   @eg:            excutedown("xxxx", "D:\\local\\image", Flase， os_method)
        #   @return:        True or Flase
        #===================================================================================
        '''
        failed_mark = os.path.join(download_dir, failed_file)
        lock_path = os.path.join(download_dir, lock_suffix)
        file_lock = FileLock()
        if isDownLoadSuccess(download_dir, suc_mark, failed_mark):
            return True
        try:
            nowtime = get_now_time_str_info()
            logger.printLog("%s Downloading, please wait" % nowtime)
            file_lock.lockFile(lock_path)
            ret = ""
            logger.info("Get lock. Start to ")
            if self.params_dict.get("bt_enable") and self.params_dict.get("bt_enable") == "True":
                ret = downloadByBitComet(source_path, download_dir, os_method)
            elif source_path.startswith('\\\\'):
                ret = downloadByCopy(source_path, download_dir, is_file)
            elif self.params_dict.get("pbiid"):
                ret = downlaodByDownloadTool(version_savepath, self.params_dict.get("version_type"), "FASTBOOT", self.params_dict.get("pbiid"))
            elif source_path.startswith("http"):
                # 临时修改下载链接
                source_path = self.modifyHttpLink(source_path)
                logger.info(f'modify path:{source_path}')
                ret = downloadFileFromDevCloud(source_path, "", "", download_dir)

            if source_path.endswith(".zip"):
                zip_name = os.path.basename(source_path)
                ret = extractZipFile(os.path.join(download_dir, zip_name), download_dir)
            if source_path.endswith(".tar.gz") or (source_path.startswith("http") and ("file_id=" in source_path)):
                if source_path.startswith("http") and ("file_id=" in source_path):
                    if source_path.endswith(".tar.gz"):
                        zip_name = source_path.split('=')[-1]
                    else:
                        zip_name = "out.tar.gz"
                else:
                    zip_name = os.path.basename(source_path)
                ret = unTarFile(os.path.join(download_dir, zip_name), download_dir)
            nowtime = get_now_time_str_info()
            logger.printLog("%s download to %s end" % (nowtime, download_dir))

            if not ret:
                with open(failed_mark, "a+") as fp:
                    fp.write("")
            return ret
        except Exception as e:
            logger.printLog(e)
            raise Exception(e)
        finally:
            file_lock.releaseFile()

    def modifyHttpLink(self,link):
        try:
            base_link = 'https://hm-verify.obs.cn-north-4.myhuaweicloud.com/'
            filename = link.split('&')[-1].split('=')[-1]
            names = filename.split('-')
            link_1 = '/'.join(names[0:-1])
            link_2 = f'/{filename}'
            final_link = base_link + link_1 + link_2
            return final_link
        except Exception as e:
            logger.info('path is errror.check the link_path is right,please')
            logger.info(str(e))


    @dec_stepmsg("upgrade")
    @timeout(900)
    def upgrade(self):
        '''
        #===================================================================================
        #   @Method:        upgrade(self)
        #   @Precondition:  none
        #   @Func:          升级相关业务逻辑
        #   @PostStatus:    none
        #   @eg:            upgrade()
        #   @return:        True or Flase
        #===================================================================================
        '''
        try:
            #time.sleep(1000)
            logger.printLog('开始升级')
            self.enterTheDevice()
            time.sleep(5)
            self.flashSystemImg()
            time.sleep(5)
            self.rebootDevice()
            return True
        except Exception as e:
            logger.error('***********************************')
            logger.error(str(e))
            logger.error('flash fail')
            logger.error('***********************************')
            return False

    #@timeout(30*3)
    def enterTheDevice(self):
        logger.info('enter fastboot status'.center(50,'*'))
        devices_status_count = 0
        while devices_status_count < 3 :
            if self.isFastbootStatus():
                break
            devices_status_count += 1
            time.sleep(20)
        else:
            logger.error(f'device {self.params_dict.get("sn")} is not exist ,please make sure the sn is True')
            raise Exception('ERROR_SN_DEVICE_NOT_EXIST')

    #@timeout(20)
    def isFastbootStatus(self):
        adb_retcode, adb_output = subprocess.getstatusoutput('adb devices')

        fastboot_retcode, fastboot_output = subprocess.getstatusoutput('fastboot devices')

        if self.params_dict.get('sn') in adb_output:
            retcode, output = subprocess.getstatusoutput('adb -s %s reboot fastboot' % self.params_dict.get('sn'))
            logger.info('adb -s %s reboot fastboot retcode: %s'% (self.params_dict.get('sn'),str(retcode)))
            logger.info('adb -s %s reboot fastboot output: %s'% (self.params_dict.get('sn'),output))
            logger.info('device from adb mode to fastboot mode,please wait')
            time.sleep(30)
            retcode, output = subprocess.getstatusoutput('fastboot devices')
            logger.info('fastboot devices fastboot_retcode: %s'% str(fastboot_retcode))
            logger.info('fastboot devices fastboot_output: %s'% fastboot_output)
            if self.params_dict.get('sn') in output:
                logger.info('device enter fastboot mode success')
                return True
            else:
                logger.info('device enter fastboot mode fail')
                return False
        elif self.params_dict.get('sn') in fastboot_output:
            logger.info('device has been in fastboot mode')
            return True
        else:
            logger.info('设备未找到，重新尝试')
            return False

    @timeout(60*2*3)
    def flashSystemImg(self):
        logger.info('flash system.img '.center(50,'*'))
        flash_count = 0
        while flash_count < 3:
            status = self.isFlashSuccess()
            logger.info(status)
            if status:
                break
            flash_count +=1
            time.sleep(3)
        else:
            logger.error('system.img镜像刷写失败')
            raise Exception('[ERROR] ERROR_FLASH_SYSTEM.IMG_FAIL')

    @timeout(60*2)
    def isFlashSuccess(self):
        #local_image_path = os.path.join(version_savepath, 'img', r'aosp\target\product\generic_arm64', 'system.img')
        local_image_path = os.path.join(version_savepath, 'img', 'system.img')
        logger.info(f'flash img path: {local_image_path}')
        if os.path.exists(local_image_path):
            retcode, output = subprocess.getstatusoutput('fastboot -s %s flash system %s' % (self.params_dict.get('sn'), local_image_path))
            logger.info('fastboot -s %s flash system %s retcode: %s'% (self.params_dict.get('sn'),local_image_path,str(retcode)))
            logger.info('fastboot -s %s flash system %s output: %s'% (self.params_dict.get('sn'),local_image_path,output))
            if retcode == 0:
                logger.info('flash system.img success')
                return True
            else:
                flash_count += 1
                logger.info('再次尝试刷system.img镜像')

        else:
            logger.error('镜像文件路径不对')
            raise Exception('[ERROR] ERROR_IMG_PATH_NOT_EXIST')

    def rebootDevice(self):
        logger.info('reboot the device '.center(50,'*'))
        reboot_count = 0
        while reboot_count < 3 :
            logger.info('start reboot')
            if self.isRebootSuccess():
                break
            else:
                reboot_count +=1
                time.sleep(10)
        else:
            raise Exception('[ERROR] ERROR_REBOOT_DEVICE_FAIL')

    def isRebootSuccess(self):
        retcode, output = subprocess.getstatusoutput('fastboot -s %s reboot' % self.params_dict.get('sn'))
        logger.info('fastboot -s %s reboot  retcode: %s'% (self.params_dict.get('sn'),str(retcode)))
        logger.info('fastboot -s %s reboot  output: %s'% (self.params_dict.get('sn'),output))
        time.sleep(50)
        retcode, output = subprocess.getstatusoutput('adb devices')
        logger.info('adb devices: %s'% str(retcode))
        logger.info('adb devices: %s'% output)
        if self.params_dict.get('sn') in output:
            logger.info('reboot success')
            return True
        else:
            logger.info('reboot fail')
            return False



if __name__ == '__main__':
    liteOsUpgrade_L3.modifyHttpLink(liteOsUpgrade_L3,
    'https://devrepo.devcloud.cn-north-4.huaweicloud.com/DevRepoServer/v1/files/download?file_id=0fe39c430a364f59bd040c53e7a3ef1c&type=archive&filename=Daily_Version#2021-05-19_06-38-31#L35_aosp_arm64.tar.gz')