上篇：https://testerhome.com/topics/19286

我们上一篇的介绍只是启动了线程，在实际项目需要中，可能需要在执行完 case 之后，停掉所有启动的线程。然而，Python 对线程的停止并不是那么友好。那么有什么方法可以更方便地达到此目的呢？经过一些尝试，发现可以通过设置标志位变量来进行操作。以下的例子中，因为涉及到多个 py 类文件，所以这个标志位就需要设置成全局变量。

前提条件：预先安装 uiautomator2，连接好手机后，在终端命令行执行以下命令：

pip3 install uiautomator2

再安装 atx：

python3 -m uiautomator2 init

1、新建 thread_deal_permission_box.py，用于点掉手机设备的相关弹窗，代码如下：

import threading
from global_parameters import GlobalParameters

# 注意：
# 如果本机未安装u2，需要先安装，命令行执行以下命令：
# pip3 install uiautomator2 # 安装u2
# python3 -m uiautomator2 init # 手机自动安装atx, u2APK
# pip3 install --pre -U weditor # 安装UI查看器
# python3 -m weditor # 启动UI查看器

import uiautomator2 as u2
driver = u2.connect("882QADT9UWT")
global_parameters = GlobalParameters()

class ThreadDealPermissionBox(threading.Thread):

    def __init__(self, driver):
        threading.Thread.__init__(self)
        self.driver = driver

    def terminate(self):
        global_parameters.set_value('_flag', True)

    def run(self): # 把要执行的代码写到run函数里面 线程在创建后会直接运行run函数
        while True:
            self.click_permission_box()
            print('Flag:' + str(global_parameters.get_value('_flag')))
            if global_parameters.get_value('_flag'):
                break

    # 处理掉安装apk时的系统弹窗
    def click_permission_box(self):
        try:
            self.driver(text="允许").click_exists(timeout=1.0)
        except:
            pass
        try:
            self.driver(text="确认").click_exists(timeout=1.0)
        except:
            pass
        try:
            self.driver(text="继续安装").click_exists(timeout=1.0)
        except:
            pass
        try:
            self.driver(text="安装").click_exists(timeout=1.0)
        except:
            pass

2、新建 execute.py，封装执行 adb shell 的方法，代码如下：

import subprocess

class Execute():

    # 执行shell
    def shell(self, cmd):
        # 执行cmd命令，如果成功，返回(0, 'xxx')；如果失败，返回(1, 'xxx')
        cmd_result = subprocess.getstatusoutput(cmd)
        # print(str(cmd_result))
        exit_code = cmd_result[0]
        output = cmd_result[1]
        if exit_code == 0:
            if 'error' in output:
                print('Maybe something wrong: ' + str(cmd_result))
            return output
        elif exit_code == 1:
            print("Failed to execute cmd: " + cmd)
            return output

3、新建 adb_completely_install.py，用于安装 apk 应用，代码如下：

from execute import Execute
import os
import time
import threading
# 注意：
# 如果本机未安装u2，需要先安装，命令行执行以下命令：
# pip3 install uiautomator2 # 安装u2
# python3 -m uiautomator2 init # 手机自动安装atx, u2APK
# pip3 install --pre -U weditor # 安装UI查看器
# python3 -m weditor # 启动UI查看器
# import uiautomator2 as u2

execute = Execute()
# driver = u2.connect("882QADT9UWT")

# class AdbInstall():
class AdbCompletelyInstall(threading.Thread):

    # 判断apk应用是否已安装到设备
    def is_apk_has_installed(self, package_name, device_id=''):
        if device_id != '':
            cmd = "adb -s " + device_id + " shell pm list package | grep '" + package_name + "'"
        else:
            cmd = "adb shell pm list package | grep '" + package_name + "'"
        result = os.popen(cmd).read()
        if package_name in result:
            return True
        else:
            return False

    # 卸载app应用
    def uninstall_app(self, package_name, device_id=''):
        if device_id != '':
            cmd = 'adb -s ' + device_id +' uninstall ' + package_name
        else:
            cmd = 'adb uninstall ' + package_name
        execute.shell(cmd)
        print("Uninstall app success~")

    # 强制安装
    def overlay_install(self, file_path, device_id=''):
        if device_id != '':
            cmd = 'adb -s ' + device_id + ' install -r ' + file_path
        else:
            cmd = 'adb install -r ' + file_path
        execute.shell(cmd)

    # 完全安装，如果已安装应用，需要卸载，重新安装
    def completely_install(self, file_path, package_name, device_id=''):
        if self.is_apk_has_installed(package_name, device_id):
            self.uninstall_app(package_name)
        print('Begin to install apk...')
        if device_id != '':
            cmd = 'adb -s ' + device_id + ' install -r ' + file_path
        else:
            cmd = 'adb install -r ' + file_path
        res = execute.shell(cmd)
        if str(res) == 'Success':
            print('The apk '+ file_path + ' install successfully')
            count = 0
            while not self.is_apk_has_installed(package_name) or count > 30:
                time.sleep(1)
                print('sleep......')
                count = count + 1
        else:
            print('The apk ' + file_path + ' install failed!!!')

4、新建 global_parameters.py，用于保存全局变量，代码如下：

class GlobalParameters():

    def __init__(self):
        global _global_dict
        _global_dict = {'_flag': False}

    def set_value(self, name, value):
        _global_dict[name] = value

    def get_value(self, name, defValue=None):
        try:
            return _global_dict[name]
        except KeyError:
            return defValue

5、新建 app_engine.py，用于安装启动 app 等，代码如下：

import uiautomator2 as u2
from thread_deal_permission_box import ThreadDealPermissionBox
from adb_completely_install import AdbCompletelyInstall

class AppEngine():

    def setUp(self):

        self.device = '882QADT9UWT'
        self.apk_file = 'xxxxxx.apk'
        self.app_package = 'com.xxx'
        self.app_activity = 'com.xxx.xxx.HomeActivity'

        driver = u2.connect(self.device)

        thread_deal_permission_box = ThreadDealPermissionBox(driver)
        thread_deal_permission_box.start()
        # 完全安装应用
        AdbCompletelyInstall().completely_install(self.apk_file, self.app_package, self.device)
        driver = self.app_start()
        thread_deal_permission_box.terminate()  # 停止线程

        return driver

    # 启动app
    def app_start(self):

        driver = u2.connect(self.device)
        driver.app_start(self.app_package, self.app_activity)

        return driver

    # 关闭app应用
    def app_stop(self):
        driver = u2.connect(self.device)
        driver.app_stop(self.app_package)

app_engine = AppEngine()
app_engine.setUp()

最后，直接在 pycharm 中执行 app_engine.py 文件即可。