影刀RPA 自动化测试入门：用RPA做接口和界面回归测试

作者：林焱

---

写在前面

影刀RPA不只能做数据采集和办公自动化，还能做自动化测试。对于没有专职QA团队的小团队来说，用影刀搭建一套轻量的接口+界面回归测试，成本极低，效果明显。

本文讲清楚两类测试的搭建方法：接口自动化测试（验证后端接口正确性）和界面自动化测试（验证前端页面交互）。

---

一、接口自动化测试

测试框架思路

import requests
import json
import datetime

class ApiTestRunner:
    """简单的接口测试框架"""
    
    def __init__(self, base_url, auth_token=None):
        self.base_url = base_url
        self.headers = {"Content-Type": "application/json"}
        if auth_token:
            self.headers["Authorization"] = f"Bearer {auth_token}"
        self.results = []
    
    def run_test(self, test_name, method, path, payload=None, 
                 expected_status=200, expected_fields=None, expected_values=None):
        """
        执行单个接口测试
        expected_fields: 期望响应中存在的字段列表
        expected_values: 期望的键值对，如 {"code": 0, "message": "success"}
        """
        url = f"{self.base_url}{path}"
        
        try:
            if method.upper() == "GET":
                resp = requests.get(url, headers=self.headers, params=payload)
            elif method.upper() == "POST":
                resp = requests.post(url, headers=self.headers, json=payload)
            elif method.upper() == "PUT":
                resp = requests.put(url, headers=self.headers, json=payload)
            elif method.upper() == "DELETE":
                resp = requests.delete(url, headers=self.headers)
            
            result = {
                "test_name": test_name,
                "url": url,
                "method": method,
                "status_code": resp.status_code,
                "response_time_ms": round(resp.elapsed.total_seconds() * 1000),
                "passed": True,
                "errors": []
            }
            
            # 检查状态码
            if resp.status_code != expected_status:
                result["passed"] = False
                result["errors"].append(f"状态码期望{expected_status}，实际{resp.status_code}")
            
            # 解析响应
            try:
                resp_data = resp.json()
            except:
                resp_data = {}
                if expected_fields or expected_values:
                    result["passed"] = False
                    result["errors"].append("响应不是有效JSON")
            
            # 检查字段存在
            if expected_fields:
                for field in expected_fields:
                    if field not in resp_data:
                        result["passed"] = False
                        result["errors"].append(f"响应缺少字段：{field}")
            
            # 检查字段值
            if expected_values:
                for key, expected_val in expected_values.items():
                    actual_val = resp_data.get(key)
                    if actual_val != expected_val:
                        result["passed"] = False
                        result["errors"].append(f"字段{key}期望{expected_val}，实际{actual_val}")
            
            result["response_body"] = str(resp_data)[:500]  # 记录部分响应
            
        except Exception as e:
            result = {
                "test_name": test_name,
                "url": url,
                "method": method,
                "passed": False,
                "errors": [str(e)],
                "response_time_ms": 0,
                "response_body": ""
            }
        
        self.results.append(result)
        status = "✅ PASS" if result["passed"] else "❌ FAIL"
        print(f"{status} [{result.get('response_time_ms', 0)}ms] {test_name}")
        if not result["passed"]:
            for err in result["errors"]:
                print(f"    ↳ {err}")
        
        return result
    
    def generate_report(self, output_file="test_report.xlsx"):
        """生成测试报告"""
        import openpyxl
        from openpyxl.styles import PatternFill, Font
        
        wb = openpyxl.Workbook()
        ws = wb.active
        ws.title = "测试结果"
        
        # 统计
        total = len(self.results)
        passed = sum(1 for r in self.results if r["passed"])
        failed = total - passed
        
        ws.append(["测试报告", f"生成时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"])
        ws.append([f"总用例：{total}", f"通过：{passed}", f"失败：{failed}"])
        ws.append([])
        ws.append(["测试名称", "请求方法", "URL", "状态码", "响应时间(ms)", "结果", "错误信息"])
        
        green_fill = PatternFill(start_color="90EE90", fill_type="solid")
        red_fill = PatternFill(start_color="FF6B6B", fill_type="solid")
        
        for result in self.results:
            row = [
                result["test_name"],
                result.get("method", ""),
                result.get("url", ""),
                result.get("status_code", ""),
                result.get("response_time_ms", ""),
                "PASS" if result["passed"] else "FAIL",
                "; ".join(result.get("errors", []))
            ]
            ws.append(row)
            # 着色
            fill = green_fill if result["passed"] else red_fill
            for cell in ws[ws.max_row]:
                cell.fill = fill
        
        wb.save(output_file)
        print(f"\n📊 测试报告已生成：{output_file}")
        print(f"通过率：{passed}/{total} ({round(passed/total*100)}%)")

定义并执行测试用例

# 初始化测试运行器
runner = ApiTestRunner("https://api.yourapp.com", auth_token="你的token")

# 用户模块测试
runner.run_test("获取用户信息", "GET", "/api/user/profile",
    expected_status=200,
    expected_fields=["id", "name", "email"],
    expected_values={"code": 0}
)

runner.run_test("更新用户信息", "PUT", "/api/user/profile",
    payload={"name": "测试用户", "phone": "13800138000"},
    expected_status=200,
    expected_values={"code": 0, "message": "更新成功"}
)

# 订单模块测试
runner.run_test("创建订单", "POST", "/api/orders",
    payload={
        "product_id": "PROD_001",
        "quantity": 2,
        "address": "北京市朝阳区xxx"
    },
    expected_status=201,
    expected_fields=["order_id", "status", "created_at"]
)

runner.run_test("获取订单列表", "GET", "/api/orders",
    payload={"page": 1, "size": 10},
    expected_status=200,
    expected_fields=["total", "items"]
)

# 边界条件测试
runner.run_test("越权访问测试", "GET", "/api/admin/users",
    expected_status=403  # 期望返回403禁止访问
)

runner.run_test("无效参数测试", "POST", "/api/orders",
    payload={"product_id": "", "quantity": -1},
    expected_status=400  # 期望返回400参数错误
)

# 生成报告
runner.generate_report("api_test_report.xlsx")

---

二、界面自动化测试

界面测试核心方法

import pyautogui
import time

class UITestRunner:
    """界面自动化测试框架"""
    
    def __init__(self):
        self.results = []
        pyautogui.FAILSAFE = True
        pyautogui.PAUSE = 0.3
    
    def find_element(self, image_file, timeout=10, confidence=0.85):
        """等待并查找界面元素"""
        start = time.time()
        while time.time() - start < timeout:
            location = pyautogui.locateOnScreen(image_file, confidence=confidence)
            if location:
                return location
            time.sleep(0.5)
        return None
    
    def click_element(self, image_file, test_step, timeout=10):
        """点击界面元素"""
        location = self.find_element(image_file, timeout)
        if location:
            pyautogui.click(pyautogui.center(location))
            return True
        print(f"⚠️ 步骤失败：{test_step} - 未找到元素 {image_file}")
        return False
    
    def input_text(self, image_file, text, clear_first=True):
        """在输入框中输入文本"""
        location = self.find_element(image_file)
        if location:
            pyautogui.click(pyautogui.center(location))
            if clear_first:
                pyautogui.hotkey('ctrl', 'a')
            pyautogui.typewrite(text, interval=0.05)
            return True
        return False
    
    def verify_text_visible(self, expected_text, screenshot=None):
        """验证文字是否在界面上可见（OCR识别）"""
        # 使用影刀的OCR指令或pytesseract
        try:
            import pytesseract
            from PIL import Image
            
            if screenshot is None:
                screenshot = pyautogui.screenshot()
            
            text_found = pytesseract.image_to_string(screenshot, lang='chi_sim+eng')
            return expected_text in text_found
        except ImportError:
            # 如果没有tesseract，用图像匹配替代
            print("提示：安装pytesseract可以做更精确的文字识别")
            return False
    
    def run_test_case(self, test_name, test_func):
        """执行测试用例并记录结果"""
        print(f"\n▶ 执行测试：{test_name}")
        start_time = time.time()
        
        try:
            success = test_func()
            elapsed = round((time.time() - start_time) * 1000)
            result = {"name": test_name, "passed": success, "time_ms": elapsed, "error": ""}
            print(f"{'✅ PASS' if success else '❌ FAIL'} [{elapsed}ms]")
        except Exception as e:
            elapsed = round((time.time() - start_time) * 1000)
            result = {"name": test_name, "passed": False, "time_ms": elapsed, "error": str(e)}
            print(f"❌ ERROR [{elapsed}ms]: {e}")
            # 截图保存错误现场
            pyautogui.screenshot(f"test_error_{test_name}_{int(time.time())}.png")
        
        self.results.append(result)
        return result["passed"]

示例：登录功能回归测试

ui_runner = UITestRunner()

def test_login_success():
    """测试正常登录流程"""
    # 点击登录按钮
    if not ui_runner.click_element("login_btn.png", "点击登录按钮"):
        return False
    time.sleep(1)
    
    # 输入用户名
    if not ui_runner.input_text("username_field.png", "test@example.com"):
        return False
    
    # 输入密码
    if not ui_runner.input_text("password_field.png", "Test@12345"):
        return False
    
    # 点击确认
    if not ui_runner.click_element("submit_btn.png", "点击提交"):
        return False
    time.sleep(2)
    
    # 验证登录成功（检查首页元素是否出现）
    dashboard = ui_runner.find_element("dashboard_title.png", timeout=10)
    return dashboard is not None

def test_login_with_wrong_password():
    """测试错误密码时的提示"""
    if not ui_runner.click_element("login_btn.png", "点击登录"):
        return False
    
    ui_runner.input_text("username_field.png", "test@example.com")
    ui_runner.input_text("password_field.png", "wrongpassword")
    ui_runner.click_element("submit_btn.png", "提交")
    time.sleep(1)
    
    # 检查是否出现错误提示
    error_msg = ui_runner.find_element("error_message.png", timeout=5)
    return error_msg is not None

# 执行测试套件
ui_runner.run_test_case("正常登录", test_login_success)
ui_runner.run_test_case("错误密码登录", test_login_with_wrong_password)

# 打印结果
passed = sum(1 for r in ui_runner.results if r["passed"])
print(f"\n=== 测试完成 ===")
print(f"通过：{passed}/{len(ui_runner.results)}")

---

三、持续集成：每次发版前自动运行测试

在影刀中创建一个定时任务，在每次发布前（比如下午5点）自动运行测试套件：

import subprocess
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders

def run_all_tests_and_notify():
    """运行所有测试并通过邮件发送报告"""
    print("开始执行自动化测试套件...")
    
    # 执行接口测试
    runner = ApiTestRunner("https://api.yourapp.com", "token")
    # ... 定义所有用例 ...
    runner.generate_report("api_report.xlsx")
    
    total = len(runner.results)
    passed = sum(1 for r in runner.results if r["passed"])
    
    # 发送邮件通知
    if passed < total:
        subject = f"⚠️ 自动化测试失败 - {passed}/{total}通过"
    else:
        subject = f"✅ 自动化测试全通过 - {total}/{total}"
    
    # 通过影刀邮件指令或Python smtplib发送
    print(f"测试完成：{subject}")

run_all_tests_and_notify()

---

四、踩坑记录

坑1：图像识别受分辨率和缩放影响
UI测试用的图片素材必须在测试运行的机器上截取，屏幕缩放比例（125%/150%）会导致图像匹配失败。建议将系统缩放设为100%来截图。

坑2：测试环境数据污染
自动化测试会创建测试订单、测试用户等数据。需要有专门的测试环境或测试完毕后清理数据，不要在生产环境跑测试。

坑3：界面测试速度太慢
界面测试比接口测试慢10-20倍。全量界面回归建议在凌晨离峰时间执行，平时只执行核心路径（冒烟测试）。

坑4：弹窗和loading遮挡
页面加载动画或意外弹窗会导致元素找不到。在关键操作后加time.sleep(2)，或等待loading消失再继续。

---

总结

用影刀做自动化测试的核心价值：不需要专职QA，发版前10分钟自动跑完回归测试，有问题立即通知。接口测试优先（快、准、稳），界面测试补充核心流程。

署名：林焱