目录:
- // # 预期的 JSON 文档结构
- {
- "name": "LiMing",
- "Courses": ["Mock", "Docker"]
- }
- // jsonschema
- {
- "$schema": "http://json-schema.org/draft-06/schema#",
- "$ref": "#/definitions/Welcome",
- "definitions": {
- "Welcome": {
- "type": "object",
- "additionalProperties": false,
- "properties": {
- "name": {
- "type": "string"
- },
- "Courses": {
- "type": "array",
- "items": {
- "type": "string"
- }
- }
- },
- "required": ["Courses", "name"],
- "title": "Welcome"
- }
- }
- }
JSON Schema 在线生成工具:https://app.quicktype.io

pip install genson- from genson import SchemaBuilder
-
-
- def gernerate_jsonschema(obj):
- builder = SchemaBuilder()
- builder.add_object(obj)
- return builder.to_schema()
-
-
- def test_generate_jsonschema():
- print(gernerate_jsonschema({"name": 1}))
运行结果:
pip install jsonschemavalidate() 进行验证。- def schema_validate(obj, schema):
- '''
- 对比 python 对象与生成的 JSONSchame 的结构是否一致
- '''
- try:
- validate(instance=obj, schema=schema)
- return True
- except Exception as e:
- return False
代码示例:
- import json
-
- from genson import SchemaBuilder
- from jsonschema.validators import validate
-
-
- def gernerate_jsonschema(obj):
- builder = SchemaBuilder()
- builder.add_object(obj)
- return builder.to_schema()
-
-
- def test_generate_jsonschema():
- print(gernerate_jsonschema({"name": 1}))
-
-
- def gernerate_jsonschema_file(obj, file_path):
- res = gernerate_jsonschema(obj)
- with open(file_path, 'w') as f:
- json.dump(res, f)
-
-
- def test_generate_jsonschema_file():
- gernerate_jsonschema_file({"name": 1}, "./datas/validate.json")
-
-
- def schema_validate(obj, schema):
- try:
- validate(instance=obj, schema=schema)
- return True
- except Exception as e:
- return False
-
-
- def test_schema_validate():
- with open("./datas/validate.json", 'r') as f:
- res = json.load(f)
- schema_validate({"name": 1}, res)
- class JSONSchemaUtils:
- @classmethod
- def generate_schema(cls, obj):
- # 实例化jsonschem
- builder = SchemaBuilder()
- # 传入被转换的对象
- builder.add_object(obj)
- # 转换成 schema 数据
- return builder.to_schema()
-
- @classmethod
- def schema_validate(cls, obj, schema):
- '''
- 对比 python 对象与生成的 json schame 的结构是否一致
- '''
- try:
- validate(instance=obj, schema=schema)
- return True
- except Exception as e:
- return False
代码示例:
test_utils_use.py
- import requests
-
- from interface_automation_testing.接口自动化测试_L3.接口鉴权的多种情况与解决方案.jsonschema_utils import JSONSchemaUtils
-
-
- def test_httpbin_generate_schema():
- r = requests.get("https://httpbin.ceshiren.com/get", verify=False)
- JSONSchemaUtils.generate_jsonschema_by_file(r.json(), "./datas/httpbin.json")
-
-
- def test_httpbin_req():
- r = requests.get("https://httpbin.ceshiren.com/get", verify=False)
- validate_res = JSONSchemaUtils.validate_schema_by_file(r.json(), "./datas/httpbin.json")
- assert validate_res == True
jsonschema_utils.py
- import json
- from genson import SchemaBuilder
- from jsonschema.validators import validate
-
-
- class JSONSchemaUtils:
-
- @classmethod
- def validate_schema(cls, data_obj, schema):
- try:
- validate(data_obj, schema=schema)
- return True
- except Exception as e:
- print(f"结构体验证失败,失败原因是{e}")
- return False
-
- @classmethod
- def generate_jsonschema(cls, obj):
- builder = SchemaBuilder()
- builder.add_object(obj)
- return builder.to_schema()
-
- @classmethod
- def validate_schema_by_file(cls, data_obj, schema_file):
- with open(schema_file) as f:
- schema_data = json.load(f)
- return cls.validate_schema(data_obj, schema_data)
-
- @classmethod
- def generate_jsonschema_by_file(cls, obj, file_path):
- json_schema_data = cls.generate_jsonschema(obj)
- with open(file_path, "w") as f:
- json.dump(json_schema_data, f)
项目结构:

如何在测试过程中验证接口没有 Bug?
自动化测试过程中,会产生大量的脏数据,如何处理?
直接对数据库做查询之外的操作是非常危险的行为
注意:只有查询权限
- import pymysql
-
-
- # 封装建立连接的对象
- def get_connnect():
- conn = pymysql.connect(
- host="litemall.hogwarts.ceshiren.com",
- port=13306,
- user="test",
- password="test123456",
- database="litemall",
- charset="utf8mb4"
- )
- return conn
-
-
- # 执行sql语句
- def execute_sql(sql):
- connect = get_connnect()
- cursor = connect.cursor()
- cursor.execute(sql) # 执行SQL
- record = cursor.fetchone() # 查询记录
- return record
-
-
- if __name__ == '__main__':
- ret = execute_sql("select * from litemall_cart where goods_name='竹语丝麻印花四件套'")
- print(ret)
- import json
- import pytest
- import requests
-
- from interface_automation_testing.接口自动化测试_L3.数据库操作与断言.db_config import execute_sql
- from interface_automation_testing.接口自动化测试_L3.数据库操作与断言.utils.log_util import logger
-
-
- class TestLitemall:
- def setup_class(self):
- # 1. 管理端登录接口
- url = "http://litemall.hogwarts.ceshiren.com/admin/auth/login"
- user_data = {"username": "admin123", "password": "admin123", "code": ""}
- r = requests.post(url, json=user_data, verify=False)
- self.token = r.json()["data"]["token"]
- # 问题: 没有执行test_get_admin_token这个方法,所以self.token 就不会被声明就会报错'TestLitemall' object has no attribute 'token'
- # 解决, self.token 的声明一定要在test_add_goods方法执行之前完成,可以使用setup_class 提前完成变量的声明
- # 2. 用户端登录接口
- url = "http://litemall.hogwarts.ceshiren.com/wx/auth/login"
- client_data = {"username": "user123", "password": "user123"}
- r = requests.post(url, json=client_data, verify=False)
- self.client_token = r.json()["data"]["token"]
-
- # ======= 数据清理,建议使用delete接口不要直接删表中的数据
- def teardown(self):
- url = "http://litemall.hogwarts.ceshiren.com/admin/goods/delete"
- r = requests.post(url, json={"id": self.goods_id}, headers={"X-Litemall-Admin-Token": self.token}, verify=False)
- logger.debug(f"删除商品的响应信息为{json.dumps(r.json(), indent=2, ensure_ascii=False)}")
-
- # 上架商品接口调试
- # ====问题2: goods_name 不能重复,所以需要添加参数化
- @pytest.mark.parametrize("goods_name", ["ADcarry12", "ADcarry13"])
- def test_add_goods(self, goods_name):
- # 3. 上架商品接口
- url = "http://litemall.hogwarts.ceshiren.com/admin/goods/create"
- goods_data = {
- "goods": {"picUrl": "", "gallery": [], "isHot": False, "isNew": True, "isOnSale": True, "goodsSn": "9001",
- "name": goods_name}, "specifications": [{"specification": "规格", "value": "标准", "picUrl": ""}],
- "products": [{"id": 0, "specifications": ["标准"], "price": "66", "number": "66", "url": ""}],
- "attributes": []}
- # 问题: token 是 手动复制进去的,一旦发生变化,还需要再次修改
- # 解决方案: token 需要自动完成获取,并且赋值
- r = requests.post(url, json=goods_data, headers={"X-Litemall-Admin-Token": self.token}, verify=False)
- # 打印响应体内容
- # print(r.json())
- # logger.debug(f"上架商品接口接口的相应信息为{r.json()}")
- logger.debug(f"上架商品接口接口的相应信息为{json.dumps(r.json(), indent=2, ensure_ascii=False)}")
- # 4. 获取商品列表
- goods_list_url = "http://litemall.hogwarts.ceshiren.com/admin/goods/list"
- # 是一个get请求,参数需要通过params也就是url参数传递
- goods_data = {
- "name": goods_name,
- "order": "desc",
- "sort": "add_time"
- }
- r = requests.get(goods_list_url, params=goods_data,
- headers={"X-Litemall-Admin-Token": self.token}, verify=False)
- self.goods_id = r.json()["data"]["list"][0]["id"]
- logger.debug(f"获取商品列表接口的响应信息为{json.dumps(r.json(), indent=2, ensure_ascii=False)}")
-
- # 5.获取商品详情接口=========
- goods_detail_url = "http://litemall.hogwarts.ceshiren.com/admin/goods/detail"
- r = requests.get(goods_detail_url, params={"id": self.goods_id},
- headers={"X-Litemall-Admin-Token": self.token}, verify=False)
- product_id = r.json()["data"]["products"][0]["id"]
- logger.debug(f"获取商品详情接口的响应信息为{json.dumps(r.json(), indent=2, ensure_ascii=False)}")
-
- # 6. 添加购物车接口
- url = "http://litemall.hogwarts.ceshiren.com/wx/cart/add"
- # 问题: goodsId 和 productId 是写死的,变量的传递没有完成
- # 解决方案: goodsId 和 productId 从其他的接口获取,并传递给添加购物车接口
- cart_data = {"goodsId": self.goods_id, "number": 1, "productId": product_id}
- r = requests.post(url, json=cart_data, headers={"X-Litemall-Token": self.client_token}, verify=False)
- res = r.json()
- logger.info(f"添加购物车接口响应信息为{json.dumps(r.json(), indent=2, ensure_ascii=False)}")
-
- # ===============问题1: 缺少断言
- ret = execute_sql(f"select * from litemall_cart where user_id=1 and deleted=0 and goods_name='{goods_name}'")
-
- assert ret is not None
- # ===============解决: 添加断言
- assert res["errmsg"] == "成功"
项目结构:

接口鉴权是指对通过接口进行的数据访问进行权限验证,以防止未授权的访问和恶意攻击。具体来说,接口鉴权可以在数据访问请求到达服务器之前,通过验证请求中的用户身份、角色、令牌等信息来确认请求者是否有权访问该接口。如果请求没有通过鉴权验证,服务器将拒绝该请求并返回相应的错误信息。
在实现接口鉴权时,需要考虑以下几个因素:
总之,接口鉴权是保障数据访问安全的重要组成部分,可以有效防止未授权的访问和恶意攻击,保护系统的稳定性和可靠性。


Session() 对象- import requests
-
-
- class TestVerify:
- def test_cookies_by_write(self):
- # 简单场景,直接写入cookie
- url = "https://httpbin.ceshiren.com/cookies"
- r = requests.get(url, verify=False, cookies={"teacher": "LiMing"})
- print(r.json())
-
- def test_cookies_without_session(self):
- # 第一次登陆,植入cookie
- set_url = "https://httpbin.ceshiren.com/cookies/set/teacher/LiMing"
- r1 = requests.get(set_url, verify=False)
- print(r1.json())
-
- # 第二次请求的时候没有携带cookie信息
- url = "https://httpbin.ceshiren.com/cookies"
- r2 = requests.get(url, verify=False)
- print(r2.json())
-
- def test_cookies_session(self):
- req = requests.Session()
-
- # 第一次登陆,植入cookie
- set_url = "https://httpbin.ceshiren.com/cookies/set/teacher/LiMing"
- r1 = req.get(set_url, verify=False)
- print(r1.json())
-
- # 第二次请求的时候即可携带cookie信息
- url = "https://httpbin.ceshiren.com/cookies"
- r2 = req.get(url, verify=False)
- print(r2.json())
- import requests
-
-
- class TestVerify:
- def test_token(self):
- # 1. 前端登录进去以后,会拿到服务器给的tocken
- url = "http://litemall.hogwarts.ceshiren.com/admin/auth/login"
- user_data = {"username": "admin123", "password": "admin123", "code": ""}
- r = requests.post(url,
- json=user_data,
- verify=False)
- self.token = r.json()["data"]["token"]
- print(r.json())
-
- # 2. 之后的请求均携带token
- goods_list_url = "http://litemall.hogwarts.ceshiren.com/admin/goods/list"
- goods_data = {"name": "LiMing", "order": "desc", "sort": "add_time"}
- r = requests.get(goods_list_url,
- params=goods_data,
- headers={"X-Litemall-Admin-Token": self.token},
- verify=False)
- print(r.json())
- import requests
- from requests.auth import HTTPBasicAuth
-
-
- class TestVerify:
- def test_basic_auth(self):
- # 正确示例
- r = requests.get("https://httpbin.ceshiren.com/basic-auth/admin/666666",
- verify=False,
- auth=HTTPBasicAuth("admin", "666666"))
- print(r.json())
-
- # 错误示例
- # r = requests.get("https://httpbin.ceshiren.com/basic-auth/admin/666666",
- # verify=False,
- # auth=HTTPBasicAuth("admin2", "666666"))
- # print(r.json())





- import logging
- import os
-
- from logging.handlers import RotatingFileHandler
-
- # 绑定绑定句柄到logger对象
- logger = logging.getLogger(__name__)
-
- # 获取当前工具文件所在的路径
- root_path = os.path.dirname(os.path.abspath(__file__))
-
- # 拼接当前要输出日志的路径
- root_len = len(root_path)
- strs = root_path[0:root_len-6]
- log_dir_path = os.sep.join([strs,f'datas\logs'])
-
- if not os.path.isdir(log_dir_path):
- os.mkdir(log_dir_path)
-
- # 创建日志记录器,指明日志保存路径,每个日志的大小,保存日志的上限
- file_log_handler = RotatingFileHandler(os.sep.join([log_dir_path, 'log.log']),
- maxBytes=1024 * 1024, backupCount=10,
- encoding="utf-8")
-
- # 设置日志的格式
- date_string = '%Y-%m-%d %H:%M:%S'
- formatter = logging.Formatter(
- '[%(asctime)s] [%(levelname)s] [%(filename)s]/[line: %(lineno)d]/[%(funcName)s] %(message)s ',
- date_string)
-
- # 日志输出到控制台的句柄
- stream_handler = logging.StreamHandler()
-
- # 将日志记录器指定日志的格式
- file_log_handler.setFormatter(formatter)
- stream_handler.setFormatter(formatter)
-
- # 为全局的日志工具对象添加日志记录器
- # 绑定绑定句柄到logger对象
- logger.addHandler(stream_handler)
- logger.addHandler(file_log_handler)
-
- # 设置日志输出级别
- logger.setLevel(level=logging.INFO)
-
-
- def prit_path():
- print(root_path)
- print(log_dir_path)
-
- prit_path()
- import json
- import pytest
- import requests
-
- from interface_automation_testing.接口自动化测试_L3.电子商城接口自动化测试实战.utils.log_util import logger
-
-
- class TestLitemall:
- def setup_class(self):
- # 1. 管理端登录接口
- url = "http://litemall.hogwarts.ceshiren.com/admin/auth/login"
- user_data = {"username": "hogwarts", "password": "test12345", "code": ""}
- r = requests.post(url, json=user_data, verify=False)
- self.token = r.json()["data"]["token"]
-
- # 2. 用户端登录接口
- url = "http://litemall.hogwarts.ceshiren.com/wx/auth/login"
- client_data = {"username": "user123", "password": "user123"}
- r = requests.post(url, json=client_data, verify=False)
- self.client_token = r.json()["data"]["token"]
-
- def teardown(self):
- url = "http://litemall.hogwarts.ceshiren.com/admin/goods/delete"
- r = requests.post(url,
- json={"id": self.goods_id},
- headers={"X-Litemall-Admin-Token": self.token},
- verify=False)
- logger.info(f"删除商品的响应信息为{json.dumps(r.json(), indent=2, ensure_ascii=False)}")
-
- @pytest.mark.parametrize("goods_name", ["ADcarry12", "ADcarry13"])
- def test_add_goods(self, goods_name):
- # 上架商品
- url = "http://litemall.hogwarts.ceshiren.com/admin/goods/create"
- goods_data = {
- "goods": {"picUrl": "", "gallery": [], "isHot": False, "isNew": True, "isOnSale": True, "goodsSn": "9001",
- "name": goods_name},
- "specifications": [{"specification": "规格", "value": "标准", "picUrl": ""}],
- "products": [{"id": 0, "specifications": ["标准"], "price": "66", "number": "66", "url": ""}],
- "attributes": []
- }
- r = requests.post(url, json=goods_data, headers={"X-Litemall-Admin-Token": self.token})
- logger.info(f"上架商品接口接口的相应信息为{json.dumps(r.json(), indent=2, ensure_ascii=False)}")
-
- # 获取商品列表
- goods_list_url = "http://litemall.hogwarts.ceshiren.com/admin/goods/list"
- goods_data = {
- "name": goods_name,
- "order": "desc",
- "sort": "add_time"
- }
- r = requests.get(goods_list_url,
- params=goods_data,
- headers={"X-Litemall-Admin-Token": self.token},
- verify=False)
- self.goods_id = r.json()["data"]["list"][0]["id"]
- logger.info(f"获取商品列表接口的响应信息为{json.dumps(r.json(), indent=2, ensure_ascii=False)}")
-
- # 获取商品详情
- goods_detail_url = "http://litemall.hogwarts.ceshiren.com/admin/goods/detail"
- r = requests.get(goods_detail_url,
- params={"id": self.goods_id},
- headers={"X-Litemall-Admin-Token": self.token},
- verify=False)
- product_id = r.json()["data"]["products"][0]["id"]
- logger.info(f"获取商品详情接口的响应信息为{json.dumps(r.json(), indent=2, ensure_ascii=False)}")
-
- # 添加购物车接口
- url = "http://litemall.hogwarts.ceshiren.com/wx/cart/add"
- cart_data = {"goodsId": self.goods_id, "number": 1, "productId": product_id}
- r = requests.post(url, json=cart_data, headers={"X-Litemall-Token": self.client_token})
- res = r.json()
- logger.info(f"添加购物车接口响应信息为{json.dumps(r.json(), indent=2, ensure_ascii=False)}")
-
- # 断言
- assert res["errmsg"] == "成功"
- # 生成报告信息
- pytest .\test_litemall.py --alluredir=./datas/report
- # 生成报告在线服务,查看报告
- allure serve ./datas/report
运行结果:

项目结构: