feat: 移除DDT模式的支持,改用POM模式
- 删除 data_loader 数据驱动加载器。 - 删除 test_keyword_sample 测试执行代码 - 新增 base_page [DDT模式极大的限制了灵活性,增加了代码的编写难度,另外项目使用者都会编码故而转用只针对POM模式进行优化]
This commit is contained in:
45
core/base_page.py
Normal file
45
core/base_page.py
Normal file
@@ -0,0 +1,45 @@
|
|||||||
|
#!/usr/bin/env python
|
||||||
|
# coding=utf-8
|
||||||
|
|
||||||
|
"""
|
||||||
|
@author: CNWei,ChenWei
|
||||||
|
@Software: PyCharm
|
||||||
|
@contact: t6g888@163.com
|
||||||
|
@file: base_page
|
||||||
|
@date: 2026/1/26 17:33
|
||||||
|
@desc:
|
||||||
|
"""
|
||||||
|
import logging
|
||||||
|
from typing import Type, TypeVar
|
||||||
|
from core.driver import CoreDriver
|
||||||
|
# 定义一个泛型,用于类型推断(IDE 依然会有补全提示)
|
||||||
|
T = TypeVar('T', bound='BasePage')
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
class BasePage:
|
||||||
|
def __init__(self, driver: CoreDriver):
|
||||||
|
self.driver = driver
|
||||||
|
# 这里放全局通用的 Page 属性和逻辑
|
||||||
|
# --- 页面工厂:属性懒加载 ---
|
||||||
|
# 这样你可以在任何页面直接通过 self.home_page 访问首页
|
||||||
|
@property
|
||||||
|
def home_page(self):
|
||||||
|
from page_objects.home_page import HomePage
|
||||||
|
return HomePage(self.driver)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def login_page(self):
|
||||||
|
from page_objects.login_page import LoginPage
|
||||||
|
return LoginPage(self.driver)
|
||||||
|
|
||||||
|
# 封装一些所有页面通用的元动作
|
||||||
|
def get_toast(self, text):
|
||||||
|
return self.driver.is_visible("text", text)
|
||||||
|
|
||||||
|
def to_page(self, page_class: Type[T]) -> T:
|
||||||
|
"""
|
||||||
|
通用的页面跳转/获取方法
|
||||||
|
:param page_class: 目标页面类
|
||||||
|
:return: 目标页面的实例
|
||||||
|
"""
|
||||||
|
logger.info(f"跳转到页面: {page_class.__name__}")
|
||||||
|
return page_class(self.driver)
|
||||||
@@ -177,8 +177,10 @@ if __name__ == "__main__":
|
|||||||
cond1 = get_condition("toast_visible", "保存成功")
|
cond1 = get_condition("toast_visible", "保存成功")
|
||||||
print(cond1)
|
print(cond1)
|
||||||
# 调用闭包生成的条件
|
# 调用闭包生成的条件
|
||||||
cond2 = get_condition("is_element_present", (By.ID, "submit"))
|
# cond2 = get_condition("is_element_present", (By.ID, "submit"))
|
||||||
print(cond2)
|
# print(cond2)
|
||||||
cond3 = get_condition(EC.presence_of_element_located, (By.ID, "submit"))
|
cond3 = get_condition(EC.presence_of_element_located, (By.ID, "submit"))
|
||||||
print(cond3)
|
print(cond3)
|
||||||
|
cond4 = get_condition("system_ready", (By.ID, "submit"))
|
||||||
|
print(cond4)
|
||||||
# WebDriverWait(driver, 10).until(cond1)
|
# WebDriverWait(driver, 10).until(cond1)
|
||||||
|
|||||||
@@ -16,7 +16,7 @@ from pathlib import Path
|
|||||||
# 项目根目录 (core 的上一级)
|
# 项目根目录 (core 的上一级)
|
||||||
# BASE_DIR = Path(__file__).parent.parent
|
# BASE_DIR = Path(__file__).parent.parent
|
||||||
BASE_DIR = Path(__file__).resolve().parents[1] # 获取根路径(绝对路径)
|
BASE_DIR = Path(__file__).resolve().parents[1] # 获取根路径(绝对路径)
|
||||||
print(BASE_DIR)
|
# print(BASE_DIR)
|
||||||
# --- 目录配置 ---
|
# --- 目录配置 ---
|
||||||
OUTPUT_DIR = BASE_DIR / "outputs"
|
OUTPUT_DIR = BASE_DIR / "outputs"
|
||||||
LOG_DIR = OUTPUT_DIR / "logs"
|
LOG_DIR = OUTPUT_DIR / "logs"
|
||||||
|
|||||||
25
page_objects/home_page.py
Normal file
25
page_objects/home_page.py
Normal file
@@ -0,0 +1,25 @@
|
|||||||
|
#!/usr/bin/env python
|
||||||
|
# coding=utf-8
|
||||||
|
|
||||||
|
"""
|
||||||
|
@author: CNWei,ChenWei
|
||||||
|
@Software: PyCharm
|
||||||
|
@contact: t6g888@163.com
|
||||||
|
@file: home_page
|
||||||
|
@date: 2026/1/26 17:37
|
||||||
|
@desc:
|
||||||
|
"""
|
||||||
|
|
||||||
|
from core.base_page import BasePage
|
||||||
|
class HomePage(BasePage):
|
||||||
|
_LOGOUT_BTN = ("text", "退出登录")
|
||||||
|
_NICKNAME = ("id", "user_nickname")
|
||||||
|
|
||||||
|
def get_nickname(self):
|
||||||
|
return self.driver.get_text(*self._NICKNAME)
|
||||||
|
|
||||||
|
def logout(self):
|
||||||
|
self.driver.click(*self._LOGOUT_BTN)
|
||||||
|
# 【核心:链式跳转】
|
||||||
|
# 退出后回到登录页
|
||||||
|
return self.login_page
|
||||||
31
page_objects/login_page.py
Normal file
31
page_objects/login_page.py
Normal file
@@ -0,0 +1,31 @@
|
|||||||
|
#!/usr/bin/env python
|
||||||
|
# coding=utf-8
|
||||||
|
|
||||||
|
"""
|
||||||
|
@author: CNWei,ChenWei
|
||||||
|
@Software: PyCharm
|
||||||
|
@contact: t6g888@163.com
|
||||||
|
@file: login_page
|
||||||
|
@date: 2026/1/26 17:34
|
||||||
|
@desc:
|
||||||
|
"""
|
||||||
|
|
||||||
|
from core.base_page import BasePage
|
||||||
|
from page_objects.home_page import HomePage
|
||||||
|
|
||||||
|
class LoginPage(BasePage):
|
||||||
|
# 定位参数私有化,不暴露给外面
|
||||||
|
_USER_FIELD = ("id", "com.app:id/username")
|
||||||
|
_PWD_FIELD = ("id", "com.app:id/password")
|
||||||
|
_LOGIN_BTN = ("id", "com.app:id/btn_login")
|
||||||
|
|
||||||
|
def login_as(self, username, password):
|
||||||
|
"""执行登录业务逻辑"""
|
||||||
|
# 调用继承自 CoreDriver 的方法(假设你的 CoreDriver 已经被注入或组合)
|
||||||
|
self.driver.input(*self._USER_FIELD, text=username)
|
||||||
|
self.driver.input(*self._PWD_FIELD, text=password, sensitive=True)
|
||||||
|
self.driver.click(*self._LOGIN_BTN)
|
||||||
|
|
||||||
|
# 【核心:链式跳转】
|
||||||
|
# 登录成功后,逻辑上应该进入首页,所以返回首页实例
|
||||||
|
return self.to_page(HomePage)
|
||||||
@@ -1,82 +0,0 @@
|
|||||||
#!/usr/bin/env python
|
|
||||||
# coding=utf-8
|
|
||||||
|
|
||||||
"""
|
|
||||||
@author: CNWei,ChenWei
|
|
||||||
@Software: PyCharm
|
|
||||||
@contact: t6g888@163.com
|
|
||||||
@file: test_keyword_sample
|
|
||||||
@date: 2026/1/23 17:48
|
|
||||||
@desc:
|
|
||||||
"""
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
import logging
|
|
||||||
from core.driver import CoreDriver
|
|
||||||
from utils.data_loader import DataLoader
|
|
||||||
from core.settings import BASE_DIR
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
|
||||||
|
|
||||||
# 假设数据文件路径
|
|
||||||
DATA_FILE = BASE_DIR / "test_cases" / "data" / "login_flow.xlsx"
|
|
||||||
|
|
||||||
|
|
||||||
# 或者 "login_flow.yaml" —— 代码不需要改动,只需要改文件名
|
|
||||||
|
|
||||||
class TestKeywordDriven:
|
|
||||||
|
|
||||||
def run_step(self, driver: CoreDriver, step: dict):
|
|
||||||
"""
|
|
||||||
核心执行引擎:反射调用 CoreDriver 的方法
|
|
||||||
"""
|
|
||||||
action_name = step.get("action")
|
|
||||||
if not action_name:
|
|
||||||
return # 跳过无效行
|
|
||||||
|
|
||||||
# 1. 获取 CoreDriver 中对应的方法
|
|
||||||
if not hasattr(driver, action_name):
|
|
||||||
raise ValueError(f"CoreDriver 中未定义方法: {action_name}")
|
|
||||||
|
|
||||||
func = getattr(driver, action_name)
|
|
||||||
|
|
||||||
# 2. 准备参数
|
|
||||||
# 你的 CoreDriver 方法签名通常是 (by, value, [args], timeout)
|
|
||||||
# 我们从 step 字典中提取这些参数
|
|
||||||
kwargs = {}
|
|
||||||
if "by" in step and step["by"]: kwargs["by"] = step["by"]
|
|
||||||
if "value" in step and step["value"]: kwargs["value"] = step["value"]
|
|
||||||
|
|
||||||
# 处理特殊参数,比如 input 方法需要的 text,或者 assert_text 需要的 expected_text
|
|
||||||
# 这里做一个简单的映射,或者在 Excel 表头直接写对应参数名
|
|
||||||
if "args" in step and step["args"]:
|
|
||||||
# 假设 input 的第三个参数是 text,这里简单处理,实际可根据 func.__code__.co_varnames 动态匹配
|
|
||||||
if action_name == "input":
|
|
||||||
kwargs["text"] = str(step["args"]) # 确保 Excel 数字转字符串
|
|
||||||
elif action_name == "assert_text":
|
|
||||||
kwargs["expected_text"] = str(step["args"])
|
|
||||||
elif action_name == "explicit_wait":
|
|
||||||
# 支持你封装的 resolve_wait_method
|
|
||||||
# 此时 method 参数就是 args 列的内容,例如 "toast_visible:成功"
|
|
||||||
kwargs["method"] = step["args"]
|
|
||||||
|
|
||||||
logger.info(f"执行步骤 [{step.get('desc', '无描述')}]: {action_name} {kwargs}")
|
|
||||||
|
|
||||||
# 3. 执行调用
|
|
||||||
func(**kwargs)
|
|
||||||
|
|
||||||
def test_execute_from_file(self, driver):
|
|
||||||
"""
|
|
||||||
主测试入口
|
|
||||||
"""
|
|
||||||
# 1. 加载数据 (自动识别 Excel/YAML)
|
|
||||||
# 注意:实际使用时建议把 load 放在 pytest.mark.parametrize 里
|
|
||||||
# 这里为了演示逻辑写在函数内
|
|
||||||
if not DATA_FILE.exists():
|
|
||||||
pytest.skip("数据文件不存在")
|
|
||||||
|
|
||||||
steps = DataLoader.load(DATA_FILE)
|
|
||||||
|
|
||||||
# 2. 遍历执行
|
|
||||||
for step in steps:
|
|
||||||
self.run_step(driver, step)
|
|
||||||
78
test_cases/test_login_demo.py
Normal file
78
test_cases/test_login_demo.py
Normal file
@@ -0,0 +1,78 @@
|
|||||||
|
#!/usr/bin/env python
|
||||||
|
# coding=utf-8
|
||||||
|
|
||||||
|
"""
|
||||||
|
@desc: 模拟登录功能测试用例
|
||||||
|
"""
|
||||||
|
import pytest
|
||||||
|
import logging
|
||||||
|
from core.driver import CoreDriver
|
||||||
|
from core.modules import AppPlatform
|
||||||
|
|
||||||
|
# 配置日志
|
||||||
|
logging.basicConfig(level=logging.INFO)
|
||||||
|
|
||||||
|
class TestLogin:
|
||||||
|
|
||||||
|
driver: CoreDriver = None
|
||||||
|
|
||||||
|
def setup_method(self):
|
||||||
|
"""
|
||||||
|
每个测试用例开始前执行:初始化 Driver 并连接设备
|
||||||
|
"""
|
||||||
|
# 定义测试设备的 Capabilities
|
||||||
|
# 注意:实际使用时,appPackage 和 appActivity 需要替换为被测 App 的真实值
|
||||||
|
caps = {
|
||||||
|
"platformName": "Android",
|
||||||
|
"automationName": "UiAutomator2",
|
||||||
|
"deviceName": "Android Emulator",
|
||||||
|
"appPackage": "com.example.android.apis", # 替换为你的 App 包名
|
||||||
|
"appActivity": ".ApiDemos", # 替换为你的 App 启动 Activity
|
||||||
|
"noReset": True, # 不清除应用数据
|
||||||
|
"newCommandTimeout": 60
|
||||||
|
}
|
||||||
|
|
||||||
|
self.driver = CoreDriver()
|
||||||
|
# 连接 Appium Server
|
||||||
|
self.driver.server_config(host="127.0.0.1", port=4723)
|
||||||
|
self.driver.connect(platform=AppPlatform.ANDROID, caps=caps)
|
||||||
|
|
||||||
|
def teardown_method(self):
|
||||||
|
"""
|
||||||
|
每个测试用例结束后执行:退出 Driver
|
||||||
|
"""
|
||||||
|
if self.driver:
|
||||||
|
self.driver.quit()
|
||||||
|
|
||||||
|
def test_login_success(self):
|
||||||
|
"""
|
||||||
|
测试场景:使用正确的用户名和密码登录成功
|
||||||
|
"""
|
||||||
|
# 1. 定位元素信息 (建议后续抽离到 Page Object 层)
|
||||||
|
# 假设登录页面的元素 ID 如下:
|
||||||
|
input_user = "id:com.example.app:id/et_username"
|
||||||
|
input_pass = "id:com.example.app:id/et_password"
|
||||||
|
btn_login = "id:com.example.app:id/btn_login"
|
||||||
|
txt_welcome = "xpath://*[@text='登录成功']"
|
||||||
|
|
||||||
|
# 2. 执行操作步骤
|
||||||
|
# 显式等待并输入用户名
|
||||||
|
self.driver.input(input_user, "", "test_user_001")
|
||||||
|
|
||||||
|
# 输入密码 (开启敏感模式,日志中脱敏)
|
||||||
|
self.driver.input(input_pass, "", "Password123!", sensitive=True)
|
||||||
|
|
||||||
|
# 点击登录按钮
|
||||||
|
self.driver.click(btn_login, "")
|
||||||
|
|
||||||
|
# 3. 断言结果
|
||||||
|
# 方式 A: 检查特定文本是否存在
|
||||||
|
# self.driver.assert_text(txt_welcome, "", "登录成功")
|
||||||
|
|
||||||
|
# 方式 B: 检查跳转后的页面元素是否可见
|
||||||
|
is_login_success = self.driver.is_visible(txt_welcome, "")
|
||||||
|
assert is_login_success, "登录失败:未检测到欢迎提示或主页元素"
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
# 允许直接运行此文件进行调试
|
||||||
|
pytest.main(["-v", "-s", __file__])
|
||||||
37
test_cases/test_settings.py
Normal file
37
test_cases/test_settings.py
Normal file
@@ -0,0 +1,37 @@
|
|||||||
|
#!/usr/bin/env python
|
||||||
|
# coding=utf-8
|
||||||
|
|
||||||
|
"""
|
||||||
|
@author: CNWei,ChenWei
|
||||||
|
@Software: PyCharm
|
||||||
|
@contact: t6g888@163.com
|
||||||
|
@file: test_settings
|
||||||
|
@date: 2026/1/16 15:56
|
||||||
|
@desc:
|
||||||
|
"""
|
||||||
|
import logging
|
||||||
|
|
||||||
|
from utils.logger import trace_step
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
@trace_step("验证失败",)
|
||||||
|
def test_settings_page_display(driver):
|
||||||
|
"""
|
||||||
|
测试设置页面是否成功加载
|
||||||
|
"""
|
||||||
|
# 此时 driver 已经通过 fixture 完成了初始化
|
||||||
|
current_act = driver.driver.current_activity
|
||||||
|
logger.info(f"捕获到当前 Activity: {current_act}")
|
||||||
|
|
||||||
|
assert ".unihome.UniHomeLauncher" in current_act
|
||||||
|
|
||||||
|
|
||||||
|
def test_wifi_entry_exists(driver):
|
||||||
|
"""
|
||||||
|
简单的元素查找示例
|
||||||
|
"""
|
||||||
|
# 这里的 driver 就是 appium.webdriver.Remote 实例
|
||||||
|
# 假设我们要查找“网络”相关的 ID
|
||||||
|
# el = driver.find_element(by='id', value='android:id/title')
|
||||||
|
assert driver.driver.session_id is not None
|
||||||
87
tests/test_finder_converter.py
Normal file
87
tests/test_finder_converter.py
Normal file
@@ -0,0 +1,87 @@
|
|||||||
|
#!/usr/bin/env python
|
||||||
|
# coding=utf-8
|
||||||
|
|
||||||
|
"""
|
||||||
|
@author: CNWei,ChenWei
|
||||||
|
@Software: PyCharm
|
||||||
|
@contact: t6g888@163.com
|
||||||
|
@file: test_finder_converter
|
||||||
|
@date: 2026/1/20 15:40
|
||||||
|
@desc: 测试 utils/finder.py 中的 FinderConverter 逻辑
|
||||||
|
"""
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from appium.webdriver.common.appiumby import AppiumBy
|
||||||
|
from utils.finder import by_converter, register_custom_finder, converter
|
||||||
|
|
||||||
|
class TestFinderConverter:
|
||||||
|
|
||||||
|
def setup_method(self):
|
||||||
|
"""每个测试用例开始前重置 converter 状态"""
|
||||||
|
converter.clear_custom_finders()
|
||||||
|
|
||||||
|
def teardown_method(self):
|
||||||
|
"""每个测试用例结束后重置 converter 状态"""
|
||||||
|
converter.clear_custom_finders()
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("input_by, expected", [
|
||||||
|
("id", "id"),
|
||||||
|
("xpath", "xpath"),
|
||||||
|
("link text", "link text"),
|
||||||
|
("aid", AppiumBy.ACCESSIBILITY_ID),
|
||||||
|
("class", AppiumBy.CLASS_NAME),
|
||||||
|
("css", AppiumBy.CSS_SELECTOR),
|
||||||
|
("uiautomator", AppiumBy.ANDROID_UIAUTOMATOR),
|
||||||
|
("predicate", AppiumBy.IOS_PREDICATE),
|
||||||
|
("chain", AppiumBy.IOS_CLASS_CHAIN),
|
||||||
|
])
|
||||||
|
def test_standard_and_shortcuts(self, input_by, expected):
|
||||||
|
"""测试标准定位方式和简写"""
|
||||||
|
assert by_converter(input_by) == expected
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("input_by, expected", [
|
||||||
|
("ID", "id"),
|
||||||
|
(" Id ", "id"),
|
||||||
|
("accessibility_id", AppiumBy.ACCESSIBILITY_ID),
|
||||||
|
("accessibility-id", AppiumBy.ACCESSIBILITY_ID),
|
||||||
|
("-ios class chain", AppiumBy.IOS_CLASS_CHAIN),
|
||||||
|
(" -Ios-Class-Chain ", AppiumBy.IOS_CLASS_CHAIN),
|
||||||
|
("UI_AUTOMATOR", AppiumBy.ANDROID_UIAUTOMATOR),
|
||||||
|
])
|
||||||
|
def test_normalization(self, input_by, expected):
|
||||||
|
"""测试归一化容错 (大小写、空格、下划线、横杠)"""
|
||||||
|
assert by_converter(input_by) == expected
|
||||||
|
|
||||||
|
def test_custom_registration(self):
|
||||||
|
"""测试自定义注册功能"""
|
||||||
|
register_custom_finder("my_text", "-android uiautomator")
|
||||||
|
assert by_converter("my_text") == "-android uiautomator"
|
||||||
|
|
||||||
|
# 测试注册后归一化依然生效
|
||||||
|
assert by_converter("MY_TEXT") == "-android uiautomator"
|
||||||
|
|
||||||
|
def test_reset_functionality(self):
|
||||||
|
"""测试重置功能"""
|
||||||
|
register_custom_finder("temp_key", "xpath")
|
||||||
|
assert by_converter("temp_key") == "xpath"
|
||||||
|
|
||||||
|
converter.clear_custom_finders()
|
||||||
|
|
||||||
|
with pytest.raises(ValueError, match="Unsupported locator strategy"):
|
||||||
|
by_converter("temp_key")
|
||||||
|
|
||||||
|
def test_invalid_strategy(self):
|
||||||
|
"""测试不支持的定位策略"""
|
||||||
|
with pytest.raises(ValueError, match="Unsupported locator strategy"):
|
||||||
|
by_converter("unknown_strategy")
|
||||||
|
|
||||||
|
def test_invalid_types(self):
|
||||||
|
"""测试非法类型输入"""
|
||||||
|
with pytest.raises(ValueError, match="Invalid selector type"):
|
||||||
|
by_converter(123) # type: ignore
|
||||||
|
|
||||||
|
with pytest.raises(ValueError, match="Invalid selector type"):
|
||||||
|
by_converter(None) # type: ignore
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
pytest.main(["-v", __file__])
|
||||||
@@ -1,130 +0,0 @@
|
|||||||
#!/usr/bin/env python
|
|
||||||
# coding=utf-8
|
|
||||||
|
|
||||||
"""
|
|
||||||
@author: CNWei,ChenWei
|
|
||||||
@Software: PyCharm
|
|
||||||
@contact: t6g888@163.com
|
|
||||||
@file: data_loader.py
|
|
||||||
@date: 2026/1/23 13:55
|
|
||||||
@desc:
|
|
||||||
"""
|
|
||||||
# !/usr/bin/env python
|
|
||||||
# coding=utf-8
|
|
||||||
|
|
||||||
"""
|
|
||||||
@author: CNWei
|
|
||||||
@desc: 数据驱动加载器 (Adapter Pattern 实现)
|
|
||||||
负责将 YAML, JSON, Excel 统一转换为 Python List[Dict] 格式
|
|
||||||
"""
|
|
||||||
import json
|
|
||||||
import logging
|
|
||||||
from pathlib import Path
|
|
||||||
from typing import List, Dict, Any, Union
|
|
||||||
|
|
||||||
import yaml
|
|
||||||
|
|
||||||
# 尝试导入 openpyxl,如果未安装则在运行时报错提示
|
|
||||||
try:
|
|
||||||
import openpyxl
|
|
||||||
except ImportError:
|
|
||||||
openpyxl = None
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
|
||||||
|
|
||||||
|
|
||||||
class DataLoader:
|
|
||||||
"""
|
|
||||||
数据加载适配器
|
|
||||||
统一输出格式: List[Dict]
|
|
||||||
"""
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def load(file_path: Union[str, Path]) -> List[Dict[str, Any]]:
|
|
||||||
"""
|
|
||||||
入口方法:根据文件后缀分发处理逻辑
|
|
||||||
"""
|
|
||||||
path = Path(file_path)
|
|
||||||
if not path.exists():
|
|
||||||
raise FileNotFoundError(f"测试数据文件未找到: {path}")
|
|
||||||
|
|
||||||
suffix = path.suffix.lower()
|
|
||||||
|
|
||||||
logger.info(f"正在加载测试数据: {path.name}")
|
|
||||||
|
|
||||||
if suffix in ['.yaml', '.yml']:
|
|
||||||
return DataLoader._load_yaml(path)
|
|
||||||
elif suffix == '.json':
|
|
||||||
return DataLoader._load_json(path)
|
|
||||||
elif suffix in ['.xlsx', '.xls']:
|
|
||||||
return DataLoader._load_excel(path)
|
|
||||||
else:
|
|
||||||
raise ValueError(f"不支持的文件格式: {suffix}。仅支持 yaml, json, xlsx")
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _load_yaml(path: Path) -> List[Dict]:
|
|
||||||
with open(path, 'r', encoding='utf-8') as f:
|
|
||||||
# safe_load 防止代码注入风险
|
|
||||||
data = yaml.safe_load(f)
|
|
||||||
# 归一化:如果根节点是字典,转为单元素列表;如果是列表则直接返回
|
|
||||||
return data if isinstance(data, list) else [data]
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _load_json(path: Path) -> List[Dict]:
|
|
||||||
with open(path, 'r', encoding='utf-8') as f:
|
|
||||||
data = json.load(f)
|
|
||||||
return data if isinstance(data, list) else [data]
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def _load_excel(path: Path) -> List[Dict]:
|
|
||||||
"""
|
|
||||||
Excel 解析规则:
|
|
||||||
1. 第一行默认为表头 (Keys)
|
|
||||||
2. 后续行为数据 (Values)
|
|
||||||
3. 自动过滤全空行
|
|
||||||
"""
|
|
||||||
if openpyxl is None:
|
|
||||||
raise ImportError("检测到 .xlsx 文件,但未安装 openpyxl。请执行: pip install openpyxl")
|
|
||||||
|
|
||||||
wb = openpyxl.load_workbook(path, read_only=True, data_only=True)
|
|
||||||
# 默认读取第一个 Sheet
|
|
||||||
ws = wb.active
|
|
||||||
|
|
||||||
# 获取所有行
|
|
||||||
rows = list(ws.rows)
|
|
||||||
if not rows:
|
|
||||||
return []
|
|
||||||
|
|
||||||
# 解析表头 (第一行)
|
|
||||||
headers = [cell.value for cell in rows[0] if cell.value is not None]
|
|
||||||
|
|
||||||
result = []
|
|
||||||
# 解析数据 (从第二行开始)
|
|
||||||
for row in rows[1:]:
|
|
||||||
# 提取当前行的数据
|
|
||||||
values = [cell.value for cell in row]
|
|
||||||
|
|
||||||
# 如果整行都是 None,跳过
|
|
||||||
if not any(values):
|
|
||||||
continue
|
|
||||||
|
|
||||||
# 组装字典: {header: value}
|
|
||||||
row_dict = {}
|
|
||||||
for i, header in enumerate(headers):
|
|
||||||
# 防止越界 (有些行可能数据列数少于表头)
|
|
||||||
val = values[i] if i < len(values) else None
|
|
||||||
# 转换处理:Excel 的数字可能需要转为字符串,视业务需求而定
|
|
||||||
# 这里保持原样,由后续逻辑处理类型
|
|
||||||
row_dict[header] = val
|
|
||||||
|
|
||||||
result.append(row_dict)
|
|
||||||
|
|
||||||
wb.close()
|
|
||||||
return result
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
# 调试代码
|
|
||||||
# 假设有一个 test.yaml
|
|
||||||
# print(DataLoader.load("test.yaml"))
|
|
||||||
pass
|
|
||||||
@@ -142,14 +142,22 @@ if __name__ == '__main__':
|
|||||||
# 5. 查看当前全量支持的归一化后的 Key
|
# 5. 查看当前全量支持的归一化后的 Key
|
||||||
print(f"当前支持的策略总数: {len(converter.get_all_finders())}")
|
print(f"当前支持的策略总数: {len(converter.get_all_finders())}")
|
||||||
print(f"前 5 个策略示例: {converter.get_all_finders()[:5]}")
|
print(f"前 5 个策略示例: {converter.get_all_finders()[:5]}")
|
||||||
|
|
||||||
# 6. 增加类型非法测试
|
# 6. 增加类型非法测试
|
||||||
print("\n--- 异常类型测试 ---")
|
print("\n--- 异常类型测试 ---")
|
||||||
try:
|
try:
|
||||||
by_converter(123) # 传入数字
|
by_converter(123) # 传入数字
|
||||||
except TypeError as e:
|
except ValueError as e:
|
||||||
print(f"验证类型拦截成功: {e}")
|
print(f"验证类型拦截成功: {e}")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
by_converter(None) # 传入 None
|
by_converter(None) # 传入 None
|
||||||
except TypeError as e:
|
except ValueError as e:
|
||||||
print(f"验证空值拦截成功: {e}")
|
print(f"验证空值拦截成功: {e}")
|
||||||
|
|
||||||
|
# 7. 验证不支持的策略 (验证 if target is None 分支)
|
||||||
|
print("\n--- 验证不支持的策略 ---")
|
||||||
|
try:
|
||||||
|
by_converter("unknown_strategy")
|
||||||
|
except ValueError as e:
|
||||||
|
print(f"验证不支持策略拦截成功: {e}")
|
||||||
|
|||||||
Reference in New Issue
Block a user