一直假装没有时间整理自动化的东西，想来这笔债不能总是拖。大概年前的时候组里说要尝试着进行自动化方面的工作，就做了相关方面知识的学习。当然对于一个普通的黑盒测试人员来说，我们选择了从UI自动化入手。

一、需求——确定框架

开始做Android自动化只为了解决“多台设备同时自动执行一套测试代码，并输出相关日志或者图片的log”。因为大家的代码能力都不高，关于使用哪种工具，并没有经过太多的探讨，选择了较容易入手的Appium + python unittest框架。

以此框架就确定了下来：以python unittest为基础，并对Appium进行封装。

后来加了临时需求，说既然要并发跑appium，那么也能用来并发运行monkey test。

以此对应的需求：UI test + monkey test

尝试了一些开源的框架，但不太适合我们产品和需求，因此只好自己写一个简单的框架出来。

二、框架——搭建

前面两篇文章已介绍了搭建过程以及环境问题，这里就不再冗余，具体请见：Android自动化 -- Appium环境搭建、Mac OSX上的python环境

官网上比较仔细地介绍了Android并发测试，详见Android并发测试

三、模块介绍——所解决的问题

由于我们的产品在使用前必须登陆，需要保证每个设备登陆不同的账号。细细想来，我们需要解决的问题大概有如下几点：

1.即可运行appium又可运行monkey，但又不能同时运行这两个任务 --> 任务划分，区分monkey和appium服务
2.根据多设备启动多个appium --> Appium Server模块
  |-- 负责处理Appium Server启动，停止，监听等 --> server的模块  
  |-- 负责处理多设备信息的模块 --> Device Object
  |-- 负责处理多个登陆用户信息的模块 --> User Object
3.封装常用方法，统一放在一个地方进行 --> Tester Object模块
4.基于unittest管理testcase --> TestCaseManager模块
  |-- Testcase --> 可测试的用例
  |-- TestLoader --> 创建test suit
  |-- TextTestRunner --> RunTestManager模块
5.处理异常 --> 沿用unittest框架的处理方式
6.测试结果报告html形式输出 --> TestResult模块
7.优化初始化如安装、卸载、登陆、处理权限、拷贝测试图片等等准备工作 --> BaseDevicePreProcess模块

1. 任务划分，区分monkey和appium服务

这里是这么构思的，构建一个SimpleHTTPServer，每次执行任务前先请求Server，Server端判断当前是否有正在执行的任务，如果有正在执行的，就返回个错误信息；如果没有，就开始执行任务。

class HttpServerHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
    //省略好多内容
    def run(self, params):
        if share.get_if_run() == True:
            result_dict = {'code':1002,"data":{"message":"已经有一个任务在执行","taskid":"%s" % share.get_taskid()}}
            self.set_response(result_dict)
            return
        if params.has_key('mode') == False:
            result_dict = {'code':1003,"data":{"message":"缺少mode参数"}}
            self.set_response(result_dict)
            return
        elif params['mode'][0] != "monkey" and params['mode'][0] != 'autotest':
            self.set_response({'code':1004, "data":{"message":"mode参数错误"}})
            return

        try:
            set_run_manager(RunTestManager(params['mode'][0]))
            self.taskid = get_run_manager().task_id
            share.set_taskid(get_run_manager().task_id) #设置全局共享taskid
            share.set_if_run(True)
            thread = threading.Thread(target=get_run_manager().start_run)
            thread.start()
            result_dict = {'code':0,"data":{"taskid fuck":"%s" % self.taskid,"message":"开始执行%s任务" % params['mode']}}
            self.set_response(result_dict)
        except Exception, e:
            traceback.print_exc()
            get_run_manager().stop_run()

2. 根据多设备启动多个appium

// 一个例子
$ appium -p 4736 -bp 4836 -U b33aa57c --session-override
// -p Appium的主要端口
// -bp Appium bootstrap端口
// -U 设备id

启动多个设备需要运行时根据不同的端口进行appium配置。所以得先有个处理devices的模块。按照惯例，把安装包和设备的详细信息和用户登陆账号等以list方式写进config文件里，后面再读出来。

// congif.yaml 文件
NiceAPK: /Users/xxxxx/com.nice.main.apk    # 测试包的路径
Devices:
 - deviceid: 5HUC9S6599999999    # 设备识别adb devices的值
   devicename: OPPO_R9M    # 设备的名称，用于区分
   serverport: 4723    # -p Appium的主要端口，设备之间不能重复
   bootstrapport: 4823    # -bp Appium bootstrap端口，设备之间不能重复
   platformname: Android    # desired_caps
   platformversion: 5.1    # desired_caps
   server: 127.0.0.1     # 地址

 - deviceid: 7c404969
   devicename: OPPO_A33
   serverport: 4724
   bootstrapport: 4824
   platformname: Android
   platformversion: 5.1.1
   server: 127.0.0.1

Users:
 - uid: 33333333333
   username: test01
   mobile: 33333333333
   password: 333333

 - uid: 44444444444
   username: test02
   mobile: 44444444444
   password: 444444

然后分别创建Device object和User object（和Device object一致）

class Device(object):
    def __init__(self, deviceid):
        self._deviceid = deviceid
        self._devicename = ""
        self._platformversion = ""
        self._platformname = ""
        self._bootstrapport = ""
        self._serverport = ""
        self._server = ""

然后我们可以通过DataProvider来实例化设备和用户信息

class DataProvider(object):
    @classmethod
    def load_devices(cls):
        cls.devicenamelist = []
        for device in cls.config['Devices']:
            deviceobject = Device(device['deviceid'])
            deviceobject.devicename = device['devicename']
            deviceobject.serverport = device['serverport']
            deviceobject.bootstrapport = device['bootstrapport']
            deviceobject.platformname = device['platformname']
            deviceobject.platformversion = device['platformversion']
            deviceobject.server = device['server']
            cls.devices[deviceobject.deviceid] = deviceobject
            cls.devicenamelist.append(device['devicename'])
        Log.logger.info(u"配置列表中一共有 %s 台设备" % len(cls.devices))

    @classmethod
    def load_users(cls):
        for user in cls.config['Users']:
            userobject = User(user['uid'])
            userobject.username = user['username']
            userobject.mobile = user['mobile']
            userobject.password = user['password']
            cls.users.append(userobject)
        Log.logger.info(u"配置列表中一共有 %s 个用户信息" % len(cls.users))

有了devices和users，后面我们就可以创建个server类来处理appium server的启动、停止、监听设备等等功能。例如根据多设备来启动多个appium

class Server:
    def __init__(self, deviceobject):
        self.logger = Log.logger
        self._deviceobject = deviceobject
        self._cmd = "appium -p %s -bp %s -U %s --session-override" % (
        self._deviceobject.serverport, self._deviceobject.bootstrapport, self._deviceobject.deviceid)

3. 封装常用方法的Tester类

这里的tester用于存放driver、共用的封装方法，如点击、滑动、截视频、图像对比等等方法

class Tester(object):
    def __init__(self, driver):
        self._driver = driver
        self._user = None
        self._device = None
        self._logger = None
        self.action = TouchAction(self._driver)
        self._screenshot_path = ""
        self.device_width = self._driver.get_window_size()['width']
        self.device_height = self._driver.get_window_size()['height']

4. 管理TestCase的TestCaseManager类

因为是基于python unittest，我们沿用unittest的方式，这里只添加一个参数化功能，用来方便我们指定所需要测试的集合。

class BaseTestCase(unittest.TestCase):
    def __init__(self, methodName='runTest', tester=None):
        super(BaseTestCase, self).__init__(methodName)
        self.tester = tester

    @staticmethod
    def parametrize(testcase_klass, tester=None):
        testloader = unittest.TestLoader()
        testnames = testloader.getTestCaseNames(testcase_klass)
        suite = unittest.TestSuite()
        for name in testnames:
            suite.addTest(testcase_klass(name, tester=tester))
        return suite

然后创建个TestCaseManager来处理不同种类的测试类型

class TestCaseManager(object):

    def __init__(self, tester):
        self.compatibility_suite = unittest.TestSuite()
        self.testcase_class = []
        self.load_case()
        self.tester = tester

    def load_case(self):
        testcase_array = []
        testsuits = unittest.defaultTestLoader.discover('testcase/', pattern='test*.py')
        for testsuite in testsuits:
            for suite in testsuite._tests:
                for test in suite:
                    testcase_array.append(test.__class__)
        self.testcase_class = sorted(set(testcase_array), key=testcase_array.index)

    # 兼容性测试用例
    def compatibility_testsuite(self):
        for testcase in self.testcase_class:
            self.compatibility_suite.addTest(BaseTestCase.parametrize(testcase, tester=self.tester))
        return self.compatibility_suite

    # monkey自动化
    def monkey_android(self):
        self.tester.run_monkey(200,1000)

    # 功能性测试用例
    def functional_testsuite(self):
        pass

    # 单独运行一条指定的用例
    def signal_case_suit(self, test_myclass):
        suite = unittest.TestSuite()
        suite.addTest(BaseTestCase.parametrize(test_myclass, tester=self.tester))
        return suite

那么实际要运行的时候，我们在TextTestRunner传个参数来指定运行的suit就可以了

  suite = TestCaseManager(tester).compatibility_testsuite()    # 运行兼容集合
  // suite = TestCaseManager(tester).functional_testsuite()      # 运行功能测试集合
  // suite = TestCaseManager(tester).signal_case_suit(test_case_001)    # 运行单条测试用例
  unittest.TextTestRunner(verbosity=2, resultclass=TheTestResult).run(suite)

TextTestRunner的执行部分写在了RunTestManager里

class RunTestManager(object):
    def start_run(self):
      //判断执行的类型，并调用start_run_test方法
    def start_run_test(self):
      //初始化tester object，并调用run方法并传tester object参数
    def init_tester_data(self, device, which_user):
      //初始化tester object
    def run(self, tester):
      //预处理（登陆、权限等流程），并调用unittest.TextTestRunner开始执行
    def stop_run(self):
      //结束运行，置server flag为false，表示当前不在有任务运行

5. 处理异常

沿用unittest框架的处理方式，在TestResult中重写addError、addFailure、addSuccess、addSkip等等一系列方法来满足我们自己的需求。特别是对addFailure的处理，我们需要详尽的知道哪台设备的哪里出了错，并且能输出截图和log日志。

   def addFailure(self, test, err):
        info = '************      - %s -!(Fail)    ***************' % self.tester.device.devicename
        self.logger.warning(info)
        info = 'Fail device:%s Run TestCase %s, Fail info:%s' % (self.tester.device.devicename, test, err[1].message)
        self.logger.warning(info)
        info = '***********************************************'
        self.logger.warning(info)

        # 失败截图
        mytest = str(test)
        simplename = clean_brackets_from_str(mytest).replace(' ', '')
        myscr = "Failure_%s" % simplename
        self.tester.screenshot2(myscr)

        # 失败日志
        list = traceback.format_exception(err[0], err[1], err[2])
        list_fail = []  # 列表包含要输出的错误日志信息
        # list_fail[0]='error:'
        # list_fail[1]=list[2:3]
        # list_fail[2]=list[-1]
        list_fail.append(list[-1])
        list_fail.append(list[2])

        self.__class__.totalresults[self.deviceid]['failtestcase'] = self.__class__.totalresults[self.deviceid]['failtestcase'] + 1

        self.__class__.detailresults[self.deviceid][test]['result'] = 'Fail'
        self.__class__.detailresults[self.deviceid][test]['reason'] = list_fail

6. 测试结果报告html形式输出

同上面的异常处理，结果的输出也放在TestResult来执行。不知道当时怎么想的，输出处理这块用了pyh。所有的表格都是一点点画出来的，心很累，还抽空搞了下css和js，美化了一下样式。代码很长就不贴了，基本是一个div一个div写出来的。直接看源码就好了，这里不展开啦。

样式上还有bug。。。。因为一些设备意外退出导致的，这个暂时won't fix。。。。

7. 关于预处理部分，优化初始化过程

这里有很多的工作，比如安装，处理每台设备登陆过程，处理登陆界面的权限问题，拷贝测试图片等等。毕竟只有登陆了才能进行测试！！！
1）因为不想每次启动appium都要安装setting\unlock\ime等apk，所以修改了Appium源码，不让他自己安装。运行的时候，由我们自己的函数处理安装过程

// 干掉自动安装
文件: /usr/local/lib/node_modules/appium/node_modules/appium-android-driver/lib/driver.js，注释以下几句代码
await this.adb.uninstallApk(this.opts.appPackage);
await helpers.installApkRemotely(this.adb, this.opts);
await helpers.resetApp(this.adb, this.opts.app, this.opts.appPackage, this.opts.fastReset);
await this.checkPackagePresent();

文件：/usr/local/lib/node_modules/appium/node_modules/appium-android-driver/build/lib/driver.js 注释以下几句代码
return _regeneratorRuntime.awrap(_androidHelpers2['default'].resetApp(this.adb, this.opts.app, this.opts.appPackage, this.opts.fastReset));
return _regeneratorRuntime.awrap(this.adb.uninstallApk(this.opts.appPackage));
return _regeneratorRuntime.awrap(_androidHelpers2['default'].installApkRemotely(this.adb, this.opts));
return _regeneratorRuntime.awrap(this.checkPackagePresent());

文件：/usr/local/lib/node_modules/appium/node_modules/appium-android-driver/lib/android-helpers.js 注释以下几句代码
await adb.install(unicodeIMEPath, false);
await helpers.pushSettingsApp(adb);
await helpers.pushUnlock(adb);

文件 /usr/local/lib/node_modules/appium/node_modules/appium-android-driver/build/lib/android-helpers.js 替换以下几句代码
return _regeneratorRuntime.awrap(helpers.initUnicodeKeyboard(adb)) 替换为return context$1$0.abrupt('return', defaultIME);
return _regeneratorRuntime.awrap(helpers.pushSettingsApp(adb)); 替换为return context$1$0.abrupt('return', defaultIME);
return _regeneratorRuntime.awrap(helpers.pushUnlock(adb)); 替换为return context$1$0.abrupt('return', defaultIME);

2）由于不同设备的安装会有极大的不同，比如有的需要确认usb安装，有的设备会询问你是否安装；高API会弹授权提示，低API没有提示等等不协调的地方有很多。因此统一写个了PreProManager类来管理设备，目的是给每一台设备分配他自己的执行函数

class PreProManager(object):
    def __init__(self, tester):
        self.tester = tester
        self.deviceid = self.tester.device.deviceid

    def device(self):
        if self.deviceid == "5HUC9S6599999999":
            return OPPOR9PreProcess(self.tester)
        elif self.deviceid =="7c404969":
            return OPPOA33PreProcess(self.tester)

然后写个BaseDevicePreProcess基类描述预处理过程，上面的各个设备的执行函数直接继承这个基类，并复写里面的一些方法就行了

class BaseDevicePreProcess(object):
    def __init__(self, tester):
        self.tester = tester
        self.driver = self.tester.driver
        self.action = TouchAction(self.driver)
        self.user = self.tester.user

    # 开始预处理流程
    def pre_process(self):
      // 卸载、安装等等

    # 安装流程
    def install_app(self):
        self.driver.install_app(DataProvider.niceapk)

    # 版本升级
    def upgrade_app(self):
      // ...

    # 该流程包括处理安装及启动过程中的各种弹窗，一直到可以点击login按钮
    def install_process(self):
        pass
      // 由子类复写 

    # 该流程包括点击login按钮到达登录页面，并登录
    def login_process(self):
      // 处理登陆流程 

    # 该流程包括登录成功后，对各种自动弹出对话框进行处理
    def login_success_process(self):
        pass
      // 由子类复写 

    # 对所有需要的权限进行处理，例如：相机、录音
    def get_permission_process(self):
        pass
      // 由子类复写      

    def data_prepare(self):
      // 写入测试data

这里举例说明每个设备如何继承基类打造自己的专属处理流程

from BaseDevicePreProcess import *
class OPPOR9PreProcess(BaseDevicePreProcess):
    def __init__(self,tester):
        super(OPPOR9PreProcess, self).__init__(tester)  

    def install_process(self):
        // OPPOR9的专属登陆处理方法
        // 如果不需要复写，则直接用基类中的默认流程执行

    def login_success_process(self):
        // 处理登陆呦
        // 如果不需要复写，则直接用基类中的默认流程执行

    def get_permission_process(self):
        //  OPPOR9的专属处理授权问题方法呦
        // 如果不需要复写，则直接用基类中的默认流程执行

总结：

搭建框架的过程中，遇到了很多困难，不过很开心的是基本都解决了。现有的这些已经能在项目中run起来，但仍有诸多地方不够完善需要持续优化。
慢慢加油吧
代码已上传至github：
https://github.com/h080294/appium_python_android.git

关注获取更多