影刀RPA 自动化测试入门：用RPA做冒烟测试

新建流程「系统接口健康检查」

添加【Python】指令

import requests
import json
import time
from datetime import datetime

# 定义要测试的接口列表
test_cases = [
{
'name': '用户登录接口',
'method': 'POST',
'url': 'https://api.yourapp.com/auth/login',
'body': {'username': 'test_user', 'password': 'test_pass'},
'expected_status': 200,
'expected_fields': ['token', 'user_id'], # 返回JSON必须包含这些字段
'timeout': 5
},
{
'name': '商品列表接口',
'method': 'GET',
'url': 'https://api.yourapp.com/products?page=1&pageSize=10',
'headers': {'Authorization': 'Bearer test_token'},
'expected_status': 200,
'expected_fields': ['data', 'total'],
'timeout': 5
},
{
'name': '下单接口',
'method': 'POST',
'url': 'https://api.yourapp.com/orders',
'body': {'product_id': 1, 'quantity': 1, 'address_id': 1},
'expected_status': 201,
'expected_fields': ['order_id'],
'timeout': 10
}
]

results = []

for tc in test_cases:
start_time = time.time()
result = {'name': tc['name'], 'status': 'PASS', 'error': None, 'response_time': None}

try:
headers = tc.get('headers', {})
headers['Content-Type'] = 'application/json'

if tc['method'] == 'GET':
resp = requests.get(tc['url'], headers=headers, timeout=tc.get('timeout', 5))
else:
resp = requests.request(
tc['method'], tc['url'],
headers=headers,
json=tc.get('body'),
timeout=tc.get('timeout', 5)
)

result['response_time'] = round(time.time() – start_time, 3)
result['status_code'] = resp.status_code

# 检查状态码
if resp.status_code != tc['expected_status']:
result['status'] = 'FAIL'
result['error'] = f'状态码不对：期望{tc["expected_status"]}，实际{resp.status_code}'
else:
# 检查返回字段
try:
body = resp.json()
missing_fields = [f for f in tc.get('expected_fields', []) if f not in body]
if missing_fields:
result['status'] = 'FAIL'
result['error'] = f'返回数据缺少字段：{missing_fields}'
except:
result['status'] = 'WARN'
result['error'] = '返回内容不是JSON'

except requests.Timeout:
result['status'] = 'FAIL'
result['error'] = f'接口超时（>{tc.get("timeout", 5)}秒）'
except Exception as e:
result['status'] = 'FAIL'
result['error'] = str(e)

results.append(result)
icon = '✅' if result['status'] == 'PASS' else '❌'
print(f'{icon} {tc["name"]}: {result["status"]} ({result.get("response_time", "N/A")}s)')
if result['error']:
print(f' 错误：{result["error"]}')

# 汇总
pass_count = sum(1 for r in results if r['status'] == 'PASS')
fail_count = sum(1 for r in results if r['status'] == 'FAIL')

print(f'\\n测试结果：{pass_count}通过，{fail_count}失败，共{len(results)}项')

yda.set_variable('test_results', json.dumps(results, ensure_ascii=False))
yda.set_variable('has_failure', fail_count > 0)
yda.set_variable('pass_count', pass_count)
yda.set_variable('fail_count', fail_count)

添加【条件判断】指令

• 判断条件：变量 has_failure 等于 True
• 如果为真：发送告警通知
• 如果为假：不操作（一切正常）

• URL：https://yourapp.com/login

• 定位：用户名输入框
• 内容：测试账号

• 定位：密码输入框
• 内容：测试密码

# 登录成功应该跳转到首页
if '/dashboard' in current_url or '/home' in current_url:
print('✅ 登录功能正常，跳转到首页')
yda.set_variable('login_test_pass', True)
elif '/login' in current_url:
# 还在登录页，登录失败
error_msg = ''
try:
error_elem = page.locator('.error-message, .alert-danger').first
if error_elem.is_visible():
error_msg = error_elem.text_content()
except:
pass
print(f'❌ 登录失败，当前URL：{current_url}，错误信息：{error_msg}')
yda.set_variable('login_test_pass', False)
yda.set_variable('login_error', error_msg)

results = json.loads(yda.get_variable('test_results'))
pass_count = yda.get_variable('pass_count')
fail_count = yda.get_variable('fail_count')

# 生成HTML测试报告
status_emoji = '🔴' if fail_count > 0 else '🟢'

rows = ''
for r in results:
status_class = 'color:green' if r['status'] == 'PASS' else 'color:red'
rows += f"""<tr>
<td>{r['name']}</td>
<td style="{status_class}"><b>{r['status']}</b></td>
<td>{r.get('response_time', 'N/A')}s</td>
<td>{r.get('error', '')}</td>
</tr>"""

report = f"""
{status_emoji} 系统健康检查 – {datetime.now().strftime('%Y-%m-%d %H:%M')}
通过：{pass_count}，失败：{fail_count}

{'⚠️ 有以下接口异常，请及时处理：' if fail_count > 0 else '✅ 所有接口正常'}
""" + '\\n'.join([f'- {r["name"]}: {r["error"]}' for r in results if r['status'] == 'FAIL'])

# 发到企微群
webhook_url = 'your_webhook_url'
requests.post(webhook_url, json={'msgtype': 'text', 'text': {'content': report}})
print('测试报告已发送')