【Web UI 自动化】05 - KDT 模式原理与实现

本文是《企业级自动化测试实战》系列的第 11 篇，也是 Web UI 自动化篇的第 5 篇。上一篇我们用 POM 模式编写了完整的测试用例。本篇将引入 KDT（Keyword-Driven Testing，关键字驱动测试）模式——测试人员不需要写 Python 代码，只需在 YAML 文件中用关键字编排测试步骤，引擎自动解析执行。

前言

POM 模式虽然解决了元素定位复用的问题，但用例仍然是 Python 代码。团队中如果有不会写代码的测试人员，他们无法直接编写和维护用例。

KDT 模式解决的就是这个问题：把所有操作封装成"关键字"，测试用例用 YAML 格式描述，不懂代码的人也能通过组合关键字来编写测试。

两种模式的对比：

POM 用例（Python 代码）：
def test_login(login_page):
login_page.login("admin", "admin123")
assert login_page.is_login_success()

KDT 用例（YAML 文件）：
– name: "管理员登录成功"
steps:
– keyword: login
params:
username: "admin"
password: "admin123"
– keyword: assert_login_success

KDT 用例更直观，非技术人员也能读懂和编写。

今天的任务：

• 理解 KDT 模式的核心思想和架构
• 实现关键字层（base_keywords + 业务关键字）
• 实现关键字注册表（keyword_registry）
• 实现驱动引擎（test_engine）
• 编写 YAML 格式的测试用例
• 实现 Pytest 集成（更新 pytest.ini 注册 marker）
• 运行验证

一、KDT 模式架构

1.1 核心组件

┌──────────────────────────────────────────────────────┐
│ YAML 测试用例 │
│ test_cases/kdt/login/test_login.yaml │
│ 用关键字描述测试步骤，不含任何 Python 代码 │
└───────────────────────┬──────────────────────────────┘
│ 被读取
┌───────────────────────▼──────────────────────────────┐
│ 驱动引擎（Engine） │
│ engine/test_engine.py │
│ 读取 YAML → 解析步骤 → 查找关键字 → 执行 │
└───────────────────────┬──────────────────────────────┘
│ 调用
┌───────────────────────▼──────────────────────────────┐
│ 关键字注册表（Registry） │
│ keywords/keyword_registry.py │
│ 维护 "关键字名 → 函数" 的映射关系 │
└───────────────────────┬──────────────────────────────┘
│ 查找
┌───────────────────────▼──────────────────────────────┐
│ 关键字层（Keywords） │
│ keywords/base_keywords.py ← 通用关键字 │
│ keywords/login_keywords.py ← 登录业务关键字 │
│ keywords/search_keywords.py ← 搜索业务关键字 │
│ keywords/cart_keywords.py ← 购物车关键字 │
│ keywords/order_keywords.py ← 订单关键字 │
└───────────────────────┬──────────────────────────────┘
│ 使用
┌───────────────────────▼──────────────────────────────┐
│ 页面对象层（POM） │
│ pages/login_page.py / home_page.py / … │
│ KDT 复用 POM 的页面对象，不重复封装 │
└──────────────────────────────────────────────────────┘

1.2 两层关键字

关键字分为两层：

底层关键字（base_keywords）：直接操作浏览器，如 navigate、click、fill、assert_visible

# 底层关键字用法（精确控制每一步）
– keyword: navigate
params:
url: "/login"
– keyword: fill
params:
selector: "#username"
value: "admin"
– keyword: click
params:
selector: "#login-btn"

业务关键字（login_keywords 等）：封装业务操作，内部调用页面对象，如 login、search、add_to_cart

# 业务关键字用法（一行搞定一个业务操作）
– keyword: login
params:
username: "admin"
password: "admin123"
– keyword: search
params:
keyword: "iPhone"

二、关键字层实现

2.1 BaseKeywords – 底层通用关键字

创建 keywords/base_keywords.py：

"""
BaseKeywords – 底层通用关键字
封装浏览器基础操作，每个关键字对应一个原子操作

这些关键字直接操作 Playwright 的 page 对象，
不依赖任何页面对象类
"""

from common.logger import get_logger

logger = get_logger("base_keywords")

class BaseKeywords:
"""
底层通用关键字类

所有方法的第一个参数都是 page（Playwright 的 page 对象）
方法名就是关键字名
"""

def __init__(self, page):
self.page = page

# ========================================
# 导航关键字
# ========================================

def navigate(self, url):
"""
导航到指定地址

YAML 用法：
– keyword: navigate
params:
url: "/login"
"""
logger.info(f"[关键字] 导航到：{url}")
self.page.goto(url)
self.page.wait_for_load_state("domcontentloaded")

def reload(self):
"""
刷新当前页面

YAML 用法：
– keyword: reload
"""
logger.info("[关键字] 刷新页面")
self.page.reload()
self.page.wait_for_load_state("domcontentloaded")

def go_back(self):
"""
浏览器后退

YAML 用法：
– keyword: go_back
"""
logger.info("[关键字] 后退")
self.page.go_back()

# ========================================
# 输入关键字
# ========================================

def fill(self, selector, value):
"""
清空并输入文本

YAML 用法：
– keyword: fill
params:
selector: "#username"
value: "admin"
"""
logger.info(f"[关键字] 输入：{selector} → '{value}'")
self.page.fill(selector, str(value))

def type_text(self, selector, value, delay=50):
"""
逐字输入（模拟真实键盘）

YAML 用法：
– keyword: type_text
params:
selector: "#search-input"
value: "iPhone"
delay: 100
"""
logger.info(f"[关键字] 逐字输入：{selector} → '{value}'")
self.page.type(selector, str(value), delay=delay)

def clear(self, selector):
"""
清空输入框

YAML 用法：
– keyword: clear
params:
selector: "#username"
"""
self.page.fill(selector, "")

# ========================================
# 点击关键字
# ========================================

def click(self, selector):
"""
点击元素

YAML 用法：
– keyword: click
params:
selector: "#login-btn"
"""
logger.info(f"[关键字] 点击：{selector}")
self.page.click(selector)

def double_click(self, selector):
"""
双击元素

YAML 用法：
– keyword: double_click
params:
selector: ".product-card"
"""
self.page.dblclick(selector)

def hover(self, selector):
"""
鼠标悬停

YAML 用法：
– keyword: hover
params:
selector: ".dropdown-menu"
"""
self.page.hover(selector)

def check(self, selector):
"""
勾选复选框

YAML 用法：
– keyword: check
params:
selector: "#remember"
"""
self.page.check(selector)

def uncheck(self, selector):
"""
取消勾选

YAML 用法：
– keyword: uncheck
params:
selector: "#remember"
"""
self.page.uncheck(selector)

def select_option(self, selector, value):
"""
下拉选择

YAML 用法：
– keyword: select_option
params:
selector: "#category"
value: "手机"
"""
self.page.select_option(selector, value)

def press_key(self, selector, key):
"""
按键

YAML 用法：
– keyword: press_key
params:
selector: "#search-input"
key: "Enter"
"""
self.page.press(selector, key)

# ========================================
# 等待关键字
# ========================================

def wait(self, milliseconds):
"""
等待指定时间

YAML 用法：
– keyword: wait
params:
milliseconds: 1000
"""
logger.info(f"[关键字] 等待 {milliseconds}ms")
self.page.wait_for_timeout(int(milliseconds))

def wait_for_visible(self, selector, timeout=10000):
"""
等待元素可见

YAML 用法：
– keyword: wait_for_visible
params:
selector: "#result"
timeout: 5000
"""
logger.info(f"[关键字] 等待元素可见：{selector}")
self.page.locator(selector).wait_for(state="visible", timeout=int(timeout))

def wait_for_hidden(self, selector, timeout=10000):
"""
等待元素消失

YAML 用法：
– keyword: wait_for_hidden
params:
selector: ".loading"
"""
self.page.locator(selector).wait_for(state="hidden", timeout=int(timeout))

def wait_for_url(self, pattern, timeout=10000):
"""
等待 URL 包含指定内容

YAML 用法：
– keyword: wait_for_url
params:
pattern: "/cart"
"""
self.page.wait_for_url(f"*{pattern}*", timeout=int(timeout))

def wait_for_load(self, state="domcontentloaded"):
"""
等待页面加载

YAML 用法：
– keyword: wait_for_load
params:
state: "networkidle"
"""
self.page.wait_for_load_state(state)

# ========================================
# 滚动关键字
# ========================================

def scroll_to_bottom(self):
"""
滚动到页面底部

YAML 用法：
– keyword: scroll_to_bottom
"""
self.page.evaluate("window.scrollTo(0, document.body.scrollHeight)")

def scroll_to_top(self):
"""
滚动到页面顶部

YAML 用法：
– keyword: scroll_to_top
"""
self.page.evaluate("window.scrollTo(0, 0)")

def scroll_to_element(self, selector):
"""
滚动到指定元素

YAML 用法：
– keyword: scroll_to_element
params:
selector: "#footer"
"""
self.page.locator(selector).scroll_into_view_if_needed()

# ========================================
# 断言关键字
# ========================================

def assert_visible(self, selector, timeout=10000):
"""
断言元素可见

YAML 用法：
– keyword: assert_visible
params:
selector: "#product-list"
"""
logger.info(f"[断言] 元素可见：{selector}")
self.page.locator(selector).wait_for(state="visible", timeout=int(timeout))

def assert_hidden(self, selector, timeout=10000):
"""
断言元素不可见

YAML 用法：
– keyword: assert_hidden
params:
selector: "#error-msg"
"""
logger.info(f"[断言] 元素不可见：{selector}")
self.page.locator(selector).wait_for(state="hidden", timeout=int(timeout))

def assert_text(self, selector, expected_text):
"""
断言元素包含文本

YAML 用法：
– keyword: assert_text
params:
selector: ".product-name"
expected_text: "iPhone"
"""
logger.info(f"[断言] 文本包含：{selector} → '{expected_text}'")
actual = self.page.text_content(selector) or ""
assert expected_text in actual, \\
f"断言失败：'{expected_text}' 不在 '{actual.strip()}' 中"

def assert_exact_text(self, selector, expected_text):
"""
断言元素文本完全匹配

YAML 用法：
– keyword: assert_exact_text
params:
selector: "#product-title"
expected_text: "iPhone 15 Pro"
"""
actual = (self.page.text_content(selector) or "").strip()
assert actual == expected_text, \\
f"断言失败：期望 '{expected_text}'，实际 '{actual}'"

def assert_url_contains(self, text):
"""
断言 URL 包含文本

YAML 用法：
– keyword: assert_url_contains
params:
text: "/login"
"""
current_url = self.page.url
logger.info(f"[断言] URL 包含：'{text}'（当前：{current_url}）")
assert text in current_url, \\
f"断言失败：URL '{current_url}' 不包含 '{text}'"

def assert_url_not_contains(self, text):
"""
断言 URL 不包含文本

YAML 用法：
– keyword: assert_url_not_contains
params:
text: "/login"
"""
current_url = self.page.url
assert text not in current_url, \\
f"断言失败：URL '{current_url}' 不应包含 '{text}'"

def assert_title(self, expected_title):
"""
断言页面标题包含文本

YAML 用法：
– keyword: assert_title
params:
expected_title: "MallLite"
"""
actual_title = self.page.title()
assert expected_title in actual_title, \\
f"断言失败：标题 '{actual_title}' 不包含 '{expected_title}'"

def assert_element_count(self, selector, expected_count):
"""
断言元素数量

YAML 用法：
– keyword: assert_element_count
params:
selector: ".product-card"
expected_count: 8
"""
actual = self.page.locator(selector).count()
assert actual == int(expected_count), \\
f"断言失败：{selector} 期望 {expected_count} 个，实际 {actual} 个"

def assert_element_count_min(self, selector, min_count):
"""
断言元素数量至少为 N

YAML 用法：
– keyword: assert_element_count_min
params:
selector: ".product-card"
min_count: 1
"""
actual = self.page.locator(selector).count()
assert actual >= int(min_count), \\
f"断言失败：{selector} 期望至少 {min_count} 个，实际 {actual} 个"

def assert_value(self, selector, expected_value):
"""
断言输入框的值

YAML 用法：
– keyword: assert_value
params:
selector: "#username"
expected_value: "admin"
"""
actual = self.page.input_value(selector)
assert actual == str(expected_value), \\
f"断言失败：期望 '{expected_value}'，实际 '{actual}'"

# ========================================
# 截图关键字
# ========================================

def screenshot(self, name="screenshot"):
"""
截图

YAML 用法：
– keyword: screenshot
params:
name: "登录成功后"
"""
from common.screenshot import Screenshot
Screenshot.take(self.page, name)

2.2 LoginKeywords – 登录业务关键字

创建 keywords/login_keywords.py：

"""
LoginKeywords – 登录业务关键字
封装登录相关的业务操作，内部使用 LoginPage 页面对象
"""

from common.logger import get_logger
from pages.login_page import LoginPage

logger = get_logger("login_keywords")

class LoginKeywords:
"""
登录业务关键字

每个方法对应一个业务级关键字
内部使用 LoginPage 页面对象完成操作
"""

def __init__(self, page):
self.page = page
self.login_page = LoginPage(page)

def open_login_page(self):
"""
打开登录页

YAML 用法：
– keyword: open_login_page
"""
logger.info("[业务关键字] 打开登录页")
self.login_page.open()

def login(self, username, password, remember=False):
"""
执行登录

YAML 用法：
– keyword: login
params:
username: "admin"
password: "admin123"
"""
logger.info(f"[业务关键字] 登录：{username}")
self.login_page.login(username, password, remember)

def input_username(self, username):
"""
输入用户名

YAML 用法：
– keyword: input_username
params:
username: "admin"
"""
self.login_page.input_username(username)

def input_password(self, password):
"""
输入密码

YAML 用法：
– keyword: input_password
params:
password: "admin123"
"""
self.login_page.input_password(password)

def click_login_button(self):
"""
点击登录按钮

YAML 用法：
– keyword: click_login_button
"""
self.login_page.click_login()

def assert_login_success(self):
"""
断言登录成功

YAML 用法：
– keyword: assert_login_success
"""
logger.info("[业务关键字] 断言登录成功")
assert self.login_page.is_login_success(), \\
f"登录失败，当前 URL：{self.login_page.get_current_url()}"

def assert_login_fail(self):
"""
断言登录失败

YAML 用法：
– keyword: assert_login_fail
"""
logger.info("[业务关键字] 断言登录失败")
assert not self.login_page.is_login_success(), \\
"不应该登录成功"

def assert_error_message(self, expected_msg):
"""
断言错误提示内容

YAML 用法：
– keyword: assert_error_message
params:
expected_msg: "密码错误"
"""
actual_msg = self.login_page.get_error_message()
assert expected_msg in actual_msg, \\
f"期望错误提示包含 '{expected_msg}'，实际为 '{actual_msg}'"

2.3 SearchKeywords – 搜索业务关键字

创建 keywords/search_keywords.py：

"""
SearchKeywords – 搜索业务关键字
"""

from common.logger import get_logger
from pages.home_page import HomePage

logger = get_logger("search_keywords")

class SearchKeywords:
"""搜索业务关键字"""

def __init__(self, page):
self.page = page
self.home_page = HomePage(page)

def open_home_page(self):
"""
打开首页

YAML 用法：
– keyword: open_home_page
"""
logger.info("[业务关键字] 打开首页")
self.home_page.open()

def search(self, keyword):
"""
搜索商品

YAML 用法：
– keyword: search
params:
keyword: "iPhone"
"""
logger.info(f"[业务关键字] 搜索：{keyword}")
self.home_page.search(keyword)

def click_category(self, category):
"""
点击分类筛选

YAML 用法：
– keyword: click_category
params:
category: "手机"
"""
logger.info(f"[业务关键字] 分类筛选：{category}")
self.home_page.click_category(category)

def click_product(self, index=0):
"""
点击第 N 个商品

YAML 用法：
– keyword: click_product
params:
index: 0
"""
logger.info(f"[业务关键字] 点击第 {int(index) + 1} 个商品")
self.home_page.click_product(int(index))

def assert_product_count(self, expected_count):
"""
断言商品数量

YAML 用法：
– keyword: assert_product_count
params:
expected_count: 8
"""
actual = self.home_page.get_product_count()
assert actual == int(expected_count), \\
f"期望 {expected_count} 个商品，实际 {actual} 个"

def assert_product_count_min(self, min_count):
"""
断言商品数量至少为 N

YAML 用法：
– keyword: assert_product_count_min
params:
min_count: 1
"""
actual = self.home_page.get_product_count()
assert actual >= int(min_count), \\
f"期望至少 {min_count} 个商品，实际 {actual} 个"

def assert_products_not_empty(self):
"""
断言商品列表不为空

YAML 用法：
– keyword: assert_products_not_empty
"""
assert self.home_page.has_products(), "商品列表不应该为空"

def assert_products_empty(self):
"""
断言商品列表为空

YAML 用法：
– keyword: assert_products_empty
"""
assert self.home_page.is_empty(), "商品列表应该为空"

2.4 CartKeywords – 购物车业务关键字

创建 keywords/cart_keywords.py：

"""
CartKeywords – 购物车业务关键字
"""

from common.logger import get_logger
from pages.home_page import HomePage
from pages.product_page import ProductPage
from pages.cart_page import CartPage

logger = get_logger("cart_keywords")

class CartKeywords:
"""购物车业务关键字"""

def __init__(self, page):
self.page = page
self.home_page = HomePage(page)
self.product_page = ProductPage(page)
self.cart_page = CartPage(page)

def open_product_detail(self, product_id):
"""
打开商品详情页

YAML 用法：
– keyword: open_product_detail
params:
product_id: 1
"""
logger.info(f"[业务关键字] 打开商品详情：ID={product_id}")
self.product_page.open(int(product_id))

def add_to_cart(self):
"""
在商品详情页加入购物车

YAML 用法：
– keyword: add_to_cart
"""
logger.info("[业务关键字] 加入购物车")
self.product_page.add_to_cart()

def add_product_from_home(self, index=0):
"""
从首页直接添加商品到购物车

YAML 用法：
– keyword: add_product_from_home
params:
index: 0
"""
logger.info(f"[业务关键字] 从首页添加第 {int(index) + 1} 个商品到购物车")
self.home_page.add_product_to_cart(int(index))

def open_cart(self):
"""
打开购物车页

YAML 用法：
– keyword: open_cart
"""
logger.info("[业务关键字] 打开购物车")
self.cart_page.open()

def clear_cart(self): # ← 新增
"""
清空购物车

YAML 用法：
– keyword: clear_cart
"""
logger.info("[业务关键字] 清空购物车")
self.cart_page.clear_cart()

def delete_cart_item(self, index=0):
"""
删除购物车中第 N 个商品

YAML 用法：
– keyword: delete_cart_item
params:
index: 0
"""
logger.info(f"[业务关键字] 删除购物车第 {int(index) + 1} 个商品")
self.cart_page.delete_item(int(index))

def checkout(self):
"""
去结算

YAML 用法：
– keyword: checkout
"""
logger.info("[业务关键字] 去结算")
self.cart_page.checkout()

def assert_cart_not_empty(self):
"""
断言购物车不为空

YAML 用法：
– keyword: assert_cart_not_empty
"""
assert not self.cart_page.is_empty(), "购物车不应该为空"

def assert_cart_empty(self):
"""
断言购物车为空

YAML 用法：
– keyword: assert_cart_empty
"""
assert self.cart_page.is_empty(), "购物车应该为空"

def assert_cart_contains(self, product_name):
"""
断言购物车包含某商品

YAML 用法：
– keyword: assert_cart_contains
params:
product_name: "iPhone 15 Pro"
"""
assert self.cart_page.has_item(product_name), \\
f"购物车中应该包含 '{product_name}'"

def assert_cart_item_count(self, expected_count):
"""
断言购物车商品种类数

YAML 用法：
– keyword: assert_cart_item_count
params:
expected_count: 2
"""
actual = self.cart_page.get_item_count()
assert actual >= int(expected_count), \\
f"购物车期望至少 {expected_count} 种商品，实际 {actual} 种"

2.5 OrderKeywords – 订单业务关键字

创建 keywords/order_keywords.py：

"""
OrderKeywords – 订单业务关键字
"""

from common.logger import get_logger
from pages.order_page import OrderPage

logger = get_logger("order_keywords")

class OrderKeywords:
"""订单业务关键字"""

def __init__(self, page):
self.page = page
self.order_page = OrderPage(page)

def open_orders(self):
"""
打开订单页

YAML 用法：
– keyword: open_orders
"""
logger.info("[业务关键字] 打开订单页")
self.order_page.open()

def assert_orders_not_empty(self):
"""
断言订单列表不为空

YAML 用法：
– keyword: assert_orders_not_empty
"""
assert not self.order_page.is_empty(), "订单列表不应该为空"

def assert_order_count_min(self, min_count):
"""
断言订单数量至少为 N

YAML 用法：
– keyword: assert_order_count_min
params:
min_count: 1
"""
actual = self.order_page.get_order_count()
assert actual >= int(min_count), \\
f"期望至少 {min_count} 个订单，实际 {actual} 个"

def assert_order_has_info(self, index=0):
"""
断言订单信息完整

YAML 用法：
– keyword: assert_order_has_info
params:
index: 0
"""
order_no = self.order_page.get_order_no(int(index))
assert order_no and "ORD" in order_no, f"订单号不正确：{order_no}"

status = self.order_page.get_order_status(int(index))
assert status, "订单状态不能为空"

total = self.order_page.get_order_total(int(index))
assert total, "订单金额不能为空"

三、关键字注册表

创建 keywords/keyword_registry.py：

"""
关键字注册表
维护 "关键字名 → 函数" 的映射关系
引擎通过注册表查找并执行关键字

注册表结构：
{
"navigate": <BaseKeywords.navigate>,
"login": <LoginKeywords.login>,
"search": <SearchKeywords.search>,
…
}
"""

from common.logger import get_logger
from keywords.base_keywords import BaseKeywords
from keywords.login_keywords import LoginKeywords
from keywords.search_keywords import SearchKeywords
from keywords.cart_keywords import CartKeywords
from keywords.order_keywords import OrderKeywords

logger = get_logger("keyword_registry")

class KeywordRegistry:
"""
关键字注册表

负责：
1. 注册所有关键字类
2. 根据关键字名查找对应的函数
3. 管理关键字实例的生命周期

用法：
registry = KeywordRegistry(page)
func = registry.get_keyword("login")
func("admin", "admin123")
"""

def __init__(self, page):
"""
初始化注册表

参数：
page: Playwright 的 page 对象
"""
self.page = page
self._keywords = {}
self._keyword_classes = {}

# 注册所有关键字类
self._register_all()

def _register_all(self):
"""注册所有关键字类"""

# 关键字类列表
keyword_classes = [
("base", BaseKeywords),
("login", LoginKeywords),
("search", SearchKeywords),
("cart", CartKeywords),
("order", OrderKeywords),
]

for prefix, cls in keyword_classes:
self._register_class(prefix, cls)

def _register_class(self, prefix, cls):
"""
注册一个关键字类

参数：
prefix: 类前缀（用于日志，不参与关键字查找）
cls: 关键字类
"""
instance = cls(self.page)
self._keyword_classes[prefix] = instance

# 遍历类的所有公开方法，注册为关键字
for attr_name in dir(instance):
# 跳过私有方法和特殊方法
if attr_name.startswith("_"):
continue

attr = getattr(instance, attr_name)
if callable(attr):
keyword_name = attr_name
self._keywords[keyword_name] = attr
logger.debug(f"注册关键字：{keyword_name} → {cls.__name__}.{attr_name}")

logger.info(f"注册关键字类：{prefix}（{cls.__name__}，{len([m for m in dir(instance) if not m.startswith('_') and callable(getattr(instance, m))])} 个关键字）")

def get_keyword(self, name):
"""
根据关键字名获取对应的函数

参数：
name: 关键字名称

返回：
可调用的函数

异常：
KeyError: 关键字不存在时抛出
"""
if name not in self._keywords:
available = sorted(self._keywords.keys())
raise KeyError(
f"关键字 '{name}' 不存在。"
f"可用关键字（{len(available)} 个）：{available}"
)
return self._keywords[name]

def has_keyword(self, name):
"""判断关键字是否存在"""
return name in self._keywords

def list_keywords(self):
"""列出所有已注册的关键字"""
return sorted(self._keywords.keys())

def get_keyword_count(self):
"""获取已注册关键字数量"""
return len(self._keywords)

3.1 注册表工作原理

KeywordRegistry(page)
│
├── 注册 BaseKeywords(page) → navigate, click, fill, assert_visible, …
├── 注册 LoginKeywords(page) → login, open_login_page, assert_login_success, …
├── 注册 SearchKeywords(page) → search, click_category, assert_product_count, …
├── 注册 CartKeywords(page) → add_to_cart, open_cart, checkout, …
└── 注册 OrderKeywords(page) → open_orders, assert_orders_not_empty, …

→ 全部汇总到 _keywords 字典：
{
"navigate": BaseKeywords.navigate,
"click": BaseKeywords.click,
"fill": BaseKeywords.fill,
"login": LoginKeywords.login,
"search": SearchKeywords.search,
"add_to_cart": CartKeywords.add_to_cart,
…
}

四、驱动引擎

创建 engine/test_engine.py：

"""
KDT 测试引擎
读取 YAML 测试用例 → 解析步骤 → 通过注册表查找关键字 → 执行 → 收集结果

引擎是 KDT 模式的核心，它把 YAML 中的步骤翻译成实际的浏览器操作
"""

import yaml
import allure
import traceback
from pathlib import Path
from dataclasses import dataclass, field
from typing import Optional

from common.logger import get_logger
from common.data_reader import DataReader
from keywords.keyword_registry import KeywordRegistry

logger = get_logger("test_engine")

@dataclass
class StepResult:
"""单步执行结果"""
step_index: int
keyword: str
params: dict
status: str # "passed" / "failed" / "skipped"
message: str = ""
duration: float = 0.0

@dataclass
class CaseResult:
"""用例执行结果"""
case_name: str
status: str # "passed" / "failed" / "error"
steps: list = field(default_factory=list)
error_message: str = ""
duration: float = 0.0

class TestEngine:
"""
KDT 测试引擎

用法：
engine = TestEngine(page)
results = engine.run_yaml("test_cases/kdt/login/test_login.yaml")

for result in results:
print(f"{result.case_name}: {result.status}")
"""

def __init__(self, page):
"""
初始化引擎

参数：
page: Playwright 的 page 对象
"""
self.page = page
self.registry = KeywordRegistry(page)
logger.info(f"测试引擎初始化完成，已注册 {self.registry.get_keyword_count()} 个关键字")

def run_yaml(self, yaml_path):
"""
执行 YAML 文件中的所有用例

参数：
yaml_path: YAML 文件路径（相对于项目根目录或绝对路径）

返回：
CaseResult 列表
"""
# 加载 YAML
cases = self._load_yaml(yaml_path)
logger.info(f"加载 YAML 用例：{yaml_path}（{len(cases)} 条）")

results = []
for case in cases:
result = self.run_case(case)
results.append(result)

# 输出汇总
passed = sum(1 for r in results if r.status == "passed")
failed = sum(1 for r in results if r.status in ("failed", "error"))
logger.info(f"执行完成：共 {len(results)} 条，通过 {passed} 条，失败 {failed} 条")

return results

def run_case(self, case_data):
"""
执行单条用例

参数：
case_data: 用例字典（从 YAML 解析）

返回：
CaseResult
"""
case_name = case_data.get("name", "未命名用例")
description = case_data.get("description", "")
steps = case_data.get("steps", [])

logger.info(f"{'=' * 50}")
logger.info(f"执行用例：{case_name}")
if description:
logger.info(f"描述：{description}")

result = CaseResult(case_name=case_name, status="passed")

import time
start_time = time.time()

for i, step_data in enumerate(steps):
step_result = self._execute_step(i + 1, step_data)

# 附加到 Allure
self._attach_step_to_allure(step_result)

result.steps.append(step_result)

if step_result.status == "failed":
result.status = "failed"
result.error_message = step_result.message
logger.error(f"用例失败：{step_result.message}")
# 失败后截图
self._take_failure_screenshot(case_name)
break

if step_result.status == "skipped":
logger.warning(f"步骤跳过：{step_data.get('keyword')}")
break

result.duration = time.time() – start_time

status_icon = "✅" if result.status == "passed" else "❌"
logger.info(f"{status_icon} 用例结果：{case_name} → {result.status}（{result.duration:.2f}s）")

return result

def _execute_step(self, step_index, step_data):
"""
执行单个步骤

参数：
step_index: 步骤序号
step_data: 步骤字典，格式：
{
"keyword": "login",
"params": {"username": "admin", "password": "admin123"}
}
或（无参数时）：
{
"keyword": "open_cart"
}

返回：
StepResult
"""
keyword_name = step_data.get("keyword")
params = step_data.get("params", {})
step_desc = step_data.get("desc", "") # 可选的步骤描述

import time
start_time = time.time()

# 查找关键字
try:
keyword_func = self.registry.get_keyword(keyword_name)
except KeyError as e:
duration = time.time() – start_time
return StepResult(
step_index=step_index,
keyword=keyword_name,
params=params,
status="failed",
message=str(e),
duration=duration,
)

# 执行关键字
try:
display = f"{keyword_name}({', '.join(f'{k}={v}' for k, v in params.items())})"
desc_display = f" [{step_desc}]" if step_desc else ""
logger.info(f" 步骤 {step_index}{desc_display}：{display}")

keyword_func(**params)

duration = time.time() – start_time
return StepResult(
step_index=step_index,
keyword=keyword_name,
params=params,
status="passed",
duration=duration,
)

except AssertionError as e:
duration = time.time() – start_time
return StepResult(
step_index=step_index,
keyword=keyword_name,
params=params,
status="failed",
message=f"断言失败：{e}",
duration=duration,
)

except Exception as e:
duration = time.time() – start_time
return StepResult(
step_index=step_index,
keyword=keyword_name,
params=params,
status="failed",
message=f"执行异常：{type(e).__name__}: {e}",
duration=duration,
)

def _load_yaml(self, yaml_path):
"""加载 YAML 文件"""
filepath = Path(yaml_path)
if not filepath.is_absolute():
filepath = Path(__file__).parent.parent / yaml_path

if not filepath.exists():
# 尝试在 test_data 目录查找
filepath = Path(__file__).parent.parent / "test_data" / yaml_path

if not filepath.exists():
raise FileNotFoundError(f"YAML 文件不存在：{yaml_path}")

with open(filepath, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)

if data is None:
return []

if isinstance(data, dict):
return data.get("cases", data.get("test_cases", [data]))

return data

def _attach_step_to_allure(self, step_result):
"""将步骤结果附加到 Allure 报告"""
try:
with allure.step(f"步骤 {step_result.step_index}: {step_result.keyword}"):
if step_result.params:
import json
allure.attach(
json.dumps(step_result.params, ensure_ascii=False, indent=2),
name="参数",
attachment_type=allure.attachment_type.JSON,
)
if step_result.status == "failed":
allure.attach(
step_result.message,
name="失败原因",
attachment_type=allure.attachment_type.TEXT,
)
except Exception:
pass

def _take_failure_screenshot(self, case_name):
"""失败时截图"""
try:
from common.screenshot import Screenshot
Screenshot.take(self.page, f"KDT失败_{case_name}", attach_to_allure=True)
except Exception as e:
logger.error(f"截图失败：{e}")

五、编写 YAML 测试用例

5.1 登录用例 – YAML

创建目录和文件：

mkdir -p test_cases/kdt/login test_cases/kdt/search test_cases/kdt/cart

创建 test_cases/kdt/login/test_login.yaml：

# KDT 登录功能测试用例
# 使用关键字描述测试步骤，无需编写 Python 代码

cases:
# ========================================
# 正向用例
# ========================================

– name: "管理员登录成功"
description: "使用管理员账号密码正常登录，验证跳转到首页"
markers: [smoke, p0, login]
steps:
– keyword: login
params:
username: "admin"
password: "admin123"
– keyword: assert_login_success

– name: "普通用户登录成功"
description: "使用普通用户账号密码正常登录"
markers: [smoke, p0, login]
steps:
– keyword: login
params:
username: "testuser"
password: "test123"
– keyword: assert_login_success

– name: "VIP 用户登录成功"
description: "使用 VIP 用户账号密码正常登录"
markers: [smoke, p0, login]
steps:
– keyword: login
params:
username: "vipuser"
password: "vip123"
– keyword: assert_login_success

# ========================================
# 反向用例
# ========================================

– name: "密码错误登录失败"
description: "输入正确用户名和错误密码，验证错误提示"
markers: [p1, login]
steps:
– keyword: login
params:
username: "admin"
password: "wrong_password"
– keyword: assert_login_fail
– keyword: assert_error_message
params:
expected_msg: "密码错误"

– name: "用户不存在登录失败"
description: "输入不存在的用户名，验证错误提示"
markers: [p1, login]
steps:
– keyword: login
params:
username: "nonexistent_user_xyz"
password: "123456"
– keyword: assert_login_fail
– keyword: assert_error_message
params:
expected_msg: "用户不存在"

– name: "用户名为空登录失败"
description: "用户名不输入任何内容，验证错误提示"
markers: [p1, login]
steps:
– keyword: login
params:
username: ""
password: "admin123"
– keyword: assert_login_fail
– keyword: assert_error_message
params:
expected_msg: "请输入用户名"

– name: "密码为空登录失败"
description: "密码不输入任何内容，验证错误提示"
markers: [p1, login]
steps:
– keyword: login
params:
username: "admin"
password: ""
– keyword: assert_login_fail
– keyword: assert_error_message
params:
expected_msg: "请输入密码"

# ========================================
# 底层关键字用例（混合使用）
# ========================================

– name: "使用底层关键字登录"
description: "演示使用底层关键字（fill、click）逐步登录"
markers: [regression, p2, login]
steps:
– keyword: navigate
params:
url: "/login"
desc: "打开登录页"
– keyword: fill
params:
selector: "#username"
value: "admin"
desc: "输入用户名"
– keyword: fill
params:
selector: "#password"
value: "admin123"
desc: "输入密码"
– keyword: click
params:
selector: "#login-btn"
desc: "点击登录按钮"
– keyword: assert_url_not_contains
params:
text: "/login"
desc: "验证已离开登录页"

5.2 搜索用例 – YAML

创建 test_cases/kdt/search/test_search.yaml：

# KDT 搜索功能测试用例

cases:
# ========================================
# 关键词搜索
# ========================================

– name: "搜索 iPhone 有结果"
description: "搜索存在的关键词，验证有结果返回"
markers: [smoke, p0, search]
steps:
– keyword: open_home_page
– keyword: search
params:
keyword: "iPhone"
– keyword: assert_products_not_empty

– name: "搜索 Pro 有多个结果"
description: "搜索通用关键词，验证有多个结果"
markers: [p1, search]
steps:
– keyword: open_home_page
– keyword: search
params:
keyword: "Pro"
– keyword: assert_product_count_min
params:
min_count: 2

– name: "搜索华为有结果"
markers: [p1, search]
steps:
– keyword: open_home_page
– keyword: search
params:
keyword: "华为"
– keyword: assert_products_not_empty

– name: "搜索 AirPods 有结果"
markers: [p1, search]
steps:
– keyword: open_home_page
– keyword: search
params:
keyword: "AirPods"
– keyword: assert_products_not_empty

– name: "搜索不存在的商品"
description: "搜索不存在的关键词，验证无结果"
markers: [p1, search]
steps:
– keyword: open_home_page
– keyword: search
params:
keyword: "xyz_not_exist_12345"
– keyword: assert_products_empty

# ========================================
# 分类筛选
# ========================================

– name: "筛选手机分类"
markers: [p1, search]
steps:
– keyword: open_home_page
– keyword: click_category
params:
category: "手机"
– keyword: assert_product_count_min
params:
min_count: 2

– name: "筛选笔记本分类"
markers: [p1, search]
steps:
– keyword: open_home_page
– keyword: click_category
params:
category: "笔记本"
– keyword: assert_product_count_min
params:
min_count: 1

– name: "筛选平板分类"
markers: [p1, search]
steps:
– keyword: open_home_page
– keyword: click_category
params:
category: "平板"
– keyword: assert_product_count_min
params:
min_count: 1

– name: "筛选配件分类"
markers: [p1, search]
steps:
– keyword: open_home_page
– keyword: click_category
params:
category: "配件"
– keyword: assert_product_count_min
params:
min_count: 2

# ========================================
# 边界用例
# ========================================

– name: "空搜索显示全部商品"
description: "搜索框留空点击搜索，显示所有商品"
markers: [regression, p2, search]
steps:
– keyword: open_home_page
– keyword: search
params:
keyword: ""
– keyword: assert_products_not_empty

5.3 购物车用例 – YAML

创建 test_cases/kdt/cart/test_cart.yaml：

# KDT 购物车功能测试用例

cases:
# ========================================
# 添加购物车
# ========================================

– name: "从首页添加商品到购物车"
description: "在首页点击加入购物车按钮，验证购物车有该商品"
markers: [smoke, p0, cart]
steps:
– keyword: login
params:
username: "admin"
password: "admin123"
desc: "登录"
– keyword: open_cart
desc: "先打开购物车"
– keyword: clear_cart
desc: "清空购物车，确保干净环境"
– keyword: open_home_page
desc: "打开首页"
– keyword: add_product_from_home
params:
index: 0
desc: "添加第一个商品到购物车"
– keyword: open_cart
desc: "打开购物车"
– keyword: assert_cart_not_empty
desc: "验证购物车不为空"

– name: "从详情页添加商品到购物车"
description: "进入商品详情页后加入购物车"
markers: [smoke, p0, cart]
steps:
– keyword: login
params:
username: "admin"
password: "admin123"
– keyword: open_cart
desc: "先打开购物车"
– keyword: clear_cart
desc: "清空购物车，确保干净环境"
– keyword: open_product_detail
params:
product_id: 1
desc: "打开商品详情"
– keyword: add_to_cart
desc: "加入购物车"
– keyword: open_cart
– keyword: assert_cart_not_empty

# ========================================
# 删除购物车
# ========================================

– name: "删除购物车商品"
description: "添加商品后删除，验证购物车为空"
markers: [smoke, p1, cart]
steps:
– keyword: login
params:
username: "admin"
password: "admin123"
– keyword: open_cart
– keyword: clear_cart
desc: "先清空购物车，确保干净环境"
– keyword: open_home_page
– keyword: add_product_from_home
params:
index: 0
– keyword: open_cart
– keyword: assert_cart_not_empty
desc: "删除前验证有商品"
– keyword: delete_cart_item
params:
index: 0
desc: "删除第一个商品"
– keyword: assert_cart_empty
desc: "删除后验证为空"

# ========================================
# 空购物车
# ========================================

– name: "未登录访问购物车"
description: "不登录直接访问购物车，验证页面正常加载"
markers: [regression, p2, cart]
steps:
– keyword: open_cart
– keyword: assert_url_contains
params:
text: "/cart"
desc: "验证购物车页面能正常加载"

# ========================================
# 底层关键字用例
# ========================================

– name: "使用底层关键字验证购物车页面"
description: "演示混合使用底层关键字和业务关键字"
markers: [regression, p2, cart]
steps:
– keyword: login
params:
username: "admin"
password: "admin123"
– keyword: open_cart
– keyword: assert_visible
params:
selector: ".container"
desc: "验证购物车页面正常加载"

六、Pytest 集成 – KDT 测试收集器

KDT 的 YAML 用例需要通过 Pytest 来执行。我们需要一个 Pytest 测试文件来收集和执行 YAML 用例。

创建 test_cases/kdt/test_kdt_runner.py：

"""
KDT 测试运行器
收集 YAML 用例并交给引擎执行，集成到 Pytest 框架中

这个文件是 KDT 用例和 Pytest 之间的桥梁
"""

import pytest
import allure
import yaml
from pathlib import Path

from engine.test_engine import TestEngine
from common.logger import get_logger

logger = get_logger("kdt_runner")

# KDT 用例目录
KDT_DIR = Path(__file__).parent

def collect_yaml_cases(yaml_dir):
"""
从目录中收集所有 YAML 用例

参数：
yaml_dir: YAML 文件目录

返回：
[(yaml_path, case_data), …] 列表
"""
cases = []
yaml_dir = Path(yaml_dir)

for yaml_file in sorted(yaml_dir.rglob("*.yaml")):
# 跳过 test_data 目录下的文件
if "test_data" in str(yaml_file):
continue

try:
with open(yaml_file, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)

if data is None:
continue

case_list = data.get("cases", data.get("test_cases", []))
if isinstance(data, list):
case_list = data

for case_data in case_list:
cases.append((str(yaml_file), case_data))
except Exception as e:
logger.error(f"解析 YAML 失败：{yaml_file}，{e}")

return cases

def yaml_case_id(case_data, yaml_path):
"""生成用例 ID（用于 Pytest 显示）"""
case_name = case_data.get("name", "unnamed")
# 用文件夹名作为前缀
folder = Path(yaml_path).parent.name
return f"{folder}/{case_name}"

# 收集所有 YAML 用例
_all_yaml_cases = collect_yaml_cases(KDT_DIR)

class TestKDTRunner:
"""
KDT 测试运行器

通过 Pytest 的 parametrize 机制，把 YAML 用例逐条交给引擎执行
"""

@pytest.mark.parametrize(
"yaml_path, case_data",
_all_yaml_cases,
ids=[yaml_case_id(c, p) for p, c in _all_yaml_cases],
)
def test_kdt_case(self, page, yaml_path, case_data):
"""
执行单条 KDT 用例

参数：
page: Playwright page 对象（来自 conftest）
yaml_path: YAML 文件路径
case_data: 用例数据字典
"""
case_name = case_data.get("name", "未命名")
description = case_data.get("description", "")
markers = case_data.get("markers", [])

# Allure 标记
allure.dynamic.title(f"[KDT] {case_name}")
if description:
allure.dynamic.description(description)

# 根据 markers 设置 Allure 属性
for marker in markers:
if marker.startswith("p0"):
allure.dynamic.severity("blocker")
elif marker.startswith("p1"):
allure.dynamic.severity("critical")
elif marker.startswith("p2"):
allure.dynamic.severity("normal")

if marker in ("login", "search", "product", "cart", "order", "e2e"):
allure.dynamic.feature(marker)
if marker in ("smoke", "regression"):
allure.dynamic.tag(marker)

# 执行用例
engine = TestEngine(page)
result = engine.run_case(case_data)

# 断言
if result.status == "failed":
pytest.fail(f"KDT 用例失败：{case_name}\\n原因：{result.error_message}")

七、更新 pytest.ini

更新 pytest.ini，添加 marker 注册和运行配置，让 Pytest 同时收集 POM 和 KDT 用例：

[pytest]
testpaths = test_cases

addopts =
-v
–tb=short
–strict-markers
–alluredir=reports/allure-results

markers =
smoke: 冒烟测试
regression: 回归测试
login: 登录模块
search: 搜索模块
product: 商品模块
cart: 购物车模块
order: 订单模块
e2e: 端到端测试
p0: 最高优先级
p1: 高优先级
p2: 中优先级
kdt: KDT 模式用例

其中 –strict-markers 确保所有用到的 marker 必须预先注册，kdt: KDT 模式用例 为本篇新增的标记。

八、运行验证

8.1 运行全部 KDT 用例

# 只运行 KDT 用例
pytest test_cases/kdt/ -v

输出：

============================= test session starts ==============================
collected 23 items

test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[login/管理员登录成功] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[login/普通用户登录成功] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[login/VIP用户登录成功] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[login/密码错误登录失败] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[login/用户不存在登录失败] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[login/用户名为空登录失败] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[login/密码为空登录失败] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[login/使用底层关键字登录] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[search/搜索iPhone有结果] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[search/搜索Pro有多个结果] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[search/搜索华为有结果] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[search/搜索AirPods有结果] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[search/搜索不存在的商品] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[search/筛选手机分类] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[search/筛选笔记本分类] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[search/筛选平板分类] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[search/筛选配件分类] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[search/空搜索显示全部商品] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[cart/从首页添加商品到购物车] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[cart/从详情页添加商品到购物车] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[cart/删除购物车商品] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[cart/未登录访问购物车] PASSED
test_cases/kdt/test_kdt_runner.py::TestKDTRunner::test_kdt_case[cart/使用底层关键字验证购物车页面] PASSED

============================= 23 passed in 85.32s ==============================

8.2 POM 和 KDT 一起运行

# 运行所有用例（POM + KDT）
pytest -v

# 冒烟测试（包含 POM 和 KDT 的 smoke 用例）
pytest -v -m smoke

# 只运行登录相关的所有用例（POM + KDT）
pytest -v -m login

8.3 查看关键字列表

# 临时脚本查看所有已注册的关键字
from playwright.sync_api import sync_playwright
from keywords.keyword_registry import KeywordRegistry

with sync_playwright() as p:
browser = p.chromium.launch()
page = browser.new_page()
registry = KeywordRegistry(page)

keywords = registry.list_keywords()
print(f"已注册 {len(keywords)} 个关键字：")
for kw in keywords:
print(f" – {kw}")

browser.close()

输出类似：

已注册 42 个关键字：
– add_product_from_home
– add_to_cart
– assert_cart_contains
– assert_cart_empty
– assert_cart_item_count
– assert_cart_not_empty
– assert_element_count
– assert_element_count_min
– assert_error_message
– assert_exact_text
– assert_hidden
– assert_login_fail
– assert_login_success
– assert_order_count_min
– assert_order_has_info
– assert_orders_not_empty
– assert_product_count
– assert_product_count_min
– assert_products_empty
– assert_products_not_empty
– assert_text
– assert_title
– assert_url_contains
– assert_url_not_contains
– assert_value
– assert_visible
– check
– checkout
– click
– click_category
– click_login_button
– click_product
– clear
– delete_cart_item
– double_click
– fill
– go_back
– hover
– input_password
– input_username
– login
– navigate
…

九、KDT 模式完整工作流

9.1 一个 YAML 用例的执行流程

以 管理员登录成功 为例：

– name: "管理员登录成功"
steps:
– keyword: login
params:
username: "admin"
password: "admin123"
– keyword: assert_login_success

执行流程：

1. TestKDTRunner 收集到这条用例
↓
2. Pytest 调用 test_kdt_case(page, yaml_path, case_data)
↓
3. 创建 TestEngine(page)
↓
4. TestEngine 初始化 KeywordRegistry(page)
→ 注册 BaseKeywords、LoginKeywords、SearchKeywords、…
→ 42 个关键字全部注册
↓
5. engine.run_case(case_data)
↓
6. 执行步骤 1：keyword=login, params={username:"admin", password:"admin123"}
→ registry.get_keyword("login") → LoginKeywords.login
→ LoginKeywords.login("admin", "admin123")
→ LoginPage.login("admin", "admin123")
→ page.goto("/login")
→ page.fill("#username", "admin")
→ page.fill("#password", "admin123")
→ page.click("#login-btn")
→ 步骤通过 ✅
↓
7. 执行步骤 2：keyword=assert_login_success
→ registry.get_keyword("assert_login_success") → LoginKeywords.assert_login_success
→ LoginPage.is_login_success()
→ 检查 page.url 是否包含 "/login"
→ 断言通过 ✅
↓
8. 用例通过 ✅

9.2 关键字层次关系

YAML 用例
│
├── 业务关键字（login / search / add_to_cart / checkout）
│ │
│ └── 页面对象（LoginPage / HomePage / CartPage）
│ │
│ └── BasePage（fill / click / wait_for_visible / …）
│ │
│ └── Playwright Page API
│
└── 底层关键字（navigate / fill / click / assert_visible）
│
└── Playwright Page API

业务关键字和底层关键字可以混合使用：

– name: "混合关键字示例"
steps:
# 用业务关键字登录
– keyword: login
params:
username: "admin"
password: "admin123"
# 用业务关键字搜索
– keyword: search
params:
keyword: "iPhone"
# 用底层关键字验证
– keyword: assert_visible
params:
selector: ".product-card"
# 用底层关键字点击
– keyword: click
params:
selector: ".product-card >> nth=0"

十、POM vs KDT 对比

10.1 同一个用例的两种写法

POM 写法（Python 代码）：

# test_cases/pom/test_login.py

@pytest.mark.smoke
@pytest.mark.p0
def test_admin_login_success(self, login_page, admin_user):
"""管理员登录成功"""
AllureHelper.title("管理员登录成功")

login_page.login(admin_user["username"], admin_user["password"])
assert login_page.is_login_success()

KDT 写法（YAML 文件）：

# test_cases/kdt/login/test_login.yaml

– name: "管理员登录成功"
markers: [smoke, p0, login]
steps:
– keyword: login
params:
username: "admin"
password: "admin123"
– keyword: assert_login_success

10.2 各自的优势

10.3 企业实践建议

推荐组合使用：

核心用例（P0 冒烟、端到端） → POM 模式（稳定、灵活）
常规回归用例（P1/P2） → KDT 模式（易维护、非技术人员可写）
探索性测试 → 手工测试

关键字层维护 → 1-2 名熟悉代码的测试工程师
YAML 用例编写 → 全体测试人员

十一、当前项目完整结构

web_ui/
├── config/
│ ├── __init__.py
│ ├── config.py
│ └── env_config.yaml
│
├── common/
│ ├── __init__.py
│ ├── logger.py
│ ├── data_reader.py
│ ├── screenshot.py
│ ├── random_data.py
│ └── allure_helper.py
│
├── pages/ ← 页面对象层
│ ├── __init__.py
│ ├── base_page.py
│ ├── login_page.py
│ ├── home_page.py
│ ├── product_page.py
│ ├── cart_page.py
│ ├── order_page.py
│ └── profile_page.py
│
├── keywords/ ← 关键字层 ✅ 本篇完成
│ ├── __init__.py
│ ├── base_keywords.py ← 底层通用关键字（25 个）
│ ├── login_keywords.py ← 登录业务关键字（7 个）
│ ├── search_keywords.py ← 搜索业务关键字（8 个）
│ ├── cart_keywords.py ← 购物车关键字（9 个）
│ ├── order_keywords.py ← 订单关键字（4 个）
│ └── keyword_registry.py ← 关键字注册表
│
├── engine/ ← 驱动引擎 ✅ 本篇完成
│ ├── __init__.py
│ └── test_engine.py ← KDT 引擎核心
│
├── test_cases/ ← 测试用例
│ ├── __init__.py
│ ├── pom/ ← POM 模式用例
│ │ ├── __init__.py
│ │ ├── test_login.py
│ │ ├── test_search.py
│ │ ├── test_product.py
│ │ ├── test_cart.py
│ │ ├── test_order.py
│ │ └── test_e2e.py
│ └── kdt/ ← KDT 模式用例 ✅ 本篇完成
│ ├── __init__.py
│ ├── test_kdt_runner.py ← KDT 运行器
│ ├── login/
│ │ └── test_login.yaml ← 8 条登录用例
│ ├── search/
│ │ └── test_search.yaml ← 10 条搜索用例
│ └── cart/
│ └── test_cart.yaml ← 5 条购物车用例
│
├── test_data/
├── reports/
├── conftest.py
├── pytest.ini
├── run.py
├── requirements.txt
└── venv/

十二、今日成果总结

今天完成了什么：

• 讲解了 KDT 模式的核心思想（两层关键字、驱动引擎、注册表）
• 实现了 base_keywords.py（25 个底层通用关键字）
• 实现了 login_keywords.py（7 个登录业务关键字）
• 实现了 search_keywords.py（8 个搜索业务关键字）
• 实现了 cart_keywords.py（9 个购物车业务关键字）
• 实现了 order_keywords.py（4 个订单业务关键字）
• 实现了 keyword_registry.py（关键字注册表，自动注册 42+ 个关键字）
• 实现了 test_engine.py（KDT 驱动引擎，解析 YAML → 查找关键字 → 执行 → 收集结果）
• 编写了 23 条 YAML 格式的测试用例（登录 8 条 + 搜索 10 条 + 购物车 5 条）
• 实现了 test_kdt_runner.py（Pytest 集成，收集 YAML 用例并执行）
• 全部 23 条 KDT 用例运行通过

十三、下篇预告

06 – KDT + DDT 实战

下一篇将在 KDT 模式的基础上引入 DDT（Data-Driven Testing，数据驱动测试）。KDT 解决了"步骤描述"的问题，DDT 解决"数据变量化"的问题——在 YAML 用例中使用变量，配合外部数据文件实现同一用例跑多组数据。同时会对比纯 KDT 和 KDT+DDT 两种写法，以及完成 POM 和 KDT 模式在同一场景下的全面对比。

系列导航