接口自动化基础知识

一、接口自动化requests

1.核心内容

1.get请求：参数 使用 params,参数拼接在url后面，多用于查询

2.post请求：json格式 使用 json=; 表单格式 使用 data=;

3.response对象的常用属性：status_code ,json(),text,headers,cookies

4.异常处理：接口 返回的不是json的时候，使用 .json()会报错误.

案例：get请求

import requests url = "http://xxx/api/user/list" payload = {"page":1,"pageSize":10} resp = requests.get(url=url, params=payload) print(resp.status_code) # 响应状态码 200 print(resp.json()) # 将响应转为字典 # 简单响应断言 assert resp.status_code == 200 assert resp.json()["code"] == 0

params参数说明：

params是GET 请求的查询参数，格式：Python 字典 dict。

requests 会自动把字典拼接成 URL 后面?key1=val1&key2=val2这一段。

(1）几种常见的写法

普通字典

params = {"page":0,"pageSize":10} requests.get(url, params=params)

值为数字、字符串、None

params = { "name": "测试", "age": 20, "beginDate": None }

同一个key多个值：有些接口需要：?id=1&id=2&id=3

案例 post-json请求（常用）

import requests url = "http://xxx/api/login" headers = {"Content‑Type":"application/json"} body = {"username":"zxj","password":"123456"} resp = requests.post(url=url, json=body, headers=headers) token = resp.json()["data"]["token"] print(token)

二、pytest核心

• 用例命名规则：文件test_xxx.py；函数def test_xx()；类TestXxx，方法test_xxx()
• 执行参数：‑v详细输出、‑s打印 print、‑x失败停止
• fixture：前置后置，scope 作用域，yield 实现前置 + 后置清理，返回数据给用例
• @pytest.mark.parametrize数据驱动
• allure 报告基础使用

案例

fixture基础。scope=session，全局登录 获取 token

import pytest import requests @pytest.fixture(scope="session") def get_token(): """session级别：整个测试只登录一次""" url = "http://xxx/api/login" data = {"username":"zxj","password":"123456"} r = requests.post(url,json=data) # 提取token token = r.json()["data"]["token"] yield token # 返回给测试用例，yield前面=前置 # yield之后，全部用例执行完成之后执行后置，这里无需清理 # 测试用例直接接收fixture返回值 def test_query_list(get_token): headers = {"token":get_token} resp = requests.get("http://xxx/api/list",headers=headers) assert resp.json()["code"] == 0

scope取值

• function默认：每个用例执行一次
• class每个测试类执行一次
• module每个 py 文件执行一次
• session整个测试会话执行一次（登录、数据库连接首选）

案例：parametrize数据驱动

同一套逻辑，多组输入生成多条用例，测试代码和数据分离

import pytest # 参数：参数名字，多组测试数据 @pytest.mark.parametrize("page,expect_code", [(0,-1), (1,0), (-1,-1)]) def test_page_param(page,expect_code,get_token): headers = {"token":get_token} params = {"page":page,"pageSize":10} res = requests.get("http://xxx/api/log",headers=headers,params=params) assert res.json()["code"] == expect_code

三、pymysql数据库操作（数据库断言）

1. 连接数据库、创建游标
2. DictCursor：查询结果返回字典，可以按字段名取值（写断言必备）
3. execute 执行 sql，占位符%s防止 sql 注入
4. fetchone () 取一条；fetchall () 取全部
5. 增删改：必须conn.commit()；异常conn.rollback()回滚
6. fixture封装全局数据库连接，不要每个用例反复打开关闭
7. 数据库断言场景：新增 / 修改接口调用后，查询数据库校验数据是否落库正确

重点：接口返回成功不等于数据库落库正确，必须数据库断言

案例：session fixture封装数据库连接

import pymysql import pytest @pytest.fixture(scope="session") def get_db(): # 建立连接 conn = pymysql.connect( host="127.0.0.1", port=3306, user="root", password="123456", database="test_command_log", charset="utf8mb4" ) # DictCursor 返回字典格式 cur = conn.cursor(cursor=pymysql.cursors.DictCursor) yield conn, cur # 给用例返回连接、游标 # 后置：全部用例跑完关闭资源 cur.close() conn.close()

开启cursor=pymysql.cursors.DictCursor，查询返回字典 / 字典列表

• fetchone()：返回1 条字典，key = 数据库字段名，value = 字段值，没有数据返回None
• fetchall()：返回列表，里面每一项是一条记录字典

cur.execute("select commandName,commandValue from command_log limit 1") row = cur.fetchone() print(row) #输出： {'commandName': '显示设置', 'commandValue': 'standard'} rows = cur.fetchall() print(rows) #输出： [ {'commandName': '显示设置', 'commandValue': 'standard'}, {'commandName': '亮度设置', 'commandValue':'high'} ]

普通 Cursor 不写 DictCursor，返回元组('显示设置','standard')，

不知道哪个值对应哪个字段，写断言很麻烦。

案例：数据库断言

场景：调用新增命令日志接口，接口返回成功，查询数据库校验数据是否正确存入

# get_token 和 get_db是 fixture def test_add_command_log(get_token, get_db): # 获取连接和游标 conn, cur = get_db # token headers = {"token":get_token} # 1、调用新增接口 req_body = {"commandName":"显示设置","commandValue":"standard"} resp = requests.post("http://xxx/api/addLog",headers=headers,json=req_body) assert resp.json()["code"] == 0 # 2、数据库断言：查询刚刚插入的数据 sql = """select commandName,commandValue from command_log where commandName=%s order by create_time desc limit 1""" cur.execute(sql, ("显示设置",)) # %s占位传参，防止SQL注入 db_result = cur.fetchone() # 获取第一条结果，字典格式 # 3、断言数据库的值和请求入参一致 assert db_result["commandName"] == "显示设置" assert db_result["commandValue"] == "standard"

注意事项

1. 不要字符串拼接 SQL，用%s占位传参，防止 SQL 注入
2. insert/update/delete 操作后需要conn.commit()；异常场景conn.rollback()回滚
3. 自动化尽量清理测试脏数据，避免脏数据导致用例失败

注意点详解：

（1）

1、禁止字符串拼接 SQL，使用%s占位符，防止 SQL 注入

❌错误：字符串拼接，有注入风险

# 千万不要这么写！！ name = "xxx" sql = f"select * from command_log where commandName = '{name}'" cur.execute(sql)

✅正确写法：SQL 语句写%s占位，execute 第二个参数传元组把值传入

sql = "select * from command_log where commandName = %s" cur.execute(sql, ("显示设置",)) # 参数放在元组里，多个参数就 ("a","b") res = cur.fetchone()

注意：这里的%s不是 python 字符串格式化，是pymysql 的占位符，

不管字段是字符串、数字，统一用%s。

多个参数案例：

sql = "select * from command_log where commandName=%s and commandValue=%s" cur.execute(sql, ("显示设置","standard"))

（2）insert /update/delete必须 commit；异常要 rollback () 回滚

• conn.commit()：提交事务，数据才真正写入数据库
• 如果程序发生异常，执行conn.rollback()回滚，撤销本次修改

案例

import pymysql conn = pymysql.connect(host="127.0.0.1",port=3306,user="root",password="123456",database="test_command_log",charset="utf8mb4") cur = conn.cursor(cursor=pymysql.cursors.DictCursor) try: # insert操作 insert_sql = "insert into command_log(commandName,commandValue) values(%s,%s)" cur.execute(insert_sql, ("日志测试","test001")) conn.commit() # ✨DML操作一定要commit，数据落库 print("插入成功") except Exception as e: print("执行出错：",e) conn.rollback() # ✨出错回滚，放弃本次修改 finally: cur.close() conn.close()

（3）清理脏 数据

自动化测试产生的测试数据，如果不删除，会堆积在库中，影响后面用例执行。 两种方案：

1. 用例执行完主动 delete 删除本次产生的数据（推荐）
2. 使用事务：用例执行完直接 rollback，不提交，
3. 测试数据根本不会保存（适合纯测试）

案例：接口自动化，用例结束清理脏数据（配合 fixture yield 后置）

import pytest import requests import pymysql @pytest.fixture(scope="function") def db_and_token(): """function级别，每个用例执行；yield后置清理脏数据""" # 前置：建立连接、获取token conn = pymysql.connect(host="127.0.0.1",user="root",password="123456",database="test_command_log",charset="utf8mb4") cur = conn.cursor(cursor=pymysql.cursors.DictCursor) #登录拿token login_resp = requests.post("http://xxx/api/login",json={"username":"zxj","password":"123456"}) token = login_resp.json()["data"]["token"] yield conn, cur, token # ✨后置：用例跑完，删除本次测试数据，清理脏数据 try: del_sql = "delete from command_log where commandName=%s" cur.execute(del_sql, ("日志测试",)) conn.commit() except: conn.rollback() cur.close() conn.close() def test_add_log_case(db_and_token): conn,cur,token = db_and_token headers = {"token":token} body = {"commandName":"日志测试","commandValue":"test001"} resp = requests.post("http://xxx/api/addLog",headers=headers,json=body) assert resp.json()["code"] ==0 #数据库断言 sql = "select commandName,commandValue from command_log where commandName=%s order by create_time desc limit 1" cur.execute(sql,("日志测试",)) db_data = cur.fetchone() assert db_data["commandValue"] == "test001"

上面这个 fixture 是scope="function"，每一条测试用例执行完成，都会执行 delete 清理测试数据，不会留脏数据。