Python+Requests+Unittest构建接口自动化测试框架实战

落日邮差

发布时间：2026-10-13 | 3人浏览过

原创

本文详解如何使用Python的Requests库与Unittest框架搭建接口自动化测试系统。针对传统工具在数据控制、加密接口及扩展性上的不足，提出通过独立测试数据库保证数据可控性的设计方案。结合Django后端示例，展示从环境配置、数据库封装到测试用例编写及HTML报告生成的完整流程，实现高效、可扩展的接口自动化测试。

1、为什么要写代码实现接口自动化

尽管Postman、JMeter、Fiddler等接口测试工具操作便捷，但编写代码实现自动化仍有其必要性。工具虽然易用，却存在明显的局限性：

测试数据不可控

接口测试的核心在于数据验证，即通过输入特定数据并校验返回结果的正确性。若依赖工具执行用例，往往需手动向数据库插入测试数据，这使得测试过程难以完全“自动化”。

无法测试加密接口

这是接口测试工具的显著短板。例如，对于需要MD5、Base64或AES等常见加密方式处理的接口，或者涉及时间戳参数的场景，工具难以模拟和处理。本书第十一章将专门介绍加密接口的测试方法。

扩展能力不足

工具带来的便利往往伴随着局限。例如，生成HTML格式测试报告、将报告发送至指定邮箱、设置定时任务或集成持续集成（CI）流程等需求，通常难以通过工具直接实现。

2、接口自动化测试设计

接口测试的调用流程可概括为下图，其中增加了测试数据库环节：

一般的接口工具测试过程如下：

1、接口工具调用被测系统接口（传入参数username="zhangsan"）。

2、系统接口根据传入参数（username="zhangsan"）向正式数据库查询数据。

3、将查询结果组装为特定格式的数据，并返回给调用者。

4、人工或通过工具的断言功能检查接口返回数据的正确性。

在接口自动化测试项目中，为确保数据可控，测试过程调整为：

1、接口测试项目先向测试数据库中插入测试数据（如zhangsan的个人信息）。

2、调用被测系统接口（传入参数username="zhangsan"）。

3、系统接口根据参数（username="zhangsan"）向测试数据库查询并获取zhangsan的个人信息。

4、将查询结果组装为特定格式的数据，并返回给调用者。

5、通过单元测试框架断言接口返回的数据（zhangsan的个人信息），并生成测试报告。

为避免污染正式数据库数据，建议使用独立的测试数据库。

2、requests库

Requests基于urllib3构建，继承了其所有特性。它支持HTTP连接保持和连接池，支持使用Cookie维持会话，支持文件上传，并能自动确定响应内容的编码。关于Requests库的更详细介绍，可参考之前的接口测试基础文章：

https://www.jb51.net/article/122571.htm?pc

https://www.jb51.net/article/108168.htm

3、接口测试代码示例

以下以之前使用Python+Django开发的用户签到系统为背景，展示接口测试代码。

为什么开发接口？开发的接口主要给谁来用？

前后端分离是近年来Web应用开发的重要趋势，其优势包括：

1、后端无需精通前端技术（HTML/JavaScript/CSS），可专注于数据处理，通过API接口对外提供服务。

2、前端专业化程度提高，通过API接口获取数据，专注于页面设计。

3、前后端分离增加了接口的应用范围，开发的接口既可应用于Web页面，也可用于移动APP。

在此开发模式下，接口测试工作变得尤为重要。

开发实现的接口代码示例：

# 添加发布会接口实现
def add_event(request):
  eid = request.POST.get('eid','')         # 发布会id
  name = request.POST.get('name','')        # 发布会标题
  limit = request.POST.get('limit','')       # 限制人数
  status = request.POST.get('status','')      # 状态
  address = request.POST.get('address','')     # 地址
  start_time = request.POST.get('start_time','')  # 发布会时间

  if eid =='' or name == '' or limit == '' or address == '' or start_time == '':
    return JsonResponse({'status':10021,'message':'parameter error'})

  result = Event.objects.filter(id=eid)
  if result:
    return JsonResponse({'status':10022,'message':'event id already exists'})

  result = Event.objects.filter(name=name)
  if result:
    return JsonResponse({'status':10023,'message':'event name already exists'})

  if status == '':
    status = 1

  try:
    Event.objects.create(id=eid,name=name,limit=limit,address=address,status=int(status),start_time=start_time)
  except ValidationError:
    error = 'start_time format error. It must be in YYYY-MM-DD HH:MM:SS format.'
    return JsonResponse({'status':10024,'message':error})

  return JsonResponse({'status':200,'message':'add event success'})

通过POST请求接收发布会参数：发布会id、标题、人数、状态、地址和时间等。

首先，判断eid、name、limit、address、start_time等字段均不能为空，否则JsonResponse()返回相应的状态码和提示信息。JsonResponse()是一个非常有用的方法，它可以直接将字典转化为Json格式返回到客户端。

接下来，判断发布会id是否存在，以及发布会名称（name）是否存在；如果存在将返回相应的状态码和提示信息。

再接下来，判断发布会状态是否为空，如果为空，将状态设置为1（True）。

最后，将数据插入到Event表，在插入过程中如果日期格式错误，将抛出ValidationError异常，接收该异常并返回相应的状态和提示，否则，插入成功，返回状态码200和“add event success”的提示。

# 发布会查询接口实现
def get_event_list(request):

  eid = request.GET.get("eid", "")   # 发布会id
  name = request.GET.get("name", "")  # 发布会名称

  if eid == '' and name == '':
    return JsonResponse({'status':10021,'message':'parameter error'})

  if eid != '':
    event = {}
    try:
      result = Event.objects.get(id=eid)
    except ObjectDoesNotExist:
      return JsonResponse({'status':10022, 'message':'query result is empty'})
    else:
      event['eid'] = result.id
      event['name'] = result.name
      event['limit'] = result.limit
      event['status'] = result.status
      event['address'] = result.address
      event['start_time'] = result.start_time
      return JsonResponse({'status':200, 'message':'success', 'data':event})

  if name != '':
    datas = []
    results = Event.objects.filter(name__contains=name)
    if results:
      for r in results:
        event = {}
        event['eid'] = r.id
        event['name'] = r.name
        event['limit'] = r.limit
        event['status'] = r.status
        event['address'] = r.address
        event['start_time'] = r.start_time
        datas.append(event)
      return JsonResponse({'status':200, 'message':'success', 'data':datas})
    else:
      return JsonResponse({'status':10022, 'message':'query result is empty'})

通过GET请求接收发布会id和name参数。两个参数都是可选的。首先，判断当两个参数同时为空，接口返回状态码10021，参数错误。

如果发布会id不为空，优先通过id查询，因为id的唯一性，所以，查询结果只会有一条，将查询结果以key:value对的方式存放到定义的event字典中，并将数据字典作为整个返回字典中data对应的值返回。

name查询为模糊查询，查询数据可能会有多条，返回的数据稍显复杂；首先将查询的每一条数据放到一个字典event中，再把每一个字典再放到数组datas中，最后再将整个数组作为返回字典中data对应的值返回。

接口测试代码示例

#查询发布会接口测试代码
import requests

url = "http://127.0.0.1:8000/api/get_event_list/"
r = requests.get(url, params={'eid':'1'})
result = r.json()
print(result)
assert result['status'] == 200
assert result['message'] == "success"
assert result['data']['name'] == "xx 产品发布会"
assert result['data']['address'] == "北京林匹克公园水立方"
assert result['data']['start_time'] == "2016-10-15T18:00:00"

因为“发布会查询接口”是GET类型，所以，通过requests库的get()方法调用，第一个参数为调用接口的URL地址，params设置接口的参数，参数以字典形式组织。

json()方法可以将接口返回的json格式的数据转化为字典。

接下来就是通过assert语句对字典中的数据进行断言。分别断言status、message和data的相关数据等。

使用unittest单元测试框架开发接口测试用例

#发布会查询接口测试代码 
import unittest
import requests

class GetEventListTest(unittest.TestCase):

  def setUp(self):
    self.base_url = "http://127.0.0.1:8000/api/get_event_list/"

  def test_get_event_list_eid_null(self):
    ''' eid 参数为空 '''
    r = requests.get(self.base_url, params={'eid':''})
    result = r.json()
    self.assertEqual(result['status'], 10021)
    self.assertEqual(result['message'], 'parameter error')

  def test_get_event_list_eid_error(self):
    ''' eid=901 查询结果为空 '''
    r = requests.get(self.base_url, params={'eid':901})
    result = r.json()
    self.assertEqual(result['status'], 10022)
    self.assertEqual(result['message'], 'query result is empty')

  def test_get_event_list_eid_success(self):
    ''' 根据 eid 查询结果成功 '''
    r = requests.get(self.base_url, params={'eid':1})
    result = r.json()
    self.assertEqual(result['status'], 200)
    self.assertEqual(result['message'], 'success')
    self.assertEqual(result['data']['name'],u'mx6发布会')
    self.assertEqual(result['data']['address'],u'北京国家会议中心')

  def test_get_event_list_nam_result_null(self):
    ''' 关键字‘abc'查询 '''
    r = requests.get(self.base_url, params={'name':'abc'})
    result = r.json()
    self.assertEqual(result['status'], 10022)
    self.assertEqual(result['message'], 'query result is empty')

  def test_get_event_list_name_find(self):
    ''' 关键字‘发布会'模糊查询 '''
    r = requests.get(self.base_url, params={'name':'发布会'})
    result = r.json()
    self.assertEqual(result['status'], 200)
    self.assertEqual(result['message'], 'success')
    self.assertEqual(result['data'][0]['name'],u'mx6发布会')
    self.assertEqual(result['data'][0]['address'],u'北京国家会议中心')
49if __name__ == '__main__':
   unittest.main()

unittest单元测试框架可以帮助组织和运行接口测试用例。

4、接口自动化测试框架实现

关于接口自动化测试，unittest已经帮我们做了大部分工作，接下来只需要集成数据库操作，以及HTMLTestRunner测试报告生成扩展即可。

框架结构如下图：

pyrequests框架：

db_fixture/：初始化接口测试数据。

interface/：用于编写接口自动化测试用例。

report/：生成接口自动化测试报告。

db_config.ini：数据库配置文件。

HTMLTestRunner.py unittest单元测试框架扩展，生成HTML格式的测试报告。

run_tests.py：执行所有接口测试用例。

4.1、数据库配置

首先，需要修改被测系统将数据库指向测试数据库。以MySQL数据库为例，针对django项目而言，修改.../guest/settings.py文件。可以在系统测试环境单独创建一个测试库。这样做的目的是让接口测试的数据不会清空或污染到功能测试库的数据。其他框架开发的项目与django项目类似，这个工作一般由开发同学完成，我们测试同学更多关注的是测试框架的代码。

4.2、框架代码实现

4.2.1、首先，创建数据库配置文件.../db_config.ini

4.2.2、接下来，简单封装数据库操作，数据库表数据的插入和清除，.../db_fixture/mysql_db.py

import pymysql.cursors
import os
import configparser as cparser


# ======== Reading db_config.ini setting ===========
base_dir = str(os.path.dirname(os.path.dirname(__file__)))
base_dir = base_dir.replace('\\', '/')
file_path = base_dir + "/db_config.ini"

cf = cparser.ConfigParser()

cf.read(file_path)
host = cf.get("mysqlconf", "host")
port = cf.get("mysqlconf", "port")
db  = cf.get("mysqlconf", "db_name")
user = cf.get("mysqlconf", "user")
password = cf.get("mysqlconf", "password")


# ======== MySql base operating ===================
class DB:

  def __init__(self):
    try:
      # Connect to the database
      self.connection = pymysql.connect(host=host,
                       port=int(port),
                       user=user,
                       password=password,
                       db=db,
                       charset='utf8mb4',
                       cursorclass=pymysql.cursors.DictCursor)
    except pymysql.err.OperationalError as e:
      print("Mysql Error %d: %s" % (e.args[0], e.args[1]))

  # clear table data
  def clear(self, table_name):
    # real_sql = "truncate table " + table_name + ";"
    real_sql = "delete from " + table_name + ";"
    with self.connection.cursor() as cursor:
      cursor.execute("SET FOREIGN_KEY_CHECKS=0;")
      cursor.execute(real_sql)
    self.connection.commit()

  # insert sql statement
  def insert(self, table_name, table_data):
    for key in table_data:
      table_data[key] = "'"+str(table_data[key])+"'"
    key  = ','.join(table_data.keys())
    value = ','.join(table_data.values())
    real_sql = "INSERT INTO " + table_name + " (" + key + ") VALUES (" + value + ")"
    #print(real_sql)

    with self.connection.cursor() as cursor:
      cursor.execute(real_sql)

    self.connection.commit()

  # close database
  def close(self):
    self.connection.close()

  # init data
  def init_data(self, datas):
    for table, data in datas.items():
      self.clear(table)
      for d in data:
        self.insert(table, d)
    self.close()


if __name__ == '__main__':

  db = DB()
  table_name = "sign_event"
  data = {'id':1,'name':'红米','`limit`':2000,'status':1,'address':'北京会展中心','start_time':'2016-08-20 00:25:42'}
  table_name2 = "sign_guest"
  data2 = {'realname':'alen','phone':12312341234,'email':'alen@mail.com','sign':0,'event_id':1}

  db.clear(table_name)
  db.insert(table_name, data)
  db.close()

首先，读取db_config.ini配置文件。创建DB类，__init__()方法初始化，通过pymysql.connect()连接数据库。

因为这里只用到数据库表的清除和插入，所以只创建clear()和insert()两个方法。其中，insert()方法对数据的插入做了简单的格式转化，可将字典转化成SQL插入语句，这样格式转化了方便了数据库表数据的创建。

最后，通过close()方法用于关闭数据库连接。

4.2.3、接下来接下来创建测试数据，.../db_fixture/test_data.py

import sys
sys.path.append('../db_fixture')
try:
  from mysql_db import DB
except ImportError:
  from .mysql_db import DB

# create data
datas = {
  'sign_event':[
    {'id':1,'name':'红米Pro发布会','`limit`':2000,'status':1,'address':'北京会展中心','start_time':'2017-08-20 14:00:00'},
    {'id':2,'name':'可参加人数为0','`limit`':0,'status':1,'address':'北京会展中心','start_time':'2017-08-20 14:00:00'},
    {'id':3,'name':'当前状态为0关闭','`limit`':2000,'status':0,'address':'北京会展中心','start_time':'2017-08-20 14:00:00'},
    {'id':4,'name':'发布会已结束','`limit`':2000,'status':1,'address':'北京会展中心','start_time':'2001-08-20 14:00:00'},
    {'id':5,'name':'小米5发布会','`limit`':2000,'status':1,'address':'北京国家会议中心','start_time':'2017-08-20 14:00:00'},
  ],
  'sign_guest':[
    {'id':1,'realname':'alen','phone':13511001100,'email':'alen@mail.com','sign':0,'event_id':1},
    {'id':2,'realname':'has sign','phone':13511001101,'email':'sign@mail.com','sign':1,'event_id':1},
    {'id':3,'realname':'tom','phone':13511001102,'email':'tom@mail.com','sign':0,'event_id':5},
  ],
}

# Inster table datas
def init_data():
  DB().init_data(datas)


if __name__ == '__main__':
  init_data()

init_data()函数用于读取datas字典中的数据，调用DB类中的clear()方法清除数据库，然后，调用insert()方法插入表数据。

4.2.4、编写接口测试用例。创建添加发布会接口测试文件.../interface/add_event_test.py

import unittest
import requests
import os, sys
parentdir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, parentdir)
from db_fixture import test_data


class AddEventTest(unittest.TestCase):
  ''' 添加发布会 '''

  def setUp(self):
    self.base_url = "http://127.0.0.1:8000/api/add_event/"

  def tearDown(self):
    print(self.result)

  def test_add_event_all_null(self):
    ''' 所有参数为空 '''
    payload = {'eid':'','':'','limit':'','address':"",'start_time':''}
    r = requests.post(self.base_url, data=payload)
    self.result = r.json()
    self.assertEqual(self.result['status'], 10021)
    self.assertEqual(self.result['message'], 'parameter error')

  def test_add_event_eid_exist(self):
    ''' id已经存在 '''
    payload = {'eid':1,'name':'一加4发布会','limit':2000,'address':"深圳宝体",'start_time':'2017'}
    r = requests.post(self.base_url, data=payload)
    self.result = r.json()
    self.assertEqual(self.result['status'], 10022)
    self.assertEqual(self.result['message'], 'event id already exists')

  def test_add_event_name_exist(self):
    ''' 名称已经存在 '''
    payload = {'eid':11,'name':'红米Pro发布会','limit':2000,'address':"深圳宝体",'start_time':'2017'}
    r = requests.post(self.base_url,data=payload)
    self.result = r.json()
    self.assertEqual(self.result['status'], 10023)
    self.assertEqual(self.result['message'], 'event name already exists')

  def test_add_event_data_type_error(self):
    ''' 日期格式错误 '''
    payload = {'eid':11,'name':'一加4手机发布会','limit':2000,'address':"深圳宝体",'start_time':'2017'}
    r = requests.post(self.base_url,data=payload)
    self.result = r.json()
    self.assertEqual(self.result['status'], 10024)
    self.assertIn('start_time format error.', self.result['message'])

  def test_add_event_success(self):
    ''' 添加成功 '''
    payload = {'eid':11,'name':'一加4手机发布会','limit':2000,'address':"深圳宝体",'start_time':'2017-05-10 12:00:00'}
    r = requests.post(self.base_url,data=payload)
    self.result = r.json()
    self.assertEqual(self.result['status'], 200)
    self.assertEqual(self.result['message'], 'add event success')


if __name__ == '__main__':
  test_data.init_data() # 初始化接口测试数据
  unittest.main()

在测试接口之前，调用test_data.py文件中的init_data()方法初始化数据库中的测试数据。

创建AddEventTest测试类继承unittest.TestCase类，通过创建测试用例，调用相关接口，并验证接口返回的数据。

4.2.5、创建run_tests.py文件

当开发的接口达到一定数量后，就需要考虑分文件分目录来划分接口测试用例，如何批量的执行不同文件目录下的用例呢？unittest单元测试框架提供的discover()方法可以帮助我们做到这一点。并使用HTMLTestRunner扩展生成HTML格式的测试报告。

import time, sys
sys.path.append('./interface')
sys.path.append('./db_fixture')
from HTMLTestRunner import HTMLTestRunner
import unittest
from db_fixture import test_data


# 指定测试用例为当前文件夹下的 interface 目录
test_dir = './interface'
discover = unittest.defaultTestLoader.discover(test_dir, pattern='*_test.py')


if __name__ == "__main__":
  test_data.init_data() # 初始化接口测试数据

  now = time.strftime("%Y-%m-%d %H_%M_%S")
  filename = './report/' + now + '_result.html'
  fp = open(filename, 'wb')
  runner = HTMLTestRunner(stream=fp,
              title='Guest Manage System Interface Test Report',
              description='Implementation Example with: ')
  runner.run(discover)
  fp.close()

首先，通过调用test_data.py文件中的init_data()函数来初始化接口测试数据。

使用unittest框架所提供的discover()方法，查找interface/目录下，所有匹配*_test.py的测试文件（*星号匹配任意字符）。

HTMLTestRunner为unittest单元测试框架的扩展，利用它所提供的HTMLTestRunner()类来替换unittest单元测试框架的TextTestRunner()类，从而生成HTML格式的测试报告。

遗憾的是HTMLTestRunner并不支持Python3.x，大家可以在网上找到适用于Python3.x的HTMLTestRunner.py文件，使用在自己的接口自动化工程中。

通过time的strftime()方法获取当前时间，并且转化成一定的时间格式。作为测试报告的名称。这样做目的是是为了避免因为生成的报告的名称重名而造成报告的覆盖。最终，将测试报告存放于report/目录下面。如下图，一张完整的接口自动化测试报告。