CNWei/InterfaceAutoTest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(exchange,case_validator),refactor(),feat(model): 解决 Pydantic 模型初始化与变量占位符的类型冲突,优化变量替换逻辑,重构 CaseInfo 模型并引入延迟校验机制

Browse Source 
- 引入 SmartInt 和 SmartDict 类型，支持 YAML 占位符与业务类型的自动转换。
- 优化 CaseInfo 互斥校验逻辑，确保 request 与 api_action 二选一。
- 统一使用 Pydantic V2 的 model_config 规范。
- 将变量替换时机提前至模型实例化之前，支持占位符在校验前完成真实值注入，
保证了 int/bool 等字段的类型转换正确性。
- 优化断言渲染时机，支持响应提取值关联。
This commit is contained in:
CNWei
2026-03-11 10:29:16 +08:00
parent 69a96a0060
commit 293b5160fe
39 changed files with 1359 additions and 1031 deletions
Download Patch File Download Diff File Expand all files Collapse all files

254
commons/cases.py
View File

@@ -9,23 +9,18 @@
@date: 2024 2024/9/16 9:57
@desc: 动态生成用例
"""
from dataclasses import asdict
from pathlib import Path
import logging
from typing import Union, Generator, Type
from unittest import TestCase
import allure
import pytest
from commons import settings
from commons.file_processors.processor_factory import get_processor_class
# from commons.models import CaseInfo
from commons.session import Session
from commons.exchange import Exchange
from commons.templates import Template
from commons.case_handler import TestCaseHandle
from utils import case_validator
from core import settings
from commons.file_processors.yaml_processor import YamlProcessor as FileHandle
from commons.models import CaseInfo
from core.session import Session
from core.exchange import Exchange
from utils import data_driver, case_validator
logger = logging.getLogger(__name__)
@@ -37,195 +32,90 @@ exchanger = Exchange(settings.exchanger)
class TestAPI:
@classmethod
def run(cls, testcase_dir: Union[Path, str] = cases_dir):
for fp in CaseFinder(testcase_dir).find_testcases():
print(fp.name)
case = CaseGenerator(fp).generate_testcases()
print(f"{case=}")
for i in case:
print(f"{i=}")
CaseRegister(cls).register_test_func(i)
# @classmethod
# def find_test_cases(cls, case_dir: Path = cases_dir):
# """
# 搜索和加载yaml文件
# :return:
# """
# case_path_list = case_dir.glob("**/test_*.yaml") # 搜索当前目录及其子目录下以test_开头yaml为后缀的文件
# for case_path in case_path_list:
# logger.info(f"加载文件：{case_path}")
#
# file = FileHandle(case_path) # 自动读取yaml文件
# try:
# CaseInfo(**file) # 校验用例格式
# logger.info(f"case_info：{FileHandle.to_string(file)}") # 把case_info 转成字符串，然后记录日志
# case_func = cls.new_case(case_path.stem, file) # 转换为pytest格式
# # print(case_path.stem)
# setattr(cls, f"{case_path.stem}", case_func) # 把pytest格式添加到类中
# except Exception as e:
# logger.error(e)
#
# @classmethod
# def new_case(cls, file_name, case_info: dict):
# test_case = data_driver.DataDriver().generate_cases(file_name, case_info)
#
# keys_list = list(test_case.keys())
# logger.info(f"keys_list：{keys_list}")
#
# values_list = list(test_case.values())
# logger.info(f"测试用例列表：{values_list}")
#
# driver_title = [i.get("title") for i in values_list]
# logger.info(f"driver_title={driver_title}")
#
# epic = case_info["epic"] if case_info["epic"] else settings.allure_epic
# logger.info(f"epic：{epic}")
#
# feature = case_info["feature"] if case_info["feature"] else settings.allure_feature
# logger.info(f"feature：{feature}")
#
# story = case_info["story"] if case_info["story"] else settings.allure_story
# logger.info(f"story：{story}")
#
# @allure.epic(epic)
# @allure.feature(feature)
# @allure.story(story)
# @pytest.mark.parametrize("case_key", keys_list, ids=driver_title)
# def test_func(self, case_key):
# logger.info(f"case_key：{case_key}")
#
# test_case_mapping = test_case.get(case_key)
# logger.info(f"测试用例：{test_case_mapping}")
#
# allure.dynamic.title(test_case_mapping.get("title"))
#
# logger.info(f"用例开始执行：{test_case_mapping.get('title')}".center(80, "="))
#
# # 0，变量替换
# new_case_info = exchanger.replace(test_case_mapping)
# logger.info(f"1，正在注入变量...")
# logger.info(f"new_case_info：{new_case_info}")
# # 1，发送请求
# logger.info(f"2，正在请求接口...")
# resp = session.request(**new_case_info.get("request"))
#
# logger.info(f"3，正在提取变量...")
# # 2，保存变量(接口关联)
# for var_name, extract_info in new_case_info.get("extract").items():
# logger.info(f"保存变量：{var_name}{extract_info}")
# exchanger.extract(resp, var_name, *extract_info)
# # 3，断言
# logger.info(f"4，正在断言...")
# assert_case_info = exchanger.replace(test_case_mapping) # 为断言加载变量
# logger.info(f"替换变量后：{assert_case_info}")
# # assert_case_info.assert_all() # 执行断言
# _validator = case_validator.CaseValidator()
# _validator.assert_all(assert_case_info.get("validate"))
#
# logger.info(f"用例执行结束：{test_case_mapping.get('title')}".center(80, "="))
#
# return test_func
def find_test_cases(cls, case_dir: Path = cases_dir):
"""
搜索和加载yaml文件
:return:
"""
case_path_list = case_dir.glob("**/test_*.yaml") # 搜索当前目录及其子目录下以test_开头yaml为后缀的文件
for case_path in case_path_list:
logger.info(f"加载文件：{case_path}")
file = FileHandle(case_path) # 自动读取yaml文件
try:
CaseInfo(**file) # 校验用例格式
logger.info(f"case_info：{FileHandle.to_string(file)}") # 把case_info 转成字符串，然后记录日志
case_func = cls.new_case(case_path.stem, file) # 转换为pytest格式
# print(case_path.stem)
setattr(cls, f"{case_path.stem}", case_func) # 把pytest格式添加到类中
except Exception as e:
logger.error(e)
class CaseFinder:
find_suffix: str = settings.test_suffix
@classmethod
def new_case(cls, file_name, case_info: dict):
test_case = data_driver.DataDriver().generate_cases(file_name, case_info)
def __init__(self, testcase_dir: Union[str, Path]):
if Path(testcase_dir).is_dir():
self.testcase_dir: Path = Path(testcase_dir)
else:
raise FileNotFoundError("不是有效的目录")
keys_list = list(test_case.keys())
logger.info(f"keys_list：{keys_list}")
def find_testcases(self) -> Generator[Path, None, None]:
testcase_files = self.testcase_dir.glob(f"**/test_*.{self.find_suffix}")
for fp in testcase_files:
logger.info(f"加载文件：{fp}")
yield fp
values_list = list(test_case.values())
logger.info(f"测试用例列表：{values_list}")
driver_title = [i.get("title") for i in values_list]
logger.info(f"driver_title={driver_title}")
class CaseGenerator:
epic = case_info["epic"] if case_info["epic"] else settings.allure_epic
logger.info(f"epic：{epic}")
def __init__(self, fp: Union[str, Path]):
self.fp: Path = Path(fp)
feature = case_info["feature"] if case_info["feature"] else settings.allure_feature
logger.info(f"feature：{feature}")
def generate_testcases(self) -> Generator[dict, None, None]:
file_name = self.fp.stem
story = case_info["story"] if case_info["story"] else settings.allure_story
logger.info(f"story：{story}")
case_info_ = get_processor_class(self.fp).load() # 自动读取yaml文件
case_info = TestCaseHandle.new(case_info_)
@allure.epic(epic)
@allure.feature(feature)
@allure.story(story)
@pytest.mark.parametrize("case_key", keys_list, ids=driver_title)
def test_func(self, case_key):
logger.info(f"case_key：{case_key}")
if not case_info.parametrize:
yield {file_name + "__": asdict(case_info)}
else:
cases = {}
args_names = case_info.parametrize[0]
for i, args_values in enumerate(case_info.parametrize[1:]):
# print(args_values)
context = dict(zip(args_names, args_values))
print(context)
# rendered = Template(FileHandle.to_string(case_info)).render(context)
rendered = Template(case_info.to_string()).render(context)
# cases.update({file_name + "[" + str(i) + "]": FileHandle.to_dict(rendered)})
cases.update({file_name + "_" + str(i): case_info.to_dict(rendered)})
test_case_mapping = test_case.get(case_key)
logger.info(f"测试用例：{test_case_mapping}")
yield cases
allure.dynamic.title(test_case_mapping.get("title"))
logger.info(f"用例开始执行：{test_case_mapping.get('title')}".center(80, "="))
class CaseRegister:
def __init__(self, register: Type["TestAPI"]):
self.register: Type["TestAPI"] = register
# 0，变量替换
new_case_info = exchanger.replace(test_case_mapping)
logger.info(f"1，正在注入变量...")
logger.info(f"new_case_info：{new_case_info}")
# 1，发送请求
logger.info(f"2，正在请求接口...")
resp = session.request(**new_case_info.get("request"))
def register_test_func(self, case: dict):
for test_filed_name, case_info in case.items():
epic = case_info["epic"] if case_info["epic"] else settings.allure_epic
logger.info(f"epic：{epic}")
logger.info(f"3，正在提取变量...")
# 2，保存变量(接口关联)
for var_name, extract_info in new_case_info.get("extract").items():
logger.info(f"保存变量：{var_name}{extract_info}")
exchanger.extract(resp, var_name, *extract_info)
# 3，断言
logger.info(f"4，正在断言...")
assert_case_info = exchanger.replace(test_case_mapping) # 为断言加载变量
logger.info(f"替换变量后：{assert_case_info}")
# assert_case_info.assert_all() # 执行断言
_validator = case_validator.CaseValidator()
_validator.assert_all(assert_case_info.get("validate"))
feature = case_info["feature"] if case_info["feature"] else settings.allure_feature
logger.info(f"feature：{feature}")
logger.info(f"用例执行结束：{test_case_mapping.get('title')}".center(80, "="))
story = case_info["story"] if case_info["story"] else settings.allure_story
logger.info(f"story：{story}")
@allure.epic(epic)
@allure.feature(feature)
@allure.story(story)
def register_func(instance, testcase=case_info):
# allure.dynamic.epic(epic)
# allure.dynamic.feature(feature)
# allure.dynamic.story(story)
allure.dynamic.title(testcase.get("title"))
logger.info(f"用例开始执行：{testcase.get('title')}".center(80, "="))
# 0，变量替换
new_case_info = exchanger.replace(testcase)
logger.info(f"1，正在注入变量...")
logger.info(f"new_case_info：{new_case_info}")
# 1，发送请求
logger.info(f"2，正在请求接口...")
resp = session.request(**new_case_info.get("request"))
logger.info(f"3，正在提取变量...")
# 2，保存变量(接口关联)
for var_name, extract_info in new_case_info.get("extract").items():
logger.info(f"保存变量：{var_name}{extract_info}")
exchanger.extract(resp, var_name, *extract_info)
# 3，断言
logger.info(f"4，正在断言...")
assert_case_info = exchanger.replace(testcase) # 为断言加载变量
logger.info(f"替换变量后：{assert_case_info}")
# assert_case_info.assert_all() # 执行断言
_validator = case_validator.CaseValidator()
_validator.assert_all(assert_case_info.get("validate"))
logger.info(f"用例执行结束：{testcase.get('title')}".center(80, "="))
# return test_func
setattr(self.register, test_filed_name, register_func) # 把pytest格式添加到类中
return test_func
# TestAPI.find_yaml_case()
if __name__ == '__main__':
TestAPI.run(cases_dir)
print(TestAPI.__dict__)
TestAPI.find_test_cases()
# print(TestAPI.__dict__)