接口自动化测试

概述

当前项目为基于 Python 的接口自动化测试，针对后端 API 接口进行全面的功能测试与验证。采用数据驱动测试（DDT）设计思路,基于pytest测试框架，将测试用例数据与测试执行逻辑分离，支持从 JSON 文件、MySQL 数据库等多种数据源加载测试用例。
项目的核心特色包括：
数据驱动测试：测试用例以 JSON 格式存储，支持参数化批量执行
AI 智能生成：集成阿里云通义千问大模型，可根据接口文档自动生成测试用例
数据模型化：使用 Python dataclass 将 JSON 数据转换为强类型对象，提升代码可读性与安全性
多层数据源：支持从 JSON 文件和 MySQL 数据库两种方式加载测试数据
Pytest 测试框架：利用 fixture 机制实现全局 Token 管理、数据库连接复用等能力
结构化测试报告：支持自定义用例 ID，测试报告清晰可读

解释:

1.什么是数据驱动测试（DDT）
数据驱动测试（Data-Driven Testing）是一种软件测试方法论，其核心思想是将测试逻辑与测试数据彻底分离——测试脚本中不硬编码任何输入值和预期结果，而是从外部数据源（Excel、CSV、JSON、YAML、数据库等）读取数据，用同一套测试逻辑驱动多组数据执行。
核心优势
在这里插入图片描述

pytest + parametrize（推荐）
pytest 原生支持参数化，无需安装额外库，语法更简洁，是目前 Python 自动化测试中更主流的选择

二、技术栈

在这里插入图片描述

在这里插入图片描述

实现:
1.测试框架Pytest: 通过 @pytest.mark.parametrize 模块参数化执行
2.API 请求层 Requests,通过 requests.request(request_method, request_url, json=request_body, headers=request_headers) 实现接口请求发送
3.@dataclass 的主要作用是：帮你自动生成“数据类”常用的初始化、打印、比较等方法，减重复代码
4.@classmethod是 Python 内置的装饰器，用于将方法定义为类方法——它绑定到类本身
核心特征
第一个参数是 cls，代表类本身（而非实例方法的 self）
可以直接通过类名调用，无需先创建实例
5.@pytest.fixture 的作用可以简单理解为：把测试用例的“前置准备、共享资源、后置清理”封装成一个可复用的依赖。

测试框架Pytest: 通过 @pytest.mark.parametrize 模块参数化执行
在 pytest 的参数化机制中，核心的规则只有一条：装饰器 @pytest.mark.parametrize 中定义的参数名，必须与测试函数括号里的形参名完全一致。
pytest 的参数化测试（parametrize） 用法，核心作用是：用多组数据自动重复运行同一个测试函数，避免写大量重复的测试代码。
case_data 执行 users_request 函数定义的请求方法
case_data 定义名称需要和 test_users_get 中的 case_data 完全一致,如果不一致，会报错
ids=[f"{idx}-{case[‘case_name’]}" for idx,case in enumerate(test_cases, start=1)] 自定义测试用例的 ID（名称）,解决用例名称重复问题
*** ids 的值必须是字符串，而且不能重复，否则 Pytest 会报错,通过idx-case-name 方式解决ids重复问题
enumerate(test_cases, start=1)
enumerate() 会同时返回索引和元素。
start=1 表示索引从 1 开始（默认是 0）。
每次迭代返回 (idx, case)，例如 (1, case1), (2, case2), …

如果不加 ids：pytest 在控制台输出的测试报告默认会显示为 test_something[case_data0]、test_something[case_data1]。如果数据很多，你根本看不出哪个用例失败了。
加了 ids 之后：这里用了一个列表推导式，从每个字典中提取 case_name 的值。测试报告就会变成非常直观的：
如果 test_something[正常登录] | test_something[异常登录]
最终输出:
test_module.py::TestUsersGet::test_users_get[正常登录] PASSED test_module.py::TestUsersGet::test_users_get[密码错误] PASSED test_module.py::TestUsersGet::test_users_get[用户不存在] PASSED

test_user_get.py
from api.users import users_request
from config import settings
import requests
import pytest
import json
from pathlib import Path
from model.model_users_get import parse_request_json

项目根目录

ROOT_DIR = Path(file).resolve().parent.parent

class TestUsersGet:
with open(ROOT_DIR / “case_data” / “users_get.json”, encoding=“utf-8”) as f:
test_cases = json.load(f)
@pytest.mark.parametrize(
“case_data”,
test_cases,
ids=[f"{idx}-{case[‘case_name’]}" for idx,case in enumerate(test_cases,start=1)]
)
#get_token 获取token,为 pytest.fixture 定义的函数 get_token，用于获取 token
def test_users_get(self,case_data,get_token):
request_obj = parse_request_json(json.dumps(case_data))
print(get_token)
print(case_data)
request_obj.request_headers[“Authorization”] = get_token
print(request_obj.request_headers)
print(type(request_obj.request_headers))
res= users_request(request_obj.request_body,
request_obj.request_headers,
request_method=request_obj.request_method,
request_url=request_obj.request_url
)
print(res)

“”"
传参格式为:
{
“case_name”: “获取用户列表_正常场景_默认分页”,
“test_purpose”: “验证在携带有效 token 且参数正确的情况下，能否成功获取用户列表数据”,
“request_method”: “GET”,
“request_url”: “/users?pagenum=1&pagesize=10”,
“request_headers”: “{“Authorization”: “Bearer <valid_token>”}”,
“request_body”: “{}”,
“exp_status_code”: 200,
“exp_response_body”: “{“data”: {“totalpage”: 5, “pagenum”: 1, “users”: []}, “meta”: {“msg”: “获取成功”, “status”: 200}}”
},

“”"

API 请求层 Requests,通过 requests.request(request_method, request_url, json=request_body, headers=request_headers) 实现接口请求发送

login_api.py
from config import settings
import requests

#实现登录请求接口，如果request_headers 为空，则默认使用application/json；如果request_method 为空，则默认使用post；如果request_url 为空，则默认使用/login
def login_request(request_body,request_headers=dict({“Content-Type”:“application/json”}), request_method=“post”, request_url=“/login”,):
request_url = settings.BASE_URL + request_url
print(request_method, request_url, request_headers, request_body)
response = requests.request(request_method, request_url, json=request_body, headers=request_headers)
return response
users.py
from config import settings
import requests

#发送获取用户列表请求，请求方法get,请求url为/users,request_headers 必须传token
def users_request(request_body,
request_headers,
request_method=“get”,
request_url=“/users”):
request_url = settings.BASE_URL + request_url
print(request_method, request_url, request_headers, request_body)
response = requests.request(request_method, request_url, json=request_body, headers=request_headers)
return response

模型层 (model/)

model_login.py
├── LoginRequest (请求模型)
│ ├── from_json(dict) -> LoginRequest
│ └── from_long_json(str) -> LoginRequest
├── LoginResponse (响应模型)
│ ├── data: LoginResponseData
│ │ ├── id, rid, username, mobile, email, token
│ └── meta: LoginResponseMeta
│ ├── msg, status
├── parse_request_json(str) -> LoginRequest
└── parse_response_json(str) -> LoginResponse

四个函数速记表
在这里插入图片描述

@dataclass 的主要作用是：帮你自动生成“数据类”常用的初始化、打印、比较等方法，减重复代码
@classmethod是 Python 内置的装饰器，用于将方法定义为类方法——它绑定到类本身
核心特征
第一个参数是 cls，代表类本身（而非实例方法的 self）
可以直接通过类名调用，无需先创建实例

model_login.py
#login的请求对象和响应对象进行python类对象转换
“”"
这是请求对象
{
“case_name”: “登录接口 - 正常登录”,
“test_purpose”: “验证使用正确的用户名和密码能够成功登录并获取 token”,
“request_method”: “POST”,
“request_url”: “/login”,
“request_headers”: “{“Content-Type”: “application/json”}”,
“request_body”: “{“username”: “admin”, “password”: “123456”}”,
“exp_status_code”: 200,
“exp_response_body”: “{“data”: {“id”: 500, “rid”: 0, “username”: “admin”, “mobile”: “123”, “email”: “123@qq.com”, “token”: “Bearer …”}, “meta”: {“msg”: “登录成功”, “status”: 200}}”
}
“”"

“”"
这是响应对象
{
“data”: {
“id”: 500,
“rid”: 0,
“username”: “admin”,
“mobile”: “123”,
“email”: “123@qq.com”,
“token”: “Bearer eyJ*****4uwnRGaIM”
},
“meta”: {
“msg”: “登录成功”,
“status”: 200
}
}
“”"

import json
from dataclasses import dataclass
from typing import Optional, Any, Dict

#@dataclass 是 dataclasses 模块提供的一个类装饰器
@dataclass
class LoginRequest:
“”“登录请求模型”“”
case_name: str
test_purpose: str
request_method: str
request_url: str
request_headers: Dict[str, str]
request_body: Dict[str, str]
exp_status_code: int
exp_response_body: Dict[str, Any]
@classmethod
def from_json(cls, json_data: dict) -> ‘LoginRequest’:
“”“从 JSON 字典创建请求对象”“”
return cls(
case_name=json_data.get(‘case_name’, ‘’),
test_purpose=json_data.get(‘test_purpose’, ‘’),
request_method=json_data.get(‘request_method’, ‘’),
request_url=json_data.get(‘request_url’, ‘’),
request_headers=json.loads(json_data.get(‘request_headers’, ‘{}’)),
request_body=json.loads(json_data.get(‘request_body’, ‘{}’)),
exp_status_code=json_data.get(‘exp_status_code’, 200),
exp_response_body=json.loads(json_data.get(‘exp_response_body’, ‘{}’))
)

# 补充一个方法，接收一个长的像json的字符串，转换成LoginRequest对象
@classmethod
def from_long_json(cls, long_json_data: str) -> 'LoginRequest':
    """从长 JSON 字符串创建请求对象"""
    json_data = json.loads(long_json_data)
    return cls.from_json(json_data)

@dataclass
class LoginResponseData:
“”“登录响应数据部分”“”
id: Optional[int] = None
rid: Optional[int] = None
username: Optional[str] = None
mobile: Optional[str] = None
email: Optional[str] = None
token: Optional[str] = None

@dataclass
class LoginResponseMeta:
“”“登录响应元数据部分”“”
msg: str
status: int

@dataclass
class LoginResponse:
“”“登录响应模型”“”
data: Optional[LoginResponseData]
meta: LoginResponseMeta
@classmethod
def from_json(cls, json_data: dict) -> ‘LoginResponse’:
“”“从 JSON 字典创建响应对象”“”
data_dict = json_data.get(‘data’)
if data_dict:
data_obj = LoginResponseData(
id=data_dict.get(‘id’),
rid=data_dict.get(‘rid’),
username=data_dict.get(‘username’),
mobile=data_dict.get(‘mobile’),
email=data_dict.get(‘email’),
token=data_dict.get(‘token’)
)
else:
data_obj = None

    meta_dict = json_data.get('meta', {})
    meta_obj = LoginResponseMeta(
        msg=meta_dict.get('msg', ''),
        status=meta_dict.get('status', 0)
    )

    return cls(data=data_obj, meta=meta_obj)

def parse_request_json(json_str: str) -> LoginRequest:
“”“解析请求 JSON 字符串为 LoginRequest 对象”“”
json_data = json.loads(json_str)
return LoginRequest.from_json(json_data)

def parse_response_json(json_str: str) -> LoginResponse:
“”“解析响应 JSON 字符串为 LoginResponse 对象”“”
json_data = json.loads(json_str)
return LoginResponse.from_json(json_data)

if name == ‘main’:
# 测试示例
request_json = ‘’’
{
“case_name”: “登录接口 - 正常登录”,
“test_purpose”: “验证使用正确的用户名和密码能够成功登录并获取 token”,
“request_method”: “POST”,
“request_url”: “/login”,
“request_headers”: “{\“Content-Type\”: \“application/json\”}”,
“request_body”: “{\“username\”: \“admin\”, \“password\”: \“123456\”}”,
“exp_status_code”: 200,
“exp_response_body”: “{\“data\”: {\“id\”: 500, \“rid\”: 0, \“username\”: \“admin\”, \“mobile\”: \“123\”, \“email\”: \“123@qq.com\”, \“token\”: \“Bearer token\”}, \“meta\”: {\“msg\”: \“登录成功\”, \“status\”: 200}}”
}

response_json = '''
{
    "data": {
        "id": 500,
        "rid": 0,
        "username": "admin",
        "mobile": "123",
        "email": "123@qq.com",
        "token": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
    },
    "meta": {
        "msg": "登录成功",
        "status": 200
    }
}
'''

req_obj = parse_request_json(request_json)
print(f"请求对象：{req_obj}")
print(f"请求体：{req_obj.request_body}")
print(f"请求头：{req_obj.request_headers}")

resp_obj = parse_response_json(response_json)
print(f"响应对象：{resp_obj}")
print(f"响应数据：{resp_obj.data}")
print(f"响应数据：{resp_obj.data.email}")
print(f"响应元数据：{resp_obj.meta.msg}")

Pytest 全局 Fixture 配置（conftest.py）

在这里插入图片描述
@pytest.fixture 的作用可以简单理解为：把测试用例的“前置准备、共享资源、后置清理”封装成一个可复用的依赖。
在这里插入图片描述

测试函数只要把 fixture 名写成参数，pytest 就会自动执行它，并把返回值注入进来
scope 控制执行次数
scope 决定 fixture 多久执行一次：
function：每个用例执行一次，默认值
class：每个测试类执行一次
module：每个 .py 文件执行一次
session：整个测试会话只执行一次

conftest.py

Pytest 全局 Fixture 配置

import pytest
import pymysql
from api.login_app import login_request
from model.model_login import parse_response_json

session级别 fixture，会话级别，只执行一次(用到的时候再加载)

@pytest.fixture(scope=“session”)
def init_db_cursor():
print(“=session_fixture前置操作=”)
conn = pymysql.connect(
host=‘192.168.31.10’, #家里环境
# host=‘192.168.6.214’,
user=‘root’,
port=31573,
password=‘pangum@2023’,
database=‘case_data’)
# 建立数据库连接会话,定义返回数据的格式为字典
cursor = conn.cursor(
# 读出来的数据为json格式，不使用时读出来的数据非json格式
pymysql.cursors.DictCursor
)
# 返回游标cursor
yield cursor
cursor.close() #关闭数据库会话
conn.close() #关闭数据库连接
print(“=session_fixture后置操作=”)

#从mysql当中读取出来所有关于login的测试用例
@pytest.fixture(scope=“session”)
def get_login_data(init_db_cursor):
cursor = init_db_cursor
cursor.execute(“select * from shop_manager_api_test_cases where request_url=‘/login’ and if_execute=1”)
data = cursor.fetchall()
# print(data)
return data

定义session级别的fixture,实现返回token

@pytest.fixture(scope=“session”,autouse=True)
def get_token():
request_body = {“username”: “admin”, “password”: “123456”}
response = login_request(request_body)
res_obj = parse_response_json(response.text)
token = res_obj.data.token
return token

存储用户id,全局加载一次

@pytest.fixture(scope=“session”,autouse=True)
def user_ids():
user_ids_list = []
return user_ids_list

config/settings.py
#全局配置 (BASE_URL、gen_api_docs)
BASE_URL = “http://127.0.0.1:8888/api/private/v1”

#生成测试用例的接口文档,用来控制每次生成哪些接口的测试用例,需要生成的直接添加到列表中
gen_api_docs =[
“docs_api/login.md”,
“docs_api/users.md”,
“docs_api/users_get.md”,
“docs_api/users_post.md”,
“docs_api/users_get_for_id.md”,
]

config全局配置
settings.py 配置全局接口请求url和测试用例接口文档测试用例生成
.env 配置API_KEY 和 BASE_URL

测试层 (test_cases/)

test_login.py
from config import settings
import requests
import json
import pytest

from conftest import get_login_data

class TestLogin:
def test_login_case(self,get_login_data):
# print(get_login_data)
for case in get_login_data:
print(case)
# print(type(case))
#解析case中所有参数,发起request请求
case_name = case.get(“case_name”)
test_purpose = case.get(“test_purpose”)
request_method = case.get(“request_method”)
request_url = case.get(“request_url”)
# 解析request_headers参数,参数为 Content-Type: application/json,需要通过 : 分割
request_headers_list = case.get(“request_headers”).split(“:”)
try:
header_key = request_headers_list[0]
#去掉第二部分的空格,str.strip() 作用是去掉字符串两端的空格
header_value = request_headers_list[1].strip()
except Exception as e:
continue
request_headers = {header_key: header_value}
request_body = json.loads(case.get(“request_body”))
exp_status_code = case.get(“exp_status_code”)
response = self.send_login_request(request_method,request_url,request_headers,request_body)
print(“>>>>>”, response.status_code, response.text)
assert response.status_code == exp_status_code,
f"{case_name} 失败：预期 {exp_status_code}，实际 {response.status_code}"
# assert 后面的逗号不是分隔符，它表示「如果断言失败，把后面的内容作为异常信息」
# 这会导致当前测试用例失败，pytest
# 会捕获这个异常并标记该测试为失败（Failed），然后继续执行下一条用例（如果有的话）

def send_login_request(self,request_method,request_url,request_headers,request_body):
    request_url = settings.BASE_URL + request_url
    print(request_method,request_url,request_headers,request_body)
    response = requests.request(request_method,request_url, json=request_body,headers=request_headers)
    return response

也可以通过函数方法执行

def test_login_case_func(get_login_data):

print(get_login_data)

#执行
#D:\PythonProject\ShopManagerApiTest 在项目目录下执行
#pytest test_cases\test_login.py -s -v

#-s：显示 print() 输出
#-v：显示详细用例信息

test_users_get.py

备注:
#在 pytest 的参数化机制中，核心的规则只有一条：装饰器 @pytest.mark.parametrize 中定义的参数名，必须与测试函数括号里的形参名完全一致。
# pytest 的参数化测试（parametrize） 用法，核心作用是：用多组数据自动重复运行同一个测试函数，避免写大量重复的测试代码。
# case_data 执行 users_request 函数定义的请求方法
# case_data 定义名称需要和 test_users_get 中的 case_data 完全一致,如果不一致，会报错
# ids=[f"{case[‘case_name’]}" for case in test_cases] 自定义测试用例的 ID（名称）
# 如果不加 ids：pytest 在控制台输出的测试报告默认会显示为 test_something[case_data0]、test_something[case_data1]。如果数据很多，你根本看不出哪个用例失败了。
# 加了 ids 之后：这里用了一个列表推导式，从每个字典中提取 case_name 的值。测试报告就会变成非常直观的：
# 如果 test_something[正常登录] | test_something[异常登录]

from api.users import users_request
from config import settings
import requests
import pytest
import json
from pathlib import Path
from model.model_users_get import parse_request_json

项目根目录

ROOT_DIR = Path(file).resolve().parent.parent

class TestUsersGet:
with open(ROOT_DIR / “case_data” / “users_get.json”, encoding=“utf-8”) as f:
test_cases = json.load(f)
@pytest.mark.parametrize(
“case_data”,
test_cases,
ids=[f"{idx}-{case[‘case_name’]}" for idx,case in enumerate(test_cases,start=1)]
)
#get_token 获取token,为 pytest.fixture 定义的函数 get_token，用于获取 token
def test_users_get(self,case_data,get_token):
#将case_data转换为json对象，json.dumps 将json对象转换为字符串，
# parse_request_json 将 JSON 字符串解析并转换为一个请求对象
request_obj = parse_request_json(json.dumps(case_data))
print(get_token)
print(case_data)
#获取请求头,请求头格式: “request_headers”: “{\“Authorization\”: \“Bearer <valid_token>\”}”
#取期中的Authorization
#request_obj.request_headers[“Authorization”] = get_token 中，
# 它的完整动作是：定位到请求头字典中的 Authorization 字段，并将它的值更新为 get_token 变量里存放的 Token 字符串
request_obj.request_headers[“Authorization”] = get_token
print(request_obj.request_headers)
print(type(request_obj.request_headers))
res= users_request(request_obj.request_body,
request_obj.request_headers,
request_method=request_obj.request_method,
request_url=request_obj.request_url
)
print(res)

“”"
传参格式为:
{
“case_name”: “获取用户列表_正常场景_默认分页”,
“test_purpose”: “验证在携带有效 token 且参数正确的情况下，能否成功获取用户列表数据”,
“request_method”: “GET”,
“request_url”: “/users?pagenum=1&pagesize=10”,
“request_headers”: “{“Authorization”: “Bearer <valid_token>”}”,
“request_body”: “{}”,
“exp_status_code”: 200,
“exp_response_body”: “{“data”: {“totalpage”: 5, “pagenum”: 1, “users”: []}, “meta”: {“msg”: “获取成功”, “status”: 200}}”
},

“”"

test_user_post.py
from api.users import users_request
from config import settings
import requests
import pytest
import json
from pathlib import Path
from model.model_users_post import parse_request_json,parse_response_json

项目根目录

ROOT_DIR = Path(file).resolve().parent.parent

class TestUsersGet:
with open(ROOT_DIR / “case_data” / “users_post.json”, encoding=“utf-8”) as f:
test_cases = json.load(f)
#在 pytest 的参数化机制中，核心的规则只有一条：装饰器 @pytest.mark.parametrize 中定义的参数名，必须与测试函数括号里的形参名完全一致。
# pytest 的参数化测试（parametrize） 用法，核心作用是：用多组数据自动重复运行同一个测试函数，避免写大量重复的测试代码。
# case_data 执行 users_request 函数定义的请求方法
# case_data 定义名称需要和 test_users_get 中的 case_data 完全一致,如果不一致，会报错
# ids=[f"{case[‘case_name’]}" for case in test_cases] 自定义测试用例的 ID（名称）
# 如果不加 ids：pytest 在控制台输出的测试报告默认会显示为 test_something[case_data0]、test_something[case_data1]。如果数据很多，你根本看不出哪个用例失败了。
# 加了 ids 之后：这里用了一个列表推导式，从每个字典中提取 case_name 的值。测试报告就会变成非常直观的：
# 如果 test_something[正常登录] | test_something[异常登录]
@pytest.mark.parametrize(
“case_data”,
test_cases,
ids=[f"{case[‘case_name’]}" for case in test_cases]
)
#get_token 获取token,为 pytest.fixture 定义的函数 get_token，用于获取 token
@pytest.mark.order(1)
def test_users_get(self,case_data,get_token,user_ids):
#将case_data转换为json对象，json.dumps 将json对象转换为字符串，parse_request_json 将str字符串转换为json对象
request_obj = parse_request_json(json.dumps(case_data))
print(get_token)
print(case_data)
test_purpose = case_data[“test_purpose”]
#获取请求头,请求头格式: “request_headers”: “{\“Authorization\”: \“Bearer <valid_token>\”}”
#取期中的Authorization
#request_obj.request_headers[“Authorization”] = get_token 中，它的完整动作是：定位到请求头字典中的 Authorization 字段，并将它的值更新为 get_token 变量里存放的 Token 字符串
request_obj.request_headers[“Authorization”] = get_token
print(request_obj.request_headers)
print(type(request_obj.request_headers))
res= users_request(request_obj.request_body,
request_obj.request_headers,
request_method=request_obj.request_method,
request_url=request_obj.request_url
)
print(res)
if res.json().get(“data”) != “null” and res.json().get(“data”) is not None:
response_obj = parse_response_json(res.text)
print(“>>>>>test_purpose>>>>>”,test_purpose)
print(“>>>>>usname>>>”,response_obj.data.username)
print(“>>>>>id>>>>>”,response_obj.data.id)
user_ids.append(response_obj.data.id)
@pytest.mark.order(2)
def test_users_get_for_id(self,get_token,user_ids):
token = {
“Authorization”: get_token
}
for user_id in user_ids:
res = requests.get(settings.BASE_URL + “/users/” + str(user_id), headers=token)
print(res.text)

安装pytest-order 控制执行顺序

pip install pytest-order

“”"
传参格式为:
{
“case_name”: “获取用户列表_正常场景_默认分页”,
“test_purpose”: “验证在携带有效 token 且参数正确的情况下，能否成功获取用户列表数据”,
“request_method”: “GET”,
“request_url”: “/users?pagenum=1&pagesize=10”,
“request_headers”: “{“Authorization”: “Bearer <valid_token>”}”,
“request_body”: “{}”,
“exp_status_code”: 200,
“exp_response_body”: “{“data”: {“totalpage”: 5, “pagenum”: 1, “users”: []}, “meta”: {“msg”: “获取成功”, “status”: 200}}”
},
“”"

test_case_json 大模型生成用例

import os
import re
import json
import glob
from pathlib import Path

import pymysql
from openai import OpenAI
from dotenv import load_dotenv

读取项目根目录下的 .env 文件

load_dotenv()

项目根目录

BASE_DIR = Path(file).parent

测试用例输出目录

TEST_CASE_DIR = BASE_DIR / “test_case_json”

接口文档目录

DOCS_API_DIR = BASE_DIR / “docs_api”

确保 test_case 目录存在,如果不存在会自动创建

os.makedirs(TEST_CASE_DIR, exist_ok=True)

client = OpenAI(
api_key=os.getenv(“DASHSCOPE_API_KEY”),
base_url=os.getenv(“DASHSCOPE_BASE_URL”),
)

读取系统提示词存在文件路径为:prompt中,文件为gen_case_prompt.md

with open(BASE_DIR / “prompt” / “gen_case_prompt_json_file.md”, “r”, encoding=“utf-8”) as f:
system_prompt = f.read()

def generate_test_cases(api_doc_path: Path) -> str:
“”“调用 AI 根据接口文档生成测试用例，返回 JSON 字符串”“”
# 从文件路径中提取不带扩展名的文件名
api_name = api_doc_path.stem
print(f"\n=========== 开始为接口 [{api_name}] 生成测试用例 =============")
with open(api_doc_path, “r”, encoding=“utf-8”) as f:
api_doc = f.read()

completion = client.chat.completions.create(
    model="qwen3.6-flash",

    messages=[
        {'role': "system", "content": system_prompt},
        {'role': "user", "content": "请根据以下接口文档生成测试用例" + api_doc}
    ],
    stream=True,
    stream_options={"include_usage": True}
)

result = []
for chunk in completion:
    try:
        #json.loads(chunk.model_dump_json())["choices"] 将流式对象转换为json,再把json转为python字典
        content = json.loads(chunk.model_dump_json())["choices"][0].get("delta").get("content", "")
    except Exception as e:
        content = ""
    if content:
        result.append(content)
        print(content, end="")

test_case_content = "".join(result)
# 清洗 AI 返回内容中可能包含的 markdown 代码块标记
test_case_content = clean_markdown_json(test_case_content)
print(f"\n----------- 接口 [{api_name}] 测试用例生成完成 -------------")
return test_case_content

#match = re.search(pattern, text, re.DOTALL)
“”"添加备注
在 text 整个字符串中，搜索第一个符合 pattern 规则的内容，并开启“跨行匹配”模式。

  1. re.search()
    re 是 Python 内置的正则表达式模块。
    search() 是 re 模块的一个函数，作用是扫描整个字符串，找到第一个匹配正则模式的子串。
    匹配成功返回一个 Match 对象，失败返回 None。
    与 re.match() 不同：match() 只从字符串开头匹配，而 search() 会扫描整个字符串。
  2. pattern
    这是你之前定义的正则表达式变量，即 r"(?:json)?\s*\n?(.*?)"。
    它告诉 search() 要匹配什么样的内容（Markdown 代码块及其内部内容）。
  3. text
    这是待搜索的目标字符串，即 AI 返回的原始文本。
  4. re.DOTALL
    这是本行代码的关键所在，也是最容易踩坑的地方。
    默认行为：正则中的 .（点号）只能匹配除换行符 \n 以外的任意字符。
    加上 re.DOTALL 后：. 可以匹配包括换行符在内的任意字符。
    为什么这里必须加 re.DOTALL？
    因为你的正则中有 (.*?) 这个捕获分组，而 JSON 内容通常是跨多行的
    re.DOTALL 就是告诉正则引擎：把整个字符串当成一行来处理，不要因为遇到换行符就停下

match.group(1).strip() 是正则匹配成功后，从结果对象中精准提取并清洗目标数据的关键操作
在正则表达式中，用圆括号 () 包裹的部分叫捕获分组。
group(0)：代表整个正则匹配到的完整字符串（包括代码块的标记）。
group(1)：代表第一个括号 (.*?) 里匹配到的内容（也就是我们真正想要的 JSON 文本）。

正则 r"(?:json)?\s*\n?(.*?)" 得到的是

{"name": "AI"}
第一个捕获分组（group(1)）是：
{"name": "AI"}
?:json 的作用是不让 json 占用分组编号，保证 group(1) 始终是你想要的内容。
最终json会被去掉
""" 添加备注完成

def clean_markdown_json(text: str) -> str:
    """去除 AI 返回内容中的 markdown 代码块标记，提取纯 JSON 内容"""
    text = text.strip()
    # 匹配 ```json ... ```或 ```... ```
    pattern = r"```(?:json)?\s*\n?(.*?)```"
    match = re.search(pattern, text, re.DOTALL)
    if match:
        return match.group(1).strip()
    return text


def save_to_json(api_name: str, test_case_content: str):
    """将测试用例保存到 test_case 目录下的 JSON 文件"""
    output_file = TEST_CASE_DIR / f"{api_name}.json"
    with open(output_file, "w", encoding="utf-8") as f:
        f.write(test_case_content)
    print(f"----------- 测试用例已保存到 {output_file} -------------")


def save_to_db(test_case_content: str, cursor):
    """将测试用例解析并存储到数据库中"""
    test_case_list = json.loads(test_case_content)
    for test_case_dict in test_case_list:
        case_name = test_case_dict["case_name"]
        test_purpose = test_case_dict["test_purpose"]
        request_method = test_case_dict["request_method"]
        request_url = test_case_dict["request_url"]
        request_headers = test_case_dict["request_headers"]
        request_body = test_case_dict["request_body"]
        exp_status_code = test_case_dict["exp_status_code"]
        sql = f"""
        INSERT INTO shop_manager_api_test_cases 
        (case_name, test_purpose, request_method, request_url, request_headers, request_body, exp_status_code)
         VALUES ('{case_name}', '{test_purpose}', '{request_method}', 
         '{request_url}', '{request_headers}', '{request_body}', '{exp_status_code}')
         """
        print(sql)
        try:
            cursor.execute(sql)
        except Exception as e:
            continue


# ===================== 主流程 =====================

# 获取 docs_api 目录下所有 .md 接口文档
# 使用 Path.glob("*.md") 自动发现 docs_api/ 下所有 Markdown 接口文档，不再硬编码 login.md
# sorted(DOCS_API_DIR.glob("*.md"))从指定目录中找出所有 .md 文件，并按文件名排序后存入列表
api_doc_files = sorted(DOCS_API_DIR.glob("*.md"))

# 检查是否找到接口文档
if not api_doc_files:
    print("docs_api 目录下没有找到任何接口文档")
    exit()

print(f"共发现 {len(api_doc_files)} 个接口文档: {[f.name for f in api_doc_files]}")

# 建立数据库连接
conn = pymysql.connect(
    # host='192.168.31.10', #家里环境
    host='192.168.31.9',
    user='root',
    port=31573,
    password='pangum@2023',
    db='case_data')
#读出来的数据为json格式，不使用时读出来的数据非json格式
cursor = conn.cursor(pymysql.cursors.DictCursor)

# 遍历每个接口文档，生成测试用例并保存
# .stem 是 pathlib.Path 对象的一个属性，作用是返回文件名中去除后缀的部分
for api_doc_path in api_doc_files:
    api_name = api_doc_path.stem  # 如 login、register 等

    # 1. 生成测试用例
    test_case_content = generate_test_cases(api_doc_path)

    # 2. 保存到 JSON 文件
    save_to_json(api_name, test_case_content)

    # 3. 保存到数据库
    try:
        save_to_db(test_case_content, cursor)
    except Exception as e:
        print(f"接口 [{api_name}] 测试用例存入数据库失败: {e}")

# 循环结束后统一提交并关闭数据库连接
conn.commit()
cursor.close()
conn.close()
print("\n=========== 所有接口测试用例生成完成 =============")
# python
Logo
北京朝阳AI社区

更多推荐

cover

每周一道 Agent 面试题 - 现在市面上的 Embedding Model 怎么选择(6)

avatar 北京朝阳AI社区
cover

防拍照系统的工程分水岭

avatar 北京朝阳AI社区
cover

向日葵CLI：批量运维从建议到执行，真正闭环还要可审计

avatar 北京朝阳AI社区