python自动化测试之接口测试

1. 项目框架的分层图:

项目层次图

2.1 common文件夹

2.1.1 do_request.py (发起http请求)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import requests
from interface_test.common.config import ReadConfig
config = ReadConfig()

class HttpRequest:
    '''
    公共使用一个session, cookies自动传递
    这是一个操作http请求的类,使用这类的request方法去完成不同的HTTP请求，并且返回响应结果
    '''
    def __init__(self):
        #保证同一个session对象,便于cookies的自动传值
        self.session = requests.session()

    def http_request(self,method,url,data=None,json=None):
        '''
        :param method: 请求的方法
        :param url: 请求的路径
        :param data: 请求的数据
        :param json: json类型的数据
        :return: 返回请求响应的文本
        '''
        method = method.lower() #强制转化为小写  方便后期操作（大写也可以）
        if type(data)==str:   # 将字符串转化为 字典形式的数据
            data = eval(data)

        # 路径的拼接，根据线上环境或测试环境 进行路径的拼接
        url = config.get_strvalue('api','pre_url')+url

        print('data的数据{}：'.format(data))
        print('url的路径：',url)

        if method == 'get':
            resp = self.session.request(method=method,url=url,params=data)
        elif method == 'post':
            if json:
                resp = self.session.request(method=method,url=url,json=json)
            else:
                resp = self.session.request(method=method,url=url,data=data)
        else:
            print('暂不支持其他的请求方式!')
        print ('响应码:', resp.status_code)
        print ('响应文本:', resp.text)
        print ('响应的cookie:', resp.cookies)
        return resp

    def close(self):
        self.session.close()

if __name__ == '__main__':
    # 注册接口
    register_url = 'http://test.lemonban.com/futureloan/mvc/api/member/register'
    params = {'mobilephone': '18871362019', 'pwd': '123456', 'regname': "test"}
    resp = HttpRequest().http_request('post',url=register_url,data=params)
    print(resp.text)
2.1.2 do_excel.py (操作Excel的读写)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from openpyxl import load_workbook

class Case:
    def __init__(self):
        self.case_id =None
        self.title = None
        self.url=None
        self.data = None
        self.method = None
        self.expected = None
        self.actual = None
        self.result = None

class DoExcel:

    def __init__(self,file_name,sheet_name):
        self.file_name = file_name
        self.sheet_name = sheet_name

    def get_data(self): #Excel的读取数据操作
        wb = load_workbook(self.file_name)
        sheet = wb[self.sheet_name]
        # 使用类与对象的思想进行操作
        cases=[]
        for i in range(2,sheet.max_row+1):
            row_case=Case()  #实例化对象
            row_case.case_id = sheet.cell(row=i,column=1).value  #对象调用属性
            row_case.title = sheet.cell (row=i, column=2).value
            row_case.url = sheet.cell (row=i, column=3).value
            row_case.data = sheet.cell (row=i, column=4).value
            row_case.method = sheet.cell (row=i, column=5).value
            row_case.expected = sheet.cell (row=i, column=6).value
            cases.append(row_case)
        wb.close ()
        return cases

    def write_back(self,row,col,value): #excel的写回操作
        wb = load_workbook(self.file_name)
        sheet = wb[self.sheet_name]
        sheet.cell(row,col).value = value
        wb.save(self.file_name)
        wb.close()
2.1.3 contants.py (文件的路径)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
import os
# os.path.abspath(__file__) 动态获取当前文件的路径
# os.path.dirname(os.path.abspath(__file__)) 动态获取当前文件的上一个文件夹的路径
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) #路径到了interface_test
# print(base_dir)

# 测试用例 Excel文件
case_file = os.path.join(base_dir,'data','cases.xlsx')
# print(case_file)

# 有关线上环境与测试环境的网址切换的配置文件
global_config_file = os.path.join(base_dir,'config','global.cfg')
online_config_file = os.path.join(base_dir,'config','online.cfg')
test_config_file = os.path.join(base_dir,'config','test.cfg')
2.1.4 config.py(配置文件的读取)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from configparser import ConfigParser
from interface_test.common.contants import *

class ReadConfig:
    def __init__(self,encoding='utf-8'):
        self.cf = ConfigParser() #打开配置文件
        self.cf.read(global_config_file,encoding) #先加载global.cfg配置文件
        switch = self.cf.getboolean('switch','on')  #读取global.cfg配置文件的switch的值
        if switch:
            # on=True 开关打开的时候,加载的是线上环境的配置
            self.cf.read(online_config_file,encoding)
        else:
            self.cf.read(test_config_file,encoding)

    def get_strvalue(self,section,option): # 获取字符串
        return self.cf.get(section,option)

    def get_intvalue(self,section,option): # 获取整数
        return self.cf.getint(section,option)

    def get_boolvalue(self,section,option): # 获取布尔值
        return self.cf.getboolean(section,option)


if __name__ == '__main__':
    config = ReadConfig()
    value = config.get_boolvalue('switch','on')
    # value = config.get_strvalue('api','pre_url')
    print(value)
2.1.5 do_mysql(操作数据库)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import pymysql
from interface_test.common.config import ReadConfig
config = ReadConfig ()


class DoMysql:
    # 1.建立连接
    # 2.新建一个查询界面
    # 3.编写sql语句
    # 4.执行SQL语句
    # 5.查看结果
    # 6.关闭查询
    # 7.关闭数据库 
    def __init__(self):
        # db_host = 'test.lemonban.com'
        # db_user = 'test'
        # db_password = 'test'
        # db_database = 'future'
        # db_port = 3306
        # 读取数据库的配置文件的值（在online.cfg的配置文件中，section：db_test 下的值）
        db_host = config.get_strvalue('db_test','db_host')
        db_user = config.get_strvalue('db_test','db_user')
        db_password = config.get_strvalue('db_test','db_password')
        db_database = config.get_strvalue('db_test','db_database')
        db_port = config.get_intvalue('db_test','db_port')

        self.db = pymysql.connect (host=db_host, user=db_user, password=db_password,
                                   database=db_database, port=db_port,charset='utf8')
        self.cursor= self.db.cursor ()  # 创建游标

    def fetch_one(self,sql):  # 查询一条数据
        self.cursor.execute(sql)
        return self.cursor.fetchone()

    def fetch_all(self,sql):  # 查询所有的数据
        self.cursor.execute(sql)
        return self.cursor.fetchall()

    def close(self):
        self.cursor.close()   # 关闭查询
        self.db.close()       # 关闭数据库连接

if __name__ == '__main__':
    my_sql = DoMysql ()
    sql = 'select max(mobilephone) from future.member'  #获取数据库中最大的手机号
    result = my_sql.fetch_one(sql)
    print(result)
2.1.6 re_context(正则匹配操作)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import re
from interface_test.common.config import ReadConfig
config = ReadConfig()


def replace_data(data):
    # 1.正则表达式
    regular = '#(.*?)#'
    # 2.查找要匹配的字符串
    # search() 扫描字符串，寻找与模式匹配的字符串,返回匹配对象，如果没有找到匹配，则为None
    while re.search(regular,data):     # 找到匹配的对象
        find_data = re.search (regular, data)
        find_key = find_data.group(1) # 拿到参数化的KEY
        find_value = config.get_strvalue('data',find_key) #拿配置文件里面的值
        # sub(pattern, repl, string, count=0, flags=0)有返回值
        # pattern正则表达式  repl要替换的数据  string--》data count 替换的次数
        data = re.sub(regular,find_value,data,count=1)
    return data

2.2 config文件夹 （配置文件的操作）

2.2.1 global.cfg（控制线上环境或者测试环境的开关）
1
2
3
[switch]
on = True
;on = False
2.2.2 online.cfg（线上环境）
1
2
3
4
5
6
7
8
9
[api]
pre_url = http://test.lemonban.com/futureloan/mvc/api

[db_test]
db_host = test.lemonban.com
db_user = test
db_password = test
db_database = future
db_port = 3306
2.2.3 test.cfg（测试环境）
1
2
[api]
pre_url = http://47.107.168.87:8080/futureloan/mvc/api

2.3 data文件夹（存放测试用例的Excel文件）

2.3.1 cases.xlsx （测试用例）
  1. 注册用例
    注册用例
  2. 登录用例
    登录用例

2.4 log文件夹（日志记录文件）

2.4.1 my_log.py（日志操作）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
import logging
class MyLog:
    '''这是一个记录log日志的类'''
    def my_log(self,level,msg):
        '''
        :param level: 日志的等级
        :param msg: 需要输出的提示信息
        :return:
        '''
        # 收集器 - --创建一个日志收集器, getLogger()函数
        my_logger = logging.getLogger('test_request')
        my_logger.setLevel('DEBUG') #设置等级
        #设定日志输出格式
        formatter = logging.Formatter ('%(asctime)s-' '[%(levelname)s]-' '[line:%(lineno)d]-''[日志信息]:%(message)s')
        #设定输出渠道--->控制台
        sh = logging.StreamHandler()
        sh.setLevel('ERROR')
        sh.setFormatter(formatter)
        #设定输出渠道---->指定文件
        fh = logging.FileHandler('test.log',encoding='utf-8')
        fh.setLevel('INFO')
        fh.setFormatter(formatter)
        #日志收集器与输出渠道进行对接
        my_logger.addHandler (sh)
        my_logger.addHandler(fh)
        if level == 'DEBUG':
            my_logger.debug(msg)
        elif level=='INFO':
            my_logger.info(msg)
        elif level == 'WARNING':
            my_logger.warning(msg)
        elif level =='ERROR':
            my_logger.error(msg)
        else:
            my_logger.critical(msg)

        # 去掉日志的重复 每次收集完毕之后 移除掉日志收集器
        my_logger.removeHandler(sh)
        my_logger.removeHandler(fh)

    #重新封装五个不同等级的日志级别的函数
    def debug(self,msg):
        self.my_log('DEBUG',msg)

    def info(self,msg):
        self.my_log('INFO',msg)

    def warning(self,msg):
        self.my_log('WARNING',msg)

    def error(self,msg):
        self.my_log('ERROR',msg)

    def critical(self,msg):
        self.my_log('CRITICAL',msg)

if __name__ == '__main__':
    log = MyLog()
    log.debug('这是个调试的信息')

2.5 report文件夹 （测试报告）

2.5.1 report.html(这是自动生成的)

2.6 testcases文件夹（存放测试用例的方法）

2.6.1 test_register.py（测试注册接口）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
import unittest
from ddt import ddt,data
from interface_test.common.do_excel import DoExcel
from interface_test.common.do_request import HttpRequest
from interface_test.common.contants import *
from interface_test.common.do_mysql import DoMysql

do_excel = DoExcel(case_file,'register')
register_cases = do_excel.get_data()

@ddt
class RegisterTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.do_request = HttpRequest()
        cls.mysql = DoMysql ()  # 创建数据库的连接

    @data(*register_cases)
    def test_register(self,case):
        if case.data.find('register_mobile')>-1: #测试用例中找到register_mobile
            sql = 'select max (mobilephone) from future.member'
            max_phone = self.mysql.fetch_one(sql)[0] # 得到的数据是元组，需要取里面的第一个值（索引）
            max_phone = int(max_phone)+1  # 将数据库中最大的手机号+1，保证该手机号未被注册
            # replace()函数 是替换之后重新返回一个新的字符串，有返回值 需要变量去接收
            case.data= case.data.replace('register_mobile',str(max_phone)) # 替换参数值

        resp = self.do_request.http_request(case.method,case.url,case.data)
        try:
            self.assertEqual(case.expected,resp.text)
            write_result = 'Pass'
        except AssertionError as e:
            write_result = 'Failed'
            print('断言出错:{}'.format(e))
            raise e
        finally:
            do_excel.write_back(case.case_id+1,7,resp.text)
            do_excel.write_back(case.case_id+1,8,write_result)

    @classmethod
    def tearDownClass(cls):
        cls.do_request.close()
        cls.mysql.close()

if __name__ == '__main__':
    unittest.main()
2.6.2 test_login.py（测试登录接口）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import unittest
from ddt import ddt,data
from interface_test.common.do_excel import DoExcel
from interface_test.common.do_request import HttpRequest
from interface_test.common.contants import *
from interface_test.common.re_context import replace_data

do_excel = DoExcel (case_file, 'login')
login_cases = do_excel.get_data ()


@ddt
class LoginTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.do_request = HttpRequest ()

    @data(*login_cases)
    def test_login(self,case):
        case.data = replace_data(case.data) #参数化 正则匹配解析
        resp =self.do_request.http_request(case.method,case.url,case.data)
        try:
            self.assertEqual(case.expected,resp.json()['msg'])
            write_result = 'Pass'
        except AssertionError as e:
            write_result = 'Failed'
            print('断言错误：{}'.format(e))
            raise e
        finally:
            do_excel.write_back(case.case_id+1,7,resp.text)
            do_excel.write_back(case.case_id+1,8,write_result)
    
    @classmethod
    def tearDownClass(cls):
        cls.do_request.close()

if __name__ == '__main__':
    unittest.main()
2.6.3 test_recharge.py（测试充值接口）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import unittest
from ddt import ddt,data
from interface_test.common.do_excel import DoExcel
from interface_test.common.do_request import HttpRequest
from interface_test.common.contants import *
from interface_test.common.re_context import replace_data

do_excel = DoExcel (case_file, 'recharge')
withdraw_cases = do_excel.get_data ()


@ddt
class WithdrawTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.do_request = HttpRequest()

    @data(*withdraw_cases)
    def test_withdraw(self,case):
        case.data = replace_data(case.data) #参数化 正则匹配解析
        resp = self.do_request.http_request(case.method,case.url,case.data)
        try:
            self.assertEqual(case.expected,resp.json()['msg'])
            write_result = 'Pass'
        except AssertionError as e:
            write_result='Failed'
            print('断言出错：',e)
            raise e
        finally:
            do_excel.write_back(case.case_id+1,7,resp.text)
            do_excel.write_back(case.case_id+1,8,write_result)

    @classmethod
    def tearDownClass(cls):
        cls.do_request.close()


if __name__ == '__main__':
    unittest.main()
2.6.4 test_withdraw.py（测试取现的接口）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import unittest
from ddt import ddt,data
from interface_test.common.do_excel import DoExcel
from interface_test.common.do_request import HttpRequest
from interface_test.common.contants import *
from interface_test.common.re_context import replace_data

do_excel = DoExcel (case_file, 'withdraw')
withdraw_cases = do_excel.get_data ()


@ddt
class WithdrawTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.do_request = HttpRequest()

    @data(*withdraw_cases)
    def test_withdraw(self,case):
        case.data = replace_data(case.data) #参数化 正则匹配解析
        resp = self.do_request.http_request(case.method,case.url,case.data)
        try:
            self.assertEqual(case.expected,resp.json()['msg'])
            write_result = 'Pass'
        except AssertionError as e:
            write_result='Failed'
            print('断言出错：',e)
            raise e
        finally:
            do_excel.write_back(case.case_id+1,7,resp.text)
            do_excel.write_back(case.case_id+1,8,write_result)

    @classmethod
    def tearDownClass(cls):
        cls.do_request.close()


if __name__ == '__main__':
    unittest.main()
2.6.5 test_addloan.py（测试添加标的接口）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import unittest
from ddt import ddt,data
from interface_practice.common.do_request import HttpRequest
from interface_practice.common.do_excel import DoExcel
from interface_practice.common.contants import *
from interface_test.common.re_context import replace_data

do_excel = DoExcel(case_file,'add_loan')
addloan_cases = do_excel.get_data()

@ddt
class AddloanTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.do_request = HttpRequest()

    @data(*addloan_cases)
    def test_addloan(self,case):
        # print('转换前的数据：',case.data) #{"mobilephone": "#normal_user#", "pwd": "#normal_pwd#"}
        case.data = replace_data(case.data)
        # print('转换后的数据：',case.data) #{"mobilephone": "17786426991", "pwd": "123456"}
        resp = self.do_request.http_request(case.method,case.url,case.data)
        try:
            self.assertEqual(case.expected,resp.json()['msg'])
            write_result = "Pass"
        except AssertionError as e:
            write_result='Failed'
            print('断言出错啦:{}'.format(e))
            raise e
        finally:
            do_excel.write_back(case.case_id+1,7,resp.text)
            do_excel.write_back(case.case_id,8,write_result)

    @classmethod
    def tearDownClass(cls):
        cls.do_request.close()


if __name__ == '__main__':
    unittest.main()
2.6.6 test_audit.py（测试审核的接口）
2.6.7 test_bidloan.py（测试投资的接口）
2.6.8 run.py