┌─────────────────────────────────────────────────────────────┐
│                    🎨 用例管理前端                          │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  用例编辑器 │  │  执行监控   │  │  结果分析   │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    🔧 用例管理服务                          │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  用例CRUD   │  │  分类管理   │  │  批量操作   │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    ⚙️ 执行引擎                              │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  API执行器  │  │  UI执行器   │  │  DB执行器   │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    💾 数据存储层                            │
│        用例数据 + 执行记录 + 结果统计                       │
└─────────────────────────────────────────────────────────────┘

数据模型设计：用例的"DNA"

核心实体关系：

Project (项目)
    ↓ 1:N
TestSuite (测试套件)
    ↓ 1:N  
TestCase (测试用例)
    ↓ 1:N
ExecutionRecord (执行记录)
    ↓ 1:N
ExecutionStep (执行步骤)

📊 数据模型详细设计

测试用例模型：灵活的数据结构

app/models/testcase.py：

from peewee import *
from .base import BaseModel
from .project import Project
import json
from enum import Enum

class TestCaseType(Enum):
    """测试用例类型枚举"""
    API = "api"
    UI = "ui"
    DATABASE = "database"
    PERFORMANCE = "performance"

class TestCasePriority(Enum):
    """测试用例优先级枚举"""
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

class TestCase(BaseModel):
    """测试用例模型 - 支持多种类型的测试用例"""
    
    # 基本信息
    project = ForeignKeyField(Project, backref='testcases', verbose_name="所属项目")
    name = CharField(max_length=200, verbose_name="用例名称")
    description = TextField(null=True, verbose_name="用例描述")
    
    # 分类信息
    type = CharField(
        max_length=20,
        choices=[(t.value, t.name) for t in TestCaseType],
        verbose_name="用例类型"
    )
    priority = CharField(
        max_length=10,
        choices=[(p.value, p.name) for p in TestCasePriority],
        default=TestCasePriority.MEDIUM.value,
        verbose_name="优先级"
    )
    tags = CharField(max_length=500, null=True, verbose_name="标签")  # 逗号分隔
    
    # 用例内容 - JSON格式存储，支持不同类型的用例结构
    content = TextField(verbose_name="用例内容")
    
    # 前置条件和后置条件
    preconditions = TextField(null=True, verbose_name="前置条件")
    postconditions = TextField(null=True, verbose_name="后置条件")
    
    # 状态管理
    status = CharField(
        max_length=20,
        choices=[('active', '启用'), ('disabled', '禁用'), ('draft', '草稿')],
        default='draft',
        verbose_name="用例状态"
    )
    
    # 执行统计
    total_executions = IntegerField(default=0, verbose_name="总执行次数")
    success_executions = IntegerField(default=0, verbose_name="成功执行次数")
    
    class Meta:
        table_name = 'test_cases'
        indexes = (
            (('project', 'name'), True),  # 项目内用例名称唯一
            (('type', 'priority'), False),  # 类型和优先级组合索引
        )
    
    def get_content_dict(self):
        """获取用例内容的字典格式"""
        try:
            return json.loads(self.content)
        except (json.JSONDecodeError, TypeError):
            return {}
    
    def set_content_dict(self, content_dict):
        """设置用例内容"""
        self.content = json.dumps(content_dict, ensure_ascii=False, indent=2)
    
    def get_tags_list(self):
        """获取标签列表"""
        if not self.tags:
            return []
        return [tag.strip() for tag in self.tags.split(',') if tag.strip()]
    
    def set_tags_list(self, tags_list):
        """设置标签列表"""
        self.tags = ','.join(tags_list) if tags_list else None
    
    @property
    def success_rate(self):
        """成功率"""
        if self.total_executions == 0:
            return 0
        return round(self.success_executions / self.total_executions * 100, 2)

class TestSuite(BaseModel):
    """测试套件模型 - 用例的集合"""
    
    project = ForeignKeyField(Project, backref='testsuites', verbose_name="所属项目")
    name = CharField(max_length=200, verbose_name="套件名称")
    description = TextField(null=True, verbose_name="套件描述")
    
    # 套件配置
    execution_order = CharField(
        max_length=20,
        choices=[('sequential', '顺序执行'), ('parallel', '并行执行')],
        default='sequential',
        verbose_name="执行顺序"
    )
    
    # 环境配置
    environment = CharField(max_length=50, null=True, verbose_name="执行环境")
    
    class Meta:
        table_name = 'test_suites'

class TestSuiteCase(BaseModel):
    """测试套件和用例的关联表"""
    
    suite = ForeignKeyField(TestSuite, backref='suite_cases', verbose_name="测试套件")
    testcase = ForeignKeyField(TestCase, backref='case_suites', verbose_name="测试用例")
    order = IntegerField(default=0, verbose_name="执行顺序")
    
    class Meta:
        table_name = 'test_suite_cases'
        indexes = (
            (('suite', 'testcase'), True),  # 套件内用例唯一
        )

执行记录模型：追踪每一次执行

app/models/execution.py：

from peewee import *
from .base import BaseModel
from .testcase import TestCase, TestSuite
import json
from datetime import datetime

class ExecutionRecord(BaseModel):
    """执行记录模型"""
    
    # 关联信息
    testcase = ForeignKeyField(TestCase, backref='executions', null=True, verbose_name="测试用例")
    testsuite = ForeignKeyField(TestSuite, backref='executions', null=True, verbose_name="测试套件")
    
    # 执行信息
    execution_id = CharField(max_length=100, unique=True, verbose_name="执行ID")
    trigger_type = CharField(
        max_length=20,
        choices=[('manual', '手动执行'), ('scheduled', '定时执行'), ('api', 'API触发')],
        verbose_name="触发方式"
    )
    executor = CharField(max_length=50, verbose_name="执行人")
    
    # 执行状态
    status = CharField(
        max_length=20,
        choices=[
            ('pending', '等待中'),
            ('running', '执行中'),
            ('success', '成功'),
            ('failed', '失败'),
            ('cancelled', '已取消')
        ],
        default='pending',
        verbose_name="执行状态"
    )
    
    # 时间信息
    start_time = DateTimeField(null=True, verbose_name="开始时间")
    end_time = DateTimeField(null=True, verbose_name="结束时间")
    duration = IntegerField(null=True, verbose_name="执行时长(秒)")
    
    # 执行结果
    result_summary = TextField(null=True, verbose_name="结果摘要")  # JSON格式
    error_message = TextField(null=True, verbose_name="错误信息")
    
    # 环境信息
    environment = CharField(max_length=50, null=True, verbose_name="执行环境")
    
    class Meta:
        table_name = 'execution_records'
    
    def get_result_summary_dict(self):
        """获取结果摘要字典"""
        try:
            return json.loads(self.result_summary) if self.result_summary else {}
        except (json.JSONDecodeError, TypeError):
            return {}
    
    def set_result_summary_dict(self, summary_dict):
        """设置结果摘要"""
        self.result_summary = json.dumps(summary_dict, ensure_ascii=False, indent=2)

class ExecutionStep(BaseModel):
    """执行步骤模型 - 记录每个步骤的详细信息"""
    
    execution = ForeignKeyField(ExecutionRecord, backref='steps', verbose_name="执行记录")
    step_name = CharField(max_length=200, verbose_name="步骤名称")
    step_order = IntegerField(verbose_name="步骤顺序")
    
    # 步骤状态
    status = CharField(
        max_length=20,
        choices=[('success', '成功'), ('failed', '失败'), ('skipped', '跳过')],
        verbose_name="步骤状态"
    )
    
    # 执行详情
    request_data = TextField(null=True, verbose_name="请求数据")
    response_data = TextField(null=True, verbose_name="响应数据")
    assertion_result = TextField(null=True, verbose_name="断言结果")
    error_message = TextField(null=True, verbose_name="错误信息")
    
    # 时间信息
    start_time = DateTimeField(default=datetime.now, verbose_name="开始时间")
    end_time = DateTimeField(null=True, verbose_name="结束时间")
    duration = IntegerField(null=True, verbose_name="执行时长(毫秒)")
    
    class Meta:
        table_name = 'execution_steps'

🎨 用例内容结构设计

API测试用例结构：标准化的接口测试

{
  "type": "api",
  "config": {
    "base_url": "https://api.example.com",
    "timeout": 30,
    "retry_count": 3
  },
  "steps": [
    {
      "name": "用户登录",
      "method": "POST",
      "url": "/auth/login",
      "headers": {
        "Content-Type": "application/json"
      },
      "body": {
        "username": "testuser",
        "password": "testpass"
      },
      "assertions": [
        {
          "type": "status_code",
          "expected": 200
        },
        {
          "type": "json_path",
          "path": "$.success",
          "expected": true
        },
        {
          "type": "response_time",
          "max_time": 2000
        }
      ],
      "extract": {
        "token": "$.data.token"
      }
    },
    {
      "name": "获取用户信息",
      "method": "GET",
      "url": "/user/profile",
      "headers": {
        "Authorization": "Bearer ${token}"
      },
      "assertions": [
        {
          "type": "status_code",
          "expected": 200
        },
        {
          "type": "json_schema",
          "schema": {
            "type": "object",
            "properties": {
              "id": {"type": "number"},
              "username": {"type": "string"}
            },
            "required": ["id", "username"]
          }
        }
      ]
    }
  ]
}

UI测试用例结构：页面操作的自动化

{
  "type": "ui",
  "config": {
    "browser": "chrome",
    "headless": false,
    "window_size": "1920x1080",
    "implicit_wait": 10
  },
  "steps": [
    {
      "name": "打开登录页面",
      "action": "navigate",
      "url": "https://example.com/login"
    },
    {
      "name": "输入用户名",
      "action": "input",
      "locator": {
        "type": "id",
        "value": "username"
      },
      "data": "testuser"
    },
    {
      "name": "输入密码",
      "action": "input",
      "locator": {
        "type": "id",
        "value": "password"
      },
      "data": "testpass"
    },
    {
      "name": "点击登录按钮",
      "action": "click",
      "locator": {
        "type": "xpath",
        "value": "//button[@type='submit']"
      }
    },
    {
      "name": "验证登录成功",
      "action": "assert",
      "assertion": {
        "type": "element_visible",
        "locator": {
          "type": "class",
          "value": "welcome-message"
        }
      }
    }
  ]
}

数据库测试用例结构：数据验证的利器

{
  "type": "database",
  "config": {
    "connection": {
      "host": "localhost",
      "port": 3306,
      "database": "test_db",
      "username": "test_user",
      "password": "test_pass"
    }
  },
  "steps": [
    {
      "name": "清理测试数据",
      "action": "execute",
      "sql": "DELETE FROM users WHERE username LIKE 'test_%'"
    },
    {
      "name": "插入测试数据",
      "action": "execute",
      "sql": "INSERT INTO users (username, email) VALUES ('test_user', 'test@example.com')"
    },
    {
      "name": "验证数据插入",
      "action": "query",
      "sql": "SELECT COUNT(*) as count FROM users WHERE username = 'test_user'",
      "assertions": [
        {
          "type": "equals",
          "field": "count",
          "expected": 1
        }
      ]
    }
  ]
}

⚙️ 执行引擎设计

执行器工厂模式：一个工厂管所有

app/services/execution_engine.py：

from abc import ABC, abstractmethod
from typing import Dict, Any, List
import json
import uuid
from datetime import datetime

class BaseExecutor(ABC):
    """执行器基类"""
    
    def __init__(self, testcase, execution_record):
        self.testcase = testcase
        self.execution_record = execution_record
        self.context = {}  # 执行上下文，用于存储变量
    
    @abstractmethod
    async def execute(self) -> Dict[str, Any]:
        """执行测试用例"""
        pass
    
    def log_step(self, step_name: str, status: str, **kwargs):
        """记录执行步骤"""
        from app.models.execution import ExecutionStep
        
        step = ExecutionStep.create(
            execution=self.execution_record,
            step_name=step_name,
            step_order=len(self.execution_record.steps) + 1,
            status=status,
            **kwargs
        )
        return step

class APIExecutor(BaseExecutor):
    """API测试执行器"""
    
    async def execute(self) -> Dict[str, Any]:
        """执行API测试用例"""
        import aiohttp
        
        content = self.testcase.get_content_dict()
        config = content.get('config', {})
        steps = content.get('steps', [])
        
        results = []
        
        async with aiohttp.ClientSession(
            timeout=aiohttp.ClientTimeout(total=config.get('timeout', 30))
        ) as session:
            
            for step in steps:
                step_result = await self._execute_api_step(session, step, config)
                results.append(step_result)
                
                # 如果步骤失败且不允许继续，则停止执行
                if not step_result['success'] and not step.get('continue_on_failure', False):
                    break
        
        # 计算总体结果
        success_count = sum(1 for r in results if r['success'])
        total_count = len(results)
        
        return {
            'success': success_count == total_count,
            'total_steps': total_count,
            'success_steps': success_count,
            'failed_steps': total_count - success_count,
            'results': results
        }
    
    async def _execute_api_step(self, session, step, config):
        """执行单个API步骤"""
        step_name = step.get('name', 'Unnamed Step')
        start_time = datetime.now()
        
        try:
            # 构建请求
            url = config.get('base_url', '') + step.get('url', '')
            method = step.get('method', 'GET').upper()
            headers = step.get('headers', {})
            
            # 处理请求体
            body = step.get('body')
            if body and isinstance(body, dict):
                body = json.dumps(body)
            
            # 发送请求
            async with session.request(
                method=method,
                url=url,
                headers=headers,
                data=body
            ) as response:
                response_text = await response.text()
                response_time = (datetime.now() - start_time).total_seconds() * 1000
                
                # 执行断言
                assertions = step.get('assertions', [])
                assertion_results = []
                
                for assertion in assertions:
                    assertion_result = self._execute_assertion(
                        assertion, response, response_text, response_time
                    )
                    assertion_results.append(assertion_result)
                
                # 提取变量
                extract_config = step.get('extract', {})
                for var_name, json_path in extract_config.items():
                    try:
                        import jsonpath
                        response_json = json.loads(response_text)
                        extracted_value = jsonpath.jsonpath(response_json, json_path)
                        if extracted_value:
                            self.context[var_name] = extracted_value[0]
                    except Exception as e:
                        print(f"变量提取失败: {e}")
                
                # 记录步骤
                success = all(ar['success'] for ar in assertion_results)
                self.log_step(
                    step_name=step_name,
                    status='success' if success else 'failed',
                    request_data=json.dumps({
                        'method': method,
                        'url': url,
                        'headers': headers,
                        'body': body
                    }, ensure_ascii=False),
                    response_data=response_text,
                    assertion_result=json.dumps(assertion_results, ensure_ascii=False),
                    duration=int(response_time)
                )
                
                return {
                    'success': success,
                    'step_name': step_name,
                    'response_time': response_time,
                    'status_code': response.status,
                    'assertions': assertion_results
                }
                
        except Exception as e:
            # 记录错误步骤
            self.log_step(
                step_name=step_name,
                status='failed',
                error_message=str(e),
                duration=int((datetime.now() - start_time).total_seconds() * 1000)
            )
            
            return {
                'success': False,
                'step_name': step_name,
                'error': str(e)
            }
    
    def _execute_assertion(self, assertion, response, response_text, response_time):
        """执行断言"""
        assertion_type = assertion.get('type')
        
        try:
            if assertion_type == 'status_code':
                expected = assertion.get('expected')
                actual = response.status
                success = actual == expected
                
            elif assertion_type == 'response_time':
                max_time = assertion.get('max_time')
                success = response_time <= max_time
                
            elif assertion_type == 'json_path':
                import jsonpath
                response_json = json.loads(response_text)
                path = assertion.get('path')
                expected = assertion.get('expected')
                actual = jsonpath.jsonpath(response_json, path)
                success = actual and actual[0] == expected
                
            else:
                success = False
                
            return {
                'success': success,
                'type': assertion_type,
                'expected': assertion.get('expected'),
                'actual': locals().get('actual', 'N/A')
            }
            
        except Exception as e:
            return {
                'success': False,
                'type': assertion_type,
                'error': str(e)
            }

class ExecutorFactory:
    """执行器工厂"""
    
    _executors = {
        'api': APIExecutor,
        # 'ui': UIExecutor,
        # 'database': DatabaseExecutor,
    }
    
    @classmethod
    def create_executor(cls, testcase, execution_record):
        """创建执行器"""
        executor_class = cls._executors.get(testcase.type)
        if not executor_class:
            raise ValueError(f"不支持的用例类型: {testcase.type}")
        
        return executor_class(testcase, execution_record)

🎯 下一步预告

今天我们设计了一个强大而灵活的测试用例管理系统，实现了：

灵活的数据模型设计
多种用例类型支持
完整的执行引擎架构
详细的执行记录追踪

下一篇我们将学习测试报告与数据可视化开发，让测试结果更直观、更有说服力。毕竟，再好的测试，如果结果展示不清楚，也是白搭！

💡 系统设计小贴士：设计系统就像盖房子，地基要稳，结构要清晰，扩展要方便。记住，过度设计是万恶之源，够用就好，迭代优化！