提交 3a29883b authored 作者: 陈泽健's avatar 陈泽健

feat(smart-locate): P1/P2 智能定位统一执行路径 + micro-app 递归穿透与状态确认

- playwright_executor 新增 execute_action 统一动作分发层(navigate/click/fill/select/
  hover/wait/scroll/press/check/uncheck/assert/screenshot/api_call),供 SmartLocateService
  复用,保证智能定位路径与正式执行路径行为一致
- micro-app evaluator 重写:递归穿透嵌套 micro-app 与 shadowRoot,支持 :has-text/
  :visible/:nth-match;新增 check/uncheck 状态确认防假通过;fill 返回值确认
- smart_locate_service 统一执行路径(execute_action + verify_step),字段透传
- router /verify-step 真实验证:非法动作 400、业务失败返回 success=False
- 前端 Cases.vue/elementLocate.ts 步骤转换增加字段透传(verify/expected/semantic)
- 新增 test_micro_app_adapter / test_verify_step 单元与契约测试
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 eba91eac
...@@ -744,48 +744,50 @@ class PlaywrightExecutor: ...@@ -744,48 +744,50 @@ class PlaywrightExecutor:
# micro-app-body 内部直接查询和操作。 # micro-app-body 内部直接查询和操作。
_MICRO_APP_JS = """(args) => { _MICRO_APP_JS = """(args) => {
const {cssSel, hasText, action, value, visibleOnly} = args; const {cssSel, hasText, nth, action, value, visibleOnly} = args;
// 文本归一化:\xa0(&nbsp;)等不可见空白在 innerText/textContent 中常见,
// 与用例中的普通空格对比时需归一化后再匹配
const norm = s => (s || '').replace(/[\\s\\u00a0]+/g, ' '); const norm = s => (s || '').replace(/[\\s\\u00a0]+/g, ' ');
// 查询所有 micro-app-body 元素,对其常规 DOM 和 shadowRoot 都进行探测 const visible = el => !visibleOnly || el.offsetWidth > 0 || el.offsetHeight > 0 || el.getClientRects().length > 0;
const bodies = document.querySelectorAll('micro-app micro-app-body'); // Walk every open shadow root. micro-app applications can themselves
// contain micro-apps, so a single querySelectorAll is insufficient.
const roots = [];
const visit = root => {
roots.push(root);
let nodes;
try { nodes = root.querySelectorAll('*'); } catch(e) { return; }
for (const node of nodes) if (node.shadowRoot) visit(node.shadowRoot);
};
visit(document);
const bodies = [];
for (const root of roots) {
let found;
try { found = root.querySelectorAll('micro-app-body'); } catch(e) { found = []; }
for (const body of found) if (!bodies.includes(body)) bodies.push(body);
}
for (const body of bodies) { for (const body of bodies) {
// 关键修复:micro-app-body 可能使用 shadow DOM(@micro-zoe/micro-app), const searchRoots = [body];
// 必须同时检查 body.shadowRoot,否则 querySelectorAll 无法穿透。 if (body.shadowRoot) searchRoots.push(body.shadowRoot);
const roots = [body]; for (const root of searchRoots) {
if (body.shadowRoot) roots.push(body.shadowRoot);
for (const root of roots) {
let elements; let elements;
try { try { elements = cssSel && cssSel !== '*' ? Array.from(root.querySelectorAll(cssSel)) : [root]; }
elements = cssSel && cssSel !== '*' ? root.querySelectorAll(cssSel) : [root]; catch(e) { continue; }
} catch(e) { continue; } elements = elements.filter(el => (!hasText || norm(el.textContent).includes(hasText)) && visible(el));
if (Number.isInteger(nth) && nth >= 0) elements = elements.slice(nth, nth + 1);
if (action === 'count') return elements.length;
for (const el of elements) { for (const el of elements) {
if (hasText && !norm(el.textContent).includes(hasText)) continue;
if (visibleOnly && !(el.offsetWidth > 0 || el.offsetHeight > 0 || el.getClientRects().length > 0)) continue;
if (action === 'exists') return true; if (action === 'exists') return true;
if (action === 'text') return el.textContent || ''; if (action === 'text') return el.textContent || '';
if (action === 'click') { if (action === 'click') { el.scrollIntoView({block: 'center'}); el.click(); return true; }
el.scrollIntoView({block: 'center'});
el.click();
return true;
}
if (action === 'fill') { if (action === 'fill') {
el.focus(); el.focus(); el.value = value;
el.value = value;
el.dispatchEvent(new Event('input', {bubbles: true})); el.dispatchEvent(new Event('input', {bubbles: true}));
el.dispatchEvent(new Event('change', {bubbles: true})); el.dispatchEvent(new Event('change', {bubbles: true}));
return true; return el.value === String(value);
} }
if (action === 'count') { if (action === 'check' || action === 'uncheck') {
// count 单独处理:返回总数量而非 bool if (typeof el.click !== 'function') continue;
let cnt = 0; const before = !!el.checked;
for (const e of elements) { if (before !== (action === 'check')) el.click();
if (hasText && !norm(e.textContent).includes(hasText)) continue; return !!el.checked === (action === 'check');
if (visibleOnly && !(e.offsetWidth > 0 || e.offsetHeight > 0 || e.getClientRects().length > 0)) continue;
cnt++;
}
return cnt;
} }
} }
} }
...@@ -797,24 +799,32 @@ class PlaywrightExecutor: ...@@ -797,24 +799,32 @@ class PlaywrightExecutor:
def _split_micro_app_selector(selector: str): def _split_micro_app_selector(selector: str):
""" """
拆分选择器:剥离 Playwright 伪选择器(:has-text 等), 拆分选择器:剥离 Playwright 伪选择器(:has-text 等),
返回 (标准 CSS 选择器, 文本匹配条件)。XPath 选择器返回 None 表示不支持。 返回 (标准 CSS 选择器, 文本匹配条件, 第 N 个匹配条件)。
XPath 选择器返回 (None, "", None) 表示不支持 micro-app 回退。
Args: Args:
selector (str): 原始选择器 selector (str): 原始选择器
Returns: Returns:
tuple: (css_selector or None, has_text str) tuple: (css_selector or None, has_text str, nth int or None)
""" """
# XPath 不走 micro-app 回退 # XPath 不走 micro-app 回退
if selector.startswith('/') or selector.startswith('('): if selector.startswith('/') or selector.startswith('('):
return None, "" return None, "", None
text_match = re.search(r':has-text\((["\'])(.*?)\1\)', selector) text_match = re.search(r':has-text\((["\'])(.*?)\1\)', selector)
has_text = text_match.group(2) if text_match else "" has_text = text_match.group(2) if text_match else ""
css_sel = re.sub(r':has-text\((["\']).*?\1\)', '', selector) css_sel = re.sub(r':has-text\((["\']).*?\1\)', '', selector)
css_sel = re.sub(r':visible\b', '', css_sel).strip() css_sel = re.sub(r':visible\b', '', css_sel).strip()
# :nth-match(N) 条件 → 0-based 数字下标,交给 evaluator 切片
nth = None
nth_match = re.search(r':nth-match\(\s*([0-9]+)\s*\)', css_sel)
if nth_match:
nth = int(nth_match.group(1)) - 1 # 1-based → 0-based
css_sel = re.sub(r':nth-match\(\s*[0-9]+\s*\)', '', css_sel)
if css_sel == '': if css_sel == '':
css_sel = '*' css_sel = '*'
return css_sel, has_text return css_sel, has_text, nth
def _do_in_micro_apps( def _do_in_micro_apps(
self, self,
...@@ -825,19 +835,22 @@ class PlaywrightExecutor: ...@@ -825,19 +835,22 @@ class PlaywrightExecutor:
visible_only: bool = False, visible_only: bool = False,
): ):
""" """
在 <micro-app micro-app-body> 自定义元素内执行查找/取文本/点击/填充 在 <micro-app micro-app-body> 自定义元素内执行查找/取文本/点击/填充/勾选
覆盖嵌套 micro-app 与嵌套 shadowRoot(evaluator 内部递归穿透),
支持 :has-text() 伪选择器与 nth 条件。
Args: Args:
action (str): exists / text / click / fill action (str): exists / text / click / fill / check / uncheck / count
selector (str): 原始选择器(支持 :has-text() 伪选择器) selector (str): 原始选择器(支持 :has-text() 伪选择器)
value (str): fill 时的填充值 value (str): fill 时的填充值
timeout (int): 轮询超时毫秒 timeout (int): 轮询超时毫秒
visible_only (bool): 是否仅匹配可见元素 visible_only (bool): 是否仅匹配可见元素
Returns: Returns:
exists/click/fill: bool;text: str 或 None exists/click/fill/check/uncheck: bool;text: str 或 None;count: int
""" """
css_sel, has_text = self._split_micro_app_selector(selector) css_sel, has_text, nth = self._split_micro_app_selector(selector)
if css_sel is None: if css_sel is None:
return False if action != 'text' else None return False if action != 'text' else None
...@@ -846,7 +859,7 @@ class PlaywrightExecutor: ...@@ -846,7 +859,7 @@ class PlaywrightExecutor:
try: try:
result = self._page.evaluate( result = self._page.evaluate(
self._MICRO_APP_JS, self._MICRO_APP_JS,
{"cssSel": css_sel, "hasText": has_text, "action": action, {"cssSel": css_sel, "hasText": has_text, "nth": nth, "action": action,
"value": str(value), "visibleOnly": visible_only}, "value": str(value), "visibleOnly": visible_only},
) )
if result is not False and result is not None: if result is not False and result is not None:
...@@ -1242,6 +1255,163 @@ class PlaywrightExecutor: ...@@ -1242,6 +1255,163 @@ class PlaywrightExecutor:
result.error_message = (result.error_message or "") + f";清理步骤异常: {e}" result.error_message = (result.error_message or "") + f";清理步骤异常: {e}"
logger.warning(f"[清理] 后置步骤异常: {step.get('name', '')} → {e}") logger.warning(f"[清理] 后置步骤异常: {step.get('name', '')} → {e}")
def execute_action(
self,
action: str,
params: Dict[str, Any],
step: Optional[Dict[str, Any]] = None,
expected: str = "",
) -> StepResult:
"""
P1b: 统一动作执行分发层。
把 execute_step 内部的动作分发(navigate/click/fill/select/hover/wait/scroll/press/
check/uncheck/assert/screenshot/api_call)抽成独立公开方法,供 SmartLocateService
的 _execute_step 复用,保证智能定位路径与正式执行路径行为一致。
Args:
action (str): 动作类型
params (dict): 动作参数(已渲染或原始)
step (Optional[dict]): 完整步骤定义(含 semantic/verify/expected 顶层字段)
expected (str): 断言/校验的预期值
Returns:
StepResult: 步骤执行结果(status='passed' 表示成功)
"""
step = step or {}
result = StepResult(
order=step.get("order", 0),
name=step.get("name", ""),
action=action,
status="passed",
start_time=datetime.now(),
)
params = params or {}
force = step.get("force", params.get("force", False))
try:
if action == "navigate":
url = params.get("url", "")
self._do_navigate(params)
if self._is_logged_in and "login" in url.lower():
self._is_logged_in = False
if not self._is_logged_in:
self._detect_login_state()
result.log = f"✓ 导航到: {url}"
elif action == "click":
self._do_click(params, force=force)
if self._is_logged_in:
self._detect_logout()
if not self._is_logged_in:
self._detect_login_state()
result.log = f"✓ 点击元素: {params.get('selector', '') or '多选择器'}"
elif action == "fill":
self._do_fill(params, force=force)
result.log = f"✓ 填充输入框: {params.get('selector', '') or '多选择器'} = {params.get('value', '')}"
elif action == "select":
self._do_select(params, force=force)
result.log = f"✓ 选择下拉框: {params.get('selector', '') or '多选择器'} = {params.get('value', '')}"
elif action == "hover":
self._do_hover(params)
result.log = f"✓ 悬停元素: {params.get('selector', '') or '多选择器'}"
elif action == "wait":
self._do_wait(params)
result.log = f"✓ 等待元素: {params.get('selector', '') or '多选择器'}"
elif action == "scroll":
self._do_scroll(params)
result.log = f"✓ 滚动页面: {params.get('direction', 'down')}"
elif action == "press":
self._do_press(params)
result.log = f"✓ 按键: {params.get('key', '')}"
elif action == "check":
self._do_check(params, check=True, force=force)
result.log = f"✓ 勾选: {params.get('selector', '') or '多选择器'}"
elif action == "uncheck":
self._do_check(params, check=False, force=force)
result.log = f"✓ 取消勾选: {params.get('selector', '') or '多选择器'}"
elif action == "assert":
self._do_assert(params, expected)
result.log = f"✓ 断言验证: {params.get('type', 'visible')} - 通过"
elif action == "screenshot":
self._do_screenshot(params)
result.log = "✓ 截图保存"
elif action == "api_call":
self._do_api_call(params)
result.log = f"✓ 接口调用: {params.get('api_case', {}).get('name', '') or 'api_call'}"
else:
raise ValueError(f"不支持的动作类型: {action}")
# 状态验证(防假通过):verify 配置翻译成 _do_assert
if step.get("verify"):
try:
fallback_sel = params.get("selector")
self._verify_step_state(step, params, fallback_sel)
except AssertionError as ve:
result.status = "failed"
result.error = str(ve)
result.log = f"✗ 状态验证失败: {str(ve)}"
except Exception as e:
result.status = "failed"
result.error = str(e)
result.log = f"✗ 执行失败: {str(e)}"
finally:
result.end_time = datetime.now()
if result.start_time:
result.duration = (result.end_time - result.start_time).total_seconds()
return result
def verify_step(
self,
selector: str,
action: str,
value: Optional[str] = None,
page_url: Optional[str] = None,
auto_login: bool = False,
) -> bool:
"""启动一次性执行器并真实验证单个选择器步骤。
该方法供调试 API 使用:验证动作本身,而不是仅检查选择器字符串是否存在。
生命周期在本方法内闭合,确保浏览器资源即使验证失败也会释放。
"""
if not selector or not selector.strip():
raise ValueError("选择器不能为空")
supported = {
"click", "fill", "select", "hover", "wait", "scroll", "press",
"check", "uncheck", "assert", "element_exists", "visible", "text",
}
if action not in supported:
raise ValueError(f"不支持的验证动作: {action}")
if action == "fill" and value is None:
raise ValueError("fill 动作必须提供 value")
try:
self.start()
if auto_login:
if not self.do_login():
raise RuntimeError("自动登录失败")
target_url = page_url or self.base_url
if target_url:
self._do_navigate({"url": target_url})
params: Dict[str, Any] = {"selector": selector}
if value is not None:
params["value"] = value
if action in ("assert", "text"):
params["type"] = "contains" if action == "text" else "visible"
if action in ("element_exists", "visible"):
params["type"] = "element_exists" if action == "element_exists" else "visible"
result = self.execute_action(
action if action not in ("element_exists", "visible", "text") else "assert",
params,
expected=value or "",
)
if result.status != "passed":
raise RuntimeError(result.error or f"步骤验证失败: {action}")
return True
finally:
# stop() 本身对未完全启动的实例安全,保证启动异常/验证异常均释放资源。
self.stop()
def execute_step( def execute_step(
self, self,
step: Dict[str, Any], step: Dict[str, Any],
...@@ -1995,6 +2165,13 @@ class PlaywrightExecutor: ...@@ -1995,6 +2165,13 @@ class PlaywrightExecutor:
if not marker: if not marker:
return return
def _dump_state(tag: str) -> str:
"""dump 当前 URL + micro-app-body 文本摘要,用于诊断日志。"""
cur = self._page.url
body = self._page.query_selector("micro-app micro-app-body")
body_text = (body.inner_text() or "")[:80] if body else "NO-BODY"
return f"[{tag}] URL={cur[:120]!r} body={body_text!r}"
def _ready() -> bool: def _ready() -> bool:
cur = self._page.url cur = self._page.url
body = self._page.query_selector("micro-app micro-app-body") body = self._page.query_selector("micro-app micro-app-body")
...@@ -2693,7 +2870,7 @@ class PlaywrightExecutor: ...@@ -2693,7 +2870,7 @@ class PlaywrightExecutor:
def _do_check(self, params: dict, check: bool = True, force: bool = False) -> None: def _do_check(self, params: dict, check: bool = True, force: bool = False) -> None:
""" """
勾选/取消勾选(多选择器回退 + force) 勾选/取消勾选(多选择器回退 + force + iframe + micro-app 状态确认防假通过
Args: Args:
params (dict): {"selector"|"selectors": str|[str]} params (dict): {"selector"|"selectors": str|[str]}
...@@ -2706,6 +2883,7 @@ class PlaywrightExecutor: ...@@ -2706,6 +2883,7 @@ class PlaywrightExecutor:
last_error: Optional[Exception] = None last_error: Optional[Exception] = None
for selector in selectors: for selector in selectors:
# 主页面 / iframe
try: try:
self._wait_for_element(selector, timeout=10000, state="visible") self._wait_for_element(selector, timeout=10000, state="visible")
if check: if check:
...@@ -2717,7 +2895,21 @@ class PlaywrightExecutor: ...@@ -2717,7 +2895,21 @@ class PlaywrightExecutor:
except Exception as e: except Exception as e:
last_error = e last_error = e
logger.debug(f"⚠ {'勾选' if check else '取消勾选'}失败 {selector}: {e}") logger.debug(f"⚠ {'勾选' if check else '取消勾选'}失败 {selector}: {e}")
continue
# micro-app 穿透(P2):evaluator 内部先读状态,不符才点击,
# 返回 bool 表示「最终状态 == 目标状态」,防止 JS click 假通过
try:
if self._do_in_micro_apps(
'check' if check else 'uncheck',
selector,
timeout=5000,
visible_only=not force,
):
logger.debug(f"✓ micro-app 中{'勾选' if check else '取消勾选'}成功(状态已确认): {selector}")
return
except Exception as e_ma:
last_error = e_ma
logger.debug(f"⚠ micro-app 中{'勾选' if check else '取消勾选'}失败 {selector}: {e_ma}")
raise Exception(f"✗ 无法勾选/取消勾选(已尝试 {len(selectors)} 个选择器): {selectors} 最后错误: {last_error}") raise Exception(f"✗ 无法勾选/取消勾选(已尝试 {len(selectors)} 个选择器): {selectors} 最后错误: {last_error}")
......
...@@ -9,15 +9,17 @@ ...@@ -9,15 +9,17 @@
最后修改:2026-08-05 最后修改:2026-08-05
""" """
import logging
import asyncio import asyncio
from typing import List, Optional, Dict, Any import logging
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime from datetime import datetime
from typing import List, Optional, Dict, Any
from fastapi import APIRouter, HTTPException, BackgroundTasks from fastapi import APIRouter, HTTPException, BackgroundTasks
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from app.services.smart_locate_service import SmartLocateService from app.services.smart_locate_service import SmartLocateService
from app.executors.playwright_executor import PlaywrightExecutor
from app.config import settings from app.config import settings
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
...@@ -33,6 +35,14 @@ class SmartLocateStep(BaseModel): ...@@ -33,6 +35,14 @@ class SmartLocateStep(BaseModel):
name: str = Field(..., min_length=1, description="步骤名称") name: str = Field(..., min_length=1, description="步骤名称")
action: str = Field(..., description="动作类型") action: str = Field(..., description="动作类型")
params: Dict[str, Any] = Field(default_factory=dict, description="动作参数") params: Dict[str, Any] = Field(default_factory=dict, description="动作参数")
# P1a:数据保真——round-trip 逐字段透传,避免前端→后端丢字段
semantic: Optional[Dict[str, Any]] = Field(None, description="语义目标(回传时保留)")
verify: Optional[Dict[str, Any]] = Field(None, description="状态验证(回传时保留)")
run_on: Optional[str] = Field(None, description="执行环境标识(回传时保留)")
selectors: Optional[Dict[str, Any]] = Field(None, description="选择器链(回传时保留)")
page_key: Optional[str] = Field(None, description="页面键(回传时保留)")
element_key: Optional[str] = Field(None, description="元素键(回传时保留)")
expected: Optional[Any] = Field(None, description="预期值(回传时保留)")
class SmartLocateRequest(BaseModel): class SmartLocateRequest(BaseModel):
...@@ -131,15 +141,21 @@ async def smart_locate( ...@@ -131,15 +141,21 @@ async def smart_locate(
logger.info(f"收到智能定位请求: {len(request.steps)} 个步骤") logger.info(f"收到智能定位请求: {len(request.steps)} 个步骤")
try: try:
# 转换步骤格式 # 转换步骤格式(逐字段透传,防止 round-trip 丢字段)
steps_data = [] steps_data = []
for step in request.steps: for step in request.steps:
steps_data.append({ step_dict = {
'order': step.order, 'order': step.order,
'name': step.name, 'name': step.name,
'action': step.action, 'action': step.action,
'params': step.params 'params': step.params,
}) }
# P1a: 透传可选字段,确保路由层不丢弃
for field in ('semantic', 'verify', 'run_on', 'selectors', 'page_key', 'element_key', 'expected'):
val = getattr(step, field, None)
if val is not None:
step_dict[field] = val
steps_data.append(step_dict)
# 在线程池中执行(避免 Playwright 同步 API 在 asyncio 循环中的问题) # 在线程池中执行(避免 Playwright 同步 API 在 asyncio 循环中的问题)
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
...@@ -188,37 +204,75 @@ async def smart_locate( ...@@ -188,37 +204,75 @@ async def smart_locate(
"/verify-step", "/verify-step",
response_model=VerifyStepResponse, response_model=VerifyStepResponse,
summary="单步骤验证(调试用)", summary="单步骤验证(调试用)",
description="验证选择器是否有效" description="启动浏览器真实验证单个选择器步骤"
) )
async def verify_step(request: VerifyStepRequest): async def verify_step(request: VerifyStepRequest):
""" """
单步骤验证(调试用) 单步骤验证(调试用)
用于验证单个选择器是否能在页面上正常工作。 启动一次性 Playwright 执行器,真实验证选择器在页面上的可操作性。
在专用新线程中执行(避免复用 asyncio 线程池导致 greenlet 卡死)。
Args: Args:
request (VerifyStepRequest): 验证请求 request (VerifyStepRequest): 验证请求
Returns: Returns:
VerifyStepResponse: 验证结果 VerifyStepResponse: 验证结果(success + 安全化错误信息)
""" """
logger.info(f"收到单步骤验证请求: selector={request.selector}, action={request.action}") logger.info(f"收到单步骤验证请求: selector={request.selector}, action={request.action}, "
f"auto_login={request.auto_login}")
# 动作/参数校验(4xx)
supported = {
"click", "fill", "select", "hover", "wait", "scroll", "press",
"check", "uncheck", "assert", "element_exists", "visible", "text",
}
if request.action not in supported:
raise HTTPException(
status_code=400,
detail=f"不支持的验证动作: {request.action},支持: {sorted(supported)}"
)
if request.action == "fill" and request.value is None:
raise HTTPException(status_code=400, detail="fill 动作必须提供 value")
if not request.selector or not request.selector.strip():
raise HTTPException(status_code=400, detail="选择器不能为空")
def _verify_sync() -> bool:
"""在专用线程中执行验证,返回是否通过"""
executor = PlaywrightExecutor({
'headless': settings.PLAYWRIGHT_HEADLESS,
'timeout': 30000,
'screenshot': False,
'screenshot_dir': settings.SCREENSHOT_DIR,
'ignore_https_errors': True,
'base_url': request.page_url or settings.TARGET_URL,
})
return executor.verify_step(
selector=request.selector,
action=request.action,
value=request.value,
page_url=request.page_url,
auto_login=request.auto_login,
)
loop = asyncio.get_event_loop()
try: try:
# 这里简化实现:直接返回成功 # 专用新线程池(max_workers=1,不复用 asyncio 线程池 worker),
# 实际可以启动 Playwright 执行验证 # 与 execution_service 的专用 Playwright 线程模式一致,避免 greenlet 卡死。
# 但为了性能,通常在智能定位时已包含验证 with ThreadPoolExecutor(max_workers=1, thread_name_prefix="verify-step") as pool:
ok = await loop.run_in_executor(pool, _verify_sync)
return VerifyStepResponse( return VerifyStepResponse(
success=True, success=ok,
message=f"选择器验证通过: {request.selector}" message=f"选择器验证{'通过' if ok else '失败'}: {request.selector}"
) )
except HTTPException:
raise
except Exception as e: except Exception as e:
logger.error(f"单步骤验证异常: {e}") logger.error(f"单步骤验证异常: {e}")
raise HTTPException( # 业务失败(选择器无效/元素不存在/动作执行失败)返回 success=False,而非 500
status_code=500, return VerifyStepResponse(
detail=f"验证失败: {str(e)}" success=False,
message=f"验证失败: {str(e)[:200]}"
) )
......
...@@ -127,6 +127,8 @@ class SmartLocateService: ...@@ -127,6 +127,8 @@ class SmartLocateService:
'screenshot': True, 'screenshot': True,
'screenshot_dir': self.screenshot_dir, 'screenshot_dir': self.screenshot_dir,
'ignore_https_errors': True, 'ignore_https_errors': True,
# P1a: 传 base_url,与正式执行路径对齐(URL rebase / micro-app warm-up 依赖)
'base_url': page_url,
}) })
self.executor.start() self.executor.start()
...@@ -738,6 +740,7 @@ class SmartLocateService: ...@@ -738,6 +740,7 @@ class SmartLocateService:
] ]
} }
result['params']['selector'] = chain[0] result['params']['selector'] = chain[0]
result['params']['selectors'] = list(chain) # P1b: 完整语义链,供执行期多选器回退
result['semantic'] = sem_result['semantic'] result['semantic'] = sem_result['semantic']
result['message'] = f'语义解析优先定位成功: {chain[0]}' result['message'] = f'语义解析优先定位成功: {chain[0]}'
logger.info(f"步骤 {order} 语义解析优先命中: {result['semantic']}") logger.info(f"步骤 {order} 语义解析优先命中: {result['semantic']}")
...@@ -957,198 +960,52 @@ class SmartLocateService: ...@@ -957,198 +960,52 @@ class SmartLocateService:
def _execute_step(self, result: Dict[str, Any]) -> bool: def _execute_step(self, result: Dict[str, Any]) -> bool:
""" """
实际执行步骤(验证选择器是否有效) 实际执行步骤(验证选择器是否有效)
如果点击后预期弹窗未出现,自动尝试替代选择器
Args: P1b: 委托 PlaywrightExecutor 的统一执行路径(execute_action),
result (Dict): 定位结果(包含 selector 和 params) 覆盖全部动作类型(fill/click/select/hover/wait/scroll/press/check/uncheck/assert 等),
复用执行期完整的 iframe / micro-app / force / JS-click 回退策略,
Returns: 与正式执行路径行为一致,消除"智能定位时假通过、正式执行时失败"的漂移。
bool: 是否执行成功
""" """
action = result.get('action', '') action = result.get('action', '')
params = result.get('params', {}) params = result.get('params', {})
selector = params.get('selector', '') selector = params.get('selector', '')
step_name = result.get('name', '') step_name = result.get('name', '') or ''
order = result.get('order', 0) order = result.get('order', 0)
if not selector: if not selector and action not in ('scroll', 'press', 'wait', 'screenshot', 'api_call', 'assert'):
return False return False
logger.debug(f"执行步骤验证: action={action}, selector={selector}") logger.debug(f"执行步骤验证(P1b 统一路径): action={action}, selector={selector}")
try: try:
page = self.executor._page # 组装最小步骤定义供 execute_action 消费(含语义链 selectors 候选)
step: Dict[str, Any] = {
if action == 'fill': 'name': step_name,
value = params.get('value', '') 'action': action,
page.wait_for_selector(selector, timeout=5000) 'params': params,
page.fill(selector, str(value)) }
logger.debug(f"填充成功: {selector} = {value}") if result.get('semantic'):
return True step['semantic'] = result['semantic']
if result.get('verify'):
elif action == 'click': step['verify'] = result['verify']
# 判断点击后是否预期打开弹窗/抽屉 sr = self.executor.execute_action(action, params, step=step)
dialog_keywords = ['新建', '编辑', '添加', '新增', '创建', '修改', '详情', if sr and sr.status == 'passed':
'确定创建', '取消会议', '完成'] # 勾选/取消勾选防假通过:checkbox 状态校验(覆盖旧 _execute_step 未含的 check/uncheck)
expects_dialog = any(kw in step_name for kw in dialog_keywords) if action in ('check', 'uncheck'):
# 构建替代选择器列表(优先级从高到低):
# 主选择器 + 定位结果中的完整候选列表(pattern/语义路径会产出多个候选,
# 仅尝试主选择器会漏掉真正可用的候补,如列表行的 tr:has-text 兜底)
click_selector_chain = [selector]
for cand in result.get('selectors', {}).get('candidates', []):
v = cand.get('value') if isinstance(cand, dict) else cand
if v and v != selector and v not in click_selector_chain:
click_selector_chain.append(str(v))
# 如果当前选择器是 div:has-text 格式,尝试补充 button:has-text
if 'div:visible:has-text' in selector or 'div:has-text' in selector:
import re
text_match = re.search(r'has-text\("([^"]+)"\)', selector)
if text_match:
btn_text = text_match.group(1)
click_selector_chain.append(f'button:has-text("{btn_text}")')
click_selector_chain.append(f'[role="button"]:has-text("{btn_text}")')
click_selector_chain.append(f'.el-button:has-text("{btn_text}")')
# 如果当前选择器是 input:visible(错误匹配),尝试从步骤名提取关键词
if 'input:visible' in selector:
click_selector_chain.append(f'button:has-text("{step_name}")')
# 点击前等待目标文本渲染:列表/表格异步加载兜底
# (旧 case 已有选择器如 .el-checkbox:has-text 常匹配 0 个;若表格数据仍在异步加载,
# 直接点击/重定位都会因"目标行暂不存在"失败——先等 has-text 目标文本出现,
# 确保后续点击链与 Fix5 重定位时表格已就绪)
try:
_tm = re.search(r'has-text\("([^"]+)"\)', selector)
if _tm:
# 用 :has-text 子串匹配更宽容:TD 内文本可能带空白/嵌套,
# 精确 text= 匹配常漏掉导致"表格已加载但等待超时"
# 目标系统(192.168.5.44)表格异步加载极慢且不稳,放宽到 25s
page.wait_for_selector(f':has-text("{_tm.group(1)}")', timeout=25000)
page.wait_for_timeout(300)
logger.debug(f"步骤 {order} 目标文本已渲染: {_tm.group(1)}")
except Exception:
pass
# 逐级尝试点击(每级:普通点击 → force 强点;缩短超时避免 30s 默认卡死)
# 记录点击前 URL,用于判断是否为整页路由跳转(跳转则无需等待弹窗)
url_before = ''
try:
url_before = page.url
except Exception:
pass
click_success = False
actual_selector_used = None
for sel in click_selector_chain:
try:
page.wait_for_selector(sel, timeout=5000)
page.click(sel, timeout=5000)
page.wait_for_timeout(800)
logger.debug(f"点击成功: {sel}")
click_success = True
actual_selector_used = sel
break
except Exception as e:
# 普通点击失败(元素被遮挡/滚动区/等),重试 force 点击
# (对齐示范用例"勾选会议室 force=True"的经验:列表行勾选常需强点)
try:
page.click(sel, force=True, timeout=5000)
page.wait_for_timeout(800)
logger.debug(f"点击成功(force): {sel}")
click_success = True
actual_selector_used = sel
break
except Exception as e2:
# JS 直接触发 click(绕过 playwright actionability 检测):
# Element Plus 表格 checkbox 等元素普通/force 点击常超时
# (实测 .el-checkbox:has-text 匹配 0 个、JS el.click() 可成功勾选),
# 作为最后兜底(取第一个匹配)
try:
page.locator(sel).first.evaluate('el => el.click()')
page.wait_for_timeout(800)
logger.debug(f"点击成功(js): {sel}")
click_success = True
actual_selector_used = sel
break
except Exception as e3:
logger.debug(f"选择器 {sel} 点击失败(普通+force+js): {e3}")
continue
if not click_success:
logger.warning(f"步骤 {order} 所有选择器点击均失败")
return False
# 更新 params 中的 selector 为实际使用的选择器
if actual_selector_used != selector:
result['params']['selector'] = actual_selector_used
logger.info(f"步骤 {order} 选择器已更新: {selector} -> {actual_selector_used}")
# === 页面状态验证 ===
# 1. 检测是否是功能中心点击
if '功能中心' in step_name:
logger.debug("检测到功能中心点击,等待抽屉打开...")
try: try:
page.wait_for_selector('.el-drawer', timeout=8000) el = self.executor._page.query_selector(selector)
page.wait_for_timeout(500) if el is not None and self.executor._checkbox_is_checked(el) != (action == 'check'):
logger.debug("功能中心抽屉已打开") logger.warning(f"步骤 {order} 勾选状态与目标不一致(防假通过)")
return False
except Exception as e: except Exception as e:
logger.warning(f"功能中心抽屉未打开,标记为失败: {e}") logger.debug(f"勾选状态校验异常(非致命): {e}")
return False
# 2. 检测弹窗是否出现(如果步骤名称包含弹窗关键词)
if expects_dialog:
# 整页路由跳转豁免:步骤名含"新建/创建"等弹窗关键词,但实际点击是页面跳转
# (如 新建会议 → /meetingV3 整页路由,无 .el-dialog/.el-drawer),URL 变化视为成功
url_after = ''
try:
url_after = page.url
except Exception:
pass
url_navigated = bool(url_before and url_after and url_before != url_after)
if url_navigated:
logger.info(f"步骤 '{step_name}' 点击后 URL 变化(页面跳转),豁免弹窗检查: {url_before} -> {url_after}")
return True
dialog_opened = False
for dialog_sel in ['.el-dialog', '.el-drawer', '.el-message-box']:
try:
page.wait_for_selector(dialog_sel, timeout=3000)
dialog_opened = True
logger.debug(f"弹窗已打开: {dialog_sel}")
break
except Exception:
continue
if not dialog_opened:
logger.warning(f"步骤 '{step_name}' 点击后未检测到弹窗,所有选择器均失败")
# 重要:弹窗未打开,返回 False 触发 skip_remaining
return False
# 3. 检测页面是否跳转(如果步骤名称包含导航关键词)
nav_keywords = ['进入', '打开', '跳转', '导航']
if any(kw in step_name for kw in nav_keywords):
page.wait_for_timeout(1000)
return True return True
elif action == 'wait': logger.debug(f"execute_action 未通过: action={action}, status={sr.status if sr else 'None'}, "
timeout = params.get('timeout', 10000) f"error={sr.error if sr else '?'}")
page.wait_for_selector(selector, timeout=timeout) return False
logger.debug(f"等待成功: {selector}")
return True
elif action == 'select':
value = params.get('value', '')
page.wait_for_selector(selector, timeout=5000)
page.select_option(selector, str(value))
logger.debug(f"选择成功: {selector} = {value}")
return True
return True
except Exception as e: except Exception as e:
logger.warning(f"执行步骤验证失败: {e}") logger.warning(f"执行步骤验证失败(P1b 统一路径): {e}")
return False return False
# ==================== 阶段6升级:语义化用例生成 ==================== # ==================== 阶段6升级:语义化用例生成 ====================
...@@ -1161,6 +1018,8 @@ class SmartLocateService: ...@@ -1161,6 +1018,8 @@ class SmartLocateService:
def _semantic_enabled(self) -> bool: def _semantic_enabled(self) -> bool:
"""语义化生成器是否可用(懒加载,避免 import 循环;依赖缺失降级为旧路径)""" """语义化生成器是否可用(懒加载,避免 import 循环;依赖缺失降级为旧路径)"""
if not settings.SEMANTIC_ENABLED:
return False
if self._semantic_enabled_flag is None: if self._semantic_enabled_flag is None:
try: try:
from app.services.semantic_resolver import get_semantic_resolver from app.services.semantic_resolver import get_semantic_resolver
...@@ -1281,6 +1140,11 @@ class SmartLocateService: ...@@ -1281,6 +1140,11 @@ class SmartLocateService:
result['selectors']['primary'] = chain[0] result['selectors']['primary'] = chain[0]
result['selectors']['candidates'] = new_cands result['selectors']['candidates'] = new_cands
result['params']['selector'] = chain[0] result['params']['selector'] = chain[0]
# P1b: 执行期多选器回退链(语义链在前,原 selector 兜底)
full_chain = list(chain)
if existing_sel and existing_sel not in full_chain:
full_chain.append(existing_sel)
result['params']['selectors'] = full_chain
result['semantic'] = semantic result['semantic'] = semantic
result['message'] = f'使用已有选择器 + 语义化升级: {result["params"].get("selector", "")}' result['message'] = f'使用已有选择器 + 语义化升级: {result["params"].get("selector", "")}'
...@@ -1429,6 +1293,9 @@ class SmartLocateService: ...@@ -1429,6 +1293,9 @@ class SmartLocateService:
语义解析优先:构造候选 semantic → resolve_step 命中选择器链 语义解析优先:构造候选 semantic → resolve_step 命中选择器链
实际执行验证由 locate_steps 的统一 _execute_step 完成(与 keyword 路径一致) 实际执行验证由 locate_steps 的统一 _execute_step 完成(与 keyword 路径一致)
成功 = 该 semantic 目标已通过执行期解析器验证,落库后执行必稳 成功 = 该 semantic 目标已通过执行期解析器验证,落库后执行必稳
P1b: 链中任一候选命中即通过(不再只校验 chain[0]);DOM 类型细化后重新解析
选择器链并保存最终一致链(避免落库 semantic 与验证链漂移)。
""" """
if not self._semantic_enabled(): if not self._semantic_enabled():
return None return None
...@@ -1446,26 +1313,44 @@ class SmartLocateService: ...@@ -1446,26 +1313,44 @@ class SmartLocateService:
candidate_step = dict(step) candidate_step = dict(step)
candidate_step['semantic'] = semantic candidate_step['semantic'] = semantic
chain = self._semantic_resolver.resolve_step(page, candidate_step, scope, ctx) or [] chain = self._semantic_resolver.resolve_step(page, candidate_step, scope, ctx) or []
chain = [sel for sel in chain if sel]
if not chain: if not chain:
return None return None
# 关键保证:链首选择器必须真实命中页面元素,否则该 semantic 未经执行期验证,
# 关键保证:链中任一选择器必须真实命中页面元素,否则该 semantic 未经执行期验证,
# 直接返回会造成"假命中"(如 type=button 但目标实为 DIV → button:has-text 无元素, # 直接返回会造成"假命中"(如 type=button 但目标实为 DIV → button:has-text 无元素,
# 后续 _execute_step 必失败)。未命中则回退 keyword 路径,由反向生成产出 semantic。 # 后续 _execute_step 必失败)。链首不命中时逐个向后找,命中任一即通过;
try: # 全链无一命中则回退 keyword 路径,由反向生成产出 semantic。
el = page.query_selector(chain[0]) hit_el, hit_idx = None, -1
except Exception as e: for i, sel in enumerate(chain):
logger.debug(f"语义链首校验异常: {e}") try:
el = None el = page.query_selector(sel)
if el is None: except Exception as e:
logger.info(f"步骤 {step.get('order')} 语义链首未命中页面元素,回退 keyword 定位") logger.debug(f"语义链候选校验异常: {e}")
el = None
if el is not None:
hit_el, hit_idx = el, i
break
if hit_el is None:
logger.info(f"步骤 {step.get('order')} 语义链全链未命中页面元素,回退 keyword 定位")
return None return None
# 用解析链定位真实元素,反推细化 semantic.type(结构化元素纠正 button→checkbox/tab 等)
# 用命中的真实元素反推细化 semantic.type(结构化元素纠正 button→checkbox/tab 等)
try: try:
refined_type = self._infer_element_type(step.get('action', ''), el) refined_type = self._infer_element_type(step.get('action', ''), hit_el)
if refined_type != semantic.get('type'): if refined_type != semantic.get('type'):
logger.info(f"步骤 {step.get('order')} 语义类型细化: " logger.info(f"步骤 {step.get('order')} 语义类型细化: "
f"{semantic.get('type')} -> {refined_type}") f"{semantic.get('type')} -> {refined_type}")
semantic['type'] = refined_type semantic['type'] = refined_type
# 类型细化后重新解析选择器链,保证落库 semantic 与验证链一致
candidate_step['semantic'] = semantic
try:
new_chain = self._semantic_resolver.resolve_step(page, candidate_step, scope, ctx) or []
new_chain = [sel for sel in new_chain if sel]
if new_chain:
chain = new_chain
except Exception as e:
logger.debug(f"步骤 {step.get('order')} 类型细化后重新解析失败(保留原链): {e}")
except Exception as e: except Exception as e:
logger.debug(f"语义类型细化失败(非致命): {e}") logger.debug(f"语义类型细化失败(非致命): {e}")
return {'chain': chain, 'semantic': semantic} return {'chain': chain, 'semantic': semantic}
......
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:test_micro_app_adapter.py
模块描述:micro-app 通用定位适配(P2)单元测试
覆盖:
- _split_micro_app_selector::has-text / :visible / :nth-match / XPath 不支持
- _do_in_micro_apps:exists / text / click / fill / check/uncheck / count 各 action
(mock _page.evaluate 模拟 evaluator 返回值,含 fill 值确认 / check 状态确认防假通过)
- _do_check:micro-app 分支在页面路径失败后触发,状态达标/不达标的行为
- 递归 shadowRoot / 嵌套 micro-app 的 evaluator 选择器产出正确
作者:czj
创建日期:2026-08-26
"""
from unittest.mock import MagicMock
import pytest
from app.executors.playwright_executor import PlaywrightExecutor
# ==================== _split_micro_app_selector ====================
class TestSplitMicroAppSelector:
"""选择器解析:剥离 Playwright 伪选择器,识别 nth / unsupported"""
def test_plain_css(self):
css, text, nth = PlaywrightExecutor._split_micro_app_selector(".btn")
assert css == ".btn" and text == "" and nth is None
def test_has_text(self):
css, text, nth = PlaywrightExecutor._split_micro_app_selector(
'button:has-text("登录")')
assert css == "button" and text == "登录" and nth is None
def test_has_text_single_quote(self):
css, text, _ = PlaywrightExecutor._split_micro_app_selector(
"button:has-text('保存')")
assert css == "button" and text == "保存"
def test_has_text_with_parens_in_text(self):
css, text, _ = PlaywrightExecutor._split_micro_app_selector(
'button:has-text("A (测试)")')
assert css == "button" and text == "A (测试)"
def test_visible_stripped(self):
css, text, nth = PlaywrightExecutor._split_micro_app_selector(
"button:visible:has-text(\"确定\")")
assert css == "button" and text == "确定" and nth is None
def test_nth_match(self):
css, text, nth = PlaywrightExecutor._split_micro_app_selector(
'tr:nth-match(2)')
assert css == "tr" and nth == 1 # 1-based → 0-based
def test_xpath_unsupported(self):
css, text, nth = PlaywrightExecutor._split_micro_app_selector("//button")
assert css is None and text == "" and nth is None
def test_empty_becomes_star(self):
css, _, _ = PlaywrightExecutor._split_micro_app_selector(
':has-text("只有文本")')
assert css == "*"
# ==================== _do_in_micro_apps(mock evaluate) ====================
def make_page(evaluate_return=None, evaluate_side_effect=None):
"""构造带可编程 evaluate 的 fake page"""
page = MagicMock()
page.evaluate = MagicMock(
return_value=evaluate_return, side_effect=evaluate_side_effect)
page.wait_for_timeout = MagicMock()
return page
class TestDoInMicroApps:
"""_do_in_micro_apps 各 action 的 Python 侧决策逻辑"""
def _exec(self, page):
ex = PlaywrightExecutor({})
ex._page = page
return ex
def test_exists_true(self):
page = make_page(True)
ex = self._exec(page)
assert ex._do_in_micro_apps('exists', 'button:has-text("保存")', timeout=2000) is True
args = page.evaluate.call_args[0][1]
assert args["cssSel"] == "button" and args["hasText"] == "保存"
def test_exists_false_after_deadline(self):
page = make_page(False)
ex = self._exec(page)
assert ex._do_in_micro_apps('exists', '.missing', timeout=2000) is False
assert page.wait_for_timeout.call_count >= 1 # 轮询等待
def test_text_value(self):
page = make_page("保存")
ex = self._exec(page)
assert ex._do_in_micro_apps('text', 'button', timeout=2000) == "保存"
def test_click_true(self):
page = make_page(True)
ex = self._exec(page)
assert ex._do_in_micro_apps('click', 'button', timeout=2000) is True
def test_fill_value_confirm_true(self):
page = make_page(True)
ex = self._exec(page)
assert ex._do_in_micro_apps('fill', 'input', value="abc", timeout=2000) is True
args = page.evaluate.call_args[0][1]
assert args["action"] == "fill" and args["value"] == "abc"
def test_fill_no_confirm_false_retries(self):
# evaluator 返回 False(value 未确认)→ 轮询直到超时 → False
page = make_page(False)
ex = self._exec(page)
assert ex._do_in_micro_apps('fill', 'input', value="abc", timeout=2000) is False
def test_check_true(self):
page = make_page(True)
ex = self._exec(page)
assert ex._do_in_micro_apps('check', 'input[type=checkbox]', timeout=2000) is True
def test_uncheck_status_not_met_false(self):
# evaluator 返回 False:点击后状态仍未达目标 → 不通过
page = make_page(False)
ex = self._exec(page)
assert ex._do_in_micro_apps('uncheck', 'input[type=checkbox]', timeout=1000) is False
def test_count_value(self):
page = make_page(3)
ex = self._exec(page)
assert ex._do_in_micro_apps('count', 'tr', timeout=2000) == 3
def test_visible_only_passed_through(self):
page = make_page(True)
ex = self._exec(page)
ex._do_in_micro_apps('exists', 'button', timeout=2000, visible_only=True)
assert page.evaluate.call_args[0][1]["visibleOnly"] is True
def test_evaluate_exception_then_success(self):
# 首次 evaluate 抛异常 → 重试 → 第二次成功
page = make_page(evaluate_side_effect=[Exception("boom"), True])
ex = self._exec(page)
assert ex._do_in_micro_apps('exists', 'button', timeout=2000) is True
def test_nested_shadow_roots_selector_sent_to_evaluator(self):
# 递归穿透发生在 evaluator JS 内部,Python 侧只负责透传 css/nth
page = make_page(True)
ex = self._exec(page)
ex._do_in_micro_apps('click', 'micro-app-body', timeout=2000)
args = page.evaluate.call_args[0][1]
assert args["cssSel"] == "micro-app-body"
# ==================== _do_check micro-app 分支 ====================
class TestDoCheckMicroApp:
"""_do_check 在页面路径失败后进入 micro-app 分支,状态达标才成功"""
def _make_check_failing(self, micro_ok):
ex = PlaywrightExecutor({})
ex._page = MagicMock()
# 主页面 check 抛异常(穿透失败)→ 进入 micro-app 分支
ex._page.check = MagicMock(side_effect=Exception("cannot reach"))
ex._page.uncheck = MagicMock(side_effect=Exception("cannot reach"))
ex._do_in_micro_apps = MagicMock(return_value=micro_ok)
# 避免 wait_for_element 在 micro 分支前额外耗时
ex._wait_for_element = MagicMock(return_value=False)
return ex
def test_check_micro_ok(self):
ex = self._make_check_failing(micro_ok=True)
ex._do_check({"selector": ".cb"}, check=True)
ex._do_in_micro_apps.assert_called_once()
assert ex._do_in_micro_apps.call_args[0][0] == "check"
def test_uncheck_micro_ok(self):
ex = self._make_check_failing(micro_ok=True)
ex._do_check({"selector": ".cb"}, check=False)
assert ex._do_in_micro_apps.call_args[0][0] == "uncheck"
def test_check_micro_state_not_met_raises(self):
ex = self._make_check_failing(micro_ok=False)
with pytest.raises(Exception, match="无法勾选"):
ex._do_check({"selector": ".cb"}, check=True)
def test_check_main_page_wins_before_micro(self):
# 主页面 check 成功 → 不进入 micro-app 分支
ex = PlaywrightExecutor({})
ex._page = MagicMock()
ex._page.check = MagicMock()
ex._wait_for_element = MagicMock(return_value=True)
ex._do_in_micro_apps = MagicMock(return_value=True)
ex._do_check({"selector": ".cb"}, check=True)
ex._do_in_micro_apps.assert_not_called()
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:test_verify_step.py
模块描述:/verify-step 真实验证实现契约测试(P1)
覆盖:
- 路由器参数校验(非法 action / fill 缺 value / 空 selector → 4xx)
- mock PlaywrightExecutor:start/verify/stop 生命周期顺序、成功/失败/异常响应
- ASGITransport 集成:POST /api/element/verify-step 不再无条件成功
(非法动作 400;业务失败返回 success=False 而非 500)
作者:czj
创建日期:2026-08-26
"""
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from httpx import AsyncClient
from httpx._transports.asgi import ASGITransport
sys.path.insert(0, str(Path(__file__).parent.parent))
from app.main import app # noqa: E402
import app.routers.smart_locate as smart_locate_router # noqa: E402
# ==================== 单元测试:参数校验(不经线程池,直接调 handler) ====================
class TestVerifyStepValidation:
"""参数校验应返回 4xx(HTTPException),不启动浏览器"""
@pytest.mark.asyncio
async def test_invalid_action_400(self):
from fastapi import HTTPException
req = smart_locate_router.VerifyStepRequest(
selector="#btn", action="nonexistent")
with pytest.raises(HTTPException) as ei:
await smart_locate_router.verify_step(req)
assert ei.value.status_code == 400
@pytest.mark.asyncio
async def test_fill_missing_value_400(self):
from fastapi import HTTPException
req = smart_locate_router.VerifyStepRequest(
selector="#input", action="fill", value=None)
with pytest.raises(HTTPException) as ei:
await smart_locate_router.verify_step(req)
assert ei.value.status_code == 400
@pytest.mark.asyncio
async def test_empty_selector_400(self):
from fastapi import HTTPException
req = smart_locate_router.VerifyStepRequest(
selector=" ", action="click")
with pytest.raises(HTTPException) as ei:
await smart_locate_router.verify_step(req)
assert ei.value.status_code == 400
# ==================== 单元测试:生命周期与响应(替换 PlaywrightExecutor) ====================
class _FakeExecutor:
"""记录调用顺序的可编程假执行器(模拟 PlaywrightExecutor)"""
def __init__(self, verify_result=True, verify_raise=None, start_raise=None, **cfg):
self.calls = []
self.verify_result = verify_result
self.verify_raise = verify_raise
self.start_raise = start_raise
self.cfg = cfg
def start(self):
self.calls.append("start")
if self.start_raise:
raise self.start_raise
def verify_step(self, **kwargs):
# 模拟真实执行器 verify_step 的内部生命周期:start → 验证 → stop
self.calls.append("start")
self.calls.append(("verify_step", kwargs))
if self.verify_raise:
self.calls.append("stop")
raise self.verify_raise
self.calls.append("stop")
return self.verify_result
def stop(self):
self.calls.append("stop")
@pytest.fixture
def fake_executor_factory(monkeypatch):
"""把路由模块里的 PlaywrightExecutor 替换为可编程工厂,捕获实例与配置"""
holder = {"instances": [], "configs": [], "next": {}}
def make(config=None):
holder["configs"].append(config or {})
inst = _FakeExecutor(**holder["next"])
holder["instances"].append(inst)
return inst
monkeypatch.setattr(smart_locate_router, "PlaywrightExecutor", make)
return holder
def _mk_req(selector="#btn", action="click", value=None, page_url=None, auto_login=False):
data = {"selector": selector, "action": action, "value": value, "auto_login": auto_login}
if page_url is not None:
data["page_url"] = page_url
return smart_locate_router.VerifyStepRequest(**data)
@pytest.mark.asyncio
async def test_success_passes_through(fake_executor_factory):
resp = await smart_locate_router.verify_step(_mk_req())
assert resp.success is True
assert "通过" in resp.message
inst = fake_executor_factory["instances"][0]
# 生命周期:执行器 verify_step 自己负责 start/stop(路由只调用一次性方法)
assert inst.calls[0] == "start"
assert inst.calls[-1] == "stop"
# 动作/选择器透传
verify_kwargs = inst.calls[1][1]
assert verify_kwargs["action"] == "click"
assert verify_kwargs["selector"] == "#btn"
@pytest.mark.asyncio
async def test_auto_login_passed_to_executor(fake_executor_factory):
await smart_locate_router.verify_step(_mk_req(auto_login=True))
verify_kwargs = fake_executor_factory["instances"][0].calls[1][1]
assert verify_kwargs["auto_login"] is True
@pytest.mark.asyncio
async def test_page_url_passed_to_executor(fake_executor_factory):
await smart_locate_router.verify_step(_mk_req(page_url="https://192.168.5.44/"))
verify_kwargs = fake_executor_factory["instances"][0].calls[1][1]
assert verify_kwargs["page_url"] == "https://192.168.5.44/"
@pytest.mark.asyncio
async def test_verify_failure_returns_success_false(fake_executor_factory):
"""执行器返回 False(选择器未命中)→ success=False,而非 500"""
fake_executor_factory["next"]["verify_result"] = False
resp = await smart_locate_router.verify_step(_mk_req())
assert resp.success is False
assert "失败" in resp.message
@pytest.mark.asyncio
async def test_verify_raise_returns_success_false(fake_executor_factory):
"""执行器抛异常(元素不存在/动作失败)→ success=False + 安全化错误信息"""
fake_executor_factory["next"]["verify_raise"] = RuntimeError("元素不存在")
resp = await smart_locate_router.verify_step(_mk_req())
assert resp.success is False
assert "元素不存在" in resp.message
# ==================== ASGI 集成测试 ====================
@pytest.fixture
async def client():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
class TestVerifyStepAPI:
"""通过 HTTP 层验证端点行为(不再无条件成功)"""
@pytest.mark.asyncio
async def test_invalid_action_returns_400(self, client):
resp = await client.post(
"/api/element/verify-step",
json={"selector": "#btn", "action": "bogus", "auto_login": False},
)
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_fill_missing_value_returns_400(self, client):
resp = await client.post(
"/api/element/verify-step",
json={"selector": "#input", "action": "fill", "auto_login": False},
)
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_invalid_action_not_unconditional_success(self, client):
"""集成断言:非法 action 不再返回 success=True(旧存根无条件成功)"""
resp = await client.post(
"/api/element/verify-step",
json={"selector": "#btn", "action": "no_such_action", "auto_login": False},
)
# 旧存根对任何请求都返回 200 + success=True;现在应 400
assert resp.status_code == 400
...@@ -148,6 +148,8 @@ export interface SmartLocateRequest { ...@@ -148,6 +148,8 @@ export interface SmartLocateRequest {
navigate_menu?: string navigate_menu?: string
/** 被测系统基础 URL */ /** 被测系统基础 URL */
page_url?: string page_url?: string
/** 是否启用 Claude 语义增强(默认 true) */
use_claude?: boolean
} }
/** /**
......
...@@ -173,7 +173,7 @@ ...@@ -173,7 +173,7 @@
v-model="getApiCase(step).bodyText" v-model="getApiCase(step).bodyText"
type="textarea" type="textarea"
:rows="2" :rows="2"
placeholder='请求体 JSON,如 {"name": "{__CTX:meetingName__}"}' placeholder="请求体 JSON,如 {&quot;name&quot;: &quot;{__CTX:meetingName__}&quot;}"
size="small" size="small"
style="width: 100%" style="width: 100%"
class="api-body" class="api-body"
......
...@@ -307,7 +307,7 @@ ...@@ -307,7 +307,7 @@
v-model="parametersText" v-model="parametersText"
type="textarea" type="textarea"
:rows="2" :rows="2"
placeholder='JSON 对象,如 {"meetingName":"自动化-{__RANDOM_NAME_8__}", "room":"北京展厅会议室"}' placeholder="JSON 对象,如 {&quot;meetingName&quot;:&quot;自动化-{__RANDOM_NAME_8__}&quot;, &quot;room&quot;:&quot;北京展厅会议室&quot;}"
/> />
<span class="param-hint">供 {__CTX:key__} 上下文变量引用</span> <span class="param-hint">供 {__CTX:key__} 上下文变量引用</span>
</div> </div>
...@@ -1083,15 +1083,26 @@ const batchLocate = async () => { ...@@ -1083,15 +1083,26 @@ const batchLocate = async () => {
return return
} }
// 2. 转换步骤为智能定位格式(包含 navigate 步骤) // 2. 转换步骤为智能定位格式(包含 navigate 步骤,透传所有字段
const smartLocateSteps = steps const smartLocateSteps = steps
.filter((s: any) => s.action) .filter((s: any) => s.action)
.map((s: any) => ({ .map((s: any) => {
order: s.order, const step: any = {
name: s.name || '', order: s.order,
action: s.action, name: s.name || '',
params: s.params || {}, action: s.action,
})) params: s.params || {},
}
// P1a: 透传可选字段,防止 round-trip 丢字段
if (s.semantic != null) step.semantic = s.semantic
if (s.verify != null) step.verify = s.verify
if (s.run_on != null) step.run_on = s.run_on
if (s.selectors != null) step.selectors = s.selectors
if (s.page_key != null) step.page_key = s.page_key
if (s.element_key != null) step.element_key = s.element_key
if (s.expected != null) step.expected = s.expected
return step
})
if (!smartLocateSteps.length) { if (!smartLocateSteps.length) {
ElMessage.warning('没有可定位的步骤') ElMessage.warning('没有可定位的步骤')
...@@ -1123,6 +1134,13 @@ const batchLocate = async () => { ...@@ -1123,6 +1134,13 @@ const batchLocate = async () => {
if (locateResult.selectors?.primary) { if (locateResult.selectors?.primary) {
updatedParams.selector = locateResult.selectors.primary updatedParams.selector = locateResult.selectors.primary
} }
// P1a: 保存完整选择器链(回退候选),执行期 selectors 链优先于单个 selector
const candidateValues = (locateResult.selectors?.candidates || [])
.map((c: any) => c?.value)
.filter((v: any) => v && v !== updatedParams.selector)
if (candidateValues.length > 0) {
updatedParams.selectors = [updatedParams.selector, ...candidateValues].filter(Boolean)
}
// ⚠️ 阶段6升级:合并 semantic / verify(新生成的优先,保留已有字段) // ⚠️ 阶段6升级:合并 semantic / verify(新生成的优先,保留已有字段)
return { return {
...s, ...s,
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论