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

feat(core): 增强 Exchange，实现智能变量替换与类型保持

Browse Source 
- 优化 conftest.py 增加异常日志记录和测试报告环境信息
 - 其他优化
This commit is contained in:
CNWei
2026-03-16 19:14:29 +08:00
parent 00791809df
commit d05757f7cc
9 changed files with 195 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
commons/case_handler.py
View File

@@ -1,80 +0,0 @@
#!/usr/bin/env python
# coding=utf-8
"""
@author: CNWei
@Software: PyCharm
@contact: t6i888@163.com
@file: case_handler
@date: 2025/5/26 22:13
@desc:
"""
import json
import logging
from dataclasses import dataclass, asdict
from commons.models import TestCaseModel
logger = logging.getLogger(__name__)
@dataclass
class TestCaseHandle(TestCaseModel):
@classmethod
def new(cls, testcase: dict) -> 'TestCaseHandle':
try:
instance = cls(**testcase)
return instance
except (TypeError, ValueError) as e:
logger.warning(f"解析错误：{e}")
raise e
def to_string(self) -> str:
"""
将 字典 转换为 json 格式的字符串。
:return:
json 格式的字符串。
"""
try:
res = json.dumps(asdict(self), ensure_ascii=False)
return res
except TypeError as e:
logger.error(f"将数据转换为 json 字符串时出错: {e}")
raise e
@staticmethod
def to_dict(json_str: str) -> dict:
"""
将 json 格式的字符串转换为 字典.
:param
json_str: json 格式的字符串。
:return:
"""
try:
res = json.loads(json_str)
return res
except json.JSONDecodeError as e:
logger.error(f"将 json 字符串转换为字典时出错: {e}")
raise e
if __name__ == '__main__':
from pathlib import Path
from commons.file_processors import processor_factory
test_data = Path(r"E:\PyP\InterfaceAutoTest\TestCases\test_1_user.yaml")
yaml_data = processor_factory.get_processor_class(test_data)
case_info = TestCaseHandle.new(yaml_data.load())
print(case_info.to_string())
print(type(case_info.to_string()))
print(case_info.to_dict(case_info.to_string()))
print(type(case_info.to_dict(case_info.to_string())))
print(type(case_info))
print(case_info.parametrize)
for i in case_info.parametrize:
print(i)

98
commons/files.py
View File

@@ -1,98 +0,0 @@
#!/usr/bin/env python
# coding=utf-8
"""
@author: chen wei
@Software: PyCharm
@contact: t6i888@163.com
@file: files.py
@date: 2024 2024/9/15 21:28
@desc: 读取和保存yaml文件
"""
import logging
from pathlib import Path
import yaml
logger = logging.getLogger(__name__)
class YamlFile(dict):
def __init__(self, path=None, data=None):
super().__init__()
self.path = Path(path) if path else None
if data:
self.update(data)
elif self.path:
if self.path.is_dir():
raise IsADirectoryError(f"The path {self.path} is a directory, not a file.")
self.load()
def load(self):
if not self.path:
logger.warning("No path specified for YamlFile, cannot load.")
return self
if self.path.exists() and self.path.is_file():
with open(self.path, "r", encoding="utf-8") as f:
loaded_data = yaml.safe_load(f) or {}
self.clear()
self.update(loaded_data)
else:
logger.warning(f"File not found at {self.path}, YamlFile initialized as empty.")
self.clear()
return self
def to_yaml(self) -> str:
return yaml.safe_dump(
dict(self),
allow_unicode=True,
sort_keys=False
)
@classmethod
def by_yaml(cls, yaml_str):
data = yaml.safe_load(yaml_str) or {}
return cls(data=data)
def save(self):
if not self.path:
raise ValueError("Cannot save YamlFile instance without a specified path.")
# 确保父目录存在
self.path.parent.mkdir(parents=True, exist_ok=True)
with open(self.path, "w", encoding="utf-8") as f:
yaml.safe_dump(
dict(self),
stream=f,
allow_unicode=True,
sort_keys=False
)
return self
if __name__ == '__main__':
from core.models import CaseInfo
from core.settings import TEST_CASE_DIR
# 1. 创建一个用于测试的临时yaml文件
dummy_path = TEST_CASE_DIR / "test_model_demo.yaml"
dummy_data = {
"title": "Get user info",
"request": {"method": "GET", "url": "/users/1"},
"validate": [{"equals": ["status_code", 200]}]
}
YamlFile(path=dummy_path, data=dummy_data).save()
print(f"--- 已创建临时测试文件: {dummy_path}")
# 2. 加载文件并使用Pydantic模型进行校验
yaml_case = YamlFile(dummy_path)
print("\n--- 已加载YAML内容 ---\n", yaml_case.to_yaml())
case_model = CaseInfo(**yaml_case)
print("\n--- Pydantic模型校验成功 ---")
print(case_model.model_dump_json(indent=2, by_alias=True))
# 3. 清理临时文件
dummy_path.unlink()
print(f"\n--- 已清理临时文件: {dummy_path}")

131
conftest.py
View File

@@ -4,57 +4,136 @@
"""
@desc: Pytest 配置文件，用于设置全局 Fixture 和钩子函数
"""
import platform
from typing import Any
import pytest
from pathlib import Path
import logging
from core import settings
from commons.files import YamlFile
from core.context import VariableStore, ExecutionEnv
from core.executor import WorkflowExecutor
from core.models import RawSchema
from core.session import Session
from core.exchange import Exchange
from core.settings import EXTRACT_CACHE
from core.settings import EXTRACT_CACHE,base_url
logger = logging.getLogger(__name__)
@pytest.fixture(scope="session")
def api_env():
# 注册命令行参数
def pytest_addoption(parser: Any) -> None:
"""
工业级资源调度器
1. 保持全局单 Session (连接池复用)
2. 变量池 L2 内存镜像化 (减少 I/O)
注册自定义命令行参数。
允许用户通过命令行传递参数来控制测试执行的行为。
Args:
parser: Pytest 的命令行参数解析器对象。
"""
parser.addoption("--test_dir", action="store", default=None, help="测试用例目录")
parser.addoption("--env", action="store", default="test", help="运行环境标识 (test/prod/dev)")
@pytest.fixture(scope="session")
def execution_context():
"""
[Session级别 Fixture] 全局执行上下文环境。
职责：
1. 生命周期管理：初始化并管理全局唯一的 Session、变量存储 (Store) 和 变量交换器 (Exchanger)。
2. 资源复用：确保 HTTP 连接池复用，减少握手开销。
3. 数据持久化：在测试结束时自动将提取的变量回写到磁盘。
Yields:
ExecutionEnv: 包含 session, store, exchanger 的环境对象实例。
"""
# Setup: 加载环境
store = VariableStore(settings.DATA_DIR / "extract.yaml")
store = VariableStore(EXTRACT_CACHE)
exchanger = Exchange(variable_cache=store.store)
session = Session(settings.base_url)
executor = WorkflowExecutor()
session = Session(base_url)
env = ExecutionEnv(session, store, executor, exchanger)
env = ExecutionEnv(session, store, exchanger)
yield env # 注入到测试用例中
# Teardown: 统一持久化与清理
store.persist()
session.close()
@pytest.fixture(scope="session")
def session():
"""全局共享的 Session Fixture"""
return Session(settings.base_url)
@pytest.fixture(scope="session")
def exchanger():
"""全局共享的 Exchange Fixture"""
return Exchange(EXTRACT_CACHE)
def pytest_exception_interact(node: Any, call: Any, report: Any) -> None:
"""
[Hook] 异常交互钩子。
当测试用例执行失败（断言错误或代码异常）时触发。
主要用于捕获详细的错误堆栈信息，并将其格式化输出到日志中，
以便于在控制台或日志文件中快速定位问题。
# @pytest.fixture(scope="session")
# def case_engine(session, exchanger):
# """全局共享的 CaseEngine Fixture"""
# return CaseEngine(session, exchanger)
Args:
node: 发生异常的测试节点（Item 或 Collector）。
call: 测试调用信息（包含 excinfo 异常信息）。
report: 测试报告对象。
"""
if report.failed:
# 获取详细的错误堆栈（包含 assert 的对比信息）
# long,short,no-locals
exc_info = call.excinfo.getrepr(style='short')
logger.error(f"\n{'=' * 40} TEST FAILED {'=' * 40}\n"
f"Node: {node.name}\n"
f"Error:\n{exc_info}"
)
logger.error("=" * 93 + "\n")
def pytest_sessionfinish(session: Any, exitstatus: int) -> None:
"""
[Hook] 会话结束钩子。
在所有测试执行完毕后调用。主要完成以下工作：
1. 根据退出状态码记录不同级别的日志信息。
2. 收集测试环境信息（如 Base URL, Python 版本, 操作系统等）。
3. 生成 `environment.properties` 文件供 Allure 报告展示。
Args:
session: Pytest 会话对象。
exitstatus: 整体测试执行的退出状态码。
"""
match exitstatus:
case pytest.ExitCode.OK:
logging.info("测试全部通过！")
case pytest.ExitCode.TESTS_FAILED:
logging.warning("部分测试用例执行失败，请检查报告。")
case pytest.ExitCode.INTERRUPTED:
logging.error("测试被用户手动中断（Ctrl+C）。")
case pytest.ExitCode.INTERNAL_ERROR:
logging.critical("Pytest 发生内部错误！")
case pytest.ExitCode.USAGE_ERROR:
logging.error("Pytest 命令行参数错误或用法不当。")
case pytest.ExitCode.NO_TESTS_COLLECTED:
logging.warning("未发现任何测试用例。")
case _:
logging.error(f"未知错误状态码: {exitstatus}")
report_dir = session.config.getoption("--alluredir")
if not report_dir:
return
report_path = Path(report_dir)
# 收集环境信息 (适配接口自动化)
env_info = {
"Base URL": base_url,
"Environment": session.config.getoption("--env"),
"Python Version": platform.python_version(),
"OS System": platform.system(),
"Project": "Interface Auto Test"
}
try:
if not report_path.exists():
report_path.mkdir(parents=True, exist_ok=True)
# 生成 environment.properties 文件
env_file = report_path / "environment.properties"
with env_file.open("w", encoding="utf-8") as f:
for k, v in env_info.items():
f.write(f"{k}={v}\n")
logging.info("Allure 环境信息已生成。")
except Exception as e:
logging.error(f"无法写入环境属性: {e}")

4
core/context.py
View File

@@ -10,7 +10,7 @@
@desc:
"""
from dataclasses import dataclass
from typing import Dict, Any
from typing import Any
from pathlib import Path
from core.exchange import Exchange
@@ -25,7 +25,7 @@ class VariableStore:
self.seed_file = seed_file
self.processor = YamlProcessor(seed_file)
# 启动时仅加载一次
self.store: Dict[str, Any] = self.processor.load() or {}
self.store: dict[str, Any] = self.processor.load() or {}
def persist(self):
"""测试结束时统一写盘"""

11
core/creator.py
View File

@@ -14,6 +14,8 @@ import logging
import allure
from pathlib import Path
from dataclasses import dataclass
from conftest import execution_context
from core import settings
from core.executor import WorkflowExecutor
from pydantic import ValidationError
@@ -142,7 +144,7 @@ class CaseGenerator:
method_name = f"test_{index:03d}_{safe_title}"
print(method_name)
setattr(target_cls, method_name, dynamic_test_method)
print(target_cls.__dict__)
# print(target_cls.__dict__)
logger.debug(f"Successfully registered: {method_name}")
@staticmethod
@@ -150,8 +152,7 @@ class CaseGenerator:
"""封装具体的 pytest 执行节点"""
case_template = entity.step_data
context = entity.row_context
def build_actual_case(instance: TestTemplateBase, api_env):
def build_actual_case(instance: TestTemplateBase, execution_context):
# --- 1. 动态设置 Allure 报告属性 ---
allure.dynamic.epic(case_template.epic or settings.allure_epic)
@@ -161,7 +162,7 @@ class CaseGenerator:
# 日志记录 (利用 instance 标注来源)
logger.info(f"[Runner] Class: {instance.__class__.__name__} | Case: {title}")
try:
WorkflowExecutor.perform(case_template, api_env, context=context)
WorkflowExecutor.perform(case_template, execution_context, context=context)
except Exception as e:
# 可以在这里记录更详细的运行上下文快照
@@ -176,4 +177,4 @@ if __name__ == '__main__':
# print(CaseDataLoader.get_all_cases(TEST_CASE_DIR))
# --- 引导执行 ---
CaseGenerator.build_and_register(TestTemplateBase, settings.TEST_CASE_DIR)
CaseGenerator.build_and_register(TestTemplateBase, settings.TEST_CASE_DIR)

2
core/executor.py
View File

@@ -139,4 +139,4 @@ class WorkflowExecutor:
rendered_validate_list = env.exchanger.replace(raw_validate_list)
# 重新通过 Adapter 触发类型修复 (str -> int)
final_validate_data = VALIDATE_LIST_ADAPTER.validate_python(rendered_validate_list)
CaseValidator.validate(resp, final_validate_data)
CaseValidator.validate(resp, final_validate_data)

2
test_cases/test_collector.py
View File

@@ -27,4 +27,4 @@ try:
except Exception as e:
logger.critical(f"--- [Collector] 动态生成测试用例时发生致命错误，测试执行中止 ---", exc_info=True)
# 抛出异常，让 pytest 捕获并报告为收集错误 (Collection Error)
raise RuntimeError("测试用例收集失败，请检查日志中的详细错误信息。") from e
raise RuntimeError("测试用例收集失败，请检查日志中的详细错误信息。") from e

32
utils/dirs_manager.py Normal file
View File

@@ -0,0 +1,32 @@
#!/usr/bin/env python
# coding=utf-8
"""
@author: CNWei,ChenWei
@Software: PyCharm
@contact: t6g888@163.com
@file: dirs_manager
@date: 2026/2/3 10:52
@desc:
"""
from pathlib import Path
from core.settings import REQUIRED_DIRS
def ensure_dirs_ok():
"""
统一管理项目目录的创建逻辑
"""
for folder in REQUIRED_DIRS:
# 使用 exist_ok=True 避免并发冲突
folder.mkdir(parents=True, exist_ok=True)
def ensure_dir(path: Path) -> Path:
"""确保路径存在并返回路径本身"""
if not isinstance(path, Path):
path = Path(path)
path.mkdir(parents=True, exist_ok=True)
return path

48
utils/report_handler.py Normal file
View File

@@ -0,0 +1,48 @@
#!/usr/bin/env python
# coding=utf-8
"""
@author: CNWei,ChenWei
@Software: PyCharm
@contact: t6g888@163.com
@file: report_handler
@date: 2026/2/3 13:51
@desc:
"""
import logging
import subprocess
import shutil
from core.settings import ALLURE_TEMP, REPORT_DIR
logger = logging.getLogger(__name__)
def generate_allure_report() -> bool:
"""
将 JSON 原始数据转换为 HTML 报告
"""
if not ALLURE_TEMP.exists() or not any(ALLURE_TEMP.iterdir()):
logger.warning("未发现 Allure 测试数据，跳过报告生成。")
return False
# 检查环境是否有 allure 命令行工具
if not shutil.which("allure"):
logger.error("系统未安装 Allure 命令行工具，请先安装：https://allurereport.org/docs/")
return False
try:
logger.info("正在生成 Allure HTML 报告...")
# --clean 会清理掉 REPORT_DIR 里的旧报告
subprocess.run(
f'allure generate "{ALLURE_TEMP}" -o "{REPORT_DIR}" --clean',
shell=True,
check=True,
capture_output=True,
text=True
)
logger.info(f"Allure 报告已生成至: {REPORT_DIR}")
return True
except subprocess.CalledProcessError as e:
logger.error(f"Allure 报告生成失败: {e.stderr}")
return False