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

docs(handoff): 精简 HANDOFF_UI自动化文档 + 更新会话37状态

- HANDOFF_UI自动化.md: 735→216行精简,删除已完成会话详情与P0/P1清单
- page_url_mapping.json: 补全 asset_info 完整配置 + 新增 meeting_room_list 页面注册
- playwright_executor.py: element_count_min 新增 micro-app 8s 轮询兜底
- 新增 sut-explore skill、generate_asset_info_cases/generate_meetingroom_cases/sync_to_560_full 脚本
- 数据库新增 10 个用例(7 会议室列表 + 3 资产信息)
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 6aa026c4
---
name: sut-explore
description: 访问被测系统探测新页面并生成可执行的 UI 自动化用例
---
访问被测系统(SUT)进行探测,发现未覆盖的功能模块,生成可直接执行的 UI 自动化测试用例。
## Usage
/sut-explore [目标功能关键词或模块]
## Description
通过真实浏览器访问被测系统,盘点已有用例覆盖范围,识别未覆盖功能,生成用例并反复验证直到可执行。
**功能:**
- 访问被测系统并登录
- 盘点现有用例库覆盖情况
- 识别未覆盖的功能点
- 通过浏览器探测真实 DOM/URL
- 生成可执行的 UI 自动化用例(写入 SQLite)
- 创建执行并迭代修复直到通过
**示例:**
```
/sut-explore 数据统计 # 探测数据统计模块
/sut-explore # 交互式全量探测
```
## 前置条件
- 本地后端已启动:`uvicorn app.main:app --reload --port 8001`(工作目录 backend/)
- SUT 可访问:https://192.168.5.44
- 登录凭据:admin@xty / Ubains@13579,验证码固定 `csba`(见 [[test-platform-credentials]])
- Chrome DevTools MCP 可用(用于真实浏览器探测)
## 执行步骤
### 1. 盘点现有用例覆盖
```bash
cd backend
python -c "
import sqlite3, json
conn = sqlite3.connect('data/test_platform.db')
cur = conn.cursor()
# 查所有模块及用例数
cur.execute('SELECT module_id, count(*) FROM test_cases GROUP BY module_id')
# 查指定功能关键词的用例数
cur.execute(\"SELECT count(*) FROM test_cases WHERE name LIKE '%会议概览%'\")
"
```
判断覆盖情况:**找到 0 个用例的关键词 = 未覆盖功能**
### 2. 浏览器探测(Chrome DevTools MCP)
1. `navigate_page``https://192.168.5.44/` → 登录页
2. `fill_form` 填 手机号/密码/验证码 → `click` 登录按钮 → 可能弹"同意协议"对话框 → `click` 确定
3. `take_snapshot` 查看首页"常用功能"区域,**所有功能入口一次列出**,可直接发现未覆盖功能
4. 点击目标功能入口 → 从地址栏获取**真实直达 URL**(Hash 双重编码,直接复制勿手拼)
5. 在页面上 `evaluate_script` 探测 DOM 结构:
- 所有 `input` 及其 `value/placeholder/readonly`(下拉框)
- 所有 `button` 及其图标 class(查询/导出按钮常为空文本)
- 标题/小部件文本
### 3. 生成用例(写 SQLite)
- 写生成脚本到 `backend/scripts/generate_xxx_cases.py`,插入 `test_cases`
- **必须配置**`case_type='ui'``status='active'``config={"auto_login": true, "headless": true}`
- **首步必须** `navigate` 直达 URL(不要 click 菜单,headless 下微前端菜单不稳定)
- 步骤 JSON 用 `ensure_ascii=False` 序列化
- 断言选择器参照「踩坑清单」(见下)
### 4. 创建执行并验证
```bash
# 用 python requests(不要用 curl 中文 body,会 400)
python -c "
import requests, json
ids = [...] # 用例 ID 列表
r = requests.post('http://localhost:8001/api/executions', json={'case_ids': ids, 'name': 'xxx验证'})
print(r.json()['id'])
"
# 启动
requests.post('http://localhost:8001/api/executions/{id}/run')
# 轮询(全量 10 用例约 400s)
requests.get('http://localhost:8001/api/executions/{id}')
# 查看失败详情
requests.get('http://localhost:8001/api/executions/{id}/results')
```
### 5. 迭代修复(2-3 轮)
- 首轮失败通常是导航/点击 → 改为 navigate 直达
- 后续失败是断言选择器不匹配 → 修正为真实 DOM 选择器
- 修复直接写脚本操作 DB(`UPDATE test_cases SET steps=? WHERE id=?`
- 每轮创建新执行,直到 100% 通过
## 踩坑清单(务必对照)
1. **Element UI 下拉框**`el-select` 渲染为 `<input readonly>`,断言用
`input[readonly][value="日"]` 优先,兜底 `input[readonly][placeholder*="选择"]`
2. **图标按钮**:查询/导出按钮是纯图标(`<i class="el-icon-download">`)无文本,
`.el-icon-download``button.el-button--primary`
3. **navigate 直达**:首页 `.block` 菜单 headless 下不稳定,首步一律 navigate
4. **Hash URL 复制**:双重 hash 路由(`%2F`=`/``%23`=`#`),从浏览器地址栏复制完整 URL
5. **curl 中文 body 400**:Git Bash 会破坏 UTF-8,改用 `python requests.post(json=body)`
6. **config 必配 auto_login**`{"auto_login": true, "headless": true}`
7. **page_url_mapping.json**:新增页面条目后需**重启 uvicorn**(单例不热加载)
8. **登录有协议弹窗**:登录后可能弹"同意《统一登录服务协议》"对话框,需点"确定"
详见 [[sut-explore-pitfalls]](完整踩坑记录)。
## 安全规则
- 只探测、不修改 SUT 数据(筛选查询可做,删除/新增真实数据不可)
- 登录凭据不写入用例文件本身(在 config 或凭据中引用)
- 生成的脚本放在 `backend/scripts/`,可在本地重复执行
\ No newline at end of file
此差异已折叠。
......@@ -3013,6 +3013,18 @@ class PlaywrightExecutor:
total += len(elements)
except Exception:
continue
# 主文档/iframe 命中不足时,轮询 micro-app 内元素计数兜底
# (@micro-zoe/micro-app 内容渲染在 <micro-app-body> 自定义元素内,
# query_selector_all 无法穿透,需 evaluate 内部统计)
if total < min_count:
deadline = time.time() + 8
while time.time() < deadline:
micro_count = self._do_in_micro_apps('count', selector, timeout=1000)
if isinstance(micro_count, int) and micro_count > 0:
total = max(total, micro_count)
if total >= min_count:
break
self._page.wait_for_timeout(800)
if total < min_count:
raise AssertionError(f"✗ 断言失败: 期望至少 {min_count} 个元素,实际 {total} 个: {selectors}")
......
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
模块名称:generate_asset_info_cases
模块描述:将资产信息 3 个用例改为 navigate 直达 Assetinformation URL 模式
作者:czj
创建日期:2026-08-21
最后修改日期:2026-08-21
"""
import json
import sqlite3
import sys
from pathlib import Path
sys.stdout.reconfigure(encoding='utf-8')
import os
os.chdir(Path(__file__).parent.parent)
ASSET_INFO_URL = "https://192.168.5.44/#/meetingV3?meetingV3=%2FmeetingV3%2F%23%2FAssetinformation"
def make_step(order, name, action, params, expected=""):
return {
"order": order,
"name": name,
"action": action,
"params": params,
"expected": expected,
"selectors": None,
"page_key": None,
"element_key": None,
"force": None,
"wait_after": None,
}
UPDATES = {}
# 1) 资产信息-页面访问验证
# 真实DOM:input[placeholder='请输入关键字查询']、input[placeholder='请选择状态'](readonly)、
# 按钮 添加条目/下载模板/导入表格/批量报修/批量报废/批量删除、.el-table
# 表格有 26 条数据,选择器 .el-table__row
UPDATES["case_eac47c7888dd425dbb18ebca95ffeb25"] = [
make_step(1, "导航到资产信息页面", "navigate", {"url": ASSET_INFO_URL}),
make_step(2, "等待页面加载", "wait", {"selector": "body", "timeout": 15000}),
make_step(3, "验证关键字查询框存在", "assert", {"type": "element_exists", "selector": "input[placeholder='请输入关键字查询']"}),
make_step(4, "验证状态筛选框存在", "assert", {"type": "element_exists", "selector": "input[placeholder='请选择状态']"}),
make_step(5, "验证添加条目按钮存在", "assert", {"type": "element_exists", "selector": "button:has-text('添加条目')"}),
make_step(6, "验证资产表格存在", "assert", {"type": "element_exists", "selector": ".el-table"}),
]
# 2) 资产信息-表格数据验证
UPDATES["case_f4f9239c31d34fc0909980eafc5f27df"] = [
make_step(1, "导航到资产信息页面", "navigate", {"url": ASSET_INFO_URL}),
make_step(2, "等待页面加载", "wait", {"selector": "body", "timeout": 15000}),
make_step(3, "验证表格有数据行", "assert", {"type": "element_count_min", "selector": ".el-table__row", "expected": "1"}),
make_step(4, "验证品牌列存在", "assert", {"type": "element_exists", "selector": "text=品牌"}),
make_step(5, "验证型号列存在", "assert", {"type": "element_exists", "selector": "text=型号"}),
make_step(6, "验证编辑操作按钮存在", "assert", {"type": "element_exists", "selector": "button:has-text('编辑')"}),
make_step(7, "验证删除操作按钮存在", "assert", {"type": "element_exists", "selector": "button:has-text('删除')"}),
]
# 3) 资产信息-筛选器交互验证
# 状态下拉框:input[placeholder='请选择状态'] readonly
# 关键字查询:input[placeholder='请输入关键字查询']
UPDATES["case_bad683575a0f4dbb9b784de63f8624e2"] = [
make_step(1, "导航到资产信息页面", "navigate", {"url": ASSET_INFO_URL}),
make_step(2, "等待页面加载", "wait", {"selector": "body", "timeout": 15000}),
make_step(3, "验证关键字查询框存在", "assert", {"type": "element_exists", "selector": "input[placeholder='请输入关键字查询']"}),
make_step(4, "填写关键字", "fill", {"selector": "input[placeholder='请输入关键字查询']", "value": "测试"}),
make_step(5, "点击确定搜索", "click", {"selector": "button:has-text('确定')"}),
make_step(6, "等待搜索结果", "wait", {"selector": "body", "timeout": 5000}),
]
NOW = "2026-08-21 16:00:00"
def main():
conn = sqlite3.connect("data/test_platform.db")
conn.row_factory = sqlite3.Row
cur = conn.cursor()
for cid, steps in UPDATES.items():
steps_json = json.dumps(steps, ensure_ascii=False)
cur.execute(
"UPDATE test_cases SET steps=?, updated_at=?, version=version+1 WHERE id=?",
(steps_json, NOW, cid),
)
cur.execute("SELECT name FROM test_cases WHERE id=?", (cid,))
r = cur.fetchone()
print(f" ✅ {cid} | {r['name']} | steps={len(steps)}")
conn.commit()
conn.close()
print(f"\n✅ 共更新 {len(UPDATES)} 个用例")
if __name__ == "__main__":
main()
\ No newline at end of file
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
模块名称:generate_meetingroom_cases
模块描述:将会议室列表 7 个用例改为 navigate 直达 MeetingRoomList URL 模式
作者:czj
创建日期:2026-08-21
最后修改日期:2026-08-21
"""
import json
import sqlite3
import sys
from pathlib import Path
sys.stdout.reconfigure(encoding='utf-8')
import os
os.chdir(Path(__file__).parent.parent)
MEETING_ROOM_LIST_URL = "https://192.168.5.44/#/meetingV3?meetingV3=%2FmeetingV3%2F%23%2FMeetingRoomList"
def make_step(order, name, action, params, expected=""):
return {
"order": order,
"name": name,
"action": action,
"params": params,
"expected": expected,
"selectors": None,
"page_key": None,
"element_key": None,
"force": None,
"wait_after": None,
}
UPDATES = {}
# 1) 会议室列表页面访问验证 — 验证页面基本元素
UPDATES["case_05d43343a6b34910a86c080450225dfc"] = [
make_step(1, "导航到会议室列表页面", "navigate", {"url": MEETING_ROOM_LIST_URL}),
make_step(2, "等待页面加载", "wait", {"selector": "body", "timeout": 15000}),
make_step(3, "验证日期输入框存在", "assert", {"type": "element_exists", "selector": "input[placeholder='日期']"}),
make_step(4, "验证会议室名称搜索框存在", "assert", {"type": "element_exists", "selector": "input[placeholder='会议室名称']"}),
make_step(5, "验证新建会议按钮存在", "assert", {"type": "element_exists", "selector": "button:has-text('新建会议')"}),
make_step(6, "验证有会议室列表卡片", "assert", {"type": "element_count_min", "selector": ".room_item", "expected": "1"}),
]
# 2) 会议室列表-查看详情 — 验证页面元素和预览功能
UPDATES["case_98abcdfcd8c644e8a086dc4444254b00"] = [
make_step(1, "导航到会议室列表页面", "navigate", {"url": MEETING_ROOM_LIST_URL}),
make_step(2, "等待页面加载", "wait", {"selector": "body", "timeout": 15000}),
make_step(3, "验证日期输入框存在", "assert", {"type": "element_exists", "selector": "input[placeholder='日期']"}),
make_step(4, "验证预览按钮存在", "assert", {"type": "element_exists", "selector": "button:has-text('预览')"}),
make_step(5, "验证有会议室列表卡片", "assert", {"type": "element_count_min", "selector": ".room_item", "expected": "1"}),
]
# 3) 会议室列表-预览功能 — 点击预览查看弹窗
UPDATES["case_5e6a80d82ea646c5b00c663aafe85494"] = [
make_step(1, "导航到会议室列表页面", "navigate", {"url": MEETING_ROOM_LIST_URL}),
make_step(2, "等待页面加载", "wait", {"selector": "body", "timeout": 15000}),
make_step(3, "验证预览按钮存在", "assert", {"type": "element_exists", "selector": "button:has-text('预览')"}),
make_step(4, "点击第一个预览按钮", "click", {"selector": "button:has-text('预览')"}),
make_step(5, "等待预览弹窗", "wait", {"selector": ".el-dialog:visible", "timeout": 5000}),
make_step(6, "验证预览弹窗存在", "assert", {"type": "element_exists", "selector": ".el-dialog"}),
make_step(7, "关闭预览弹窗", "click", {"selector": ".el-dialog__headerbtn"}),
]
# 4) 会议室列表-搜索功能 — 搜索会议室名称
UPDATES["case_099ce65993ea47b98aca16ca5adcb796"] = [
make_step(1, "导航到会议室列表页面", "navigate", {"url": MEETING_ROOM_LIST_URL}),
make_step(2, "等待页面加载", "wait", {"selector": "body", "timeout": 15000}),
make_step(3, "验证搜索框存在", "assert", {"type": "element_exists", "selector": "input[placeholder='会议室名称']"}),
make_step(4, "填写会议室名称", "fill", {"selector": "input[placeholder='会议室名称']", "value": "会议室"}),
make_step(5, "点击确定搜索", "click", {"selector": "button:has-text('确定')"}),
make_step(6, "等待搜索结果", "wait", {"selector": "body", "timeout": 5000}),
]
# 5) 会议室列表-筛选功能验证 — 条件筛选交互
UPDATES["case_98bdb3f6b5054f3db383bd8a6aeda2b2"] = [
make_step(1, "导航到会议室列表页面", "navigate", {"url": MEETING_ROOM_LIST_URL}),
make_step(2, "等待页面加载", "wait", {"selector": "body", "timeout": 15000}),
make_step(3, "验证有筛选输入框", "assert", {"type": "element_count_min", "selector": "input.el-input__inner", "expected": "2"}),
make_step(4, "填写会议室名称", "fill", {"selector": "input[placeholder='会议室名称']", "value": "会议室"}),
make_step(5, "点击确定筛选", "click", {"selector": "button:has-text('确定')"}),
make_step(6, "等待筛选结果", "wait", {"selector": "body", "timeout": 5000}),
]
# 6) 会议室列表-预约会议验证 — 点击新建会议跳转到新建会议页
UPDATES["case_6e00f1983f91441395f468cf967c49ba"] = [
make_step(1, "导航到会议室列表页面", "navigate", {"url": MEETING_ROOM_LIST_URL}),
make_step(2, "等待页面加载", "wait", {"selector": "body", "timeout": 15000}),
make_step(3, "点击新建会议按钮", "click", {"selector": "button:has-text('新建会议')"}),
make_step(4, "等待跳转到新建会议", "wait", {"selector": "input[placeholder='会议名称/会议议题']", "timeout": 10000}),
make_step(5, "验证进入新建会议页面", "assert", {"type": "element_exists", "selector": "input[placeholder='会议名称/会议议题']"}),
]
# 7) 会议室列表-预览详情验证 — 点击预览查看弹窗详情
UPDATES["case_8e89f99b86ba441ca2030beac86ec93f"] = [
make_step(1, "导航到会议室列表页面", "navigate", {"url": MEETING_ROOM_LIST_URL}),
make_step(2, "等待页面加载", "wait", {"selector": "body", "timeout": 15000}),
make_step(3, "验证预览按钮存在", "assert", {"type": "element_exists", "selector": "button:has-text('预览')"}),
make_step(4, "点击第一个预览按钮", "click", {"selector": "button:has-text('预览')"}),
make_step(5, "等待预览弹窗", "wait", {"selector": ".el-dialog:visible", "timeout": 5000}),
make_step(6, "验证弹窗有会议室名称", "assert", {"type": "element_exists", "selector": ".el-dialog:visible .room_name, .el-dialog:visible .el-dialog__body"}),
make_step(7, "关闭预览弹窗", "click", {"selector": ".el-dialog__headerbtn"}),
]
NOW = "2026-08-21 12:00:00"
def main():
conn = sqlite3.connect("data/test_platform.db")
conn.row_factory = sqlite3.Row
cur = conn.cursor()
for cid, steps in UPDATES.items():
steps_json = json.dumps(steps, ensure_ascii=False)
cur.execute(
"UPDATE test_cases SET steps=?, updated_at=?, version=version+1 WHERE id=?",
(steps_json, NOW, cid),
)
cur.execute("SELECT name FROM test_cases WHERE id=?", (cid,))
r = cur.fetchone()
print(f" ✅ {cid} | {r['name']} | steps={len(steps)}")
conn.commit()
conn.close()
print(f"\n✅ 共更新 {len(UPDATES)} 个用例")
if __name__ == "__main__":
main()
\ No newline at end of file
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
同步脚本 — 将本地所有增量内容同步到 5.60 生产服务器
包括:
1. 后端文件(playwright_executor.py / smart_locate_service.py / execution_service.py / scheduled_tasks.py)
2. page_url_mapping.json(25 页)
3. 用例数据(本地有服务器没有的 18 条 + 更新已有)
4. 重启容器
用法:
cd backend && python scripts/sync_to_560_full.py
"""
import json
import os
import sys
import time
from pathlib import Path
sys.stdout.reconfigure(encoding='utf-8')
os.chdir(Path(__file__).parent.parent) # 切到 backend/
import paramiko
import pymysql
SERVER = ('192.168.5.60', 'ubains', 'Ubains@123')
MYSQL_PORT = 3307
MYSQL_USER = 'platapp'
MYSQL_PWD = 'PlatApp2026'
MYSQL_DB = 'plat_auto_test'
REMOTE_BASE = '/data/third_party/plat-auto-test'
REMOTE_BACKEND = f'{REMOTE_BASE}/backend'
# 需要同步的后端文件(本地路径 → 远程相对路径)
BACKEND_FILES = {
'app/executors/playwright_executor.py': 'app/executors/playwright_executor.py',
'app/services/smart_locate_service.py': 'app/services/smart_locate_service.py',
'app/services/execution_service.py': 'app/services/execution_service.py',
'app/routers/scheduled_tasks.py': 'app/routers/scheduled_tasks.py',
'app/data/page_url_mapping.json': 'app/data/page_url_mapping.json',
}
INSERT_SQL = (
"INSERT INTO test_cases "
"(id, module_id, name, description, status, priority, tags, steps, config, "
"depends_on, parameters, `condition`, `order`, version, created_at, updated_at, case_type) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
"ON DUPLICATE KEY UPDATE "
"name=VALUES(name), description=VALUES(description), priority=VALUES(priority), "
"tags=VALUES(tags), steps=VALUES(steps), config=VALUES(config), "
"depends_on=VALUES(depends_on), parameters=VALUES(parameters), "
"`condition`=VALUES(`condition`), `order`=VALUES(`order`), version=VALUES(version), "
"updated_at=VALUES(updated_at), case_type=VALUES(case_type)"
)
def load_local_cases():
"""从本地 SQLite 加载所有用例"""
import sqlite3
conn = sqlite3.connect('data/test_platform.db')
conn.row_factory = sqlite3.Row
cur = conn.cursor()
cur.execute("SELECT * FROM test_cases ORDER BY module_id, name")
rows = [dict(r) for r in cur.fetchall()]
conn.close()
return rows
def load_local_mapping():
with open('app/data/page_url_mapping.json', encoding='utf-8') as f:
return json.load(f)
def sync_backend_files():
"""通过 SFTP 上传后端文件到服务器"""
cli = paramiko.SSHClient()
cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
cli.connect(SERVER[0], username=SERVER[1], password=SERVER[2], timeout=10)
sftp = cli.open_sftp()
results = {}
try:
for local_rel, remote_rel in BACKEND_FILES.items():
local_path = local_rel
remote_path = f'{REMOTE_BACKEND}/{remote_rel}'
try:
sftp.put(local_path, remote_path)
size = os.path.getsize(local_path)
results[local_rel] = f'OK ({size}B)'
except Exception as e:
results[local_rel] = f'FAIL: {e}'
finally:
sftp.close()
cli.close()
return results
def get_remote_case_ids():
"""获取服务器 MySQL 中所有用例 ID"""
conn = pymysql.connect(host=SERVER[0], port=MYSQL_PORT, user=MYSQL_USER,
password=MYSQL_PWD, database=MYSQL_DB,
charset='utf8mb4', connect_timeout=8)
try:
cur = conn.cursor()
cur.execute("SELECT id, module_id, name FROM test_cases")
return {r[0]: {'module_id': r[1], 'name': r[2]} for r in cur.fetchall()}
finally:
conn.close()
def sync_cases(local_cases, remote_ids):
"""批量 UPSERT 用例到服务器 MySQL"""
conn = pymysql.connect(host=SERVER[0], port=MYSQL_PORT, user=MYSQL_USER,
password=MYSQL_PWD, database=MYSQL_DB,
charset='utf8mb4', connect_timeout=8)
try:
cur = conn.cursor()
inserted = updated = skipped = 0
for c in local_cases:
cid = c['id']
if cid in remote_ids:
# 检查是否有变化(简单判断 updated_at 不同则更新)
# 由于 MySQL datetime 精度可能不同,直接 UPSERT
cur.execute(INSERT_SQL, (
cid, c['module_id'], c['name'], c.get('description', ''),
c['status'], c['priority'], c['tags'], c['steps'], c['config'],
c['depends_on'], c['parameters'], c['condition'],
c['order'], c['version'], c['created_at'], c['updated_at'],
c['case_type'],
))
if cur.rowcount == 1:
inserted += 1
else:
updated += 1
else:
cur.execute(INSERT_SQL, (
cid, c['module_id'], c['name'], c.get('description', ''),
c['status'], c['priority'], c['tags'], c['steps'], c['config'],
c['depends_on'], c['parameters'], c['condition'],
c['order'], c['version'], c['created_at'], c['updated_at'],
c['case_type'],
))
inserted += 1
conn.commit()
return inserted, updated
finally:
conn.close()
def restart_container():
"""重启 5.60 容器"""
cli = paramiko.SSHClient()
cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
cli.connect(SERVER[0], username=SERVER[1], password=SERVER[2], timeout=10)
try:
stdin, stdout, stderr = cli.exec_command(
'cd /data/third_party/plat-auto-test/deploy && docker compose restart app',
timeout=60
)
out = stdout.read().decode('utf-8', 'replace')
err = stderr.read().decode('utf-8', 'replace')
return out.strip(), err.strip()
finally:
cli.close()
def wait_healthy(retries=12, interval=5):
"""等待容器健康"""
import urllib.request
for i in range(retries):
try:
resp = urllib.request.urlopen('http://192.168.5.60/health', timeout=5)
if resp.status == 200:
return True, resp.read().decode('utf-8')
except Exception:
pass
print(f' 等待容器启动... ({i+1}/{retries})')
time.sleep(interval)
return False, ''
def main():
print('===== 1. 同步后端文件 =====')
file_results = sync_backend_files()
for f, r in file_results.items():
print(f' {f}: {r}')
print('\n===== 2. 同步用例数据 =====')
local_cases = load_local_cases()
remote_ids = get_remote_case_ids()
print(f' 本地: {len(local_cases)} 条, 服务器: {len(remote_ids)} 条')
# 统计差异
only_local = [c for c in local_cases if c['id'] not in remote_ids]
common_ids = [c['id'] for c in local_cases if c['id'] in remote_ids]
print(f' 本地新增: {len(only_local)} 条, 共同: {len(common_ids)} 条')
inserted, updated = sync_cases(local_cases, remote_ids)
print(f' 同步结果: 新增 {inserted} 条, 更新 {updated} 条')
# 验证
remote_ids_after = get_remote_case_ids()
print(f' 同步后服务器: {len(remote_ids_after)} 条')
# 按模块统计
from collections import Counter
module_cnt = Counter()
for c in local_cases:
if c['id'] not in remote_ids:
module_cnt[c['module_id']] += 1
if module_cnt:
print(f' 新增用例按模块分布:')
for mid, cnt in module_cnt.most_common():
print(f' {mid}: {cnt} 条')
print('\n===== 3. 重启容器 =====')
out, err = restart_container()
print(f' docker compose restart 输出: {out or "(空)"}')
if err:
print(f' stderr: {err}')
print('\n===== 4. 等待健康检查 =====')
ok, body = wait_healthy()
if ok:
print(f' ✅ 容器健康: {body[:100]}')
else:
print(f' ❌ 容器未在预期时间内恢复健康')
print('\n===== 同步完成 =====')
if __name__ == '__main__':
main()
\ No newline at end of file
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论