提交 7cf4f03f authored 作者: 陈泽健's avatar 陈泽健

feat: 复杂用例通用执行机制-语义化用例执行(阶段1-6)

参考文档:_PRD_复杂用例通用执行机制_语义化用例执行.md

范式转变:从"记录选择器"到"记录业务意图"。semantic 语义目标 +
执行期现场解析 + 上下文变量 + 状态验证 + 数据自建/自清理。

- 阶段1 数据模型+上下文:StepDefinition 新增 semantic/verify/run_on 顶层
  字段;ExecutionContext + {__CTX:key__} 渲染(复用 TemplateResolver)
- 阶段2 页面注册表:page_url_mapping 升级 url_patterns+fingerprint;
  recognize_current_page 运行时页面身份感知
- 阶段3 语义解析器:semantic_resolver 解析链(lookup_by_key→模式选择器
  →关键词规则→Claude仲裁),wire 进 execute_step;debug 只读接口
  /api/debug/resolve
- 阶段4 状态验证+api_call+finally:verify 断言防假通过、finally/on_failure
  清理步骤、api_call 数据自清理
- 阶段5 映射表v2显式字典:lookup_by_key 纯字典查找、USE_FUZZY_MAPPING
  默认关停、migrate_mapping_v2 迁移脚本
- 阶段6 前端+示范:CaseStepEditor 语义目标编辑、stepParser 语义推断、
  语义化重写「新建会议」示范用例(create_semantic_meeting_case.py)
- 修复参数持久化 bug:case_service.create + _case_to_response 补 parameters
- 新增映射别名 创建会议-查看详情(lookup_by_key 原 MISS,不回归旧路径)
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 83561256
...@@ -82,6 +82,16 @@ class Settings: ...@@ -82,6 +82,16 @@ class Settings:
# 被测系统配置 # 被测系统配置
TARGET_URL: str = os.getenv("TARGET_URL", "https://192.168.5.44") TARGET_URL: str = os.getenv("TARGET_URL", "https://192.168.5.44")
# 语义化用例执行配置(复杂用例通用执行机制,各阶段可独立开关回退)
# SEMANTIC_ENABLED: 语义目标解析(阶段 3)
# USE_KEYWORD_MATCHER: legacy 自然语言步骤走 keyword_matcher(阶段 3 过渡路径)
# USE_FUZZY_MAPPING: 旧模糊评分(阶段 5 关停,默认 false)
# ENFORCE_VERIFICATION: 强制状态验证(阶段 4,默认 false)
SEMANTIC_ENABLED: bool = os.getenv("SEMANTIC_ENABLED", "true").lower() == "true"
USE_KEYWORD_MATCHER: bool = os.getenv("USE_KEYWORD_MATCHER", "false").lower() == "true"
USE_FUZZY_MAPPING: bool = os.getenv("USE_FUZZY_MAPPING", "false").lower() == "true"
ENFORCE_VERIFICATION: bool = os.getenv("ENFORCE_VERIFICATION", "false").lower() == "true"
# 文档管理配置 # 文档管理配置
DOCUMENT_UPLOAD_DIR: str = os.getenv("DOCUMENT_UPLOAD_DIR", "./data/documents/upload") DOCUMENT_UPLOAD_DIR: str = os.getenv("DOCUMENT_UPLOAD_DIR", "./data/documents/upload")
DOCUMENT_OUTPUT_DIR: str = os.getenv("DOCUMENT_OUTPUT_DIR", "./data/documents/output") DOCUMENT_OUTPUT_DIR: str = os.getenv("DOCUMENT_OUTPUT_DIR", "./data/documents/output")
......
...@@ -1446,6 +1446,12 @@ ...@@ -1446,6 +1446,12 @@
"type": "css", "type": "css",
"tier": "会议预约-创建会议", "tier": "会议预约-创建会议",
"description": "创建会议 - 左侧会议室框编辑模式下的完成按钮(保存参会人配置)" "description": "创建会议 - 左侧会议室框编辑模式下的完成按钮(保存参会人配置)"
},
"创建会议-查看详情": {
"selector": ".el-message-box button:has-text('查看详情')",
"type": "css",
"tier": "会议预约-创建会议",
"description": "创建会议成功弹窗 - 查看详情按钮(semantic 键:与 会议创建成功-查看详情 同选择器,供 lookup_by_key(创建会议, 查看详情) 命中)"
} }
} }
} }
\ No newline at end of file
此差异已折叠。
{ {
"version": "1.0", "version": "2.0",
"description": "页面URL映射表:根据用例步骤识别目标页面,直达URL跳过中间导航点击", "description": "页面注册表:Phase 4 用例步骤识别直达URL(match_rules/skip_steps/url)+ 语义化运行时页面身份识别(url_patterns/fingerprint)",
"pages": [ "pages": [
{ {
"id": "create_meeting", "id": "create_meeting",
"name": "新建会议", "name": "新建会议",
"url": "https://192.168.5.44/#/meetingV3?meetingV3=%2FmeetingV3%2F%23%2FCreateMeeting", "url": "https://192.168.5.44/#/meetingV3?meetingV3=%2FmeetingV3%2F%23%2FCreateMeeting",
"url_patterns": [
"meetingV3"
],
"fingerprint": [
{"selector": ".topic_item input", "min_count": 1}
],
"match_rules": [ "match_rules": [
{"type": "step_name_contains", "value": "新建会议"}, {"type": "step_name_contains", "value": "新建会议"},
{"type": "selector_contains", "value": "新建会议"} {"type": "selector_contains", "value": "新建会议"}
...@@ -14,6 +20,42 @@ ...@@ -14,6 +20,42 @@
"step_name_contains": ["新建会议", "功能中心", "会议预约"], "step_name_contains": ["新建会议", "功能中心", "会议预约"],
"action_is": ["click"] "action_is": ["click"]
} }
},
{
"id": "login",
"name": "登录页",
"url": "",
"url_patterns": [
"login",
"Login"
],
"fingerprint": [
{"selector": "input[placeholder*='手机号']", "min_count": 1}
]
},
{
"id": "home",
"name": "主页",
"url": "",
"url_patterns": [
"#/home",
"#/dashboard"
],
"fingerprint": [
{"selector": ".home_nav_left", "min_count": 1}
]
},
{
"id": "meeting_detail",
"name": "会议详情",
"url": "",
"url_patterns": [
"meetingV3.*Detail",
"MeetingDetail"
],
"fingerprint": [
{"selector": ".room:has-text('会议')", "min_count": 1}
]
} }
] ]
} }
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:ui_context.py
模块描述:UI 用例执行上下文 — 维护用例级上下文变量,支持模板变量渲染
语义化用例执行机制的一部分:
- 用例参数(parameters)在 execute_case 入口初始化为上下文变量
- 步骤 params / semantic 目标 / api_call 请求体在 execute_step 渲染
- 支持两种占位符:
- 内置生成器(复用 TemplateResolver):{__NOW__} / {__NOW_+N_H__} /
{__RANDOM_NAME_N__} / {__UUID__} / {__RANDOM_INT_N_M__} / {__INDEX__} / {__REF:task:key__}
- 上下文变量:{__CTX:key__} 引用用例参数或前面步骤 save_as 写入的值
渲染原则:不含 "__" 的字符串直接返回(老用例零影响);
未知占位符保留原文;渲染返回新容器,不修改原模板。
作者:czj
创建日期:2026-08-18
"""
import re
from typing import Any, Dict, Optional
from app.executors.template_resolver import TemplateResolver
logger = __import__("logging").getLogger(__name__)
# 上下文变量占位符:{__CTX:key__}
CTX_PATTERN = re.compile(r"\{__CTX:([^}]+?)__\}")
class ExecutionContext:
"""
用例级执行上下文
维护变量表(初始化为用例 parameters + 内置变量),
提供递归渲染:dict/list/str 中的占位符 → 实际值。
Usage:
>>> ctx = ExecutionContext({"meetingName": "自动化-{__RANDOM_NAME_8__}"})
>>> ctx.render("主题:{__CTX:meetingName__}")
'主题:自动化-aBcD1234'
>>> ctx.set("createdId", "meeting_123")
>>> ctx.get("createdId")
'meeting_123'
"""
def __init__(self, parameters: Optional[Dict[str, Any]] = None):
"""
Args:
parameters: 用例参数(来自 case["parameters"])
其值中的占位符在初始化时渲染(支持跨键引用,迭代收敛)
"""
self._resolver = TemplateResolver()
self._values: Dict[str, Any] = dict(parameters or {})
# 内置变量缺省值(execute_case 入口会覆盖为真实值)
self._values.setdefault("caseName", "")
self._values.setdefault("caseId", "")
# 渲染 parameters 自身的占位符(如 {"meetingName": "自动化-{__RANDOM_NAME_8__}"})
# 两阶段,避免 CTX 引用的值被二次独立渲染生成不一致的随机值:
# 阶段 A:渲染所有键的"内置生成器"({__NOW__}/{__RANDOM_NAME_N__} 等),
# {__CTX:key__} 占位符保留原样,作为 CTX 查找表。
# 阶段 B:仅做 {__CTX:key__} 替换(不调用生成器),引用的是阶段 A 已渲染值。
# 迭代最多 5 轮:支持前向引用(引用定义在后面的键)跨轮收敛。
lookup: Dict[str, Any] = {
key: self._render_value(value, {}) # 空查找表:生成器解析,CTX 保留
for key, value in self._values.items()
}
for _ in range(5):
new_lookup: Dict[str, Any] = dict(lookup) # 以本轮起点作查找表:前向引用跨轮收敛
changed = False
for key, value in lookup.items():
substituted = self._substitute_recursive(value, new_lookup)
new_lookup[key] = substituted
if substituted != value:
changed = True
lookup = new_lookup
if not changed:
break
self._values = lookup
# ---------- 变量读写 ----------
def set(self, key: str, value: Any) -> None:
"""写入上下文变量(save_as 使用)"""
self._values[key] = value
def get(self, key: str) -> Optional[Any]:
"""读取上下文变量;不存在返回 None"""
return self._values.get(key)
def all(self) -> Dict[str, Any]:
"""返回当前全部上下文变量快照(调试用)"""
return dict(self._values)
# ---------- 渲染 ----------
def render(self, template: Any) -> Any:
"""
递归渲染模板中的占位符,返回新容器(不修改原模板)
先由 TemplateResolver 处理内置生成器({__NOW__} 等),
再替换 {__CTX:key__} 上下文变量。
Args:
template: dict/list/str,支持嵌套
Returns:
渲染后的 dict/list/str
"""
return self._render_value(template)
def _render_value(self, value: Any, lookup: Optional[Dict[str, Any]] = None) -> Any:
"""单值递归渲染;lookup 为 CTX 变量查找表(缺省用 self._values)"""
if lookup is None:
lookup = self._values
if isinstance(value, dict):
return {k: self._render_value(v, lookup) for k, v in value.items()}
if isinstance(value, list):
return [self._render_value(v, lookup) for v in value]
if isinstance(value, str) and "__" in value:
# 1. 内置生成器(CTX 占位符会被 TemplateResolver 原样保留)
resolved = self._resolver.resolve(value)
if not isinstance(resolved, str):
return resolved
# 2. 上下文变量替换
return self._substitute_ctx(resolved, lookup)
return value
@staticmethod
def _substitute_ctx(text: str, lookup: Dict[str, Any]) -> str:
"""替换 {__CTX:key__} → 上下文变量值;未知键保留原文"""
def _repl(match: "re.Match[str]") -> str:
key = match.group(1)
if key in lookup:
return str(lookup[key])
return match.group(0) # 未知键保留占位符,调用方记录警告
return CTX_PATTERN.sub(_repl, text)
@staticmethod
def _substitute_recursive(value: Any, lookup: Dict[str, Any]) -> Any:
"""
仅递归替换 {__CTX:key__}(不调用生成器)
用于 parameters 初始化阶段 B:保证被引用的值已是阶段 A 渲染结果,
避免已渲染随机值被二次独立重新生成导致引用不一致。
"""
if isinstance(value, dict):
return {k: ExecutionContext._substitute_recursive(v, lookup) for k, v in value.items()}
if isinstance(value, list):
return [ExecutionContext._substitute_recursive(v, lookup) for v in value]
if isinstance(value, str):
return ExecutionContext._substitute_ctx(value, lookup)
return value
...@@ -27,7 +27,7 @@ from fastapi.responses import FileResponse ...@@ -27,7 +27,7 @@ from fastapi.responses import FileResponse
from app.config import settings from app.config import settings
from app.database import init_db from app.database import init_db
from app.routers import modules, cases, executions, recorder, stats, reports, cleanup, batch, dependencies, security, security_erp, security_erp_upload, security_erp_task, ai_locator, device_sim, system, element_locator, login_template, smart_locate, performance, performance_output, api_preset, deploy, functional_report, report_template, api_test from app.routers import modules, cases, executions, recorder, stats, reports, cleanup, batch, dependencies, security, security_erp, security_erp_upload, security_erp_task, ai_locator, device_sim, system, element_locator, login_template, smart_locate, performance, performance_output, api_preset, deploy, functional_report, report_template, api_test, debug
# 配置日志 # 配置日志
logging.basicConfig( logging.basicConfig(
...@@ -277,6 +277,13 @@ app.include_router( ...@@ -277,6 +277,13 @@ app.include_router(
tags=["接口测试"] tags=["接口测试"]
) )
# 调试只读接口(语义解析验证,无写入)
app.include_router(
debug.router,
prefix="/api/debug",
tags=["调试"]
)
# ==================== 根路径 ==================== # ==================== 根路径 ====================
......
...@@ -51,6 +51,7 @@ def _case_to_response(c) -> TestCaseResponse: ...@@ -51,6 +51,7 @@ def _case_to_response(c) -> TestCaseResponse:
steps=c.steps or [], steps=c.steps or [],
config=c.config or {}, config=c.config or {},
case_type=getattr(c, "case_type", None) or "ui", case_type=getattr(c, "case_type", None) or "ui",
parameters=c.parameters or {},
created_at=c.created_at, created_at=c.created_at,
updated_at=c.updated_at, updated_at=c.updated_at,
) )
......
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:debug.py
模块描述:调试只读 API 路由(语义解析器验证用,无数据库写入)
提供 `/api/debug/resolve`:在真实被测系统页面(TARGET_URL)上对单个步骤
做语义解析,返回选择器链 + 页面识别结果 + save_as 提取值,用于验证
"语义化用例执行"各阶段规则是否按预期命中。
作者:czj
创建日期:2026-08-18
"""
import asyncio
import logging
from typing import Any, Dict, List, Optional
from fastapi import APIRouter
from pydantic import BaseModel, Field
from app.config import settings
logger = logging.getLogger(__name__)
router = APIRouter(tags=["调试"])
# ==================== Schema 定义 ====================
class DebugResolveRequest(BaseModel):
"""语义解析调试请求"""
url: str = Field(default_factory=lambda: settings.TARGET_URL, description="要打开的页面 URL")
step: Dict[str, Any] = Field(..., description="步骤定义(含可选 semantic 字段)")
scope: Optional[str] = Field(None, description="期望页面 scope id(页面注册表,可选)")
parameters: Dict[str, Any] = Field(default_factory=dict, description="用例参数(渲染语义文本)")
class DebugResolveResponse(BaseModel):
"""语义解析调试响应"""
ok: bool = Field(..., description="是否解析到选择器链")
chain: List[str] = Field(default_factory=list, description="解析出的选择器链")
path: str = Field(default="", description="命中路径: semantic/pattern/keyword/mapping/none")
page_id: Optional[str] = Field(None, description="运行时识别的页面 scope id")
page_name: Optional[str] = Field(None, description="运行时识别的页面名称")
expected_scope: Optional[str] = Field(None, description="步骤声明的期望 scope")
scope_matched: Optional[bool] = Field(None, description="scope 是否匹配")
save_as_values: Dict[str, Any] = Field(default_factory=dict, description="save_as 提取的上下文变量")
message: str = Field(default="", description="说明信息")
# ==================== 同步解析(Playwright 须在线程中执行) ====================
def _resolve_on_live_page(
url: str,
step: Dict[str, Any],
scope: Optional[str],
parameters: Dict[str, Any],
) -> DebugResolveResponse:
"""
打开真实页面并做语义解析(同步 API,须在工作线程调用)
返回值:
DebugResolveResponse —— 解析结果
"""
from playwright.sync_api import sync_playwright
from app.executors.ui_context import ExecutionContext
from app.services.semantic_resolver import get_semantic_resolver
from app.services.page_url_service import get_page_url_service
resp = DebugResolveResponse(ok=False)
semantic = step.get("semantic") if isinstance(step, dict) else None
resp.expected_scope = semantic.get("scope") if isinstance(semantic, dict) else None
try:
with sync_playwright() as p:
browser = p.chromium.launch(
headless=True,
args=["--ignore-certificate-errors"],
)
try:
page = browser.new_page()
page.goto(url, wait_until="domcontentloaded", timeout=15000)
page.wait_for_timeout(1200)
# 运行时页面身份识别
rec = get_page_url_service().recognize_current_page(page)
scope_id = rec["id"] if rec else scope
resp.page_id = rec["id"] if rec else None
resp.page_name = rec["name"] if rec else None
if isinstance(semantic, dict):
resp.scope_matched = bool(
scope_id and semantic.get("scope") and scope_id == semantic.get("scope")
)
# 语义解析
ctx = ExecutionContext(parameters or {})
resolver = get_semantic_resolver()
chain = resolver.resolve_step(page, step, scope_id, ctx) or []
# save_as 提取验证
if chain and isinstance(semantic, dict):
resolver.extract_save_as(page, semantic, chain, ctx)
resp.ok = bool(chain)
resp.chain = chain
resp.save_as_values = ctx.all()
if chain and isinstance(semantic, dict) and semantic.get("type") in (
"tab", "row_checkbox", "checkbox", "row",
):
resp.path = "pattern"
elif chain and isinstance(semantic, dict):
resp.path = "semantic"
elif chain:
resp.path = "keyword"
else:
resp.path = "none"
resp.message = (
f"页面识别: {resp.page_name or '未识别'}({resp.page_id or '-'})"
+ (",scope 匹配" if resp.scope_matched else "")
+ ("" if chain else ";未解析到选择器链")
)
finally:
browser.close()
except Exception as e:
resp.ok = False
resp.message = f"解析失败: {e}"
logger.warning(f"[debug/resolve] 解析失败: {e}")
return resp
# ==================== API ====================
@router.post("/resolve", response_model=DebugResolveResponse)
async def debug_resolve(req: DebugResolveRequest) -> DebugResolveResponse:
"""
语义解析调试(只读,启动一次无头浏览器)
在真实被测系统页面上运行语义解析器,返回选择器链 + 页面身份 + save_as 值。
用于阶段验证,不写入任何数据。
"""
try:
loop = asyncio.get_event_loop()
resp = await loop.run_in_executor(
None,
_resolve_on_live_page,
req.url,
req.step,
req.scope,
req.parameters,
)
return resp
except Exception as e:
logger.error(f"[debug/resolve] 接口异常: {e}")
return DebugResolveResponse(ok=False, message=f"接口异常: {e}")
...@@ -40,6 +40,11 @@ class StepDefinition(BaseModel): ...@@ -40,6 +40,11 @@ class StepDefinition(BaseModel):
wait_after (Optional[int]): 自定义操作后延时(ms) wait_after (Optional[int]): 自定义操作后延时(ms)
locator_type (Optional[str]): 定位类型: css/xpath/text/id/name locator_type (Optional[str]): 定位类型: css/xpath/text/id/name
locator_value (Optional[str]): 定位值 locator_value (Optional[str]): 定位值
semantic (Optional[dict]): 语义目标(语义化用例执行):
{type, text, contains, scope, role, nth, container, save_as}
verify (Optional[dict]): 状态验证(防假通过):
{type, selector, text, timeout, must}
run_on (Optional[str]): 执行时机: always/finally/on_failure/on_success(缺省 always)
""" """
order: int = Field(..., ge=1, description="步骤顺序") order: int = Field(..., ge=1, description="步骤顺序")
...@@ -55,6 +60,18 @@ class StepDefinition(BaseModel): ...@@ -55,6 +60,18 @@ class StepDefinition(BaseModel):
wait_after: Optional[int] = Field(None, description="自定义操作后延时(ms)") wait_after: Optional[int] = Field(None, description="自定义操作后延时(ms)")
locator_type: Optional[str] = Field(None, description="定位类型: css/xpath/text/id/name") locator_type: Optional[str] = Field(None, description="定位类型: css/xpath/text/id/name")
locator_value: Optional[str] = Field(None, description="定位值") locator_value: Optional[str] = Field(None, description="定位值")
semantic: Optional[Dict[str, Any]] = Field(
None,
description="语义目标(语义化用例执行):{type, text, contains, scope, role, nth, container, save_as}",
)
verify: Optional[Dict[str, Any]] = Field(
None,
description="状态验证(防假通过):{type, selector, text, timeout, must}",
)
run_on: Optional[str] = Field(
None,
description="执行时机: always/finally/on_failure/on_success(缺省 always)",
)
class TestCaseBase(BaseModel): class TestCaseBase(BaseModel):
...@@ -77,6 +94,10 @@ class TestCaseBase(BaseModel): ...@@ -77,6 +94,10 @@ class TestCaseBase(BaseModel):
description="执行配置" description="执行配置"
) )
case_type: str = Field(default="ui", description="用例类型: ui/api/security/deploy") case_type: str = Field(default="ui", description="用例类型: ui/api/security/deploy")
parameters: Dict[str, Any] = Field(
default_factory=dict,
description="用例参数(上下文变量初始值,支持 {__CTX:key__} 引用)"
)
class TestCaseCreate(TestCaseBase): class TestCaseCreate(TestCaseBase):
...@@ -120,6 +141,7 @@ class TestCaseUpdate(BaseModel): ...@@ -120,6 +141,7 @@ class TestCaseUpdate(BaseModel):
) )
config: Optional[Dict[str, Any]] = Field(None, description="执行配置") config: Optional[Dict[str, Any]] = Field(None, description="执行配置")
case_type: Optional[str] = Field(None, description="用例类型: ui/api/security/deploy") case_type: Optional[str] = Field(None, description="用例类型: ui/api/security/deploy")
parameters: Optional[Dict[str, Any]] = Field(None, description="用例参数(上下文变量初始值)")
class TestCaseResponse(TestCaseBase): class TestCaseResponse(TestCaseBase):
...@@ -193,3 +215,7 @@ class TestCaseImport(BaseModel): ...@@ -193,3 +215,7 @@ class TestCaseImport(BaseModel):
description="执行配置" description="执行配置"
) )
case_type: str = Field(default="ui", description="用例类型: ui/api/security/deploy") case_type: str = Field(default="ui", description="用例类型: ui/api/security/deploy")
parameters: Dict[str, Any] = Field(
default_factory=dict,
description="用例参数(上下文变量初始值,支持 {__CTX:key__} 引用)"
)
\ No newline at end of file
...@@ -210,6 +210,7 @@ class CaseService: ...@@ -210,6 +210,7 @@ class CaseService:
steps=steps_data, steps=steps_data,
config=data.config or {"timeout": 30000, "retry": 0, "screenshot": True}, config=data.config or {"timeout": 30000, "retry": 0, "screenshot": True},
case_type=data.case_type or "ui", case_type=data.case_type or "ui",
parameters=data.parameters or {},
) )
self.db.add(test_case) self.db.add(test_case)
......
...@@ -85,7 +85,8 @@ class ClaudeService: ...@@ -85,7 +85,8 @@ class ClaudeService:
params: dict, params: dict,
candidates: List[Dict[str, Any]], candidates: List[Dict[str, Any]],
page_url: str = "", page_url: str = "",
previous_steps: List[Dict[str, Any]] = None previous_steps: List[Dict[str, Any]] = None,
min_confidence: float = 0.0
) -> Tuple[int, float, str, bool]: ) -> Tuple[int, float, str, bool]:
""" """
对候选元素进行语义排序(Phase 5 增强) 对候选元素进行语义排序(Phase 5 增强)
...@@ -108,6 +109,8 @@ class ClaudeService: ...@@ -108,6 +109,8 @@ class ClaudeService:
] ]
page_url: 当前页面 URL(Phase 5 新增) page_url: 当前页面 URL(Phase 5 新增)
previous_steps: 前面的步骤列表(Phase 5 新增) previous_steps: 前面的步骤列表(Phase 5 新增)
min_confidence: 调用方预判的高置信度阈值;>0 且单候选时直接短路不调 Claude
(语义解析器"规则优先,Claude 按需":规则可判时跳过 Claude 仲裁)
Returns: Returns:
Tuple[int, float, str, bool]: (选中索引, 置信度, 选择理由, 是否确认) Tuple[int, float, str, bool]: (选中索引, 置信度, 选择理由, 是否确认)
...@@ -124,6 +127,11 @@ class ClaudeService: ...@@ -124,6 +127,11 @@ class ClaudeService:
if len(candidates) == 0: if len(candidates) == 0:
raise Exception("候选元素列表为空") raise Exception("候选元素列表为空")
# 规则优先:调用方判定唯一/高置信候选时跳过 Claude 仲裁(语义化执行机制)
if min_confidence > 0 and len(candidates) == 1:
logger.debug(f"Claude 语义排序短路: 唯一候选(min_confidence={min_confidence})")
return 0, 1.0, "唯一候选元素(min_confidence 短路)", True
# Phase 5: 单候选也调用 Claude 验证 # Phase 5: 单候选也调用 Claude 验证
logger.info(f"Claude 语义排序: step='{step_name}', candidates={len(candidates)}, url={page_url[:50] if page_url else 'N/A'}") logger.info(f"Claude 语义排序: step='{step_name}', candidates={len(candidates)}, url={page_url[:50] if page_url else 'N/A'}")
......
...@@ -1177,4 +1177,43 @@ def find_element_by_semantic( ...@@ -1177,4 +1177,43 @@ def find_element_by_semantic(
except Exception: except Exception:
continue continue
def resolve_selectors(
page,
step: Dict[str, Any],
max_chain: int = 4,
) -> List[str]:
"""
keyword_matcher 选择器链 facade(legacy 自然语言步骤的过渡路径)
把"步骤名关键词 → 候选选择器"包成执行器可直接注入 params["selectors"] 的
字符串链(最高分在前)。供 SemanticResolver legacy 路径复用,也便于独立接入。
Args:
page: Playwright Page 对象
step: 步骤 dict(name/action)
max_chain: 选择器链最大长度
Returns:
List[str] 选择器字符串链(降序),未匹配返回 []
"""
name = (step.get("name") or "") if isinstance(step, dict) else ""
action = (step.get("action") or "click") if isinstance(step, dict) else "click"
if not name:
return []
keywords = extract_keywords(name)
if not keywords:
return []
elements, selectors = match_element_by_keywords(page, keywords, action)
if elements and selectors:
return [str(s.get("value")) for s in selectors if s.get("value")][:max_chain]
el, sels = find_element_by_semantic(page, keywords, action)
if sels:
return [str(s.get("value")) for s in sels if s.get("value")][:max_chain]
return []
return None, [] return None, []
\ No newline at end of file
...@@ -17,6 +17,7 @@ ...@@ -17,6 +17,7 @@
import json import json
import logging import logging
import os import os
import re
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
...@@ -70,7 +71,8 @@ class PageUrlService: ...@@ -70,7 +71,8 @@ class PageUrlService:
for page in pages: for page in pages:
pid = page.get("id") pid = page.get("id")
if pid and page.get("url"): # 注册表按 id 存储全部页面(含无直达 URL 的登录/主页,供运行时识别 scope)
if pid:
self._pages[pid] = page self._pages[pid] = page
self._enabled = True self._enabled = True
...@@ -150,9 +152,11 @@ class PageUrlService: ...@@ -150,9 +152,11 @@ class PageUrlService:
keywords.extend([str(s) for s in selectors]) keywords.extend([str(s) for s in selectors])
entry_keywords.append((action, keywords)) entry_keywords.append((action, keywords))
# 对每个页面匹配规则打分 # 对每个页面匹配规则打分(仅考虑有直达 URL 的页面)
matches: List[Dict[str, Any]] = [] matches: List[Dict[str, Any]] = []
for pid, page in self._pages.items(): for pid, page in self._pages.items():
if not page.get("url"):
continue
conf = self._score_page(page, entry_keywords) conf = self._score_page(page, entry_keywords)
if conf > 0: if conf > 0:
matches.append({"page_id": pid, "name": page.get("name", ""), matches.append({"page_id": pid, "name": page.get("name", ""),
...@@ -255,6 +259,93 @@ class PageUrlService: ...@@ -255,6 +259,93 @@ class PageUrlService:
return skip_orders return skip_orders
def recognize_current_page(self, page) -> Optional[Dict[str, Any]]:
"""
运行时根据当前浏览器 URL + DOM 指纹识别页面身份(语义化 scope)
识别策略:
1. 先用 url_patterns(正则数组)对当前 URL 计分,唯一命中直接返回
2. 0 或歧义时,用 fingerprint(DOM 指纹,min_count 个匹配)判存
3. 都失败返回 None(执行器零行为变化)
Args:
page: Playwright Page 对象(鸭子类型:.url / .query_selector_all)
Returns:
{"id", "name", "confidence"} 或 None
"""
if not self._enabled:
return None
try:
current_url = str(page.url or "")
except Exception:
return None
# 1. URL patterns 计分
candidates: List[Dict[str, Any]] = []
for pid, page_conf in self._pages.items():
score = 0
for pat in page_conf.get("url_patterns", []):
try:
if re.search(pat, current_url):
score += 1
except re.error:
continue
if score > 0:
candidates.append((score, pid, page_conf))
if len(candidates) == 1:
pid = candidates[0][1]
p = candidates[0][2]
return {"id": pid, "name": p.get("name", ""), "confidence": 1.0}
# 2. fingerprint DOM 指纹判存(URL 缺失/歧义时)
if page is not None:
for pid, page_conf in self._pages.items():
fp = page_conf.get("fingerprint", [])
if not fp:
continue
hits = 0
required = 0
for item in fp:
sel = item.get("selector")
if not sel:
continue
try:
min_count = int(item.get("min_count", 1))
except (TypeError, ValueError):
min_count = 1
required += 1
try:
matched = page.query_selector_all(sel)
if len(matched) < min_count:
continue
# 文本校验(首个匹配的文本必须一致)
expected_text = item.get("text")
if expected_text:
if not matched or (matched[0].text_content() or "").strip() != expected_text:
continue
hits += 1
except Exception:
continue
if required > 0 and hits >= required:
return {"id": pid, "name": page_conf.get("name", ""), "confidence": 0.8}
return None
def get_scope_page(self, scope: str) -> Optional[Dict[str, Any]]:
"""
按 scope id 返回页面注册表配置(语义解析器容器约束用)
Args:
scope: 页面注册表 id(如 create_meeting)
Returns:
页面配置 dict 或 None
"""
return self._pages.get(scope)
def is_enabled(self) -> bool: def is_enabled(self) -> bool:
"""映射表是否可用""" """映射表是否可用"""
return self._enabled return self._enabled
......
此差异已折叠。
...@@ -133,6 +133,7 @@ class SmartLocateService: ...@@ -133,6 +133,7 @@ class SmartLocateService:
# 4. 逐步骤定位 # 4. 逐步骤定位
skip_remaining = False skip_remaining = False
seen_steps: List[Dict[str, Any]] = [] # 已完成步骤(Claude 上下文)
for step in steps: for step in steps:
# 如果前置步骤失败,跳过后续步骤 # 如果前置步骤失败,跳过后续步骤
if skip_remaining: if skip_remaining:
...@@ -148,8 +149,9 @@ class SmartLocateService: ...@@ -148,8 +149,9 @@ class SmartLocateService:
results.append(result) results.append(result)
continue continue
result = self._locate_single_step(step) result = self._locate_single_step(step, previous_steps=seen_steps)
results.append(result) results.append(result)
seen_steps.append(step)
# 如果定位成功且操作类型需要执行,则实际执行(确保页面状态正确) # 如果定位成功且操作类型需要执行,则实际执行(确保页面状态正确)
if result.get('success') and result.get('action') != 'navigate': if result.get('success') and result.get('action') != 'navigate':
...@@ -358,12 +360,17 @@ class SmartLocateService: ...@@ -358,12 +360,17 @@ class SmartLocateService:
logger.error(f"导航到菜单异常: {e}") logger.error(f"导航到菜单异常: {e}")
return False return False
def _locate_single_step(self, step: Dict[str, Any]) -> Dict[str, Any]: def _locate_single_step(
self,
step: Dict[str, Any],
previous_steps: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
""" """
为单个步骤定位元素 为单个步骤定位元素
Args: Args:
step (Dict): 步骤定义 step (Dict): 步骤定义
previous_steps (Optional[List[Dict]]): 当前步骤之前的步骤(Claude 语义排序上下文)
Returns: Returns:
Dict: 定位结果 Dict: 定位结果
...@@ -607,7 +614,9 @@ class SmartLocateService: ...@@ -607,7 +614,9 @@ class SmartLocateService:
page_url = page.url if page else "" page_url = page.url if page else ""
previous_steps_info = [] previous_steps_info = []
# 获取前面步骤的名称作为上下文 # 获取前面步骤的名称作为上下文
for prev_step in self.steps[:step_idx]: # (修复:原 self.steps[:step_idx] 引用未定义的 self.steps/step_idx,
# 导致 Claude 增强路径恒抛异常回退;改为方法参数传入)
for prev_step in (previous_steps or []):
previous_steps_info.append({ previous_steps_info.append({
'name': prev_step.get('name', ''), 'name': prev_step.get('name', ''),
'action': prev_step.get('action', '') 'action': prev_step.get('action', '')
......
此差异已折叠。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:migrate_mapping_v2.py
模块描述:元素映射表 v1 扁平结构 → v2 显式字典结构迁移脚本(离线)
v2 格式(前端开发可维护,见 PRD 第五章):
{
"version": "2.0", "source": "迁移自 v1 231 条",
"pages": [
{ "id": "创建会议", "name": "新建会议",
"elements": { "预定会议": {"selectors": [".time_btn .btns:has-text(\"预定会议\")"], "type": "tab"} } }
],
"shared_elements": { "登录按钮": {"selectors": ["button:has-text(\"登录\")"], "type": "button"} }
}
说明:
- 按页面 token 归组(与 element_mapping_service._page_id_for 同规则)
- 键保留原语义名(形如"创建会议-预定会议"),值为 {selectors:[原selector], type, description}
- 纯显式字典,无任何评分规则
用法:
cd backend && python scripts/migrate_mapping_v2.py [输出路径]
默认输出 backend/data/elements_mapping_v2.json
作者:czj
创建日期:2026-08-18
"""
import json
import os
import sys
# 允许从任意工作目录运行
# 脚本位于 backend/scripts/,上两级 = backend;映射表在 backend/app/data/
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SRC_FILE = os.path.join(BASE_DIR, "app", "data", "elements_mapping.json")
OUT_DIR = os.path.join(BASE_DIR, "app", "data")
# 页面 token(与 element_mapping_service._page_id_for 保持一致)
_PAGE_TOKENS = ("创建会议", "会议详情", "会议列表", "会议预约", "登录页", "首页",
"会议室", "会议模板", "个人日程", "管理看板", "信息发布", "预订2.0")
def page_id_for(key: str, info: dict) -> str:
tier = info.get("tier") or ""
for tok in _PAGE_TOKENS:
if tok in tier:
return tok
for tok in _PAGE_TOKENS:
if key.startswith(f"{tok}-"):
return tok
return tier or "默认"
def migrate() -> dict:
with open(SRC_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
elements = data.get("elements", {})
pages: dict = {}
for key, info in elements.items():
pid = page_id_for(key, info)
page = pages.setdefault(pid, {
"id": pid,
"name": pid,
"elements": {},
})
# 保留原 selector(v1 每个条目一个 selector)
page["elements"][key] = {
"selectors": [info["selector"]] if isinstance(info.get("selector"), str) else (info.get("selectors") or []),
"type": info.get("type", "css"),
"description": info.get("description", ""),
}
return {
"version": "2.0",
"source": f"迁移自 v1 {len(elements)} 条",
"pages": [pages[pid] for pid in sorted(pages)],
"shared_elements": {},
}
def main() -> None:
out = sys.argv[1] if len(sys.argv) > 1 else os.path.join(OUT_DIR, "elements_mapping_v2.json")
result = migrate()
with open(out, "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
n_pages = len(result["pages"])
n_elements = sum(len(p["elements"]) for p in result["pages"])
print(f"迁移完成: {n_pages} 页, {n_elements} 元素 → {out}")
if __name__ == "__main__":
main()
...@@ -20,9 +20,23 @@ import pytest ...@@ -20,9 +20,23 @@ import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app")) sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))
from app.config import settings
from app.services.element_mapping_service import ElementMappingService from app.services.element_mapping_service import ElementMappingService
@pytest.fixture(autouse=True)
def _legacy_fuzzy_enabled():
"""阶段 5 起 USE_FUZZY_MAPPING 默认 false(模糊评分关停,改用 v2 显式字典)。
本模块测试的是 legacy 模糊评分路径——显式开启该 flag,既保留对旧算法
的回归覆盖,也验证"部署可一键回切"的逃生舱仍然有效。
"""
old = settings.USE_FUZZY_MAPPING
settings.USE_FUZZY_MAPPING = True
yield
settings.USE_FUZZY_MAPPING = old
@pytest.fixture @pytest.fixture
def meeting_mapping(tmp_path): def meeting_mapping(tmp_path):
"""创建会议流程 P1 步骤映射(含干扰项,验证评分能正确区分)""" """创建会议流程 P1 步骤映射(含干扰项,验证评分能正确区分)"""
...@@ -38,6 +52,7 @@ def meeting_mapping(tmp_path): ...@@ -38,6 +52,7 @@ def meeting_mapping(tmp_path):
# P1 表步骤 13-21 目标条目 # P1 表步骤 13-21 目标条目
"创建会议-议题输入框": {"selector": ".topic_item input", "type": "css", "tier": "创建会议", "description": "创建会议 - 议题输入框"}, "创建会议-议题输入框": {"selector": ".topic_item input", "type": "css", "tier": "创建会议", "description": "创建会议 - 议题输入框"},
"创建会议-编辑会议室": {"selector": ".room .delimg", "type": "css", "tier": "创建会议", "description": "创建会议 - 参会人配置模式开关(编辑/完成共用)"}, "创建会议-编辑会议室": {"selector": ".room .delimg", "type": "css", "tier": "创建会议", "description": "创建会议 - 参会人配置模式开关(编辑/完成共用)"},
"创建会议-完成按钮": {"selector": ".room .delimg", "type": "css", "tier": "创建会议", "description": "创建会议 - 参会人配置完成按钮(v2 显式条目,替代模糊评分特例)"},
"创建会议-参会人勾选": {"selector": ".el-table:has-text('姓名') .el-checkbox", "type": "css", "tier": "创建会议", "description": "创建会议 - 参会人表格复选框"}, "创建会议-参会人勾选": {"selector": ".el-table:has-text('姓名') .el-checkbox", "type": "css", "tier": "创建会议", "description": "创建会议 - 参会人表格复选框"},
"创建会议-确定创建": {"selector": ".btn button.el-button", "type": "css", "tier": "创建会议", "description": "创建会议 - 底部确定创建按钮"}, "创建会议-确定创建": {"selector": ".btn button.el-button", "type": "css", "tier": "创建会议", "description": "创建会议 - 底部确定创建按钮"},
"会议创建成功-查看详情": {"selector": ".el-message-box button:has-text('查看详情')", "type": "css", "tier": "会议创建成功", "description": "创建成功弹窗查看详情按钮"}, "会议创建成功-查看详情": {"selector": ".el-message-box button:has-text('查看详情')", "type": "css", "tier": "会议创建成功", "description": "创建成功弹窗查看详情按钮"},
...@@ -197,11 +212,11 @@ class TestP1MeetingFlow: ...@@ -197,11 +212,11 @@ class TestP1MeetingFlow:
assert result["selector"] == ".room .delimg" assert result["selector"] == ".room .delimg"
def test_complete_button(self, meeting_mapping): def test_complete_button(self, meeting_mapping):
"""'完成按钮'与编辑按钮共用 .room .delimg,不应误配会议室-区域树""" """'完成按钮'走 v2 显式字典 → .room .delimg(阶段 5 不再靠模糊评分特例)"""
svc = _make_service(meeting_mapping) svc = _make_service(meeting_mapping)
result = svc.get_selector("点击左侧会议室框的【完成】按钮", "click") result = svc.lookup_by_key("创建会议", "完成按钮")
assert result is not None assert result is not None
assert result["key"] == "创建会议-编辑会议室" assert result["selectors"] == [".room .delimg"]
def test_check_user(self, meeting_mapping): def test_check_user(self, meeting_mapping):
"""'勾选右侧的用户'应命中参会人表格复选框,而非参会人表格或编辑按钮""" """'勾选右侧的用户'应命中参会人表格复选框,而非参会人表格或编辑按钮"""
......
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:test_element_mapping_v2.py
模块描述:映射表 v2 显式字典单元测试(阶段 5)
覆盖:
- lookup_by_key:v1 flat 自动归组 / scope 别名对齐 / 页前缀键 / 共享元素 / 缺失
- USE_FUZZY_MAPPING 关停:_exact_match/_fuzzy_match 返回 None
- v2 显式 pages 结构加载(前端开发提供格式)
- _build_v2_result 归一
作者:czj
创建日期:2026-08-18
"""
import json
import tempfile
import os
from unittest.mock import MagicMock
import pytest
from app.config import settings
from app.services.element_mapping_service import ElementMappingService
@pytest.fixture(scope="module")
def v1_service():
"""加载真实 v1 映射表(231 条),验证自动归组"""
s = ElementMappingService()
assert s.is_enabled()
return s
# ==================== v1 自动归组 → lookup_by_key ====================
class TestLookupByKeyV1:
def test_create_meeting_exact_semantic_key(self, v1_service):
"""scope 别名 + 语义键 → 确定性选择器链(无评分)"""
r = v1_service.lookup_by_key("create_meeting", "预定会议")
assert r is not None
assert r["selectors"] and isinstance(r["selectors"], list)
assert r["key"] == "预定会议"
def test_participant_checkbox_chain(self, v1_service):
"""参会人勾选 → 行级复选框选择器链(此前靠评分规则才能命中)"""
r = v1_service.lookup_by_key("创建会议", "参会人勾选")
assert r is not None
assert any(".el-checkbox" in s for s in r["selectors"])
def test_login_tier_alias(self, v1_service):
"""page_id 'login' → tier 别名 '登录页' → 命中"""
r = v1_service.lookup_by_key("login", "密码登录Tab")
assert r is not None
assert r["selectors"]
def test_missing_key_returns_none(self, v1_service):
assert v1_service.lookup_by_key("create_meeting", "不存在的键XYZ") is None
def test_missing_page_returns_none(self, v1_service):
assert v1_service.lookup_by_key("不存在的页", "预定会议") is None
def test_none_page_id_returns_none(self, v1_service):
assert v1_service.lookup_by_key(None, "预定会议") is None
# ==================== 评分路径关停 ====================
class TestFuzzyGating:
def test_exact_match_disabled(self, v1_service):
old = settings.USE_FUZZY_MAPPING
settings.USE_FUZZY_MAPPING = False
try:
assert v1_service._exact_match("点击预定会议") is None
assert v1_service._fuzzy_match("点击预定会议") is None
finally:
settings.USE_FUZZY_MAPPING = old
def test_get_selector_gated(self, v1_service):
old = settings.USE_FUZZY_MAPPING
settings.USE_FUZZY_MAPPING = False
try:
assert v1_service.get_selector("点击预定会议") is None
finally:
settings.USE_FUZZY_MAPPING = old
def test_get_selector_enabled_when_flag_on(self, v1_service):
old = settings.USE_FUZZY_MAPPING
settings.USE_FUZZY_MAPPING = True
try:
assert v1_service.get_selector("点击预定会议") is not None
finally:
settings.USE_FUZZY_MAPPING = old
# ==================== v2 显式结构加载 ====================
class TestV2ExplicitStructure:
def _write_v2(self):
data = {
"version": "2.0",
"source": "前端开发提供",
"pages": [
{
"id": "create_meeting",
"name": "新建会议",
"elements": {
"预定会议": {"selectors": ['.time_btn .btns:has-text("预定会议")'], "type": "tab"},
"参会人勾选": {
"selectors": ['.el-table__row:has-text("admin@xty12") .el-checkbox'],
"type": "row_checkbox",
},
},
},
{
"id": "meeting_detail",
"name": "会议详情",
"elements": {
"取消会议": {"selectors": ["button:has-text(\"取消会议\")"], "type": "button"},
},
},
],
"shared_elements": {
"登录按钮": {"selectors": ["button:has-text(\"登录\")"], "type": "button"},
},
}
fd, path = tempfile.mkstemp(suffix=".json")
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False)
return path
def test_explicit_pages_lookup(self):
path = self._write_v2()
try:
s = ElementMappingService(path)
assert s.is_enabled()
r = s.lookup_by_key("create_meeting", "预定会议")
assert r["selectors"] == ['.time_btn .btns:has-text("预定会议")']
assert r["type"] == "tab"
r2 = s.lookup_by_key("meeting_detail", "取消会议")
assert r2["selectors"] == ['button:has-text("取消会议")']
finally:
os.remove(path)
def test_shared_elements_fallback(self):
path = self._write_v2()
try:
s = ElementMappingService(path)
r = s.lookup_by_key("create_meeting", "登录按钮")
assert r["selectors"] == ['button:has-text("登录")']
finally:
os.remove(path)
# ==================== _build_v2_result 归一 ====================
class TestBuildV2Result:
def _s(self):
return ElementMappingService.__new__(ElementMappingService)
def test_string_entry(self):
r = ElementMappingService._build_v2_result(".btn", "x")
assert r["selectors"] == [".btn"]
assert r["type"] == "css"
def test_dict_with_selectors(self):
r = ElementMappingService._build_v2_result(
{"selectors": [".a", ".b"], "type": "tab"}, "x"
)
assert r["selectors"] == [".a", ".b"]
assert r["type"] == "tab"
def test_dict_with_single_selector(self):
r = ElementMappingService._build_v2_result({"selector": ".a"}, "x")
assert r["selectors"] == [".a"]
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:test_page_url_service.py
模块描述:PageUrlService 单元测试(页面注册表 + 运行时页面身份识别)
覆盖:
- 阶段 2 新功能 recognize_current_page:URL patterns 唯一命中 / fingerprint 兜底 / 未识别 None
- get_scope_page 按 scope id 返回页面配置
- 旧功能 recognize_page(步骤识别直达 URL)不回归
作者:czj
创建日期:2026-08-18
"""
import json
import sys
import tempfile
from pathlib import Path
# 加入项目路径,确保可导入
sys.path.insert(0, str(Path(__file__).parent.parent))
from app.services.page_url_service import PageUrlService
# ==================== 测试夹具 ====================
REGISTRY = {
"version": "2.0",
"pages": [
{
"id": "create_meeting",
"name": "新建会议",
"url": "https://192.168.5.44/#/meetingV3?meetingV3=%2FmeetingV3%2F%23%2FCreateMeeting",
"url_patterns": ["meetingV3"],
"fingerprint": [{"selector": ".topic_item input", "min_count": 1}],
"match_rules": [{"type": "step_name_contains", "value": "新建会议"}],
},
{
"id": "login",
"name": "登录页",
"url": "",
"url_patterns": ["login"],
"fingerprint": [{"selector": "input[placeholder*='手机号']", "min_count": 1}],
},
{
"id": "home",
"name": "主页",
"url": "",
"url_patterns": ["#/home"],
"fingerprint": [{"selector": ".home_nav_left", "min_count": 1}],
},
],
}
class FakePage:
"""鸭子类型 Playwright Page(只实现 .url / .query_selector_all)"""
def __init__(self, url: str = "", matches=None):
self._url = url
# selector -> 匹配元素数(query_selector_all 返回 [1]*n)
self._matches = matches or {}
@property
def url(self):
return self._url
def query_selector_all(self, selector: str):
count = self._matches.get(selector, 0)
return [object() for _ in range(count)]
def make_service():
"""创建指向临时注册表文件的 PageUrlService"""
with tempfile.NamedTemporaryFile(
"w", suffix=".json", delete=False, encoding="utf-8"
) as f:
json.dump(REGISTRY, f, ensure_ascii=False)
path = f.name
svc = PageUrlService(path)
# 清理临时文件(PageUrlService 已加载进内存)
import os
try:
os.unlink(path)
except OSError:
pass
return svc
# ==================== recognize_current_page:URL patterns ====================
def test_recognize_url_pattern_unique_hit():
"""URL patterns 唯一命中 → 直接返回该页面"""
svc = make_service()
page = FakePage(url="https://192.168.5.44/#/meetingV3?x=%2FCreateMeeting")
rec = svc.recognize_current_page(page)
assert rec is not None
assert rec["id"] == "create_meeting"
assert rec["confidence"] == 1.0
def test_recognize_url_pattern_login():
"""登录页 URL 命中 → login"""
svc = make_service()
page = FakePage(url="https://192.168.5.44/platform/2.0.2/#/login")
rec = svc.recognize_current_page(page)
assert rec is not None
assert rec["id"] == "login"
# ==================== recognize_current_page:fingerprint 兜底 ====================
def test_recognize_fingerprint_fallback():
"""URL 未命中任何 url_patterns,但 DOM 指纹命中 → fingerprint 判存"""
svc = make_service()
# create_meeting URL 与 login/home 都含 login(歧义),改用指纹区分
page = FakePage(
url="https://example.com/somewhere",
matches={".topic_item input": 1},
)
rec = svc.recognize_current_page(page)
assert rec is not None
assert rec["id"] == "create_meeting"
def test_recognize_fingerprint_min_count():
"""fingerprint min_count 未满足 → 不判存"""
svc = make_service()
page = FakePage(
url="https://example.com/unknown",
matches={".topic_item input": 0, ".home_nav_left": 1},
)
# home 指纹命中,create_meeting 的 min_count 未满足
rec = svc.recognize_current_page(page)
assert rec is not None
assert rec["id"] == "home"
def test_recognize_no_match_returns_none():
"""URL 与指纹都未命中 → None(零行为变化)"""
svc = make_service()
page = FakePage(url="https://example.com/random", matches={})
assert svc.recognize_current_page(page) is None
def test_recognize_empty_url():
"""URL 为空 → 仅靠指纹识别"""
svc = make_service()
page = FakePage(url="", matches={".home_nav_left": 2})
rec = svc.recognize_current_page(page)
assert rec is not None
assert rec["id"] == "home"
# ==================== get_scope_page ====================
def test_get_scope_page():
"""按 scope id 返回页面配置;不存在返回 None"""
svc = make_service()
cfg = svc.get_scope_page("create_meeting")
assert cfg is not None
assert cfg["id"] == "create_meeting"
assert svc.get_scope_page("not_exists") is None
# ==================== 旧功能不回归 ====================
def test_recognize_page_legacy_still_works():
"""步骤识别(直达 URL)老功能不回归;无直达 URL 的页面不参与"""
svc = make_service()
steps = [{"order": 1, "name": "进入新建会议", "action": "click", "params": {"selector": "新建会议"}}]
rec = svc.recognize_page(steps)
assert rec is not None
assert rec["page_id"] == "create_meeting"
assert rec["url"].startswith("https")
def test_recognize_page_ignores_no_url_pages():
"""login/home 无直达 URL,不应被步骤识别选中"""
svc = make_service()
steps = [{"order": 1, "name": "登录页", "action": "click", "params": {}}]
rec = svc.recognize_page(steps)
# 无 url 的页面被排除;无 match_rules 匹配时返回 None
assert rec is None or rec["page_id"] == "create_meeting"
此差异已折叠。
此差异已折叠。
此差异已折叠。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:test_ui_context.py
模块描述:ExecutionContext 单元测试(语义化用例执行上下文)
覆盖:
- parameters 初始化渲染({__RANDOM_NAME_8__} / 跨键 CTX 引用)
- {__CTX:key__} 替换 / 未知键保留
- {__NOW__} 内置生成器复用
- 嵌套 dict/list 递归渲染
- 无 "__" 字符串零改动(老用例兼容)
- 原模板不被修改
作者:czj
创建日期:2026-08-18
"""
import re
import sys
from datetime import datetime
from pathlib import Path
# 加入项目路径,确保可导入
sys.path.insert(0, str(Path(__file__).parent.parent))
from app.executors.ui_context import ExecutionContext
# ==================== parameters 初始化渲染 ====================
def test_parameters_random_name_rendered():
"""parameters 值中的 {__RANDOM_NAME_8__} 应在初始化时渲染为 8 位随机串"""
ctx = ExecutionContext({"meetingName": "自动化-{__RANDOM_NAME_8__}"})
value = ctx.get("meetingName")
assert value.startswith("自动化-"), f"应保留前缀: {value}"
suffix = value.split("-", 1)[1]
assert len(suffix) == 8, f"后缀应为 8 位: {value}"
assert suffix.isalnum(), f"后缀应为字母数字: {value}"
def test_parameters_cross_key_reference():
"""参数键之间的 {__CTX:...__} 交叉引用应迭代收敛"""
ctx = ExecutionContext({
"base": "自动化-{__RANDOM_NAME_8__}",
"title": "主题-{__CTX:base__}",
})
title = ctx.get("title")
base = ctx.get("base")
assert title == f"主题-{base}", f"交叉引用未收敛: {title} != 主题-{base}"
def test_parameters_forward_reference():
"""前向引用({__CTX:B__} 引用定义在后面的键 B)应跨轮收敛"""
ctx = ExecutionContext({
"name": "会议-{__CTX:uid__}",
"uid": "U-{__RANDOM_NAME_4__}",
})
uid = ctx.get("uid")
assert ctx.get("name") == f"会议-{uid}", f"前向引用未收敛: name={ctx.get('name')} != 会议-{uid}"
def test_parameters_plain_value_untouched():
"""普通值(无占位符)应原样保留"""
ctx = ExecutionContext({"room": "北京展厅会议室", "participant": "admin@xty12"})
assert ctx.get("room") == "北京展厅会议室"
assert ctx.get("participant") == "admin@xty12"
# ==================== {__CTX:key__} 替换 ====================
def test_ctx_placeholder_substituted():
"""{__CTX:meetingName__} 应替换为参数值"""
ctx = ExecutionContext({"meetingName": "自动化-test"})
result = ctx.render("{__CTX:meetingName__}")
assert result == "自动化-test"
def test_ctx_embedded_in_text():
"""{__CTX:key__} 可嵌入任意字符串"""
ctx = ExecutionContext({"meetingName": "自动化-test"})
result = ctx.render("会议《{__CTX:meetingName__}》的议题")
assert result == "会议《自动化-test》的议题"
def test_ctx_unknown_key_preserved():
"""未知键应保留占位符原文"""
ctx = ExecutionContext({})
result = ctx.render("{__CTX:not_exists__}")
assert result == "{__CTX:not_exists__}"
def test_ctx_after_set():
"""save_as 写入的变量应可被后续渲染引用"""
ctx = ExecutionContext({})
ctx.set("createdId", "meeting_123")
assert ctx.get("createdId") == "meeting_123"
result = ctx.render("ID={__CTX:createdId__}")
assert result == "ID=meeting_123"
# ==================== 内置生成器复用 ====================
def test_now_generator_reused():
"""{__NOW__} 应复用 TemplateResolver 内置生成器"""
ctx = ExecutionContext({})
result = ctx.render({"time": "{__NOW__}"})
now_prefix = datetime.now().strftime("%Y-%m-%d %H:%M:%S")[:16]
assert result["time"].startswith(now_prefix), f"{result['time']} != {now_prefix}"
def test_builtin_and_ctx_combined():
"""内置生成器与 CTX 可混合使用"""
ctx = ExecutionContext({"tag": "summit"})
result = ctx.render("{__CTX:tag__}-{__RANDOM_NAME_4__}")
assert result.startswith("summit-")
assert len(result) == len("summit-") + 4
# ==================== 嵌套渲染 ====================
def test_nested_dict_list_rendered():
"""嵌套 dict/list 应递归渲染"""
ctx = ExecutionContext({"meetingName": "自动化-test", "participant": "admin@xty12"})
template = {
"body": {
"name": "{__CTX:meetingName__}",
"participants": ["{__CTX:participant__}"],
"meta": {"seq": "{__RANDOM_INT_1_100__}"},
},
}
result = ctx.render(template)
assert result["body"]["name"] == "自动化-test"
assert result["body"]["participants"] == ["admin@xty12"]
assert 1 <= int(result["body"]["meta"]["seq"]) <= 100
# ==================== 兼容性(老用例零影响) ====================
def test_no_placeholder_unchanged():
"""不含 '__' 的字符串应原样返回(老用例零影响)"""
ctx = ExecutionContext({"meetingName": "自动化-test"})
params = {"selector": "#submit", "value": "固定值", "url": "/login"}
result = ctx.render(params)
assert result == params
def test_original_template_not_modified():
"""渲染返回新容器,原模板不被修改(防重复渲染放大动态值)"""
ctx = ExecutionContext({})
params = {"value": "{__RANDOM_NAME_8__}"}
result = ctx.render(params)
assert result["value"] != params["value"]
assert params["value"] == "{__RANDOM_NAME_8__}", "原模板不应被修改"
def test_empty_render():
"""空容器渲染应返回空"""
ctx = ExecutionContext({})
assert ctx.render({}) == {}
assert ctx.render([]) == []
assert ctx.render("") == ""
assert ctx.render(42) == 42
assert ctx.render(None) is None
...@@ -6,6 +6,44 @@ ...@@ -6,6 +6,44 @@
* @date 2026-07-09 * @date 2026-07-09
*/ */
/** 语义目标(语义化用例执行) */
export interface SemanticTarget {
/** 元素类型: button/input/textarea/checkbox/row_checkbox/row/tab/link/menu/icon/text/dialog */
type?: string
/** 主匹配文本(渲染后用于映射精确键 + DOM 文本/placeholder) */
text?: string
/** 行/单元格次级过滤文本 */
contains?: string
/** 页面注册表 id(scope) */
scope?: string
/** 角色/辅助提示(如 tab 的 aria-selected) */
role?: string
/** 第 N 个匹配(1 起) */
nth?: number
/** 父容器约束 → 组合选择器 */
container?: {
type?: string
selector?: string
title?: string
}
/** 提取元素值写入上下文变量(save_as) */
save_as?: string | Record<string, string>
}
/** 状态验证(防假通过) */
export interface StepVerify {
/** 验证类型: text_visible/element_visible/url_contains/dialog 等 */
type?: string
/** 验证选择器 */
selector?: string
/** 期望文本 */
text?: string
/** 超时 ms */
timeout?: number
/** true 失败则步骤真实 failed(缺省 false 只 warn) */
must?: boolean
}
/** 测试步骤定义 */ /** 测试步骤定义 */
export interface StepDefinition { export interface StepDefinition {
/** 步骤顺序 */ /** 步骤顺序 */
...@@ -34,6 +72,12 @@ export interface StepDefinition { ...@@ -34,6 +72,12 @@ export interface StepDefinition {
locator_type?: string locator_type?: string
/** 定位值 */ /** 定位值 */
locator_value?: string locator_value?: string
/** 语义目标(语义化用例执行,后端 snake_case 序列化) */
semantic?: SemanticTarget
/** 状态验证(防假通过,后端 snake_case 序列化) */
verify?: StepVerify
/** 执行时机: always/finally/on_failure/on_success(缺省 always) */
run_on?: 'always' | 'finally' | 'on_failure' | 'on_success'
} }
/** 测试用例接口 */ /** 测试用例接口 */
...@@ -58,6 +102,8 @@ export interface TestCase { ...@@ -58,6 +102,8 @@ export interface TestCase {
config: Record<string, any> config: Record<string, any>
/** 用例类型 ui/api/security/deploy/performance/functional */ /** 用例类型 ui/api/security/deploy/performance/functional */
caseType?: string caseType?: string
/** 用例参数(上下文变量初始值,支持 {__CTX:key__} 引用) */
parameters?: Record<string, any>
/** 创建时间 */ /** 创建时间 */
createdAt: string createdAt: string
/** 更新时间 */ /** 更新时间 */
...@@ -84,6 +130,8 @@ export interface CreateCaseRequest { ...@@ -84,6 +130,8 @@ export interface CreateCaseRequest {
config?: Record<string, any> config?: Record<string, any>
/** 用例类型 ui/api/security/deploy/performance/functional */ /** 用例类型 ui/api/security/deploy/performance/functional */
case_type?: string case_type?: string
/** 用例参数(上下文变量初始值) */
parameters?: Record<string, any>
} }
/** 用例列表响应 */ /** 用例列表响应 */
......
此差异已折叠。
此差异已折叠。
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论