api课堂笔记

 1 # 框架分层设计
 2 1、common 公用层
 3 2、config 配置层
 4 3、outputs 输出层
 5 4、testcase 测试脚本层
 6 5、testdatas 存放excle用例层
 7 6.mian.py --入口
 8 # 1、数据库的使用
 9 Python与各大数据库的连接
10 http://testingpai.com/article/1596527686073
11 pymysql 安装
12 pip install pymysql
13 # 包引入
14 import pymysql
15 操作步骤：根据自己连接数据库的工具
16 # 2、对手机号码的随机生成处理
17 # 3、结合代码在一定数据位置加上log.info --便于查看日志
18   # 入参，断言，数据库入库校验

19  # 1、连接MySQL数据库 -占用数据库资源
db = pymmysql.connect(
　　host = "xxxx"
　　user ="xxxx"
　　password = "xxxx"
   port = xxx
　　database ="xxxx"
　　charset = "utf8"
   cursourclass = pymysql.cursours.DictCursor
)
#2、 创建游标
cur = db.cursor()
#3、执行SQL语句
sql = "select xx form xxxx where xxxx = 'xxx'"
# 返回的是affect_rows表示执行后的结果，条数
affect_rows = cur.ececute(sql)
# 4、获取查询结果
# 获取第一个结果，返回时一个字典
# data = cur.fetchone（）
# cur.fetchmay(size=2)#获取前2行
# 获取所有的结果，返回时一个列表
data = cur.fetchall()
print(data)
# 5、关闭游标，关闭数据库连接
cur.close()
db.close()
# 对数据库进行封装：放入common层

import pymysql
import os

from py_37.Py_Api接口自动化.class_api07.common.myConf import MyConf
from py_37.Py_Api接口自动化.class_api07.common.my_path import conf_dir


class MyMysql:

    def __init__(self):
        # 实例化配置类对象
        conf = MyConf(os.path.join(conf_dir, "mysql.ini"))
        # 连接数据库/生成游标
        self.db = pymysql.connect(
            host=conf.get("mysql", "host"),
            user=conf.get("mysql", "user"),
            password=conf.get("mysql", "passwd"),
            port=conf.getint("mysql", "port"),
            database=conf.get("mysql", "database"),
            charset="utf8",
            cursorclass=pymysql.cursors.DictCursor
        )

        # 2、创建游标
        self.cur = self.db.cursor()

    def get_count(self,sql):
        count = self.cur.execute(sql)
        return count

    def get_one_data(self,sql):
        self.cur.execute(sql)
        return self.cur.fetchone()

    def get_many_data(self,sql, size=None):
        self.cur.execute(sql)
        if size:
            return self.cur.fetchmany(size)
        else:
            return self.cur.fetchall()

    # def update_data(self):
    #     事务
    #     提交commit  回滚 rollback
    #     pass

    def close_conn(self):
        self.cur.close()
        self.db.close()

if __name__ == '__main__':
    conn = MyMysql()
    sql = "select id from member where mobile_phone='xxxxx'"
    count = conn.get_count(sql)
    print(count)
    conn.close_conn()
相关的数据连接配置采用conf 配置层：ini或者yaml方式存储

[mysql]
host=xxxx
database=xxxx
port=xxxx
user=xxxx
passwd=xxxx

[log]
name=xxxx
level=xxxx
file=xxxx
# 对conf配置数据的读取；查看封装的方式呈现；放在了common层

from configparser import ConfigParser

class MyConf(ConfigParser):

    def __init__(self,filename):
        super().__init__()
        self.read(filename,encoding="utf-8")


# 将以前的绝对路径替换为相对路径：放在common层

import os

# 1、basedir
basedir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

# 拼到配置文件路径
conf_dir = os.path.join(basedir, "Conf")

# 拼接  测试数据路径
testdata_dir = os.path.join(basedir, "testdatas")

# 日志路径
log_dir = os.path.join(basedir, "outputs", "logs")

# 报告路径
report_dir = os.path.join(basedir, "outputs", "reports")



# 通过Faker函数方式生成随机的手机号码：

from faker import Faker
from py_37.Py_Api接口自动化.class_api07.common.my_mysql import MyMysql

def get_new_phone():
    """
    # 得到没有注册过的手机号码。
    # 1、使用faker生成手机号码
    # 2、调用mysql数据库操作，去判断是否在数据中存在。如果不在，表示没有注册
    :return:
    """
    while True:
        phone = Faker("zh_CN").phone_number()
        sql = "select id from member where mobile_phone='{}'".format(phone)
        res = MyMysql().get_count(sql)
        if res == 0:
            return phone
# 在测试用例Excel中相关的手机号码特换：#phone# testdatas
#在测试脚本中相关的字段又手机号码的替换

import pytest
import json
import os

from py_37.Py_Api接口自动化.class_api07.common.my_requests import MyRequests
from py_37.Py_Api接口自动化.class_api07.common.my_excel import MyExcel
from py_37.Py_Api接口自动化.class_api07.common.my_assert import MyAssert
from py_37.Py_Api接口自动化.class_api07.common.mylogger import logger
from py_37.Py_Api接口自动化.class_api07.common.handle_phone import get_new_phone
from py_37.Py_Api接口自动化.class_api07.common.my_path import testdata_dir


# 第一步：读取注册接口的测试数据 - 是个列表，列表中的每个成员，都是一个接口用例的数据。
excel_path = os.path.join(testdata_dir, "测试用例.xlsx")
me = MyExcel(excel_path, "注册接口")
cases = me.read_data()

# 第二步：遍历测试数据，每一组数据，发起一个http的接口
# 实例化请求对象
mq = MyRequests()
massert = MyAssert()

class TestRegister:

    @pytest.mark.parametrize("case", cases)
    def test_regiser(self,case):
        logger.info("=========== 注册接口测试  ===============")
        new_phone = get_new_phone()
        if case["req_data"] and case["req_data"].find('#phone#') != -1:
            # 替换掉占位符 - 请求数据和断言里面替换掉#phone#，替换成未注册手机号码
            logger.info("新生成的手机号码是：{}".format(new_phone))
            case["req_data"] = case["req_data"].replace('#phone#', new_phone)

        # 替换掉占位符 - 请求数据和断言里面替换掉#phone#，替换成未注册手机号码
        if case["assert_list"] and case["assert_list"].find('#phone#') != -1:
            case["assert_list"] = case["assert_list"].replace('#phone#', new_phone)

        # 把json格式的字符串，转换成一个字典
        req_dict = json.loads(case["req_data"])
        resp = mq.send_requests(case["method"], case["url"], req_dict)
        # print(resp.json())
        # 要断言
        if case["assert_list"]:
            massert.assert_response_value(case["assert_list"], resp.json())

api课堂笔记_day07

推荐阅读