[开发测试]Appium之生成服务端appium指令+Python执行命令行+获取设备信息+端口设定+多线程启动Appium+多进程执行测试用例

Appium之生成服务端appium指令+Python执行命令行+获取设备信息+端口设定+多线程启动Appium+多进程执行测试用例

python命令行获取devicesID

import os

print(os.system('adb devices'))
print("os.system只能打印出设备信息来，但是并不能让我们通过命令提取出设备信息")
print(type(os.system('adb devices')))
print(os.popen('adb devices'))
print("直接执行os.popen出来的是一个os._wrap_close类型的，我也不知道是啥")
print(type(os.popen('adb devices')))
print(os.popen('adb devices').readlines())
print("readlines之后就成了一个list")
print(type(os.popen('adb devices').readlines()))

结果

List of devices attached
JPF4C19123011893	device

0
os.system只能打印出设备信息来，但是并不能让我们通过命令提取出设备信息
List of devices attached
JPF4C19123011893	device

<class 'int'>
<os._wrap_close object at 0x7ffa7f2014c0>
直接执行os.popen出来的是一个os._wrap_close类型的，我也不知道是啥
<class 'os._wrap_close'>
['List of devices attached\n', 'JPF4C19123011893\tdevice\n', '\n']
readlines之后就成了一个list
<class 'list'>

封装执行命令和获取命令结果的方法

import os


class DosCmd:
    # 获取执行命令的结果
    def excute_cmd_result(self, cmd):
        result_list = []
        result = os.popen(cmd).readlines()
        for i in result:
            if i == '\n':
                continue
            # strip() 方法用于移除字符串头尾指定的字符
            result_list.append(i.strip('\n'))
        return result_list

    # 直接执行命令
    def excute_cmd(self, cmd):
        os.system(cmd)


if __name__ == "__main__":
    dos = DosCmd()
    print(dos.excute_cmd_result('adb devices'))

结果

['List of devices attached', 'JPF4C19123011893\tdevice']

获取手机ID

from util.dos_cmd import DosCmd


class Driver:
    def __init__(self):
        self.cmd = DosCmd()

    def get_driver(self):
        devices_list = []
        device_info = self.cmd.excute_cmd_result('adb devices')
        # 第一部分为一段文案，从第二段开始才是设备信息，所以只有大于等于2的时候才有设备连接到电脑
        if len(device_info) >= 2:
            for i in device_info:
                if 'List' in i:
                    continue
                # 将一条设备信息分割成一个list,第一部分为设备信息，第二部分为device，也就是设备的状态
                device = i.split('\t')
                # 只有状态为device的时候这个设备才是可用的
                if device[1] == 'device':
                    devices_list.append(device[0])
            return devices_list
        else:
            return None


if __name__ == "__main__":
    driver = Driver()
    print(driver.get_driver())

结果

['JPF4C19123011893']

检查端口是否被占用

from util.dos_cmd import DosCmd


class Port:
    def __init__(self):
        self.cmd = DosCmd()

    def windows_port_was_used(self, port):
        # 但是上面这个命令在Mac上是没有的
        result = self.cmd.excute_cmd_result('netstat -ano | findstr ' + port)
        if len(result) > 0:
            return True
        else:
            return True

    def mac_port_was_used(self, port):
        result = self.cmd.excute_cmd_result('netstat -anp  tcp| grep ' + port)
        print(result)
        if len(result) > 0:
            return True
        else:
            return False


if __name__ == "__main__":
    port = Port()
    print(port.mac_port_was_used('9999'))

结果

[]
False

生成可用的端口列表

from util.dos_cmd import DosCmd


class Port:
    def __init__(self):
        self.cmd = DosCmd()

    def windows_port_was_used(self, port):
        # 但是上面这个命令在Mac上是没有的
        result = self.cmd.excute_cmd_result('netstat -ano | findstr ' + str(port))
        if len(result) > 0:
            return True
        else:
            return True

    def mac_port_was_used(self, port):
        result = self.cmd.excute_cmd_result('netstat -anp  tcp| grep ' + str(port))
        if len(result) > 0:
            return True
        else:
            return False

    def create_port_list(self, start_port, device_list):
        # start_port，我们从哪个端口开始占用端口,但是注意，我们传进来的这个值是
        # device_list，我们有了设备的个数，就知道占用多少个端口
        port_list = []

        # 只有设备信息不为空的时候我们才继续
        if device_list is not None:
            # for i in len(device_list):
            # 我们不能用for循环来决定，因为如果有端口被占用了怎么办，那就会报错，所以我们用while
            # 只要我们的端口个数不等于设备个数，那我们就继续
            while len(port_list) != len(device_list):
                # 当初始端口没有被占用的时候我们才继续
                if not self.mac_port_was_used(start_port):
                    port_list.append(start_port)
                # 由于我们这里是需要进行假发运算，所以这里是int类型的，但是mac_port_was_used里的port需要的是str类型的
                # 所以我们需要修改一下mac_port_was_used里的port参数
                start_port = start_port + 1
            return port_list
        else:
            print("生成端口失败")
            return None


if __name__ == "__main__":
    port = Port()
    print(port.mac_port_was_used(9999))
    listone = [1, 2, 3, 4]
    print(port.create_port_list(8888, listone))

结果

False
[8889, 8890, 8891, 8892]

整合生成端口和获取手机ID，方便生成服务端appium启动命令

import threading

from util.dos_cmd import DosCmd
from util.port import Port


class Driver:
    def __init__(self):
        self.cmd = DosCmd()

    def get_driver(self):
        devices_list = []
        device_info = self.cmd.excute_cmd_result('adb devices')
        # 第一部分为一段文案，从第二段开始才是设备信息，所以只有大于等于2的时候才有设备连接到电脑
        if len(device_info) >= 2:
            for i in device_info:
                if 'List' in i:
                    continue
                # 将一条设备信息分割成一个list,第一部分为设备信息，第二部分为device，也就是设备的状态
                device = i.split('\t')
                # 只有状态为device的时候这个设备才是可用的
                if device[1] == 'device':
                    devices_list.append(device[0])
            return devices_list
        else:
            return None

    def creat_port_list(self, start_port):
        port = Port()
        port_list = port.create_port_list(start_port, self.get_driver())
        return port_list

    def create_command(self):
        # appium -p 4700 -bp 4901 -U JPF4C19123011893
        command_list = []
        appium_port_list = self.creat_port_list(4700)
        bp_port_list = self.creat_port_list(4900)
        driver_list = self.get_driver()
        if driver_list is not None:
            for i in range(len(driver_list)):
                # 　　--no-reset 不要每次都安装apk
                #    --session-override 是指覆盖之前的session
                command = "appium -p " + str(appium_port_list[i]) + " -bp " + str(bp_port_list[i]) + " -U " + str(
                    driver_list[i]) + " --no-reset  --session-override"
                command_list.append(command)
            return command_list
        else:
            return None


if __name__ == "__main__":
    driver = Driver()
    print(driver.get_driver())
    print(driver.create_command())

结果

['JPF4C19123011893']
['appium -p 4700 -bp 4901 -U JPF4C19123011893 --no-reset  --session-override']

多线程启动

import threading

from util.dos_cmd import DosCmd
from util.port import Port


class Driver:
    def __init__(self):
        self.cmd = DosCmd()

    def get_driver(self):
        devices_list = []
        device_info = self.cmd.excute_cmd_result('adb devices')
        # 第一部分为一段文案，从第二段开始才是设备信息，所以只有大于等于2的时候才有设备连接到电脑
        if len(device_info) >= 2:
            for i in device_info:
                if 'List' in i:
                    continue
                # 将一条设备信息分割成一个list,第一部分为设备信息，第二部分为device，也就是设备的状态
                device = i.split('\t')
                # 只有状态为device的时候这个设备才是可用的
                if device[1] == 'device':
                    devices_list.append(device[0])
            return devices_list
        else:
            return None

    def creat_port_list(self, start_port):
        port = Port()
        port_list = port.create_port_list(start_port, self.get_driver())
        return port_list

    def create_command(self):
        # appium -p 4700 -bp 4901 -U JPF4C19123011893
        command_list = []
        appium_port_list = self.creat_port_list(4700)
        bp_port_list = self.creat_port_list(4900)
        driver_list = self.get_driver()
        if driver_list is not None:
            for i in range(len(driver_list)):
                # 　　--no-reset 不要每次都安装apk
                #    --session-override 是指覆盖之前的session
                command = "appium -p " + str(appium_port_list[i]) + " -bp " + str(bp_port_list[i]) + " -U " + str(
                    driver_list[i]) + " --no-reset  --session-override"
                command_list.append(command)
            return command_list
        else:
            return None

    def start_driver(self, i):
        start_list = self.create_command()
        self.cmd.excute_cmd(start_list[i])

    def main(self):
        for i in range(len(self.create_command())):
            appium_start = threading.Thread(target=self.start_driver, args=(i,))
            appium_start.start()


if __name__ == "__main__":
    driver = Driver()
    print(driver.get_driver())
    print(driver.create_command())
    print(driver.main())

结果

['JPF4C19123011893']
['appium -p 4700 -bp 4901 -U JPF4C19123011893 --no-reset  --session-override']
None
(node:83653) [DEP0022] DeprecationWarning: os.tmpDir() is deprecated. Use os.tmpdir() instead.
(node:83653) [DEP0005] DeprecationWarning: Buffer() is deprecated due to security and usability issues. Please use the Buffer.alloc(), Buffer.allocUnsafe(), or Buffer.from() methods instead.
[Appium] Welcome to Appium v1.5.3
[Appium] Non-default server args:
[Appium]   port: 4700
[Appium]   bootstrapPort: 4901
[Appium]   sessionOverride: true
[Appium]   udid: 'JPF4C19123011893'
[Appium]   noReset: true
[Appium] Deprecated server args:
[Appium]   -U,--udid => --default-capabilities '{"udid":"JPF4C19123011893"}'
[Appium]   --no-reset => --default-capabilities '{"noReset":true}'
[Appium] Default capabilities, which will be added to each request unless overridden by desired capabilities:
[Appium]   udid: 'JPF4C19123011893'
[Appium]   noReset: true
[Appium] Appium REST http interface listener started on 0.0.0.0:4700

将上面生成的端口号及设备号写入到文件里

然后在BaseDriver里通过读取yaml文件里的内容再生成driver

封装将内容变成字典格式写入到文件

import os

import yaml
from yaml import FullLoader


class WriteDeviceCommand:
    def read_data(self):
        file_path = os.path.join(os.path.join(os.path.dirname(os.path.dirname(__file__)), "config"),
                                 "deviceconfig.yaml")
        with open(file_path) as f:
            data = yaml.load(f, Loader=FullLoader)
        return data

    # 拼接数据
    def join_data(self, i, device, bp, port):
        data = {
            "device_info_" + str(i): {
                "deviceName": device,
                "bp": bp,
                "port": port
            }
        }
        return data

    def write_data(self, i, device, bp, port):
        data = self.join_data(i, device, bp, port)
        file_path = os.path.join(os.path.join(os.path.dirname(os.path.dirname(__file__)), "config"),
                                 "deviceconfig.yaml")
        with open(file_path, 'a') as f:
            yaml.dump(data, f)

在我们的create_command创建命令的时候就写入到文件里，重点中的重点

import threading

from util.dos_cmd import DosCmd
from util.port import Port
from util.write_device_yaml import WriteDeviceCommand


class Driver:
    def __init__(self):
        self.cmd = DosCmd()
        # 重点7，将device_list放到构造函数里，因为get_driver的作用就是获取devices_list，那么就需要把所有的调用get_driver的地方都替换成devices_list
        self.devices_list = self.get_driver()

    def get_driver(self):
        devices_list = []
        device_info = self.cmd.excute_cmd_result('adb devices')
        # 第一部分为一段文案，从第二段开始才是设备信息，所以只有大于等于2的时候才有设备连接到电脑
        if len(device_info) >= 2:
            for i in device_info:
                if 'List' in i:
                    continue
                # 将一条设备信息分割成一个list,第一部分为设备信息，第二部分为device，也就是设备的状态
                device = i.split('\t')
                # 只有状态为device的时候这个设备才是可用的
                if device[1] == 'device':
                    devices_list.append(device[0])
            return devices_list
        else:
            return None

    def creat_port_list(self, start_port):
        port = Port()
        # 重点8，这里用到了get_driver，要替换一下
        # port_list = port.create_port_list(start_port, self.get_driver())
        port_list = port.create_port_list(start_port, self.devices_list)
        return port_list

    def create_command(self, i):
        # 重点4，既然我们把i传进来了，就把所有调用这个地方的都加上i这个参数
        # appium -p 4700 -bp 4901 -U JPF4C19123011893
        write = WriteDeviceCommand()
        command_list = []
        appium_port_list = self.creat_port_list(4700)
        bp_port_list = self.creat_port_list(4900)
        # 重点9，这里用到了get_driver，要替换一下
        # driver_list = self.get_driver()
        driver_list = self.devices_list

        if driver_list is not None:
            # 重点3， 既然在多线程的时候我们已经循环过create_command了，那这里就不应该再根据driver_list的长度循环了
            # 而这个i其实我们在多线程那块已经有了，我们可以直接拿来用，并将for循环删除掉
            # for i in range(len(driver_list)):
            # 　　--no-reset 不要每次都安装apk
            #    --session-override 是指覆盖之前的session
            command = "appium -p " + str(appium_port_list[i]) + " -bp " + str(bp_port_list[i]) + " -U " + str(
            driver_list[i]) + " --no-reset  --session-override"
            command_list.append(command)
            # 重点3， 每一次调用create_command都会写入文件一次，调用两次，写了两次
            write.write_data(i, driver_list[i], bp_port_list[i], appium_port_list[i])
            return command_list
        else:
            return None

    def start_driver(self, i):
        # 重点5，这里要加上i
        # start_list = self.create_command()
        start_list = self.create_command(i)
        # 重点12，上面已经是的时候我们已经通过i来控制了start_list的内容了，就只可能是1条，那我们下面这一行就不能用i了，直接用0即可
        # self.cmd.excute_cmd(start_list[i])
        self.cmd.excute_cmd(start_list[0])


    def main(self):
        # 重点1，我们在这里循环了一次create_command
        # 重点6， start_driver里面有i，可是在main的入参里可没有i怎么办，我们加上了这个i就会报错。我们发现每次需要获取设备个数的时候都要执行self.create_command(i)
        # 我们可以把设备个数这一部分放到构造函数里，设备个数即为devices_list，在get_driver方法里
        # 重点10，既然我们已经有了devices_list，就不用再调用create_command获取长度了，换成devices_list即可
        for i in range(len(self.devices_list)):
            # 重点2，这里的target是多线程的，并没有直接调用
            # 重点11，我们这里start_driver传的是什么，传的是多线程,如果我们有两个手机，那么这里就会i就会是0，1
            appium_start = threading.Thread(target=self.start_driver, args=(i,))
            appium_start.start()


if __name__ == "__main__":
    driver = Driver()
    print(driver.main())

结果

None
(node:94809) [DEP0022] DeprecationWarning: os.tmpDir() is deprecated. Use os.tmpdir() instead.
(node:94809) [DEP0005] DeprecationWarning: Buffer() is deprecated due to security and usability issues. Please use the Buffer.alloc(), Buffer.allocUnsafe(), or Buffer.from() methods instead.
[Appium] Welcome to Appium v1.5.3
[Appium] Non-default server args:
[Appium]   port: 4700
[Appium]   bootstrapPort: 4901
[Appium]   sessionOverride: true
[Appium]   udid: 'JPF4C19123011893'
[Appium]   noReset: true
[Appium] Deprecated server args:
[Appium]   -U,--udid => --default-capabilities '{"udid":"JPF4C19123011893"}'
[Appium]   --no-reset => --default-capabilities '{"noReset":true}'
[Appium] Default capabilities, which will be added to each request unless overridden by desired capabilities:
[Appium]   udid: 'JPF4C19123011893'
[Appium]   noReset: true
[Appium] Appium REST http interface listener started on 0.0.0.0:4700

在yaml文件中的内容也只有一条

device_info_0:
  bp: 4901
  deviceName: JPF4C19123011893
  port: 4700

写文件之前先清空文件内容

但是我们每次执行都会写入一次文件，我们需要在写之前先清空yaml文件内容

import os

import yaml
from yaml import FullLoader


class WriteDeviceCommand:
    def read_data(self):
        file_path = os.path.join(os.path.join(os.path.dirname(os.path.dirname(__file__)), "config"),
                                 "deviceconfig.yaml")
        with open(file_path) as f:
            data = yaml.load(f, Loader=FullLoader)
        return data

    # 拼接数据
    def join_data(self, i, device, bp, port):
        data = {
            "device_info_" + str(i): {
                "deviceName": device,
                "bp": bp,
                "port": port
            }
        }
        return data

    def write_data(self, i, device, bp, port):
        data = self.join_data(i, device, bp, port)
        file_path = os.path.join(os.path.join(os.path.dirname(os.path.dirname(__file__)), "config"),
                                 "deviceconfig.yaml")
        with open(file_path, 'a') as f:
            yaml.dump(data, f)

    # 清空数据
    def clear_data(self):
        file_path = os.path.join(os.path.join(os.path.dirname(os.path.dirname(__file__)), "config"),
                                 "deviceconfig.yaml")
        # 注意这里是w
        with open(file_path, 'w') as f:
            f.truncate()


if __name__ == "__main__":
    file = WriteDeviceCommand()
    file.clear_data()

在主函数中执行最开始就清空文件

import threading

from util.dos_cmd import DosCmd
from util.port import Port
from util.write_device_yaml import WriteDeviceCommand


class Driver:
    def __init__(self):
        self.cmd = DosCmd()
        # 重点7，将device_list放到构造函数里，因为get_driver的作用就是获取devices_list，那么就需要把所有的调用get_driver的地方都替换成devices_list
        self.devices_list = self.get_driver()
        ## 将WriteDeviceCommand放到构造函数里来
        self.write_file = WriteDeviceCommand()

    def get_driver(self):
        devices_list = []
        device_info = self.cmd.excute_cmd_result('adb devices')
        # 第一部分为一段文案，从第二段开始才是设备信息，所以只有大于等于2的时候才有设备连接到电脑
        if len(device_info) >= 2:
            for i in device_info:
                if 'List' in i:
                    continue
                # 将一条设备信息分割成一个list,第一部分为设备信息，第二部分为device，也就是设备的状态
                device = i.split('\t')
                # 只有状态为device的时候这个设备才是可用的
                if device[1] == 'device':
                    devices_list.append(device[0])
            return devices_list
        else:
            return None

    def creat_port_list(self, start_port):
        port = Port()
        # 重点8，这里用到了get_driver，要替换一下
        # port_list = port.create_port_list(start_port, self.get_driver())
        port_list = port.create_port_list(start_port, self.devices_list)
        return port_list

    def create_command(self, i):
        # 重点4，既然我们把i传进来了，就把所有调用这个地方的都加上i这个参数
        # appium -p 4700 -bp 4901 -U JPF4C19123011893
        ## 已经添加到了构造函数里，所以直接用self即可
        ## write = WriteDeviceCommand()
        write = self.write_file
        command_list = []
        appium_port_list = self.creat_port_list(4700)
        bp_port_list = self.creat_port_list(4900)
        # 重点9，这里用到了get_driver，要替换一下
        # driver_list = self.get_driver()
        driver_list = self.devices_list

        if driver_list is not None:
            # 重点3， 既然在多线程的时候我们已经循环过create_command了，那这里就不应该再根据driver_list的长度循环了
            # 而这个i其实我们在多线程那块已经有了，我们可以直接拿来用，并将for循环删除掉
            # for i in range(len(driver_list)):
            # 　　--no-reset 不要每次都安装apk
            #    --session-override 是指覆盖之前的session
            command = "appium -p " + str(appium_port_list[i]) + " -bp " + str(bp_port_list[i]) + " -U " + str(
                driver_list[i]) + " --no-reset  --session-override"
            command_list.append(command)
            # 重点3， 每一次调用create_command都会写入文件一次，调用两次，写了两次
            write.write_data(i, driver_list[i], bp_port_list[i], appium_port_list[i])
            return command_list
        else:
            return None

    def start_driver(self, i):
        # 重点5，这里要加上i
        # start_list = self.create_command()
        start_list = self.create_command(i)
        # 重点12，上面已经是的时候我们已经通过i来控制了start_list的内容了，就只可能是1条，那我们下面这一行就不能用i了，直接用0即可
        # self.cmd.excute_cmd(start_list[i])
        self.cmd.excute_cmd(start_list[0])

    def main(self):
        ## 我们将清理文件的功能放到主函数里来，还记得在create_command我们也调用了写文件的操作，那我们直接将WriteDeviceCommand写到构造函数里，然后再将所有的自己定义替换掉即可
        self.write_file.clear_data()
        # 重点1，我们在这里循环了一次create_command
        # 重点6， start_driver里面有i，可是在main的入参里可没有i怎么办，我们加上了这个i就会报错。我们发现每次需要获取设备个数的时候都要执行self.create_command(i)
        # 我们可以把设备个数这一部分放到构造函数里，设备个数即为devices_list，在get_driver方法里
        # 重点10，既然我们已经有了devices_list，就不用再调用create_command获取长度了，换成devices_list即可
        for i in range(len(self.devices_list)):
            # 重点2，这里的target是多线程的，并没有直接调用
            # 重点11，我们这里start_driver传的是什么，传的是多线程,如果我们有两个手机，那么这里就会i就会是0，1
            appium_start = threading.Thread(target=self.start_driver, args=(i,))
            appium_start.start()


if __name__ == "__main__":
    driver = Driver()
    print(driver.main())

设计BaseDriver和Unittest

BaseDriver

from appium import webdriver
import time

from util.write_device_yaml import WriteDeviceCommand


class Base_Driver:
    def __init__(self):
        # 已经把设备信息写入到了文件里，那我们就再从文件里取出来
        self.device_info = WriteDeviceCommand()

    # 由于获取device信息需要一个参数i，所以我们需要在方法的入参上添加i，并且从page到handle和bussiness全要新增这个i
    def get_android_driver(self, i):
        device_info_list = self.device_info.read_data()
        device = device_info_list.get("device_info_" + str(i)).get("deviceName")
        port = device_info_list.get("device_info_" + str(i)).get("port")

        # 需要定制化
        # server = 'http://localhost:4723/wd/hub'
        server = "http://localhost:" + str(port) + "/wd/hub"
        caps = {
            "platformName": "Android",
            # 需要定制化
            "deviceName": device,
            "platformVersion": "10.0",
            "appPackage": "com.czb.webczbdemo",
            "appActivity": "com.czb.webczbdemo.MainActivity",
            "automationName": "uiautomator2",
            "noReset": "true",
            "unicodeKeyboard": "true",
            "resetKeyboard": "true"
        }
        android_driver = webdriver.Remote(server, caps)
        time.sleep(10)
        return android_driver

    def get_ios_driver(self):
        pass

Unittest

但是我们发现这个i从哪里传过来呢

import threading
import unittest
from bussiness.select_bussinessthree import *

from util import HTMLTestRunner


class CaseTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        # 案例说我们应该在这里传入的，cls.select_buss = Select_Bussiness(i)，但是这样传入会报错
        cls.select_buss = Select_Bussiness()
        print("测试开始")

    def test_select(self):
        self.select_buss.select_success()
        time.sleep(3)

    @classmethod
    def tearDownClass(cls):
        print("测试结束")


def get_suite(z):
    suite = unittest.TestSuite()
    suite.addTest(CaseTest("test_select"))
    print("执行了没")

    html_file = os.path.join(os.path.join(os.path.dirname(__file__), "report"), "report" + str(z) + ".html")
    with open(html_file, 'wb') as f:
        HTMLTestRunner.HTMLTestRunner(f).run(suite)


if __name__ == "__main__":
    threads = []
    for i in range(4):
        t = threading.Thread(target=get_suite, args=(i, ))
        threads.append(t)
    for j in threads:
        j.start()

我们进入到TestCase的源码里看一下，发现是没有地方放我们i参数的

    def __init__(self, methodName='runTest'):
        """Create an instance of the class that will use the named test
           method when executed. Raises a ValueError if the instance does
           not have a method with the specified name.
        """
        self._testMethodName = methodName
        self._outcome = None
        self._testMethodDoc = 'No test'
        try:
            testMethod = getattr(self, methodName)
        except AttributeError:
            if methodName != 'runTest':
                # we allow instantiation with no explicit method name
                # but not an *incorrect* or missing method name
                raise ValueError("no such test method in %s: %s" %
                      (self.__class__, methodName))
        else:
            self._testMethodDoc = testMethod.__doc__
        self._cleanups = []
        self._subtest = None

        # Map types to custom assertEqual functions that will compare
        # instances of said type in more detail to generate a more useful
        # error message.
        self._type_equality_funcs = {}
        self.addTypeEqualityFunc(dict, 'assertDictEqual')
        self.addTypeEqualityFunc(list, 'assertListEqual')
        self.addTypeEqualityFunc(tuple, 'assertTupleEqual')
        self.addTypeEqualityFunc(set, 'assertSetEqual')
        self.addTypeEqualityFunc(frozenset, 'assertSetEqual')
        self.addTypeEqualityFunc(str, 'assertMultiLineEqual')

怎么办，我们先用一个测试脚本来实验，等成功了之后再放到case层

"""
@File    :   test_unittest_czb.py    
@Contact :   xxx@xxx.com
@Modify Time      @Author    @Version
------------      -------    --------    
2021/7/26 2:07 下午   xxx        1.0
@Desciption： 演示unittest基本用法
"""
import os
import threading
import unittest

from util import HTMLTestRunner


class ParamTestCase(unittest.TestCase):
    # 重写TestCase的构造函数，添加上一个param，用来放我们的i
    def __init__(self, methodName='runTest', param=None):
        # 其他的构造函数内容不变。
        super(ParamTestCase, self).__init__(methodName)
        # 但是这样还不行，因为setUpClass是类方法，我们这里是self，setUpClass是取不到这个值的，需要将param设置为全局变量
        # self.param = param
        global params
        params = param


class Czb_Test_One(ParamTestCase):
    @classmethod  # 声明为类方法（必须）
    def setUpClass(cls):  # 类方法，注意后面是cls，整个类只执行一次
        print("setUPClass--->" + str(params))

    @classmethod
    def tearDownClass(cls):
        print("在类执行完成后调用，一般用来清理环境，不过我一般都是用来打印一句话，提示一下用例执行完了就可以了")

    def setUp(self):  # 每个用例执行前先执行setUp里的内容
        print("setUp--->" + str(params))

    def tearDown(self):
        print("tearDown(self)")

    def test_one(self):
        print("test_one------>"+str(params))


def get_suite(z):
    suite = unittest.TestSuite()
    suite.addTest(Czb_Test_One("test_one", param=z))
    unittest.TextTestRunner().run(suite)


if __name__ == "__main__":
    threads = []
    for i in range(4):
        # 为什么要加一个逗号，这个args不是必传的，如果没有参数的话也可以不传
        # https://blog.csdn.net/yasyal515/article/details/86014915
        # 这里记得不要写成threading.Thread(target=get_suite())
        t = threading.Thread(target=get_suite, args=(i,))
        threads.append(t)
    for j in threads:
        j.start()

结果,由于是多线程，所以结果可能不是每次都一样

setUPClass--->0
setUp--->0
test_one------>0
tearDown(self)
在类执行完成后调用，一般用来清理环境，不过我一般都是用来打印一句话，提示一下用例执行完了就可以了
setUPClass--->1
setUp--->1
test_one------>1
tearDown(self)
在类执行完成后调用，一般用来清理环境，不过我一般都是用来打印一句话，提示一下用例执行完了就可以了
setUPClass--->2
setUp--->2
test_one------>2
tearDown(self)
在类执行完成后调用，一般用来清理环境，不过我一般都是用来打印一句话，提示一下用例执行完了就可以了
setUPClass--->3
setUp--->3
test_one------>3
tearDown(self)
在类执行完成后调用，一般用来清理环境，不过我一般都是用来打印一句话，提示一下用例执行完了就可以了
.
----------------------------------------------------------------------
.
----------------------------------------------------------------------
.
----------------------------------------------------------------------
Ran 1 test in 0.000s
Ran 1 test in 0.000s

OK

OK
Ran 1 test in 0.000s

OK
.
----------------------------------------------------------------------
Ran 1 test in 0.000s

OK

然后 我们就将上面改造后的ParamTestCase移植到test层，并且我们现在在case层传入了i，也只是调用了BaseDriver，而BaseDriver是根据文件里的数据来的，文件里的数据又是在Driver调用main之后生成的，所以，我们现在要将这个main也引入进来，同时我们我们多线程的次数在上面是写死的，那么我们要根据手机的个数来确定循环的次数，我们在写入文件的时候不就知道有多少台手机了么

所以我们改造一下读写yaml文件的类

import os

import yaml
from yaml import FullLoader


class WriteDeviceCommand:
    def read_data(self):
        file_path = os.path.join(os.path.join(os.path.dirname(os.path.dirname(__file__)), "config"),
                                 "deviceconfig.yaml")
        with open(file_path) as f:
            data = yaml.load(f, Loader=FullLoader)
        return data

    # 拼接数据
    def join_data(self, i, device, bp, port):
        data = {
            "device_info_" + str(i): {
                "deviceName": device,
                "bp": bp,
                "port": port
            }
        }
        return data

    def write_data(self, i, device, bp, port):
        data = self.join_data(i, device, bp, port)
        file_path = os.path.join(os.path.join(os.path.dirname(os.path.dirname(__file__)), "config"),
                                 "deviceconfig.yaml")
        with open(file_path, 'a') as f:
            yaml.dump(data, f)

    def get_value(self, key, port):
        dataone = self.read_data()
        value = dataone[key][port]
        return value

    # 清空数据
    def clear_data(self):
        file_path = os.path.join(os.path.join(os.path.dirname(os.path.dirname(__file__)), "config"),
                                 "deviceconfig.yaml")
        # 注意这里是w
        with open(file_path, 'w') as f:
            f.truncate()

    # 返回数据长度，这样多线程循环的时候才知道循环多少次
    def get_file_lines(self):
        data = self.read_data()
        return len(data)


if __name__ == "__main__":
    file = WriteDeviceCommand()
    data = file.read_data()
    print(file.get_value("device_info_0", "port"))

多进程执行测试用例

这样在case层再将get_file_lines封装一下，并在执行的时候加上多进程，看清楚是多进程，因为多线程在执行测试用例的时候会有各种各样的问题，那我们就用多进程来替代

import unittest
import multiprocessing
from bussiness.select_bussinessthree import *

# 这样做有一个问题就是用的不是同一个driver，会启动两次app
from util.HTMLTestRunner import HTMLTestRunner
from util.driver import Driver
from util.write_device_yaml import WriteDeviceCommand


class ParamTestCase(unittest.TestCase):
    # 重写TestCase的构造函数，添加上一个param，用来放我们的i
    def __init__(self, methodName='runTest', param=None):
        # 其他的构造函数内容不变。
        super(ParamTestCase, self).__init__(methodName)
        # 但是这样还不行，因为setUpClass是类方法，我们这里是self，setUpClass是取不到这个值的，需要将param设置为全局变量
        # self.param = param
        global params
        params = param


# class CaseTest(unittest.TestCase):
# 这样我们也不继承unittest了，直接继承ParamTestCase
class CaseTest(ParamTestCase):
    @classmethod
    def setUpClass(cls):
        # 案例说我们应该在这里传入的，cls.select_buss = Select_Bussiness(i)，但是这样传入会报错
        # 将param设置为全局变量之后，这里就可以传入i了,但是这个i，现在就是全局变量params，所以我们要传params
        cls.select_buss = Select_Bussiness(params)
        print(params)
        print("测试开始")

    def test_select(self):
        self.select_buss.select_success()
        time.sleep(3)

    @classmethod
    def tearDownClass(cls):
        print("测试结束")


def get_suite(z):
    suite = unittest.TestSuite()
    suite.addTest(CaseTest("test_select", z))
    html_file = os.path.join(os.path.join(os.path.dirname(os.path.dirname(__file__)), "report"), "report"+str(z)+".html")
    with open(html_file, 'wb') as f:
        HTMLTestRunner(f).run(suite)


# 将driver引进并封装成一个方法，这样只要在这里调用这个方法，就可以直接执行appium服务端命令，并写入文件，这样BaseDriver才会拿到数据
def start_server():
    driver = Driver()
    # 由于真正启动appium服务端不会很快，所以我们在main函数最后加上一个time.sleep(20)
    driver.main()


def get_count():
    write_device_command = WriteDeviceCommand()
    count = write_device_command.get_file_lines()
    return count
    print("count"+str(count))


if __name__ == "__main__":
    start_server()
    threads = []
    for i in range(get_count()):
        t = multiprocessing.Process(target=get_suite, args=(i,))
        threads.append(t)
    for j in threads:
        j.start()