第27讲：Skill 测试与质量保证

掌握 Skill 的测试方法和质量保证技巧，确保 Skill 稳定可靠运行。

一、测试策略

1.1 测试金字塔

        /\
       /  \     E2E 测试（端到端）
      /____\        少量，关键路径
     /      \
    /   /\\\   \   集成测试
   /   /  \\   \      中等数量
  /___/____\___\
 /              \
/   /\\  /\\  /\\  \ 单元测试
    \/    \/    \/     大量，快速

1.2 测试类型

测试类型	目的	工具
单元测试	验证单个函数正确性	pytest
集成测试	验证模块间协作	pytest
端到端测试	验证完整流程	手动/自动化
性能测试	验证响应速度	locust
安全测试	发现安全漏洞	bandit

二、单元测试

2.1 测试框架

# tests/test_invoice.py
import pytest
from unittest.mock import Mock, patch
from services.invoice_service import InvoiceService

class TestInvoiceService:
    """发票服务测试"""
    
    @pytest.fixture
    def service(self):
        """测试夹具"""
        mock_ocr = Mock()
        mock_db = Mock()
        return InvoiceService(mock_ocr, mock_db)
    
    def test_extract_invoice_code(self, service):
        """测试发票代码提取"""
        text = "发票代码：123456789012"
        result = service._extract_pattern(text, r'发票代码[:：]\s*(\d{12})')
        assert result == '123456789012'
    
    def test_extract_invoice_code_not_found(self, service):
        """测试发票代码不存在"""
        text = "这是一段普通文本"
        result = service._extract_pattern(text, r'发票代码[:：]\s*(\d{12})')
        assert result is None
    
    def test_validate_invoice_success(self, service):
        """测试发票验证成功"""
        info = {
            'invoice_code': '123456789012',
            'invoice_number': '12345678',
            'amount': '1000.00'
        }
        service.db.check_duplicate.return_value = False
        
        result = service._validate_invoice(info)
        
        assert result['is_valid'] is True
        assert len(result['errors']) == 0
    
    def test_validate_invoice_duplicate(self, service):
        """测试重复发票"""
        info = {
            'invoice_code': '123456789012',
            'invoice_number': '12345678',
            'amount': '1000.00'
        }
        service.db.check_duplicate.return_value = True
        
        result = service._validate_invoice(info)
        
        assert result['is_valid'] is False
        assert '发票已存在' in result['errors']

2.2 Mock 测试

# tests/test_with_mock.py
from unittest.mock import Mock, patch, MagicMock

class TestWithMock:
    """Mock 测试示例"""
    
    def test_mock_function(self):
        """Mock 函数"""
        mock_func = Mock(return_value=42)
        result = mock_func()
        
        assert result == 42
        mock_func.assert_called_once()
    
    def test_patch_decorator(self):
        """使用 patch 装饰器"""
        with patch('services.invoice_service.OCRService') as mock_ocr:
            mock_instance = MagicMock()
            mock_instance.recognize.return_value = "识别结果"
            mock_ocr.return_value = mock_instance
            
            # 执行测试
            from services.invoice_service import InvoiceService
            service = InvoiceService(mock_instance, Mock())
            result = service.ocr.recognize("test.jpg")
            
            assert result == "识别结果"
    
    @patch('services.invoice_service.requests.get')
    def test_api_call(self, mock_get):
        """测试 API 调用"""
        mock_get.return_value.json.return_value = {'status': 'ok'}
        
        # 执行 API 调用
        import requests
        response = requests.get('http://api.example.com')
        
        assert response.json()['status'] == 'ok'

三、集成测试

3.1 测试数据库

# tests/test_integration.py
import pytest
import sqlite3
from services.invoice_service import InvoiceService

class TestIntegration:
    """集成测试"""
    
    @pytest.fixture(scope='function')
    def db(self):
        """创建测试数据库"""
        conn = sqlite3.connect(':memory:')
        conn.execute('''
            CREATE TABLE invoices (
                id INTEGER PRIMARY KEY,
                invoice_code TEXT,
                invoice_number TEXT,
                amount REAL
            )
        ''')
        yield conn
        conn.close()
    
    def test_save_and_query_invoice(self, db):
        """测试保存和查询发票"""
        # 保存发票
        db.execute('''
            INSERT INTO invoices (invoice_code, invoice_number, amount)
            VALUES (?, ?, ?)
        ''', ('123456789012', '12345678', 1000.00))
        db.commit()
        
        # 查询发票
        cursor = db.execute('SELECT * FROM invoices WHERE invoice_code = ?', 
                          ('123456789012',))
        result = cursor.fetchone()
        
        assert result[1] == '123456789012'
        assert result[3] == 1000.00

四、端到端测试

4.1 对话流程测试

# tests/test_e2e.py
class TestEndToEnd:
    """端到端测试"""
    
    def test_invoice_recognition_flow(self):
        """测试发票识别完整流程"""
        skill = FinanceAssistantSkill()
        
        # 用户请求识别发票
        response1 = skill.handle_message("帮我识别这张发票", {})
        assert "请上传发票图片" in response1
        
        # 用户上传图片
        context = {'last_intent': 'recognize_invoice'}
        response2 = skill.handle_file("test_invoice.jpg", context)
        
        # 验证响应包含关键信息
        assert "发票识别成功" in response2 or "发票代码" in response2
    
    def test_report_generation_flow(self):
        """测试报表生成流程"""
        skill = FinanceAssistantSkill()
        
        # 用户请求生成报表
        response = skill.handle_message("生成2024年1月的资产负债表", {})
        
        # 验证响应
        assert "资产负债表" in response or "已生成" in response

五、性能测试

5.1 负载测试

# tests/test_performance.py
import time
import statistics
from concurrent.futures import ThreadPoolExecutor

class TestPerformance:
    """性能测试"""
    
    def test_response_time(self):
        """测试响应时间"""
        skill = FinanceAssistantSkill()
        
        times = []
        for _ in range(10):
            start = time.time()
            skill.handle_message("查询发票", {})
            elapsed = time.time() - start
            times.append(elapsed)
        
        avg_time = statistics.mean(times)
        print(f"平均响应时间: {avg_time:.3f}s")
        
        # 断言平均响应时间小于 1 秒
        assert avg_time < 1.0
    
    def test_concurrent_requests(self):
        """测试并发请求"""
        skill = FinanceAssistantSkill()
        
        def make_request(i):
            start = time.time()
            skill.handle_message(f"请求 {i}", {})
            return time.time() - start
        
        with ThreadPoolExecutor(max_workers=5) as executor:
            times = list(executor.map(make_request, range(20)))
        
        avg_time = statistics.mean(times)
        print(f"并发平均响应时间: {avg_time:.3f}s")
        
        assert avg_time < 2.0

六、测试最佳实践

6.1 测试原则

独立性：每个测试独立运行，不依赖其他测试
可重复：测试结果可重复，不受外部环境影响
快速：单元测试应在毫秒级完成
全面：覆盖正常、异常、边界情况

6.2 测试覆盖率

# 运行测试并生成覆盖率报告
pytest --cov=services --cov-report=html tests/

# 查看覆盖率报告
open htmlcov/index.html

6.3 持续集成

# .github/workflows/test.yml
name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      
      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: '3.9'
      
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-cov
      
      - name: Run tests
        run: pytest --cov=services tests/
      
      - name: Upload coverage
        uses: codecov/codecov-action@v2