提交 1a3d52ad authored 作者: 陈泽健's avatar 陈泽健

fix(executor): 修复执行中无法取消(MySQL 1205 行锁)并加固 SUT URL 校验

- run_execution 置 running 后立即 commit,释放 InnoDB 行锁,
  取消接口不再被 Lock wait timeout 1205 阻塞(实机 91ms 返回)
- 结果写入循环增加取消检测:执行期间被取消则用例结果保持 skipped
- 收尾状态判定改列查询 + FOR UPDATE:修复 SQLAlchemy 身份映射过期实例
  (expire_on_commit=False 不刷新)将已取消执行覆盖为 completed 的问题
- 新增 _validate_sut_url:拒绝非法 IP 段/端口/字符(如 192.168.5.4202),
  前端 Settings.vue 同步表单校验
- 新增回归测试 test_execution_cancel.py(4 条)+ test_sut_url_validation.py(8 条)
- HANDOFF_UI自动化.md 记录会话29进度
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 ac823858
......@@ -2,8 +2,8 @@
> **生成时间**: 2026-08-19
> **当前分支**: `platform-auto-test`
> **最近提交**: `1bb8756a` chore: .gitignore 遮蔽历史会话遗留的一次性脚本与临时输出
> **状态**: 🟢 **会话28:系统配置切换被测系统后用例自动跟随(方案A·执行期地址重写)已实现并部署 5.60 生产——执行器新增 `_rebase_sut_url`,导航/接口步骤若指向已知被测系统主机(默认 5.44)则自动替换为当前配置的 SUT 地址,130 条存量硬编码用例自动跟随,不动任何数据。后端 182 测试全过、已部署 5.60**
> **最近提交**: `ac823858` feat(executor): 系统配置切换被测系统后用例执行自动跟随新地址
> **状态**: 🟢 **会话29:修复「执行中无法取消」(MySQL 1205 行锁)+ SUT URL 校验加固,已实现并部署 5.60 生产——`run_execution` 置 running 后立即 commit 释放 InnoDB 行锁(取消接口由 ~50s 1205 超时降至 91ms 返回);实机发现收尾状态判定命中 SQLAlchemy 身份映射(expire_on_commit=False 实例不刷新)导致 cancelled 被覆盖为 completed,改为列查询 + FOR UPDATE 修复;新增 `_validate_sut_url` 拒绝非法地址(如 192.168.5.4202)。后端 186 测试全过、实机取消演练通过、恢复用例跑通 100%**
---
......@@ -211,10 +211,48 @@
---
### ✅ 修复「执行中无法取消」(MySQL 1205 行锁)+ SUT URL 校验加固(会话 29,2026-08-19)
**问题**
- 用户反馈:执行中心有执行任务一直「运行中」且**无法取消**;此前一段时期用例执行也瘫痪
- **双根因**:(1) 系统配置 SUT URL 配错 `https://192.168.5.4202`(IP 段越界)→ 执行器 `Cannot navigate to invalid URL`,全部 UI 用例执行瘫痪;(2) 取消接口被 MySQL InnoDB 行锁阻塞
**根因 2 详解(取消 1205 阻塞)**
- `run_execution``status=running` 后只 `flush``commit` → 整个 Playwright 执行期间(数十秒)持有 `executions` 行锁(InnoDB 排他锁)
- 取消接口的 `UPDATE executions` 被同一行锁阻塞 → `Lock wait timeout exceeded (1205)`(约 50s 超时),执行中永远无法取消
**修复**`backend/app/services/execution_service.py`,三处):
| # | 修复点 | 说明 |
|---|--------|------|
| 1 | 置 running 后**立即 commit** | 释放行锁,取消接口随时可写;commit 后实例过期,重新查询刷新 |
| 2 | 结果写入循环加**取消检测** | 执行期间已取消 → 用例结果保持 skipped,不写真实结果 |
| 3 | 收尾状态判定改**列查询 + FOR UPDATE** | ⚠️ 实机发现的隐藏 bug(见下) |
**实机发现的隐藏 bug(SQLAlchemy 身份映射)**
- 收尾原用完整实体查询 `select(Execution)` 判定是否已取消 → 命中 SQLAlchemy **身份映射**,返回内存中过期的 execution 实例(`expire_on_commit=False` 下 commit 后实例不刷新),`.status` 读到旧值 `running`
- `FOR UPDATE` 虽执行了 SQL 但结果未灌入对象 → 已取消的执行被覆盖回 `completed`
- 修复:改为 `select(Execution.status)` **列查询(标量)+ FOR UPDATE**,绕过身份映射;回归测试 `test_cancelled_preserved_despite_stale_identity_map`(fake DB 模拟身份映射返回过期实体)
**SUT URL 校验加固**(防止再次配错导致全部用例瘫痪):
- 后端 `routers/system.py` 新增 `_validate_sut_url()`:http/https + 主机 + 端口 1-65535 + IP 段 0-255(拦 `192.168.5.4202`)+ 域名标签合法性
- 前端 `Settings.vue` 同步表单校验(`new URL` + 端口/IP 段检查)
**验证**
- 后端全量 **186 passed**(新增 `tests/test_sut_url_validation.py` 8 条 + `tests/test_execution_cancel.py` 4 条含身份映射回归);前端 `npm run build` 通过
- **已部署 5.60 生产**`routers/system.py` + `services/execution_service.py` + 前端 dist 90 文件 → 容器重启 healthy)
- 实机验证:
- `PUT /api/system/sut-config` 非法 URL `192.168.5.4202`**400 拒绝**;恢复 `https://192.168.5.44` → 200 持久化
- **实机取消演练**:创建执行 → 运行 → 取消 → 取消接口 **91ms 返回**(修复前 ~50s 1205 超时),最终状态稳定 **`cancelled`**(未被覆盖为 completed)
- 恢复用例「会议模板页面访问验证」完整跑通:**通过 1/1,100%,23.4s**(验证原有用例正常执行)
---
## 📊 会话历史时间线
| 会话 | 日期 | 主题 | 结果 |
|------|------|------|------|
| 29 | 08-19 | **修复「执行中无法取消」(MySQL 1205 行锁)+ SUT URL 校验加固**:置 running 后立即 commit 释放行锁;实机发现身份映射过期实例覆盖 cancelled 为 completed(改列查询 + FOR UPDATE);`_validate_sut_url` 拒绝非法地址(192.168.5.4202);结果保持 skipped | ✅ 后端 186 测试全过、已部署 5.60、实机取消演练通过(91ms + cancelled 保持)、恢复用例跑通 100% |
| 28 | 08-19 | **系统配置切换被测系统后用例自动跟随(方案A·执行期地址重写)**`_rebase_sut_url` 导航/接口主机重写(已知 SUT 主机→当前配置地址),130 条存量硬编码用例自动跟随、不动数据 | ✅ 后端 182 测试全过、已部署 5.60 生产 |
| 27 | 08-19 | **UI自动化定时任务模块 + 系统配置被测系统URL**:调度引擎(30s 轮询 + 到期异步触发 + 冲突跳过)+ 定时任务 CRUD/启停/立即执行/运行记录 + 报告按模块分组并展示被测系统 URL + SUT URL 保存/重启持久化 + 前端定时任务页与系统配置卡片 | ✅ 按 PRD 工作流实现,后端 179 测试全过、前端 build 通过、真实端到端验收通过(已部署生产) |
| 26 | 08-19 | **新建会议用例 step 11/17 执行失败修复**:保留页面直达 + 执行器策略5 增强(轮询等待/force勾选/校验防假通过)+ 用例 step 12 改「保持立即开始」(选择器修正零匹配坑);部署 5.60 | ✅ 本地 + 生产各连跑 3 次全绿、零残留,提交 `c0dedbdf` |
......@@ -322,6 +360,11 @@ cd frontend && npm run build
### P0(高优先级)
- [x] **修复「执行中无法取消」(MySQL 1205 行锁)+ SUT URL 校验加固**(会话29 实现 + 部署)
- ✅ 置 running 后立即 commit 释放 InnoDB 行锁(取消接口 91ms 返回,修复前 ~50s 1205 超时)
- ✅ 实机发现身份映射过期实例覆盖 cancelled 为 completed → 收尾状态判定改列查询 + FOR UPDATE,回归测试覆盖
-`_validate_sut_url` 拒绝非法地址(192.168.5.4202 段越界 / 非法端口 / 非法字符),前端同步表单校验
- ✅ 后端 186 测试全过;**已部署 5.60 生产**,实机取消演练通过、恢复用例跑通 1/1 100%
- [x] **系统配置切换被测系统后用例自动跟随(方案A·执行期地址重写)**(会话28 实现 + 部署)
- ✅ 执行器 `_rebase_sut_url`:导航/接口步骤若指向已知被测系统主机(默认 5.44)→ 自动替换为当前配置 SUT 地址(保留 scheme/路径/查询/hash)
- ✅ 130 条存量硬编码 `https://192.168.5.44` 的 UI 用例全部自动跟随,**不动任何用例数据**
......@@ -362,4 +405,4 @@ cd frontend && npm run build
---
*本文档由 Claude Code 于 2026-08-19 更新(会话 28:系统配置切换被测系统后用例自动跟随(方案A·执行期地址重写)——`_rebase_sut_url` 导航/接口主机重写,130 条存量硬编码用例自动跟随,后端 182 测试全过,已部署 5.60 生产)。*
*本文档由 Claude Code 于 2026-08-19 更新(会话 29:修复「执行中无法取消」(MySQL 1205 行锁,置 running 后立即 commit 释放行锁)+ 实机发现身份映射过期实例覆盖 cancelled 为 completed(改列查询 + FOR UPDATE)+ SUT URL 校验加固,后端 186 测试全过、实机取消演练通过、已部署 5.60 生产)。*
......@@ -9,12 +9,14 @@
最后修改:2026-07-29
"""
import ipaddress
import logging
import platform
import sys
import time
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import func, select, text
......@@ -40,6 +42,64 @@ router = APIRouter()
_START_TIME = time.time()
def _validate_sut_url(url: str) -> Optional[str]:
"""
校验被测系统 URL(防止无效地址保存后导致全部 UI 用例执行瘫痪)。
规则:
- 必须 http/https 且带主机名
- 端口(如有)必须是合法 1-65535 数字(urlparse.port 会拒绝如 :abc)
- 主机必须是合法 IPv4/IPv6 或合法域名;纯数字点分形式必须是合法 IPv4
(如 `192.168.5.4202` 每段须 0-255,否则拒绝)
Args:
url (str): 待校验 URL(应已 strip)
Returns:
Optional[str]: 错误信息;校验通过返回 None
"""
if not url.startswith(("http://", "https://")):
return "URL 须以 http:// 或 https:// 开头"
parsed = urlparse(url)
host = parsed.hostname
if not host:
return "URL 缺少主机名"
try:
port = parsed.port
except ValueError:
return "URL 端口无效(端口须为 1-65535 的数字)"
if port is not None and not (1 <= port <= 65535):
return "URL 端口必须在 1-65535 之间"
# 主机为合法 IP(IPv4/IPv6)则通过
try:
ipaddress.ip_address(host)
return None
except ValueError:
pass
# 纯数字点分形式(如 192.168.5.4202)必须是合法 IPv4——ipaddress 已失败,直接拒绝
if all(ch in "0123456789." for ch in host):
return "URL 主机无效(IP 地址每段须为 0-255)"
# 域名校验:字母/数字/中划线/点,标签 1-63 字符,不以中划线开头结尾
if len(host) > 253:
return "URL 主机名过长"
labels = host.rstrip(".").split(".")
if not labels:
return "URL 主机名无效"
for label in labels:
if not (1 <= len(label) <= 63):
return "URL 主机名标签长度须为 1-63"
if not all(ch.isalnum() or ch in "-_" for ch in label):
return "URL 主机名包含非法字符"
if label.startswith("-") or label.endswith("-"):
return "URL 主机名标签不能以中划线开头或结尾"
return None
@router.get("/info", response_model=dict)
async def get_system_info(db: AsyncSession = Depends(get_db)):
"""
......@@ -228,8 +288,9 @@ async def update_sut_config(
(lifespan 启动时从数据库加载)。
"""
url = (request.url or "").strip().rstrip("/")
if not url.startswith(("http://", "https://")):
raise HTTPException(status_code=400, detail="URL 须以 http:// 或 https:// 开头")
url_err = _validate_sut_url(url)
if url_err:
raise HTTPException(status_code=400, detail=url_err)
# 从数据库读取或创建单行记录(id=1)
result = await db.execute(select(DingTalkConfig).where(DingTalkConfig.id == 1))
......
......@@ -276,10 +276,15 @@ class ExecutionService:
# 获取执行锁
set_running_execution(execution_id)
# 更新状态为运行中
# 更新状态为运行中并立即提交:
# 若不提交,长执行期间事务会一直持有 executions 行锁(MySQL InnoDB),
# 取消接口的 UPDATE 将被阻塞(Lock wait timeout exceeded 1205),
# 导致执行中无法取消。立即提交可释放行锁,让取消接口随时可写。
execution.status = "running"
execution.start_time = datetime.now()
await self.db.flush()
await self.db.commit()
# commit 后 ORM 实例已过期,重新查询刷新,避免后续异步属性访问触发刷新
execution = await self.get_execution(execution_id)
# 广播执行开始
await manager.broadcast(execution_id, {
......@@ -424,9 +429,23 @@ class ExecutionService:
raise exec_holder["err"]
all_results = exec_holder.get("data", [])
# 若执行期间已被用户取消,结果保持取消时的 skipped 状态,不再写入真实结果
# with_for_update 为当前读:MySQL REPEATABLE READ 下普通 SELECT 走快照读,
# 看不到本事务之外已提交的取消(会读到 running 旧快照),必须加锁读最新已提交状态
cancel_check = await self.db.execute(
select(Execution.status).where(Execution.id == execution_id).with_for_update()
)
exec_was_cancelled = cancel_check.scalar_one_or_none() == "cancelled"
# 更新结果
for i, case_info in enumerate(cases_to_run):
case_result = case_info["case_result"]
if exec_was_cancelled:
# 执行已在运行中被取消:保留 skipped,不覆盖真实结果
case_result.status = "skipped"
case_result.error_message = "执行任务已取消"
await self.db.flush()
continue
result = all_results[i]
# 更新状态为运行中(广播用例开始)
......@@ -481,8 +500,25 @@ class ExecutionService:
# 释放执行锁
set_running_execution(None)
# 更新执行状态
execution.status = "completed"
# 读取最新执行状态(避免使用 commit 后的过期实例):
# 执行期间可能已被用户取消,若已取消则保持 cancelled,不被覆盖为 completed。
# 必须用【列查询(标量)+ FOR UPDATE 当前读】:
# 1. MySQL REPEATABLE READ 快照读看不到本事务之外已提交的取消;
# 2. 完整实体查询会命中 SQLAlchemy 身份映射,返回内存中过期的 execution
# 实例(expire_on_commit=False,commit 后不刷新),FOR UPDATE 读了数据库
# 也不灌入对象,.status 仍是旧值 running(已实机复现)。
final_q = await self.db.execute(
select(Execution.status).where(Execution.id == execution_id).with_for_update()
)
final_status = final_q.scalar_one_or_none()
if final_status is None:
return execution
was_cancelled = final_status == "cancelled"
if was_cancelled:
logger.info(f"执行已在运行中被取消,保持 cancelled: {execution_id}")
# 用当前会话中的 execution 实例写回(内存字段可能过期,此处显式覆盖全部字段)
execution.status = "cancelled" if was_cancelled else "completed"
execution.end_time = datetime.now()
execution.duration = (execution.end_time - execution.start_time).total_seconds()
execution.passed = await self._count_results(execution_id, "passed")
......
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:test_execution_cancel.py
模块描述:执行取消 / 行锁释放 单元测试
覆盖:
- run_execution 在置 status=running 后立即 commit(否则长执行期间持有
executions 行锁,MySQL InnoDB 下取消接口报 Lock wait timeout 1205)
- 执行期间被取消时,finally 保持 cancelled 状态(不被覆盖为 completed)
作者:czj
创建日期:2026-08-19
"""
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch
from app.models.execution import Execution
from app.services.execution_service import ExecutionService
class _Scalars:
def __init__(self, rows):
self._rows = rows
def all(self):
return list(self._rows)
class _Result:
def __init__(self, rows):
self._rows = rows
def scalars(self):
return _Scalars(self._rows)
def scalar_one_or_none(self):
return self._rows[0] if self._rows else None
class _FakeDB:
"""最小异步 DB 假件:仅服务 run_execution 需要的 execute/flush/commit"""
def __init__(self, exec_obj, simulate_cancel=False, stale_entity_status=None):
self.exec_obj = exec_obj
self.simulate_cancel = simulate_cancel
self.stale_entity_status = stale_entity_status
self.commit_calls = 0
self.executed_sql = []
async def execute(self, stmt):
sql = str(stmt)
self.executed_sql.append(sql)
# 用例结果 / 用例定义查询 → 空(用例列表为空,worker 快速结束)
if "case_results" in sql or "test_cases" in sql:
return _Result([])
# 仅查状态列(执行期间取消检测 / 收尾状态读取)→ 返回当前状态字符串
if sql.lstrip().startswith("SELECT executions.status"):
return _Result([self.exec_obj.status])
# 完整实体查询 → 模拟 SQLAlchemy 身份映射:expire_on_commit=False 时
# commit 后实例不刷新,返回内存中过期的对象(stale_entity_status)
if self.stale_entity_status is not None:
stale = _make_execution(status=self.stale_entity_status)
stale.id = self.exec_obj.id
return _Result([stale])
# 默认返回执行对象
return _Result([self.exec_obj])
async def flush(self):
pass
async def commit(self):
self.commit_calls += 1
if self.simulate_cancel and self.commit_calls == 1:
# 首次 commit 是"置 running"那一次;其后模拟执行期间用户并发取消
self.exec_obj.status = "cancelled"
def _make_execution(status="running", case_type="ui"):
ex = Execution()
ex.id = "exec_test_cancel_001"
ex.name = "test"
ex.trigger_type = "manual"
ex.trigger_by = "user"
ex.total_cases = 1
ex.passed = 0
ex.failed = 0
ex.skipped = 0
ex.pass_rate = 0.0
ex.duration = 0.0
ex.status = status
ex.error_message = None
ex.start_time = datetime.now()
ex.end_time = None
ex.environment = "default"
ex.config = {}
ex.case_type = case_type
ex.created_at = datetime.now()
return ex
def _make_service(db, exec_obj):
service = ExecutionService(db)
service.get_execution = AsyncMock(return_value=exec_obj)
service._count_results = AsyncMock(return_value=0)
return service
# 真实线程运行 worker(PlaywrightExecutor 已 mock,无需浏览器)
@patch("app.services.execution_service.PlaywrightExecutor", autospec=True)
@patch("app.services.execution_service.manager.broadcast", new_callable=AsyncMock)
async def test_run_execution_commits_running_status_early(mock_broadcast, mock_executor):
"""置 status=running 后立即 commit(释放行锁,取消接口才不会被 1205 阻塞)"""
exec_obj = _make_execution(status="running")
db = _FakeDB(exec_obj, simulate_cancel=False)
service = _make_service(db, exec_obj)
result = await service.run_execution(exec_id := exec_obj.id)
# 关键:run_execution 内部至少 commit 过一次(置 running 后的立即提交)
assert db.commit_calls >= 1, "run_execution 应提交 status=running(否则 MySQL 下取消接口被行锁阻塞)"
# 未被取消 → 正常收尾为 completed
assert result.status == "completed"
# 真实线程运行 worker(PlaywrightExecutor 已 mock,无需浏览器)
@patch("app.services.execution_service.PlaywrightExecutor", autospec=True)
@patch("app.services.execution_service.manager.broadcast", new_callable=AsyncMock)
async def test_run_execution_preserves_cancelled_status(mock_broadcast, mock_executor):
"""执行期间被取消 → finally 保持 cancelled,不被覆盖为 completed"""
exec_obj = _make_execution(status="running")
db = _FakeDB(exec_obj, simulate_cancel=True)
service = _make_service(db, exec_obj)
result = await service.run_execution(exec_id := exec_obj.id)
assert db.commit_calls >= 1
# 模拟并发取消后,finally 应保留 cancelled
assert result.status == "cancelled", "执行期间被取消后状态应保持 cancelled"
# 真实线程运行 worker(PlaywrightExecutor 已 mock,无需浏览器)
@patch("app.services.execution_service.PlaywrightExecutor", autospec=True)
@patch("app.services.execution_service.manager.broadcast", new_callable=AsyncMock)
async def test_status_reads_use_locking_read(mock_broadcast, mock_executor):
"""取消检测与收尾的状态读取必须使用锁定读(FOR UPDATE)。
背景:MySQL InnoDB REPEATABLE READ 下,普通 SELECT 走快照读,看不到
本事务之外已提交的取消(会读到 running 旧快照),导致 cancelled 被覆盖为
completed(已实机复现)。必须用 with_for_update 做当前读。
"""
exec_obj = _make_execution(status="running")
db = _FakeDB(exec_obj, simulate_cancel=True)
service = _make_service(db, exec_obj)
await service.run_execution(exec_obj.id)
# 取消检测(SELECT executions.status)与收尾重查(FROM executions WHERE executions.id)
# 都必须带 FOR UPDATE;否则快照读看不到已提交的取消
exec_reads = [
s for s in db.executed_sql
if "SELECT executions.status" in s or "FROM executions WHERE executions.id" in s
]
assert exec_reads, "应存在对 executions 的状态读取"
for sql in exec_reads:
assert "FOR UPDATE" in sql, f"状态读取必须为锁定读(快照读会把 cancelled 覆盖为 completed): {sql}"
# 状态判定必须是【列查询】,不能是完整实体查询——实体查询会命中身份映射
# 返回内存过期实例(.status 仍为 running),导致 cancelled 被覆盖为 completed
assert "SELECT executions.status" in sql, \
f"状态读取必须为列查询(完整实体查询命中身份映射返回过期实例): {sql}"
@patch("app.services.execution_service.PlaywrightExecutor", autospec=True)
@patch("app.services.execution_service.manager.broadcast", new_callable=AsyncMock)
async def test_cancelled_preserved_despite_stale_identity_map(mock_broadcast, mock_executor):
"""身份映射返回过期实体(status=running)时,取消状态仍须保持。
回归:收尾曾用完整实体查询判定状态,命中 SQLAlchemy 身份映射返回 commit 后
未刷新的过期实例(expire_on_commit=False),.status 仍是 running,把已取消的
执行覆盖为 completed(5.60 实机复现)。现改为列查询(SELECT executions.status)
+ FOR UPDATE 当前读,绕过身份映射。
"""
exec_obj = _make_execution(status="running")
# 模拟生产:完整实体查询返回内存中过期的 running 对象,列查询返回实时状态
db = _FakeDB(exec_obj, simulate_cancel=True, stale_entity_status="running")
service = _make_service(db, exec_obj)
result = await service.run_execution(exec_obj.id)
assert db.commit_calls >= 1
assert result.status == "cancelled", \
"身份映射过期实体下,已取消的执行仍被覆盖为 completed(状态判定必须用列查询)"
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:test_sut_url_validation.py
模块描述:被测系统 SUT URL 校验单元测试
覆盖:_validate_sut_url 拒绝无效地址(如 192.168.5.4202 段越界 / 非法端口 /
非法字符 / 缺主机名),放行合法地址(内网 IP / 域名 / 带端口)。防止无效
SUT 地址被保存后导致全部 UI 用例执行瘫痪(Cannot navigate to invalid URL)。
作者:czj
创建日期:2026-08-19
"""
import pytest
from app.routers.system import _validate_sut_url
class TestSutUrlValidation:
"""被测系统 URL 校验"""
def test_rejects_invalid_ip_octet(self):
"""纯数字点分主机非合法 IPv4(段 4202 > 255)必须拒绝"""
assert _validate_sut_url("https://192.168.5.4202") is not None
assert _validate_sut_url("https://192.168.5.4202/") is not None
def test_rejects_bad_scheme(self):
assert _validate_sut_url("ftp://192.168.5.44") is not None
assert _validate_sut_url("192.168.5.44") is not None
def test_rejects_missing_host(self):
assert _validate_sut_url("https://") is not None
def test_rejects_invalid_port(self):
assert _validate_sut_url("https://192.168.5.44:abc") is not None
assert _validate_sut_url("https://192.168.5.44:99999") is not None
def test_rejects_invalid_host_chars(self):
assert _validate_sut_url("https://exa mple.com/") is not None
assert _validate_sut_url("https://exa*mple.com/") is not None
def test_accepts_valid_ipv4(self):
assert _validate_sut_url("https://192.168.5.44") is None
assert _validate_sut_url("https://192.168.5.44/") is None
def test_accepts_ipv4_with_port(self):
assert _validate_sut_url("https://192.168.5.44:4202") is None
assert _validate_sut_url("http://192.168.5.44:8080/") is None
def test_accepts_domain_and_localhost(self):
assert _validate_sut_url("https://example.com") is None
assert _validate_sut_url("https://meeting.ubains.com.cn/#/Home") is None
assert _validate_sut_url("https://localhost:3000") is None
......@@ -133,14 +133,46 @@ const sutForm = reactive({
url: '',
})
/** 校验被测系统 URL:http/https + 合法主机(IP 段 0-255 / 合法域名)+ 合法端口 */
function validateSutUrl(_rule: any, value: string, callback: (err?: Error) => void) {
const url = (value || '').trim()
if (!url) {
callback(new Error('请输入被测系统 URL'))
return
}
if (!/^https?:\/\//.test(url)) {
callback(new Error('URL 需以 http:// 或 https:// 开头'))
return
}
try {
const parsed = new URL(url)
const host = parsed.hostname
if (!host) {
callback(new Error('URL 缺少主机名'))
return
}
if (parsed.port && !(Number(parsed.port) >= 1 && Number(parsed.port) <= 65535)) {
callback(new Error('URL 端口必须在 1-65535 之间'))
return
}
// 纯数字点分形式必须是合法 IPv4(每段 0-255)
if (/^[\d.]+$/.test(host)) {
const ok = host.split('.').every((seg) => seg !== '' && Number(seg) >= 0 && Number(seg) <= 255)
if (!ok) {
callback(new Error('URL 主机无效(IP 地址每段须为 0-255)'))
return
}
}
} catch {
callback(new Error('URL 格式无效'))
return
}
callback()
}
const sutRules = {
url: [
{ required: true, message: '请输入被测系统 URL', trigger: 'blur' },
{
pattern: /^https?:\/\/.+/,
message: 'URL 需以 http:// 或 https:// 开头',
trigger: 'blur',
},
{ required: true, validator: validateSutUrl, trigger: 'blur' },
],
}
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论