影刀RPA自动化测试：让你的机器人更可靠的完整测试指南

作者：林焱
标签： 影刀RPA、自动化测试、单元测试、回归测试、RPA质量保障
适合读者： 希望提升RPA流程可靠性、减少上线后故障的开发者

---

为什么RPA也需要测试？

软件开发有测试已成共识，但很多RPA开发者认为"RPA是自动化工具，不需要测试"。

这个认知会带来什么后果：

• 🔴 部署到生产后才发现：某个条件分支从未被执行过，里面有严重bug
• 🔴 上线一周后网站改版，大量错误处理逻辑失效
• 🔴 接手同事项目，不知道改了某个子流程后会不会破坏其他功能
• 🔴 线上运行4小时后悄悄中断，没有任何告警

测试是RPA工程化的重要环节， 本文系统介绍如何为影刀RPA项目建立测试体系。

---

第一章：RPA测试的特殊性

1.1 RPA测试 vs 传统软件测试

维度	传统软件测试	RPA测试
测试对象	函数/API	流程/子流程
环境依赖	可Mock	依赖真实UI/网络
执行速度	毫秒级	秒/分钟级

| 稳定性 | 高 | 受网络、网站变更影响 |
| 成本 | 低 | 相对较高 |

1.2 RPA测试的层次划分

RPA测试金字塔

          ┌─────────────────┐
          │   E2E测试（少量）│  完整流程端到端测试
          │     占10-20%    │
          ├─────────────────┤
          │   集成测试（中量）│  子流程组合测试
          │     占20-30%    │
          ├─────────────────┤
          │  单元测试（大量）│  工具函数/数据处理测试
          │     占50-70%    │
          └─────────────────┘

核心原则：

• 工具函数（Python代码部分）：大量单元测试，快速验证
• 子流程组合：关键路径集成测试
• 完整流程：关键业务场景E2E测试（少量、高价值）

---

第二章：单元测试（工具函数层）

2.1 使用pytest进行单元测试

# 安装pytest
pip install pytest pytest-cov

# 运行测试
pytest tests/ -v

2.2 数据处理函数的单元测试

# ============================================================
# 文件：tests/test_data_utils.py
# 对数据清洗工具函数进行单元测试
# ============================================================

import pytest
import sys
sys.path.insert(0, "..")

from utils.data_utils import (
    clean_price, 
    clean_phone, 
    format_date, 
    extract_numbers,
    validate_email
)


class TestCleanPrice:
    """价格清洗函数测试"""
    
    def test_basic_price(self):
        """基础价格字符串"""
        assert clean_price("99") == 99.0
        assert clean_price("99.9") == 99.9
        assert clean_price("1299.00") == 1299.0
    
    def test_with_currency_symbol(self):
        """带货币符号"""
        assert clean_price("¥99") == 99.0
        assert clean_price("￥1,299.00") == 1299.0
        assert clean_price("$29.9") == 29.9
    
    def test_with_chinese_text(self):
        """带中文描述"""
        assert clean_price("价格：99元") == 99.0
        assert clean_price("售价¥1,299.00元") == 1299.0
    
    def test_with_comma_separator(self):
        """千位分隔符"""
        assert clean_price("1,299.00") == 1299.0
        assert clean_price("10,000") == 10000.0
    
    def test_empty_returns_zero(self):
        """空值处理"""
        assert clean_price("") == 0.0
        assert clean_price("暂无价格") == 0.0
    
    def test_negative_price(self):
        """负数价格（退款场景）"""
        assert clean_price("-99.0") == -99.0


class TestCleanPhone:
    """手机号清洗函数测试"""
    
    def test_valid_phones(self):
        """有效手机号"""
        valid_phones = [
            "13812345678",
            "15987654321",
            "17012345678",
            "19912345678"
        ]
        for phone in valid_phones:
            assert clean_phone(phone) == phone
    
    def test_phone_with_spaces(self):
        """带空格的手机号"""
        assert clean_phone("138 1234 5678") == "13812345678"
        assert clean_phone("138-1234-5678") == "13812345678"
    
    def test_invalid_phone_returns_none(self):
        """无效手机号返回None"""
        assert clean_phone("12312345678") is None  # 不以1开头
        assert clean_phone("1381234567") is None   # 位数不足
        assert clean_phone("138123456789") is None  # 位数超出
        assert clean_phone("") is None


class TestFormatDate:
    """日期格式化测试"""
    
    def test_slash_format(self):
        """斜杠分隔格式"""
        assert format_date("2024/01/15") == "2024-01-15"
        assert format_date("2024/1/5") == "2024-01-05"
    
    def test_chinese_format(self):
        """中文日期格式"""
        assert format_date("2024年1月15日") == "2024-01-15"
    
    def test_custom_output_format(self):
        """自定义输出格式"""
        assert format_date("2024/01/15", "%Y%m%d") == "20240115"
        assert format_date("2024/01/15", "%m/%d/%Y") == "01/15/2024"
    
    def test_invalid_date(self):
        """无效日期抛出异常"""
        with pytest.raises(ValueError):
            format_date("not-a-date")


class TestValidateEmail:
    """邮箱验证测试"""
    
    def test_valid_emails(self):
        valid_emails = [
            "user@example.com",
            "user.name+tag@sub.domain.org",
            "rpa@company.cn",
        ]
        for email in valid_emails:
            assert validate_email(email) == True, f"应该是有效邮箱：{email}"
    
    def test_invalid_emails(self):
        invalid_emails = [
            "not-an-email",
            "missing@.com",
            "@no-user.com",
            "spaces in@email.com",
        ]
        for email in invalid_emails:
            assert validate_email(email) == False, f"应该是无效邮箱：{email}"

2.3 Excel处理函数的单元测试

# tests/test_excel_utils.py

import pytest
import os
import pandas as pd
from utils.excel_utils import (
    read_excel_to_list,
    write_list_to_excel,
    merge_excel_files,
    get_sheet_names
)


@pytest.fixture
def sample_excel(tmp_path):
    """创建测试用Excel文件（pytest fixture）"""
    df = pd.DataFrame({
        "姓名": ["张三", "李四", "王五"],
        "年龄": [25, 30, 28],
        "部门": ["技术", "产品", "技术"],
        "薪资": [15000, 20000, 18000]
    })
    
    excel_path = str(tmp_path / "test_data.xlsx")
    df.to_excel(excel_path, index=False)
    return excel_path


class TestReadExcel:
    
    def test_basic_read(self, sample_excel):
        """基础读取测试"""
        data = read_excel_to_list(sample_excel)
        assert len(data) == 3
        assert data[0]["姓名"] == "张三"
    
    def test_read_specific_sheet(self, sample_excel):
        """读取指定工作表"""
        data = read_excel_to_list(sample_excel, sheet_name="Sheet1")
        assert len(data) > 0
    
    def test_file_not_found(self):
        """文件不存在时的错误处理"""
        with pytest.raises(FileNotFoundError):
            read_excel_to_list("nonexistent.xlsx")
    
    def test_empty_file_returns_empty_list(self, tmp_path):
        """空Excel返回空列表"""
        df = pd.DataFrame()
        empty_path = str(tmp_path / "empty.xlsx")
        df.to_excel(empty_path, index=False)
        data = read_excel_to_list(empty_path)
        assert data == []


class TestWriteExcel:
    
    def test_basic_write(self, tmp_path):
        """基础写入测试"""
        data = [
            {"name": "Alice", "score": 95},
            {"name": "Bob", "score": 87}
        ]
        output_path = str(tmp_path / "output.xlsx")
        write_list_to_excel(data, output_path)
        
        # 验证文件存在
        assert os.path.exists(output_path)
        
        # 验证内容
        df = pd.read_excel(output_path)
        assert len(df) == 2
        assert df.iloc[0]["name"] == "Alice"
    
    def test_creates_directory(self, tmp_path):
        """自动创建不存在的目录"""
        output_path = str(tmp_path / "subdir" / "nested" / "output.xlsx")
        data = [{"key": "value"}]
        write_list_to_excel(data, output_path)
        assert os.path.exists(output_path)

---

第三章：集成测试（子流程层）

3.1 子流程集成测试框架

# ============================================================
# 文件：tests/test_login_module.py
# 测试登录子流程的各种场景
# ============================================================

import pytest
from unittest.mock import Mock, patch, MagicMock


class TestLoginModule:
    """登录模块集成测试"""
    
    @pytest.fixture
    def mock_page(self):
        """Mock浏览器页面对象"""
        page = MagicMock()
        page.is_element_exist.return_value = True
        page.get_text.return_value = ""
        return page
    
    def test_successful_login(self, mock_page):
        """正常登录场景"""
        # 配置Mock行为：登录成功后出现用户头像
        mock_page.is_element_exist.side_effect = lambda sel: (
            sel == ".user-avatar"  # 登录成功后的元素
        )
        
        from modules.login import login_to_erp
        result = login_to_erp(
            page=mock_page,
            username="test_user",
            password="test_pass"
        )
        
        assert result["success"] == True
        assert result["user_info"] is not None
    
    def test_wrong_password(self, mock_page):
        """密码错误场景"""
        # 配置Mock：出现错误提示
        mock_page.is_element_exist.side_effect = lambda sel: (
            sel == ".error-message"
        )
        mock_page.get_text.side_effect = lambda sel: (
            "用户名或密码错误" if sel == ".error-message" else ""
        )
        
        from modules.login import login_to_erp
        result = login_to_erp(
            page=mock_page,
            username="wrong_user",
            password="wrong_pass"
        )
        
        assert result["success"] == False
        assert "密码错误" in result["error"]
    
    def test_captcha_required(self, mock_page):
        """需要验证码场景"""
        mock_page.is_element_exist.side_effect = lambda sel: (
            sel == "#captcha-input"
        )
        
        from modules.login import login_to_erp
        result = login_to_erp(
            page=mock_page,
            username="test_user",
            password="test_pass"
        )
        
        # 遇到验证码应返回特定状态码
        assert result["status"] == "captcha_required"
    
    def test_network_error_triggers_retry(self, mock_page):
        """网络错误时触发重试"""
        call_count = [0]
        
        def side_effect(*args):
            call_count[0] += 1
            if call_count[0] < 3:
                raise ConnectionError("网络不稳定")
            return True
        
        mock_page.click.side_effect = side_effect
        
        from modules.login import login_to_erp
        result = login_to_erp(
            page=mock_page,
            username="test_user",
            password="test_pass",
            max_retries=3
        )
        
        # 应该在第3次重试后成功
        assert call_count[0] == 3

---

第四章：端到端测试

4.1 关键业务场景E2E测试

# ============================================================
# 文件：tests/test_e2e_order_scrape.py
# 端到端测试：完整的订单采集流程
# 需要连接到测试环境（沙箱环境）
# ============================================================

import pytest
import os
from pathlib import Path


@pytest.mark.e2e
@pytest.mark.slow  # 标记为耗时测试
class TestOrderScrapingE2E:
    """订单采集端到端测试
    
    注意：此测试需要：
    1. 有效的测试账号（在 config/test_config.json 中配置）
    2. 连接到测试环境（而非生产环境）
    3. 约2-5分钟运行时间
    """
    
    @pytest.fixture(autouse=True)
    def setup(self, tmp_path):
        """测试前准备"""
        self.output_dir = str(tmp_path / "test_output")
        os.makedirs(self.output_dir, exist_ok=True)
        
        # 使用测试配置（而非生产配置）
        os.environ["RPA_CONFIG"] = "config/test_config.json"
    
    def test_full_order_scraping_flow(self):
        """完整订单采集流程测试"""
        from main.order_scrape import run_order_scrape
        
        # 执行只采集最近1天的订单（测试用）
        result = run_order_scrape(
            date_range="last_1_day",
            output_dir=self.output_dir,
            max_orders=10  # 测试时限制数量
        )
        
        # 断言1：流程执行成功
        assert result["status"] == "completed"
        
        # 断言2：有数据产出
        assert result["total_scraped"] > 0
        
        # 断言3：输出文件存在
        output_files = list(Path(self.output_dir).glob("*.xlsx"))
        assert len(output_files) > 0
        
        # 断言4：数据质量检查
        import pandas as pd
        df = pd.read_excel(output_files[0])
        
        # 必须包含关键字段
        required_columns = ["订单号", "下单时间", "金额", "状态"]
        for col in required_columns:
            assert col in df.columns, f"缺少必要字段：{col}"
        
        # 订单号不能为空
        assert df["订单号"].notna().all(), "存在空订单号"
        
        # 金额必须为正数
        assert (df["金额"] > 0).all(), "存在异常金额"
    
    def test_handles_empty_date_range(self):
        """处理空日期范围（无数据场景）"""
        from main.order_scrape import run_order_scrape
        
        # 查询很久以前的数据（应该没有订单）
        result = run_order_scrape(
            date_range="2010-01-01:2010-01-02",
            output_dir=self.output_dir
        )
        
        # 无数据时应优雅处理，不崩溃
        assert result["status"] == "completed"
        assert result["total_scraped"] == 0
        assert result.get("output_file") is None  # 无数据不创建文件

---

第五章：测试自动化与CI集成

5.1 pytest配置

# pytest.ini
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*

# 标记定义
markers =
    e2e: 端到端测试（需要外部依赖）
    slow: 耗时测试（超过30秒）
    smoke: 冒烟测试（最基础的功能验证）

# 默认排除E2E和慢速测试（可用 -m e2e 显式运行）
addopts = -m "not e2e and not slow" -v --tb=short

5.2 测试覆盖率报告

# 生成测试覆盖率报告
pytest tests/ --cov=utils --cov=modules --cov-report=html

# 查看报告（在 htmlcov/index.html）
![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/9268872dfb4d467ca5c77b09c4c9d1a5.png#pic_center)
![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/d3563122ac6f486291e45aeb9cb0e37d.png#pic_center)


# 目标：工具函数覆盖率 > 90%，核心模块 > 70%

5.3 自动化测试脚本

# ============================================================
# 文件：scripts/run_tests.py
# 每日自动执行测试并生成报告
# ============================================================

import subprocess
import datetime
import json
import os

def run_tests_and_report():
    """运行测试并生成报告"""
    
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    report_dir = f"test_reports/{timestamp}"
    os.makedirs(report_dir, exist_ok=True)
    
    # 运行快速测试（单元+集成）
    result = subprocess.run([
        "pytest", "tests/",
        "-m", "not e2e and not slow",
        "--json-report",
        f"--json-report-file={report_dir}/report.json",
        "--tb=short",
        "-q"
    ], capture_output=True, text=True)
    
    # 解析结果
    with open(f"{report_dir}/report.json") as f:
        report = json.load(f)
    
    summary = report["summary"]
    total = summary.get("total", 0)
    passed = summary.get("passed", 0)
    failed = summary.get("failed", 0)
    
    success_rate = passed / total if total > 0 else 0
    
    print(f"\n{'='*50}")
    print(f"测试完成 [{timestamp}]")
    print(f"通过：{passed} / 总数：{total} = {success_rate:.1%}")
    print(f"失败：{failed}")
    
    # 测试失败时发送告警
    if failed > 0:
        print(f"\n⚠️ 有 {failed} 个测试失败，请查看报告：{report_dir}/report.json")
        send_test_failure_alert(failed, report_dir)
    
    return success_rate >= 1.0  # 所有测试通过才返回True


if __name__ == "__main__":
    success = run_tests_and_report()
    exit(0 if success else 1)

---

第六章：测试数据管理

6.1 测试数据工厂

# tests/factories.py
import random
import string
import datetime
from typing import List, Dict


class DataFactory:
    """测试数据生成工厂"""
    
    @staticmethod
    def make_employee(**overrides) -> dict:
        """生成员工测试数据"""
        default = {
            "employee_id": f"EMP{random.randint(10000, 99999)}",
            "name": f"测试员工_{random.randint(1, 100)}",
            "department": random.choice(["技术", "产品", "运营", "销售"]),
            "position": random.choice(["工程师", "经理", "总监"]),
            "salary": random.randint(8000, 50000),
            "join_date": "2023-01-01",
            "email": f"test{random.randint(1,999)}@example.com",
            "phone": f"1{''.join(random.choices(string.digits, k=10))}"
        }
        default.update(overrides)
        return default
    
    @staticmethod
    def make_order(**overrides) -> dict:
        """生成订单测试数据"""
        statuses = ["待付款", "已付款", "已发货", "已完成", "已取消"]
        default = {
            "order_id": f"JD{datetime.date.today().strftime('%Y%m%d')}{random.randint(10000, 99999)}",
            "customer_id": f"C{random.randint(10000, 99999)}",
            "amount": round(random.uniform(10, 10000), 2),
            "status": random.choice(statuses),
            "created_at": datetime.datetime.now().isoformat(),
            "items": [
                {"product_id": f"P{i}", "quantity": random.randint(1, 5), "price": random.randint(10, 500)}
                for i in range(random.randint(1, 5))
            ]
        }
        default.update(overrides)
        return default
    
    @classmethod
    def make_batch(cls, factory_method, count: int, **common_attrs) -> List[dict]:
        """批量生成测试数据"""
        return [factory_method(**common_attrs) for _ in range(count)]


# 使用示例
factory = DataFactory()

# 生成10个员工数据
employees = factory.make_batch(factory.make_employee, 10, department="技术")

# 生成指定状态的订单
paid_orders = factory.make_batch(factory.make_order, 20, status="已付款")

---

总结

建立完整的RPA测试体系，按优先级排序：

优先级	测试类型	工作量	回报
P0	工具函数单元测试	低	极高
P1	关键子流程集成测试	中	高
P2	核心业务E2E测试	高	高
P3	性能/压力测试	高	中

从今天开始： 为你现有的工具函数补充单元测试，1-2小时的投入就能显著提升项目质量。

---

作者：林焱 | 影刀RPA技术专栏 | 如果本文对你有帮助，欢迎点赞收藏！