代码实现自动化相关理论

• 代码编写脚本和工具实现脚本区别是啥?

  • 代码：
    • 优点：代码灵活方便
    • 缺点：学习成本高
  • 工具：
    • 优点：易上手
    • 缺点：灵活度低，有局限性。
  • 总结：
    • 功能脚本：工具
    • 自动化脚本：代码

• 代码接口自动化怎么做的？

  第一步：python+request+unittest;
  具体描述？
  第二步：封装、调用、数据驱动、日志、报告;
  详细举例:
  第三步：api\scripts\data\log\report\until...
  

脚本实现

• 使用代码编写自动化脚本的流程

1、抽取功能用例转为自动化用例
2、搭建环境（测试工具、）
3、搭建目录结构
4、编写脚本
5、执行脚本
6、配置持续集成

抽取功能转为自动化用例

搭建环境（测试工具）

1、python、Pycharm、requests、pymysql、parametrize
2、jenkins、jdk
提示：由于编写的自动化脚本，而自动化脚本编写之前功能已测试完毕，所以不需要在单独搭建项目环境。

搭建目录结构

代码编写

• api(api_register_login.py)

  from config import Host
  class ApiRegisterLogin:
      #初始化
      def __init__(self,session):
          # 获取session对象
          self.session =session
         # 图片验证码url
          self.__url_img_code=Host+"/common/public/verifycode1/{}"
         # 短信验证码url
          self.__url_phone_code = Host + "/member/public/sendSms"
         # 注册url
          self.__url_register= Host + "/member/public/reg"
         # 登录url
          self.__url_login=Host + "/member/public/login"
         # 登录状态url
          self.__url_login_status = Host + "/member/public/islogin"
  
      #1.获取图片验证码接口 封装
      def api_img_code(self):
          pass
      #2.获取短信验证码接口 封装
      def api_phone_code(self):
          pass
      #3.注册接口封装
      def api_register(self):
          pass
      #4.登录接口封装
      def api_login(self):
          pass
      #5.查询登录状态接口 封装
      def api_login_status(self):
          pass
  

• 实现

    #1.获取图片验证码接口 封装
    def api_img_code(self,random):
        # 调用get方法，返回响应对象
        return self.session.get(url=self.__url_img_code.format(random))

    #2.获取短信验证码接口 封装
    def api_phone_code(self,phone,imgVerifyCode):
        #1.定义请求参数
        data={
        "phone":phone,
        "imgVerifyCode":imgVerifyCode,
        "type":"reg"
        }
        # 调用请求方法
        return self.session.post(url=self.__url_phone_code,data=data)

    #3.注册接口封装
    def api_register(self,phone,password,imgVerifyCode,phone_code):
        # 1.定义请求参数
        data = {
            "phone": phone,
            "password":password,
            "imgVerifyCode": imgVerifyCode,
            "phone_code":phone_code,
            "dy_server":"on",
            "invite_phone": ""
        }
        # 调用请求方法
        return self.session.post(url=self.__url_register, data=data)

    #4.登录接口封装
    def api_login(self,phone,password):
        #1.定义请求参数
        data={
        "phone":phone,
        "password":password,
        }
        # 调用请求方法
        return self.session.post(url=self.__url_login,data=data)

    #5.查询登录状态接口 封装
    def api_login_status(self):
        return self.session.post(url=self.__url_login_status)

• script(test01_register_login.py)

import unittest
import requests
from api.api_register_login import ApiRegisterLogin
class TestRegisterLogin(unittest.TestCase):
    #初始化
    def setUp(self) -> None:
        # 获取session对象
        self.session=requests.session()
        #获取ApiRegisterLogin实例
        self.reg=ApiRegisterLogin(self.session)

    #结束
    def tearDownClass(cls) -> None:
        # 关闭session对象
        self.session.close()

    #1.获取图片验证码接口 测试
    def test01_img_code(self):
        pass

    #2.获取短信验证码接口 测试
    def test02_phone_code(self):
        pass
    #3.注册接口 测试
    def test03_register(self):
        pass
    #4.登录接口 测试
    def test04_login(self):
        pass
    #5.查询登录状态接口 测试
    def test05_login_status(self):
        pass

• 封装

import unittest
import requests
from api.api_register_login import ApiRegisterLogin
class TestRegisterLogin(unittest.TestCase):
    #初始化
    def setUp(self) -> None:
        # 获取session对象
        self.session=requests.session()
        #获取ApiRegisterLogin实例
        self.reg=ApiRegisterLogin(self.session)

    #结束
    def tearDown(self) -> None:
        # 关闭session对象
        self.session.close()

    #1.获取图片验证码接口 测试
    def test01_img_code(self):
       # 调用图片验证码
        r= self.reg.api_img_code(123)
       #查看响应状态码
        print(r.status_code)

    #2.获取短信验证码接口 测试
    def test02_phone_code(self,phone=13312345678,imgVerifyCode=8888):
        #1.调用获取短信验证码接口--目的：让session对象记录cookie
        self.reg.api_img_code(123)
        #2.调用短信验证码接口
        r=self.reg.api_phone_code(phone=phone,imgVerifyCode=imgVerifyCode)
        print(r.json())
    #3.注册接口 测试
    def test03_register(self,phone=13312345678,imgVerifyCode=8888,password="test123",phone_code=666666):
        #1.图片验证码接口
        self.reg.api_img_code(123)

        #2.短信验证码接口
        self.reg.api_phone_code(phone=phone, imgVerifyCode=imgVerifyCode)
        #3.注册接口
        r=self.reg.api_register(phone=phone, password=password,verifycode=imgVerifyCode,phone_code=phone_code)
        #4.查看结果
        print(r.json())
    #4.登录接口 测试
    def test04_login(self,keywords="13312345678",password="test123"):
        #1.调用登录接口
        r = self.reg.api_login(keywords=keywords, password=password)
        # 2.查看结果
        print(r.json())
    #5.查询登录状态接口 测试
    def test05_login_status(self):
        #调用登录接口
        self.reg.api_login(keywords="13312345678",password="test123")
        #调用查询登录接口状态
        r = self.reg.api_login_status()
        #看结果
        print(r.json())

断言

• 说明：判断程序执行实际结果是否符合预期结果
• 示例：

import unittest
import requests
from api.api_register_login import ApiRegisterLogin
class TestRegisterLogin(unittest.TestCase):
    #初始化
    def setUp(self) -> None:
        # 获取session对象
        self.session=requests.session()
        #获取ApiRegisterLogin实例
        self.reg=ApiRegisterLogin(self.session)

    #结束
    def tearDown(self) -> None:
        # 关闭session对象
        self.session.close()

    #1.获取图片验证码接口 测试
    def test01_img_code(self):
        try:
           # 调用图片验证码
            r= self.reg.api_img_code(123)
           #查看响应状态码
            self.assertEqual(200,r.status_code)
        except Exception as e:
            #日志
            print(e)
            #抛异常
            raise


    #2.获取短信验证码接口 测试
    def test02_phone_code(self,phone=13312345678,imgVerifyCode=8888,expect_text="发送成功"):
        try:
            #1.调用获取短信验证码接口--目的：让session对象记录cookie
            self.reg.api_img_code(123)
            #2.调用短信验证码接口
            r=self.reg.api_phone_code(phone=phone,imgVerifyCode=imgVerifyCode)
            #3.查看响应结果
            print(r.json())
            #self.assertIn("发送成功",r.json().get("description"))
            self.assertIn(expect_text,r.text)
        except Exception as e:
            #日志
            print(e)
            #抛异常
            raise
    #3.注册接口 测试
    def test03_register(self,phone=13312345679,imgVerifyCode=8888,password="test123",phone_code=666666,expect_text="注册成功"):
        try:
            #1.图片验证码接口
            self.reg.api_img_code(123)

            #2.短信验证码接口
            self.reg.api_phone_code(phone=phone, imgVerifyCode=imgVerifyCode)
            #3.注册接口
            r=self.reg.api_register(phone=phone, password=password,verifycode=imgVerifyCode,phone_code=phone_code)
            #4.查看结果
            print(r.json())
            self.assertIn(expect_text, r.text)
        except Exception as e:
            #日志
            print(e)
            #抛异常
            raise
    #4.登录接口 测试
    def test04_login(self,keywords="13312345678",password="test123",expect_text="登录成功"):
        try:
            #1.调用登录接口
            r = self.reg.api_login(keywords=keywords, password=password)
            # 2.查看结果
            print(r.json())
            self.assertIn(expect_text, r.text)
        except Exception as e:
            #日志
            print(e)
            #抛异常
            raise
    #5.查询登录状态接口 测试
    def test05_login_status(self,expect_text="OK"):
        try:
            #调用登录接口
            self.reg.api_login(keywords="13312345678",password="test123")
            #调用查询登录接口状态
            r = self.reg.api_login_status()
            #看结果
            print(r.json())
            self.assertIn(expect_text, r.text)
        except Exception as e:
            #日志
            print(e)
            #抛异常
            raise

提示：捕获异常的目的是为了将错误信息记录下来，捕获信息完成后，必须抛出异常。

参数化

• 步骤

1.编写数据存储文件json
2.编写读取工具方法 read_json()
3.使用参数化组件进行引用 parametrize

• 1.编写参数化文件(register_login.json)

心得：
1、根据模块来新建json文件（1个模块1个json文件）
2、最外侧使用{}，模块下几个接口，编写几个key，值为列表
3、列表值中，有几组数据，就写几个{}
4、每组数据{}中，组成格式：说明+参数+预期结果

{
  "img_code": [
    {
      "desc": "获取图片验证码成功（随机小数）",
      "random": 0.123,
      "expect_code": 200
  },
    {
      "desc": "获取图片验证码成功（随机整数）",
      "random": 1,
      "expect_code": 200
    },
    {
      "desc": "获取图片验证码成功（随机为空）",
      "random": " ",
      "expect_code": 404

    },
    {
      "desc": "获取图片验证码成功（随机数为字符串）",
      "random": "a123",
      "expect_code": 400
    }],
  "phone_code": [
    {
      "desc":"获取短信验证码成功 ",
      "phone":"16612345678",
      "imgVerifyCode":8888,
      "expect_code": "发送成功"
    },
    {
      "desc":"获取短信验证码成功 ",
      "phone":"16612345678",
      "imgVerifyCode":8889,
      "expect_code": "验证码错误"
    }

  ],
  "register": [
    {
      "desc":"注册成功(必填参数) ",
      "phone":"16612341111",
      "password":"test123",
      "verifycode":8888,
      "phone_code": 666666,
      "expect_code": "注册成功"

    },
    {
      "desc":"注册失败(图片验证码错误)",
      "phone":"16612341111",
      "password":"test123",
      "verifycode":8889,
      "phone_code": 666666,
      "expect_code": "验证码错误!"

    },
    {
      "desc":"注册失败(短信验证码错误)",
      "phone":"16612341111",
      "password":"test123",
      "verifycode":8888,
      "phone_code": 666667,
      "expect_code": "验证码错误"
    },
    {
      "desc":"注册失败(手机号已存在)",
      "phone":"13312345678",
      "password":"test123",
      "verifycode":8888,
      "phone_code": 666666,
      "expect_code": "已存在"
    }
  ],
  "login": [
    {
      "desc": "登录成功",
      "keywords": "13312345678",
      "password":"test123",
      "expect_code": "登录成功"
    },
    {
      "desc": "登录失败（密码为空）",
      "keywords": "13312345678",
      "password":"",
      "expect_code": "不能为空"
    },
        {
      "desc": "登录失败（解锁）",
      "keywords": "13312345678",
      "password":"error123",
      "expect_code": "登录成功"
    }

  ],
  "login_status": [
    {
      "desc": "查询登录状态（已登录） ",
      "status": "已登录",
      "expect_code": "OK"
    },
        {
      "desc": "查询登录状态（未登录） ",
      "status": "未登录",
      "expect_code": "未登"
    }
  ]
}

• 编写读取数据工具(util.py)

#读取json工具
import os
import json

from config import DIR_PATH
def read_json(filename,key):
    #拼接读取文件的完整路径 os.sep动态获取/\
    file_path=DIR_PATH+os.sep+"data"+os.sep+filename
    arr=[]
    with open(file_path,"r",encoding="utf-8") as f:
        for data in json.load(f).get(key):
            arr.append(tuple(data.values())[1:])
        return  arr

if __name__ == '__main__':
    print(read_json("register_login.json", "register"))

• 参数化引用

  • 难点1：错误次数锁定

  # 如果 password=="error123":
               i= 1
                while i<=3:
                #调用登录
                   r=self.xxxlogin()
                  #改变计数器
                  i+=1
                  #断言锁定
                  #暂停60秒
                  #测试登录成功(注意：登录时必须给正确密码)
                  r= self.reg.api_login(keywords="13312345678",password="test123")
              else:
              #调用登录（参数数据）
              #断言   
  

  • 难点2：查询登录状态，不同结果。

      #如果 status=="已登录":
       # 调用登录
      #调用查询登录状态接口
      #断言
  

断言代码示例：（test01_register_login.py）

import unittest
import requests
from  time import sleep
from parameterized import parameterized
from api.api_register_login import ApiRegisterLogin
from util import read_json
class TestRegisterLogin(unittest.TestCase):
    #初始化
    def setUp(self) -> None:
        # 获取session对象
        self.session=requests.session()
        #获取ApiRegisterLogin实例
        self.reg=ApiRegisterLogin(self.session)

    #结束
    def tearDown(self) -> None:
        # 关闭session对象
        self.session.close()

    #1.获取图片验证码接口 测试
    @parameterized.expand(read_json("register_login.json", "img_code"))
    def test01_img_code(self,random,expect_code):
        try:
           # 调用图片验证码
            r= self.reg.api_img_code(random)
           #查看响应状态码
            self.assertEqual(expect_code,r.status_code)
        except Exception as e:
            #日志
            print(e)
            #抛异常
            raise
 

    #2.获取短信验证码接口 测试
    @parameterized.expand(read_json("register_login.json", "phone_code"))
    def test02_phone_code(self,phone,imgVerifyCode,expect_text):
        try:
            #1.调用获取短信验证码接口--目的：让session对象记录cookie
            self.reg.api_img_code(123)
            #2.调用短信验证码接口
            r=self.reg.api_phone_code(phone=phone,imgVerifyCode=imgVerifyCode)
            #3.查看响应结果
            print(r.json())
            #self.assertIn("发送成功",r.json().get("description"))
            self.assertIn(expect_text,r.text)
        except Exception as e:
            #日志
            print(e)
            #抛异常
            raise

    #3.注册接口 测试
    @parameterized.expand(read_json("register_login.json", "register"))
    def test03_register(self,phone,password,imgVerifyCode,phone_code,expect_text):
        try:
            #1.图片验证码接口
            self.reg.api_img_code(123)

            #2.短信验证码接口
            self.reg.api_phone_code(phone=phone, imgVerifyCode=imgVerifyCode)
            #3.注册接口
            r=self.reg.api_register(phone=phone, password=password,verifycode=imgVerifyCode,phone_code=phone_code)
            #4.查看结果
            print(r.json())
            self.assertIn(expect_text, r.text)
        except Exception as e:
            #日志
            print(e)
            #抛异常
            raise

    @parameterized.expand(read_json("register_login.json", "login"))
    #4.登录接口 测试
    def test04_login(self,keywords,password,expect_text):
        try:
            i=1
            r=None
            if "error" in  password:
                while i<=3:
                    r=self.reg.api_login(keywords,password)
                    i+=1
                #断言锁定
                print("测试锁定：",r.text)
                self.assertIn("锁定",r.text)
                #暂停60秒
                sleep(60)
                #测试登录成功
                r= self.reg.api_login(keywords="13312345678",password="test123")
             # 断言登录成功
                self.assertIn(expect_text,r.text)
            else:
                #1.调用登录接口
                r = self.reg.api_login(keywords=keywords, password=password)
                # 2.查看结果
                print(r.json())
                self.assertIn(expect_text, r.text)
        except Exception as e:
            #日志
            print(e)
            #抛异常
            raise

    @parameterized.expand(read_json("register_login.json", "login_status"))
    #5.查询登录状态接口 测试
    def test05_login_status(self,status,expect_text):
        try:
            if status=="已登录":
                #调用登录接口
                self.reg.api_login(keywords="13312345678",password="test123")
            #调用查询登录接口状态
            r = self.reg.api_login_status()
            #看结果
            print(r.json())
            self.assertIn(expect_text, r.text)
        except Exception as e:
            #日志
            print(e)
            #抛异常
            raise

​

日志

• 日志的作用？

  记录程序运行的步骤和错误。
  

  日志的场景

• 1、调试bug
  2、查看程序运行轨迹
  

• 日志的基本应用

  #1、导包
  import logging
  #2、调用日志入口
  logging.error("出错啦，错误原因：{}".format(e))
  

  import logging
  #设置日志级别及保存文件名
  logging.basicConfig(level=logging.DEBUG,filename="../log/p2p.log")
  #调用日志
  logging.debug("调试信息")
  logging.info("信息级别")
  logging.warning("警告")
  logging.error(("断言错误！"))
  logging.critical("严重错误")
  

• 测试人员使用的日志的入口

  info:记录运行步骤
  error:记录运行错误
  

• 日志底层组成介绍

  说明：logging库底层有4大组件（日志器、处理器、格式器、过滤器）
  1、日志器：接受日志信息，设置日志显示级别
  2、处理器：控制日志显示位置或文件
  3、格式器：控制日志输出的显示样式
  关系：
  格式器必须关联处理器
  处理器必须关联日志器
  

• 日志封装应用

重组封装的目的：解决日志显示的样式、存储方式

import logging.handlers
# 日志工具
class GetLog:
    @classmethod
    def get_log(cls):
        cls.log = None
        if cls.log is None:
            #1.获取日志器
            cls.log =logging.getLogger()
            #设置日志级别info
            cls.log.setLevel(logging.INFO)
            file_path = DIR_PATH + os.sep + "log" + os.sep + "p2p.log"
            #2.获取处理器TimedRotatingFileHandler:日志保存到文件且根据时间去分隔
            tf =  logging.handlers.TimedRotatingFileHandler(filename=file_path,
                                                            when="midnight",
                                                            interval=1,
                                                            backupCount=3,
                                                            encoding="utf-8")
            #3.获取格式器
            fmt ="%(asctime)s %(levelname)s[%(filename)s(%(funcName)s:%(lineno)d)]-%(message)s"
            fm = logging.Formatter(fmt)
            #4.将格式器添加到处理器中
            tf.setFormatter(fm)
            #5.将处理器添加到日志器中
            cls.log.addHandler(tf)
        #返回日志器
        return cls.log

if __name__ == '__main__':
    GetLog.get_log().info("信息级别测试")

• 应用(init.py)

应用的级别：info、error
info:info记录程序运行的步骤
error:记录程序错误

from api import log
from config import Host

class ApiRegisterLogin:
    #初始化
    def __init__(self,session):
        # 获取session对象
        self.session =session
       # 图片验证码url
        self.__url_img_code=Host+"/common/public/verifycode1/{}"
       # 短信验证码url
        self.__url_phone_code = Host + "/member/public/sendSms"
       # 注册url
        self.__url_register= Host + "/member/public/reg"
       # 登录url
        self.__url_login=Host + "/member/public/login"
       # 登录状态url
        self.__url_login_status = Host + "/member/public/islogin"

    #1.获取图片验证码接口 封装
    def api_img_code(self,random):
        log.info("正在调用获取图片验证码接口，请求方法:{},请求url:{}".format("get",self.__url_img_code.format(random)))
        # 调用get方法，返回响应对象
        return self.session.get(url=self.__url_img_code.format(random))

    #2.获取短信验证码接口 封装
    def api_phone_code(self,phone,imgVerifyCode):
        #1.定义请求参数
        data={
        "phone":phone,
        "imgVerifyCode":imgVerifyCode,
        "type":"reg"
        }
        log.info("正在调用短信验证码接口，请求方法:{},请求url:{},请求参数:{}".format("post",self.__url_phone_code,data=data))
        # 调用请求方法
        return self.session.post(url=self.__url_phone_code,data=data)

    #3.注册接口封装
    def api_register(self,phone,password,verifycode,phone_code):
        # 1.定义请求参数
        data = {
            "phone": phone,
            "password":password,
            "verifycode": verifycode,
            "phone_code":phone_code,
            "dy_server":"on",
            "invite_phone": ""
        }
        log.info("正在调用注册接口，请求方法:{},请求url:{},请求参数:{}".format("post",self.__url_register, data=data))
        # 调用请求方法
        return self.session.post(url=self.__url_register, data=data)

    #4.登录接口封装
    def api_login(self,keywords,password):
        #1.定义请求参数
        data={
        "keywords":keywords,
        "password":password,
        }
        log.info("正在调用登录接口，请求方法:{},请求url:{},请求参数:{}".format("post",self.__url_login,data=data))
        # 调用请求方法
        return self.session.post(url=self.__url_login,data=data)

    #5.查询登录状态接口 封装
    def api_login_status(self):
        log.info("正在调用查询登录状态接口，请求方法：{},请求url:{}".format("post",self.__url_login_status))
        return self.session.post(url=self.__url_login_status)

import unittest
import requests
import logging
from api import log
from  time import sleep
from parameterized import parameterized
from api.api_register_login import ApiRegisterLogin
from util import read_json
class TestRegisterLogin(unittest.TestCase):
    #初始化
    def setUp(self) -> None:
        # 获取session对象
        self.session=requests.session()
        log.info("正在初始化session对象：{}".format(self.session))
        #获取ApiRegisterLogin实例
        self.reg=ApiRegisterLogin(self.session)

    #结束
    def tearDown(self) -> None:
        # 关闭session对象
        self.session.close()
        log.info("正在关闭session对象：{}".format(self.session))

    #1.获取图片验证码接口 测试
    @parameterized.expand(read_json("register_login.json", "img_code"))
    def test01_img_code(self,random,expect_code):
        try:
           # 调用图片验证码
            r= self.reg.api_img_code(random)
            log.info("执行图片验证码响应状态码：{}".format(r.status_code))
           #查看响应状态码
            self.assertEqual(expect_code,r.status_code)
            log.info("执行图片验证码断言通过")
        except Exception as e:
            #日志
            log.error("断言失败，原因：{}".format(e))
            #抛异常
            raise
 

    #2.获取短信验证码接口 测试
    @parameterized.expand(read_json("register_login.json", "phone_code"))
    def test02_phone_code(self,phone,imgVerifyCode,expect_text):
        try:
            #1.调用获取短信验证码接口--目的：让session对象记录cookie
            self.reg.api_img_code(123)
            #2.调用短信验证码接口
            r=self.reg.api_phone_code(phone=phone,imgVerifyCode=imgVerifyCode)
            log.info("执行结果接口为：{}".format(r.text))
            #3.查看响应结果
            print(r.json())
            #self.assertIn("发送成功",r.json().get("description"))
            self.assertIn(expect_text,r.text)
            log.info("执行断言通过")
        except Exception as e:
            #日志
            log.error("断言失败，原因：{}".format(e))
            #抛异常
            raise

    #3.注册接口 测试
    @parameterized.expand(read_json("register_login.json", "register"))
    def test03_register(self,phone,password,imgVerifyCode,phone_code,expect_text):
        try:
            #1.图片验证码接口
            self.reg.api_img_code(123)

            #2.短信验证码接口
            self.reg.api_phone_code(phone=phone, imgVerifyCode=imgVerifyCode)
            #3.注册接口
            r=self.reg.api_register(phone=phone, password=password,verifycode=imgVerifyCode,phone_code=phone_code)
            log.info("执行结果接口为：{}".format(r.text))
            #4.查看结果
            print(r.json())
            self.assertIn(expect_text, r.text)
            log.info("执行断言通过")
        except Exception as e:
            #日志
            log.error("断言失败，原因：{}".format(e))
            #抛异常
            raise

    @parameterized.expand(read_json("register_login.json", "login"))
    #4.登录接口 测试
    def test04_login(self,keywords,password,expect_text):
        try:
            i=1
            r=None
            if "error" in  password:
                while i<=3:
                    r=self.reg.api_login(keywords,password)
                    i+=1
                #断言锁定
                print("测试锁定：",r.text)
                self.assertIn("锁定",r.text)
                #暂停60秒
                sleep(60)
                #测试登录成功
                r= self.reg.api_login(keywords="13312345678",password="test123")
             # 断言登录成功
                self.assertIn(expect_text,r.text)
            else:
                #1.调用登录接口
                r = self.reg.api_login(keywords=keywords, password=password)
                log.info("执行结果接口为：{}".format(r.text))
                # 2.查看结果
                print(r.json())
                self.assertIn(expect_text, r.text)
                log.info("执行断言通过")
        except Exception as e:
            #日志
            log.error("断言失败，原因：{}".format(e))
            #抛异常
            raise

    @parameterized.expand(read_json("register_login.json", "login_status"))
    #5.查询登录状态接口 测试
    def test05_login_status(self,status,expect_text):
        try:
            if status=="已登录":
                #调用登录接口
                self.reg.api_login(keywords="13312345678",password="test123")
            #调用查询登录接口状态
            r = self.reg.api_login_status()
            log.info("执行结果接口为：{}".format(r.text))
            #看结果
            print(r.json())
            self.assertIn(expect_text, r.text)
            log.info("执行断言通过")
        except Exception as e:
            #日志
            log.error("断言失败，原因：{}".format(e))
            #抛异常
            raise

认证开户

封装（api_approve_trust.py）

from config import Host
class ApiApproveTrust:
    # 初始化
    def __init__(self,session):
        #获取session对象
        self.session=session
        # 认证url
        self.__url_approve = Host+"/member/realname/approverealname"
        # 查询认证状态url
        self.__url_approve_status = Host+"/member/member/getapprove"
        #开户url
        self.__url_trust = Host + "/trust/trust/register"
        #图片验证码url
        self.__url_img_code = Host + "/common/public/verifycode1/{}"
        #充值url
        self.__url_recharge = Host + "/trust/trust/recharge"
    # 1.认证接口 封装
    def api_approve(self):
        pass

     #2.查询认证状态接口 封装
    def api_approve_status(self):
        pass

     #3.开户接口 封装
    def api_trust(self):
        pass

     #4.获取图片验证码接口 封装
    def api_img_code(self):
        pass

     #5.充值接口 封装
    def api_recharge(self):
        pass

• 难点：认证接口请求参数类型为：multipart/form-data多消息类型，如何实现？

解决：请求使用data+files两种参数格式，消息头会自动切换到multipart接口
示例：self.session.post(url=self.__url_approve,data=data,files={"x":"y"})

from config import Host
class ApiApproveTrust:
    # 初始化
    def __init__(self,session):
        #获取session对象
        self.session=session
        # 认证url
        self.__url_approve = Host+"/member/realname/approverealname"
        # 查询认证状态url
        self.__url_approve_status = Host+"/member/member/getapprove"
        #开户url
        self.__url_trust = Host + "/trust/trust/register"
        #图片验证码url
        self.__url_img_code = Host + "/common/public/verifycode1/{}"
        #充值url
        self.__url_recharge = Host + "/trust/trust/recharge"
    # 1.认证接口 封装
    def api_approve(self):
        #1.请求参数
        data={
            "realname":"华仔",
             "card_id":"110101200007289809"
        }
        #2.调用请求方法 难题：multipart/form-data使用：data+files来实现多消息体类型
        return self.session.post(url=self.__url_approve,data=data,files={"x":"y"})

     #2.查询认证状态接口 封装
    def api_approve_status(self):
        return self.session.post(url=self.__url_approve_status)

     #3.开户接口 封装
    def api_trust(self):
        return self.session.post(url=self.__url_trust)

     #4.获取图片验证码接口 封装
    def api_img_code(self,random):
        return self.session.get(url=self.__url_img_code.format(random))

     #5.充值接口 封装
    def api_recharge(self,valicode):
        # 1.请求参数
        data = {
            "paymentType": "chinapnrTrust",
            "amount": "10000",
            "formStr":"reForm",
            "valicode":valicode
        }
        # 2.调用请求方法
        return self.session.post(url=self.__url_recharge, data=data)

调用（test02_approve_trust.py）

• 结构

import unittest
import requests
from api.api_approve_trust import ApiApproveTrust
from api.api_register_login import ApiRegisterLogin
class TestApproveTrust(unittest.TestCase):
    #初始化
    def setUp(self) -> None:
        #1. 获取session
        self.session=requests.session()
        #2.获取ApiApproveTrust对象
        self.ApiApproveTrust=ApiApproveTrust(self.session)
        #3.调用登录成功
        ApiRegisterLogin(self.session).api_login()
    # 结束
    def tearDown(self) -> None:
        self.session.close()
        
    #1、认证接口 测试
    def test01_approve(self):
        pass
    #2、查询认证状态接口 测试
    def test_02_status(self):
        pass
    #3、开户接口 测试
    def test03_trust(self):
        pass
    #4、获取图片验证码 测试
    def test04_img_code(self):
        pass
    #5、充值接口 测试
    def test05_recharge(self):
        pass

• 实现

import unittest
import requests
from api import log
from api.api_approve_trust import ApiApproveTrust
from api.api_register_login import ApiRegisterLogin
class TestApproveTrust(unittest.TestCase):
    #初始化
    def setUp(self) -> None:
        #1. 获取session
        self.session=requests.session()
        #2.获取ApiApproveTrust对象
        self.approve=ApiApproveTrust(self.session)
        #3.调用登录成功
        ApiRegisterLogin(self.session).api_login()
    # 结束
    def tearDown(self) -> None:
        self.session.close()

    #1、认证接口 测试
    def test01_approve(self):
        #调用认证接口
        try:
            r=self.approve.api_approve()
            log.info("接口执行结果为：{}".format(r.text))
            #断言
            self.assertIn("提交成功",r.text)
            log.info("断言通过")
        except Exception as e:
            # 日志
            log.error("断言错误！原因：{}".format(e))
            # 抛异常
            raise

    #2、查询认证状态接口 测试
    def test_02_status(self):
        # 调用查询认证状态接口
        try:
            r = self.approve.api_approve_status()
            log.info("接口执行结果为：{}".format(r.text))
            # 断言
            self.assertIn("华", r.text)
            log.info("断言通过")
        except Exception as e:
            # 日志
            log.error("断言错误！原因：{}".format(e))
            # 抛异常
            raise

    #3、开户接口 测试
    def test03_trust(self):
        # 调用开户接口
        try:
            r = self.approve.api_trust()
            log.info("接口执行结果为：{}".format(r.json()))
            # 断言
            self.assertIn("form", r.text)
            print("请求后台开户结果为：", r.json())
            log.info("断言通过!")
            #三方开户
        except Exception as e:
            # 日志
            log.error("断言错误！原因：{}".format(e))
            # 抛异常
            raise

    #4、获取图片验证码 测试
    def test04_img_code(self,random=123):
        # 调用获取图片验证码接口
        try:
            r = self.approve.api_img_code(random)
            log.info("接口执行结果为：{}".format(r.status_code))
            # 断言
            self.assertEqual(200, r.status_code)
            log.info("断言通过")
        except Exception as e:
            # 日志
            log.error("断言错误！原因：{}".format(e))
            # 抛异常
            raise
            
    #5、充值接口 测试
    def test05_recharge(self,valicode=8888):
        # 调用充值接口
        try:
            #调用图片验证码 获取cookie
            self.approve.api_img_code(123)
            r = self.approve.api_recharge(valicode)
            log.info("接口执行结果为：{}".format(r.json()))
            print("请求后台充值结果为:{}".format(r.json()))
            # 断言
            self.assertIn("form", r.text)
            log.info("断言通过")
            # 三方充值
        except Exception as e:
            # 日志
            log.error("断言错误！原因：{}".format(e))
            # 抛异常
            raise

问题：

三方开户、三方充值如何操作？--从响应数据中提取input标签name属性值和value属性，在python中如何提取？

BeautifSoup库

• 一个python解析html/xml的三方库

• 安装pip install beautifulsoup4

• 应用步骤：

  1、导包
  2、实例化
  3、调用方法
  

• 示例：

#1、导包
from bs4 import BeautifulSoup
html="""
    <html>
        <head>
             <title>某马程序员</title>
        </head>
        <body>
            <p id ="test01">软件测试</p>
            <p id ="test02">2024年</p>
            <a href="/api.html">接口测试</a>
            <a href="/web.html">Web自动化测试</a>
            <a href="/app.html">App自动化测试</a>
        </body>
    </html>
"""
#2、获取bs对象 告诉BeautifulSoup类 你要解析的是html格式
bs=BeautifulSoup(html,"html.parser")
#3、调用方法
"""重点：
1、查找所有标签  bs.find_all("标签名")==元素的集合==("元素1","元素2")
2、查找属性 元素.get("属性名")
"""
for a in bs.find_all('a'):
    print(a.get("href"))

#4、扩展其他方法
#获取单个元素
print(bs.a)
#获取文本
print(bs.a.string)
#获取属性
print(bs.a.get("href"))
#获取标签名
print(bs.a.name)

• 重点

1、查找所有标签  bs.find_all("标签名")==元素的集合==("元素1","元素2")
2、查找属性 元素.get("属性名")

• 其他方法

#4、扩展其他方法
#获取单个元素
print(bs.a)
#获取文本
print(bs.a.string)
#获取属性
print(bs.a.get("href"))
#获取标签名
print(bs.a.name)

提取html数据根据封装

• 思路

# 提取html
# 获取BeautifulSoup对象
# 提取url
# 遍历查找所有的input标签
# 提取name和value的值，并组装到新的字典中
# 返回url和字段

• 示例

# 提取html数据
def parser_html(result):
    #1、提取html
    html =result.json().get("description").get("form")
    #2、获取bs对象
    bs=BeautifulSoup(html,"html.parser")
    #3、提取url
    url= bs.form.get("action")
    print("提取的url:",url)
    #4、查找所有的input标签
    data={}
    for input in bs.find_all("input"):
        data[input.get("name")]=input.get("value")
    return url,data

    #3、开户接口 测试
    def test03_trust(self):
        # 调用开户接口
        try:
            r = self.approve.api_trust()
            log.info("接口执行结果为：{}".format(r.json()))
            # 断言
            self.assertIn("form", r.text)
            print("请求后台开户结果为：", r.json())
            log.info("断言通过!")
            #三方开户
            result = parser_html(r)
            #期望 http://xxx,{"version":10}
            r=self.session.post(url=result[0],data=result[1])
            print("三方开户的结果为：",r.text)
            self.assertIn("OK",r.text)
            log.info("接口执行结果为：{}".format(r.text))
        except Exception as e:
            # 日志
            log.error("断言错误！原因：{}".format(e))
            # 抛异常
            raise

认证开户参数化

• 参数化文件（approve_trust.json）

{
  "img_code": [ {
      "desc": "获取图片验证码成功（随机小数）",
      "random": 0.123,
      "expect_code": 200
  },
    {
      "desc": "获取图片验证码成功（随机整数）",
      "random": 1,
      "expect_code": 200
    },
    {
      "desc": "获取图片验证码成功（随机为空）",
      "random": " ",
      "expect_code": 404

    },
    {
      "desc": "获取图片验证码成功（随机数为字符串）",
      "random": "a123",
      "expect_code": 400
    }],
  "recharge": [
    {"desc":"后台充值响应成功",
     "valicode": 8888,
      "expect": "OK"
    },
       {"desc":"后台充值响应失败",
     "valicode": 8889,
      "expect": "验证码错误"
    }
  ]
}

• 实现难点：

1、充值需要判断验证码不同，执行步骤和结果不同
思路：
1、调用图片验证码-->记录cookie
2、调用充值接口（验证码）
3、判断验证码为8888
	# 1、提取响应数据
	# 2、三方充值
	# 3、断言 
4、否则
	# 直接断言 验证码错误

 #4、获取图片验证码 测试
    @parameterized.expand(read_json("approve_trust.json", "img_code"))
    def test04_img_code(self,random,expect_text):
        # 调用获取图片验证码接口
        try:
            r = self.approve.api_img_code(random)
            log.info("接口执行结果为：{}".format(r.status_code))
            # 断言
            self.assertEqual(expect_text, r.status_code)
            log.info("断言通过")
        except Exception as e:
            # 日志
            log.error("断言错误！原因：{}".format(e))
            # 抛异常
            raise
    @parameterized.expand(read_json("approve_trust.json", "recharge"))
    #5、充值接口 测试
    def test05_recharge(self,valicode,expect_text):
        # 调用充值接口
        try:
            #调用图片验证码 获取cookie
            self.approve.api_img_code(123)
            r = self.approve.api_recharge(valicode)
            log.info("接口执行结果为：{}".format(r.json()))
            if valicode==8888:
                # 断言
                self.assertIn("form", r.text)
                log.info("断言通过")
                # 三方充值
                result = parser_html(r)
                # 期望 http://xxx,{"version":10}
                r = self.session.post(url=result[0], data=result[1])
                print("三方充值的结果为：", r.text)
                self.assertIn(expect_text, r.text)
                log.info("接口执行结果为：{}".format(r.text))
            else:
                self.assertIn(expect_text,r.text)
                print("验证码错误响应结果为：",r.text)
        except Exception as e:
            # 日志
            log.error("断言错误！原因：{}".format(e))
            # 抛异常
            raise

投资模块

封装(api_tender.py)

from config import Host
class ApiTender:
    #初始化
    def __init__(self,session):
        #session
        self.session=session
        #url
        self.__url_tender = Host + "/trust/trust/tender"
    #1、投资方法
    def api_tender(self,amount):
        #1、参数
        data={
            "id":8071,
            "depositCertificate":-1,
            "amount":amount
        }
        #2、调用请求方法
        return self.session.post(url=self.__url_tender,data=data)

调用（test03_tender.py）

import unittest
import requests
from util import parser_html
from api import log
from api.api_tender import ApiTender
from api.api_register_login import ApiRegisterLogin
class TestTender(unittest.TestCase):
    #初始化
    def setUp(self) -> None:
        #获取session对象
        self.session=requests.session()
        #获取ApiTender对象
        self.tender=ApiTender(self.session)
        #调用登录
        ApiRegisterLogin(self.session).api_login()
    #结束
    def tearDown(self) -> None:
        self.session.close()

    #测试方法
    def test01_tender(self,amount=100,expect_text="OK"):
        try:
            #调用投资方法
            r=self.tender.api_tender(amount)
            #调用三方投资
            result = parser_html(r)
            # 期望 http://xxx,{"version":10}
            r = self.session.post(url=result[0], data=result[1])
            print("三方投资的结果为：", r.text)
            self.assertIn("OK", r.text)
            log.info("接口执行结果为：{}".format(r.text))
            #断言
            self.assertIn(expect_text,r.text)
        except Exception as e:
            # 日志
            log.error(e)
            # 抛异常
            raise

参数化

• 文件(tender.json)

{
  "tender": [
    {
      "desc": "请求后台投资响应成功",
      "amount": 100,
      "expect_text": "OK"
    },
        {
      "desc": "请求后台投资响应失败(金额为空)",
      "amount": " ",
      "expect_text": "不是正确的金额"
    }
  ]
}

• 调用(test03_tender.py)

import unittest
import requests
from util import parser_html
from api import log
from api.api_tender import ApiTender
from api.api_register_login import ApiRegisterLogin
from util import read_json
from parameterized import parameterized
class TestTender(unittest.TestCase):
    #初始化
    def setUp(self) -> None:
        #获取session对象
        self.session=requests.session()
        #获取ApiTender对象
        self.tender=ApiTender(self.session)
        #调用登录
        ApiRegisterLogin(self.session).api_login()
    #结束
    def tearDown(self) -> None:
        self.session.close()

    #测试方法
    @parameterized.expand(read_json("tender.json", "tender"))
    def test01_tender(self,amount,expect_text):
        try:
            #调用投资方法
            r=self.tender.api_tender(amount)
            if amount==100:
                #调用三方投资
                result = parser_html(r)
                # 期望 http://xxx,{"version":10}
                r = self.session.post(url=result[0], data=result[1])
                print("三方投资的结果为：", r.text)
                self.assertIn("OK", r.text)
                log.info("接口执行结果为：{}".format(r.text))
                #断言
                self.assertIn(expect_text,r.text)
            else:
                self.assertIn(expect_text, r.text)
        except Exception as e:
            # 日志
            log.error(e)
            # 抛异常
            raise

投资业务

import unittest,requests
from util import parser_html
from api import log
from api.api_tender import ApiTender
from api.api_register_login import ApiRegisterLogin
from api.api_approve_trust import ApiApproveTrust

class TestTenderList(unittest.TestCase):
    def setUp(self) -> None:
        #获取session对象
        self.session = requests.session()
        #获取ApiRegisterLogin对象
        self.reg = ApiRegisterLogin(self.session)
        #获取ApiApproveTrust对象
        self.approve = ApiApproveTrust(self.session)
        #获取ApiTender对象
        self.tender = ApiTender(self.session)

    def tearDown(self) -> None:
        #关闭session
        self.session.close()

    def test01_tender_list(self):
        phone="18822228888"
        img_code=8888
        pwd="test123"
        phone_code=666666
        card_id="120101200007288410"
        #1、获取图片验证码
        self.reg.api_img_code(123123)
        #2、获取短信验证码
        self.reg.api_phone_code(phone,img_code)
        #3、注册
        self.reg.api_register(phone,pwd,img_code,phone_code)
        #4、登录
        self.reg.api_login(phone,pwd)
        #5、认证
        self.approve.api_approve(card_id)
        #6、后台开户
        r = self.approve.api_trust()
        #7、三方开户
        result = parser_html(r)
        r = self.session.post(url=result[0], data=result[1])
        print("三方开户的结果为：", r.text)
        self.assertIn("OK", r.text)
        log.info("接口执行结果为:{}".format(r.text))
        #8、获取充值验证码
        self.approve.api_img_code(123123)
        #9、后台充值
        r= self.approve.api_recharge(img_code)
        #10、三方充值
        result = parser_html(r)
        r = self.session.post(url=result[0], data=result[1])
        print("三方充值的结果为：", r.text)
        self.assertIn("OK", r.text)
        log.info("接口执行结果为:{}".format(r.text))
        #11、后台投资
        r=self.tender.api_tender(100)
        #12、三方投资
        result = parser_html(r)
        r = self.session.post(url=result[0], data=result[1])
        print("三方投资的结果为：", r.text)
        self.assertIn("OK", r.text)
        log.info("接口执行结果为:{}".format(r.text))

连接数据库工具封装

• util.py

#连接数据库工具
def conn_pymsql(sql):
    conn= None
    cursor=None

    try:
        #1、获取连接对象
        conn=pymysql.connect(host="121.43.169.97",user="root",password="123456",database="czbk_member",
                             port=3306,charset="utf8",autocommit=True)
        #2、获取游标对象
        cursor=conn.cursor()
        #3、执行sql语句
        cursor.execute(sql)
        #判断sql语句是否为查询
        if sql.split()[0].lower()=="select":
            #返回所有结果
            return cursor.fetchall()
        #否则
        else:
            #返回受影响的行数
            return "受影响的行数为：{}".format(cursor.rowcount)

    except Exception as e:
        GetLog.get_log().error(e)
        raise
    finally:
        #4、关闭游标
        cursor.close()
        #5、关闭连接
        conn.close()

清除测试数据应用

• util.py

def clear_data():
    sql1="""
    delete i.* from mb_member_info i INNER JOIN mb_member m on
i.member_id=m.id where m.phone in
("16612345671","16612345672","16612345673","16612345674");
"""
    conn_pymsql(sql1)
    sql2 = """
     delete l.* from mb_member_login_log l INNER JOIN mb_member m on
l.member_id=m.id where m.phone in
("16612345671","16612345672","16612345673","16612345674");
    """
    conn_pymsql(sql2)
    sql3= """
       delete from mb_member_register_log where phone in
("16612345671","16612345672","16612345673","16612345674");
    """
    conn_pymsql(sql3)
    sql4 = """
        delete from mb_member where phone in
("16612345671","16612345672","16612345673","16612345674");
    """
    conn_pymsql(sql4)

• 调用（test01_register_login.py）

    @classmethod
    def setUpClass(cls) -> None:
        #清除数据
       clear_data()
    #提示：必须在test01_xxx.py中类方法调用

运行

• 测试报告

• run_suite.py

"""
报告：htmltestreport
"""
#1、导包
from htmltestreport import HTMLTestReport
import unittest
import os
from config import DIR_PATH
#组合测试套件
suite=unittest.defaultTestLoader.discover("./script")
#指定测试报告存储目录
report_path=DIR_PATH+os.sep+"report"+os.sep+"p2p.html"
#执行测试套件
HTMLTestReport(report_path,title="p2p接口自动化测试报告").run(suite)