pytest+requests+excel 接口自动化框架

Posted 2021-02-13 guo2733

一、项目框架如图：

1、common ：这个包都是一些公共的方法，如：手机号加解密，get/post接口请求的方法封装，接口鉴权，发邮件，读写excel文件方法等等

2、result：存放每次运行的log和测试报告

3、testcase：这个包放test开头的测试用例

4、testFile：这个包放接口上传的图片（如注册需要上传头像），excel管理的接口参数

5、caselist.txt：需要运行的用例

6、config.ini ：放一些配置信息，如发邮件的房间账户，接口需要使用的固定参数和版本，发送邮件的开关，以及动态的数据，如动态更改了就将最新的写在这个配置里面，其他地方需要用的就从配置文件里面取最新生成的。

7、readConfig.py ：操作配置文件的一些方法，获取数据，修改数据等

8、runAll.py：运行所有用例 

2、excel表里面的参数：

　　后面这个code和msg是预期结果，用例里面会获取实际结果，然后跟这个预期结果比较判断接口是否成功。

 

3、接口的写法

接口是采用参数化实现的，传入不用的参数对比不同的结果

aes = aesCryter.AesCrypter()
authen= authentication.getAuthentication()
login_xls = commons.get_xls("userCase.xlsx", "login")
localReadConfig = readConfig.ReadConfig()
configHttp = ConfigHttp.ConfigHttp()
info = {}

@paramunittest.parametrized(*login_xls)
class Login(unittest.TestCase):
    def setParameters(self, case_name,url, method, token, par, code, msg):
        self.case_name = str(case_name)
        self.url = str(url)
        self.method = str(method)
        self.token = str(token)
        self.par = eval(str(par))
        self.code = str(code)
        self.msg = str(msg)
        self.return_json = None
        self.info = None

    def setUp(self):
        self.log = Log.MyLog.get_log()
        self.logger = self.log.get_logger()
        print(self.case_name+"测试开始前准备")

    def testLogin(self):
        get_url = authen.get_Url("userCase.xlsx", "login",0,1)
        ydtoken = authen.get_ydtoken(url = get_url)        #获取ydtoken
        new_url =localReadConfig.get_http(‘url‘)+get_url+"&ydtoken="+ydtoken  #拼接接口的请求链接
        configHttp.set_url(new_url)
        print("第一步：设置url  "+new_url)

        # get visitor token
        if self.token == ‘0‘:
            token_v = None
        elif self.token == ‘1‘:
            businessCommon.set_visitor_token_to_config()
            token_v = localReadConfig.get_headers(‘token_v‘)
            header = {"cookie": "SSO=%s" % str(token_v)}
            configHttp.set_headers(header)
        print("第二步：设置header(token等)")

        # set params
        data =self.par
        data[‘mobile‘] = aes.encrypt(str(data[‘mobile‘]))          #对手机号和密码继续加密
        data[‘password‘] = aes.encrypt(str(data[‘password‘]))
        data["token"] = str(token_v)
        configHttp.set_data(data)
        print("第三步：设置发送请求的参数")

        # test interface
        self.return_json = configHttp.post()
        print(self.return_json)
        print("第四步：发送请求方法")

        # check result
        self.checkResult()
        print("第五步：检查结果")

    def tearDown(self):
        """

        :return:
        """
        self.log.build_case_line(self.case_name, actualcode, actualmsg)
        print("测试结束，输出log完结

")

    def checkResult(self):
        """
        check test result
        :return:
        """
        global actualcode,actualmsg
        actualcode = commons.get_value_from_return_json(self.return_json, ‘code‘)
        print(‘实际code:%s‘%actualcode)
        actualmsg = commons.get_value_from_return_json(self.return_json, ‘msg‘)
        print(‘实际msg:%s‘%actualmsg)
        self.assertEqual(actualcode, self.code)
        self.assertEqual(actualmsg, self.msg)


if __name__ == ‘__main__‘:
    unittest.main

4、接口关联，

接口和接口之间有这依赖关系，必须上一个接口请求完成，后面这个接口才能才能，所以在common公共包里面写了一个通用的方法，如果涉及到有依赖关系的，就调用这个接口，把需要先请求的接口先完成了，再请求要测试的接口

比如：接下来这个忘记密码接口，更改密码，必须先请求发送验证码接口，再请求验证已验证接口，最后才能请求更改密码接口

import unittest
import paramunittest
import readConfig as readConfig
from common import Log as Log
from common import commons
from common import configHttp as ConfigHttp
from common import businessCommon
from common import get_Authentication as authentication


authen= authentication.getAuthentication()
forgotcheck = commons.get_xls("userCase.xlsx", "forgotcheck")
localReadConfig = readConfig.ReadConfig()
configHttp = ConfigHttp.ConfigHttp()
info = {}


@paramunittest.parametrized(*forgotcheck)
class ForgotCheck(unittest.TestCase):
    def setParameters(self, case_name, url,method, token, par, code, msg):
        self.case_name = str(case_name)
        self.url = str(url)
        self.method = str(method)
        self.token = str(token)
        self.par = eval(str(par))
        self.code = str(code)
        self.msg = str(msg)
        self.return_json = None
        self.info = None


    def setUp(self):
        self.log = Log.MyLog.get_log()
        self.logger = self.log.get_logger()
        print(self.case_name+"测试开始前准备")

    def testforgotcheck(self):
        # set url
        get_url = authen.get_Url("userCase.xlsx", "forgotcheck", 0, 1)
        ydtoken = authen.get_ydtoken(url=get_url)  # 获取ydtoken
        new_url = localReadConfig.get_http(‘url‘) + get_url + "&ydtoken=" + ydtoken  # 拼接鉴权的请求链接
        configHttp.set_url(new_url)
        print("第一步：设置url  " + new_url)

        # get visitor token
        if self.token == ‘0‘:
            token_v = None
        elif self.token == ‘1‘:
            businessCommon.set_visitor_token_to_config()
            token_v = localReadConfig.get_headers(‘token_v‘)
            header = {"cookie": "SSO=%s" % str(token_v)}
            configHttp.set_headers(header)
        print("第二步：设置header(token等)")

        # set params
        data = self.par
        data["token"] = str(token_v)
        configHttp.set_data(data)
        print("第三步：设置发送请求的参数")

        # test interface
        self.return_json = configHttp.post()
        print("第四步：发送请求方法")

        # check result
        self.checkResult()
        print("第五步：检查结果")

    def tearDown(self):
        """

        :return:
        """
        self.log.build_case_line(self.case_name, actualcode, actualmsg)
        print("测试结束，输出log完结

")

    def checkResult(self):
        """
        check test result
        :return:
        """
        global actualcode,actualmsg
        actualcode = commons.get_value_from_return_json(self.return_json, ‘code‘)
        print(‘实际code:%s‘%actualcode)
        actualmsg = commons.get_value_from_return_json(self.return_json, ‘msg‘)
        print(‘实际msg:%s‘%actualmsg)
        self.assertEqual(actualcode, self.code)
        self.assertEqual(actualmsg.strip(), self.msg)

5、控制接口是否需要运行 ，每次新增用例把用例名字写在这个 caselist.txt  ，不需要运行就注释即可。

#reqSer/test_001_getRequst
#reqSer/test_002_addRequst
#reqSer/test_003_userRadar
user/test_001_Login
#user/test_002_Sign_sms
#user/test_003_Forgotsms
#user/test_004_Forgotcheck
#user/test_005_forgotpass
#user/test_006_NoParameterization

6、运行所有用例 runAll.py

import os
import unittest
from common.Log import MyLog as Log
import readConfig as readConfig
from common import  htmlTestRunner
from common.configEmail import MyEmail

localReadConfig = readConfig.ReadConfig()


class AllTest:
    def __init__(self):
        global log, logger, resultPath, on_off
        log = Log.get_log()
        logger = log.get_logger()
        resultPath = log.get_report_path()
        print(resultPath)
        on_off = localReadConfig.get_email("on_off")
        self.caseListFile = os.path.join(readConfig.proDir, "caselist.txt")
        self.caseFile = os.path.join(readConfig.proDir, "testCase")
        # self.caseFile = None
        self.caseList = []
        self.email = MyEmail.get_email()

    def set_case_list(self):
        """
        set case list
        :return:
        """
        fb = open(self.caseListFile)
        for value in fb.readlines():
            data = str(value)
            if data != ‘‘ and not data.startswith("#"):
                self.caseList.append(data.replace("
", ""))
        fb.close()

    def set_case_suite(self):
        """
        set case suite
        :return:
        """
        self.set_case_list()
        test_suite = unittest.TestSuite()
        suite_module = []

        for case in self.caseList:
            case_name = case.split("/")[-1]
            print(case_name+".py")
            discover = unittest.defaultTestLoader.discover(self.caseFile, pattern=case_name + ‘.py‘, top_level_dir=None)
            suite_module.append(discover)

        if len(suite_module) > 0:

            for suite in suite_module:
                for test_name in suite:
                    test_suite.addTest(test_name)
        else:
            return None
        return test_suite

    def run(self):
        """
        run test
        :return:
        """
        try:
            suit = self.set_case_suite()
            if suit is not None:
                logger.info("********TEST START********")
                fps = open(resultPath, ‘wb‘)
                runner = HTMLTestRunner.HTMLTestRunner(stream=fps, title=‘Test Report‘, description=‘Test Description‘)
                runner.run(suit)
            else:
                logger.info("Have no case to test.")
        except Exception as ex:
            logger.error(str(ex))
        logger.info("*********TEST END*********")
        fps.close()
        # send test report by email
        if on_off == ‘on‘:
            self.email.send_email()
        elif on_off == ‘off‘:
            logger.info("Doesn‘t send report email to developer.")
        else:
            logger.info("Unknow state.")


if __name__ == ‘__main__‘:
    obj = AllTest()
    obj.run()

6、生成的报告：

7、发送邮件，邮件附件是report和log，所以压缩了一下。