测试自动化

学习目标

通过本文档的学习，你将能够：

• 理解核心概念和原理
• 掌握实际应用方法
• 了解最佳实践和注意事项

前置知识

在学习本文档之前，建议你已经掌握：

• 基础的嵌入式系统知识
• C/C++编程基础
• 相关领域的基本概念

概述

测试自动化是使用软件工具自动执行测试用例、比较实际结果与预期结果、生成测试报告的过程。对于医疗设备软件，自动化测试可以提高测试效率、确保测试一致性、支持持续集成和快速迭代。

为什么需要测试自动化？

医疗设备软件的挑战

• 频繁的回归测试: 每次代码变更都需要验证现有功能
• 复杂的测试场景: 需要测试大量的输入组合和边界条件
• 监管要求: 需要可追溯的测试记录和文档
• 质量保证: 确保每个版本都经过充分测试

自动化的优势

方面	手动测试	自动化测试
执行速度	慢	快
可重复性	低	高
人为错误	高	低
成本（长期）	高	低
覆盖率	有限	广泛
CI/CD集成	困难	容易

测试自动化策略

测试金字塔

        /\
       /  \  UI测试 (10%)
      /____\  手动/自动化
     /      \
    / API测试 \ (30%)
   /___________\  自动化
  /             \
 /   单元测试    \ (60%)
/_________________\  自动化

自动化测试框架选择

Python测试框架

pytest - 推荐用于医疗设备软件

# test_ecg_processor.py
import pytest
from medical_device.ecg import ECGProcessor

class TestECGProcessor:
    """ECG处理器测试"""

    @pytest.fixture
    def processor(self):
        """测试夹具 - 创建处理器实例"""
        return ECGProcessor(sample_rate=500)

    @pytest.fixture
    def sample_ecg_data(self):
        """测试夹具 - 生成样本ECG数据"""
        import numpy as np
        # 生成60秒的模拟ECG数据
        t = np.linspace(0, 60, 30000)
        ecg = np.sin(2 * np.pi * 1.2 * t)  # 72 bpm
        return ecg

    def test_heart_rate_calculation(self, processor, sample_ecg_data):
        """测试心率计算"""
        heart_rate = processor.calculate_heart_rate(sample_ecg_data)
        assert 70 <= heart_rate <= 75, f"心率应在70-75之间，实际: {heart_rate}"

    def test_r_peak_detection(self, processor, sample_ecg_data):
        """测试R波检测"""
        r_peaks = processor.detect_r_peaks(sample_ecg_data)
        assert len(r_peaks) > 0, "应该检测到R波"
        assert len(r_peaks) >= 70, "60秒应检测到至少70个R波"

    @pytest.mark.parametrize("noise_level", [0.1, 0.2, 0.5])
    def test_noise_tolerance(self, processor, sample_ecg_data, noise_level):
        """测试噪声容忍度"""
        import numpy as np
        noisy_data = sample_ecg_data + np.random.normal(0, noise_level, len(sample_ecg_data))
        heart_rate = processor.calculate_heart_rate(noisy_data)
        assert 60 <= heart_rate <= 85, f"噪声水平{noise_level}下心率计算失败"

    def test_invalid_input(self, processor):
        """测试无效输入处理"""
        with pytest.raises(ValueError):
            processor.calculate_heart_rate([])  # 空数据

        with pytest.raises(ValueError):
            processor.calculate_heart_rate(None)  # None

    @pytest.mark.slow
    def test_long_duration_processing(self, processor):
        """测试长时间数据处理"""
        import numpy as np
        # 24小时的数据
        long_data = np.random.randn(43200000)
        result = processor.process_continuous(long_data)
        assert result is not None

# pytest配置文件
# pytest.ini
[pytest]
markers =
    slow: marks tests as slow (deselect with '-m "not slow"')
    integration: marks tests as integration tests
    unit: marks tests as unit tests
    critical: marks tests as critical for patient safety

testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*

# 覆盖率配置
addopts = 
    --cov=medical_device
    --cov-report=html
    --cov-report=term
    --cov-fail-under=80
    --strict-markers
    -v

Java测试框架

JUnit 5 + Mockito

// ECGProcessorTest.java
import org.junit.jupiter.api.*;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.*;

@DisplayName("ECG处理器测试")
class ECGProcessorTest {

    private ECGProcessor processor;

    @Mock
    private DataLogger logger;

    @BeforeEach
    void setUp() {
        MockitoAnnotations.openMocks(this);
        processor = new ECGProcessor(500, logger);
    }

    @Test
    @DisplayName("应该正确计算心率")
    @Tag("unit")
    void shouldCalculateHeartRateCorrectly() {
        // Arrange
        double[] ecgData = generateSampleECG(60, 72);

        // Act
        int heartRate = processor.calculateHeartRate(ecgData);

        // Assert
        assertTrue(heartRate >= 70 && heartRate <= 75,
            "心率应在70-75之间，实际: " + heartRate);
        verify(logger).log(contains("Heart rate calculated"));
    }

    @ParameterizedTest
    @ValueSource(ints = {60, 80, 100, 120})
    @DisplayName("应该处理不同心率")
    void shouldHandleDifferentHeartRates(int expectedHR) {
        double[] ecgData = generateSampleECG(60, expectedHR);
        int actualHR = processor.calculateHeartRate(ecgData);
        assertEquals(expectedHR, actualHR, 5);
    }

    @Test
    @DisplayName("应该抛出异常当输入为空")
    void shouldThrowExceptionForEmptyInput() {
        assertThrows(IllegalArgumentException.class, () -> {
            processor.calculateHeartRate(new double[0]);
        });
    }

    @RepeatedTest(10)
    @DisplayName("应该产生一致的结果")
    void shouldProduceConsistentResults() {
        double[] ecgData = generateSampleECG(60, 72);
        int firstResult = processor.calculateHeartRate(ecgData);
        int secondResult = processor.calculateHeartRate(ecgData);
        assertEquals(firstResult, secondResult);
    }

    @Nested
    @DisplayName("R波检测测试")
    class RPeakDetectionTests {

        @Test
        @DisplayName("应该检测到所有R波")
        void shouldDetectAllRPeaks() {
            double[] ecgData = generateSampleECG(60, 72);
            List<Integer> rPeaks = processor.detectRPeaks(ecgData);
            assertTrue(rPeaks.size() >= 70);
        }

        @Test
        @DisplayName("R波间隔应该合理")
        void rPeakIntervalsShouldBeReasonable() {
            double[] ecgData = generateSampleECG(60, 72);
            List<Integer> rPeaks = processor.detectRPeaks(ecgData);

            for (int i = 1; i < rPeaks.size(); i++) {
                int interval = rPeaks.get(i) - rPeaks.get(i-1);
                assertTrue(interval > 200 && interval < 600,
                    "R-R间隔应该合理");
            }
        }
    }

    private double[] generateSampleECG(int durationSeconds, int heartRate) {
        // 生成模拟ECG数据
        int sampleRate = 500;
        int numSamples = durationSeconds * sampleRate;
        double[] data = new double[numSamples];

        double frequency = heartRate / 60.0;
        for (int i = 0; i < numSamples; i++) {
            double t = (double) i / sampleRate;
            data[i] = Math.sin(2 * Math.PI * frequency * t);
        }

        return data;
    }
}

CI/CD集成

GitHub Actions工作流

# .github/workflows/medical-device-ci.yml
name: Medical Device CI/CD

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

env:
  PYTHON_VERSION: '3.9'

jobs:
  # 单元测试
  unit-tests:
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v3

    - name: 设置Python
      uses: actions/setup-python@v4
      with:
        python-version: ${{ env.PYTHON_VERSION }}

    - name: 安装依赖
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install -r requirements-dev.txt

    - name: 运行单元测试
      run: |
        pytest tests/unit \
          --cov=medical_device \
          --cov-report=xml \
          --cov-report=html \
          --junitxml=junit/test-results.xml

    - name: 上传覆盖率报告
      uses: codecov/codecov-action@v3
      with:
        files: ./coverage.xml
        flags: unittests
        name: codecov-umbrella

    - name: 发布测试结果
      uses: EnricoMi/publish-unit-test-result-action@v2
      if: always()
      with:
        files: junit/test-results.xml

  # 集成测试
  integration-tests:
    runs-on: ubuntu-latest
    needs: unit-tests

    services:
      postgres:
        image: postgres:13
        env:
          POSTGRES_PASSWORD: postgres
          POSTGRES_DB: medical_device_test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

      redis:
        image: redis:6
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 6379:6379

    steps:
    - uses: actions/checkout@v3

    - name: 设置Python
      uses: actions/setup-python@v4
      with:
        python-version: ${{ env.PYTHON_VERSION }}

    - name: 安装依赖
      run: |
        pip install -r requirements.txt
        pip install -r requirements-dev.txt

    - name: 运行集成测试
      env:
        DATABASE_URL: postgresql://postgres:postgres@localhost:5432/medical_device_test
        REDIS_URL: redis://localhost:6379
      run: |
        pytest tests/integration -v --tb=short

  # 性能测试
  performance-tests:
    runs-on: ubuntu-latest
    needs: integration-tests

    steps:
    - uses: actions/checkout@v3

    - name: 设置Python
      uses: actions/setup-python@v4
      with:
        python-version: ${{ env.PYTHON_VERSION }}

    - name: 安装依赖
      run: |
        pip install -r requirements.txt
        pip install locust

    - name: 运行性能测试
      run: |
        locust -f tests/performance/locustfile.py \
          --headless \
          --users 100 \
          --spawn-rate 10 \
          --run-time 5m \
          --host https://staging.medical-device.com \
          --html performance-report.html

    - name: 上传性能报告
      uses: actions/upload-artifact@v3
      with:
        name: performance-report
        path: performance-report.html

  # 安全扫描
  security-scan:
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v3

    - name: 运行Bandit安全扫描
      run: |
        pip install bandit
        bandit -r medical_device/ -f json -o bandit-report.json

    - name: 运行Safety依赖检查
      run: |
        pip install safety
        safety check --json > safety-report.json

    - name: 上传安全报告
      uses: actions/upload-artifact@v3
      with:
        name: security-reports
        path: |
          bandit-report.json
          safety-report.json

  # 构建Docker镜像
  build-docker:
    runs-on: ubuntu-latest
    needs: [unit-tests, integration-tests, security-scan]

    steps:
    - uses: actions/checkout@v3

    - name: 设置Docker Buildx
      uses: docker/setup-buildx-action@v2

    - name: 登录Docker Hub
      uses: docker/login-action@v2
      with:
        username: ${{ secrets.DOCKER_USERNAME }}
        password: ${{ secrets.DOCKER_PASSWORD }}

    - name: 构建并推送
      uses: docker/build-push-action@v4
      with:
        context: .
        push: true
        tags: |
          medical-device:${{ github.sha }}
          medical-device:latest
        cache-from: type=gha
        cache-to: type=gha,mode=max

    - name: 扫描Docker镜像
      run: |
        docker run --rm \
          -v /var/run/docker.sock:/var/run/docker.sock \
          aquasec/trivy image medical-device:${{ github.sha }}

GitLab CI/CD配置

# .gitlab-ci.yml
stages:
  - test
  - security
  - build
  - deploy

variables:
  PYTHON_VERSION: "3.9"
  PIP_CACHE_DIR: "$CI_PROJECT_DIR/.cache/pip"

cache:
  paths:
    - .cache/pip
    - venv/

before_script:
  - python -V
  - pip install virtualenv
  - virtualenv venv
  - source venv/bin/activate
  - pip install -r requirements.txt

# 单元测试
unit-test:
  stage: test
  script:
    - pip install pytest pytest-cov
    - pytest tests/unit --cov=medical_device --cov-report=term --cov-report=xml
  coverage: '/TOTAL.*\s+(\d+%)$/'
  artifacts:
    reports:
      coverage_report:
        coverage_format: cobertura
        path: coverage.xml
    paths:
      - htmlcov/
    expire_in: 1 week

# 集成测试
integration-test:
  stage: test
  services:
    - postgres:13
    - redis:6
  variables:
    POSTGRES_DB: medical_device_test
    POSTGRES_USER: postgres
    POSTGRES_PASSWORD: postgres
    DATABASE_URL: postgresql://postgres:postgres@postgres:5432/medical_device_test
  script:
    - pytest tests/integration -v
  only:
    - merge_requests
    - main

# 代码质量检查
code-quality:
  stage: test
  script:
    - pip install pylint flake8 mypy
    - pylint medical_device/
    - flake8 medical_device/
    - mypy medical_device/
  allow_failure: true

# 安全扫描
security-sast:
  stage: security
  script:
    - pip install bandit
    - bandit -r medical_device/ -f json -o bandit-report.json
  artifacts:
    reports:
      sast: bandit-report.json

# 依赖扫描
dependency-scanning:
  stage: security
  script:
    - pip install safety
    - safety check --json > safety-report.json
  artifacts:
    paths:
      - safety-report.json

# 构建
build:
  stage: build
  script:
    - python setup.py bdist_wheel
  artifacts:
    paths:
      - dist/
    expire_in: 1 month
  only:
    - tags
    - main

# 部署到测试环境
deploy-staging:
  stage: deploy
  script:
    - echo "Deploying to staging environment"
    - pip install dist/*.whl
    - python scripts/deploy_staging.py
  environment:
    name: staging
    url: https://staging.medical-device.com
  only:
    - develop

# 部署到生产环境
deploy-production:
  stage: deploy
  script:
    - echo "Deploying to production environment"
    - pip install dist/*.whl
    - python scripts/deploy_production.py
  environment:
    name: production
    url: https://medical-device.com
  when: manual
  only:
    - tags

测试数据管理

测试数据生成

# test_data_factory.py
from faker import Faker
import random
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List

fake = Faker('zh_CN')

@dataclass
class PatientData:
    """患者数据"""
    patient_id: str
    name: str
    gender: str
    birth_date: str
    phone: str
    address: str

@dataclass
class VitalSigns:
    """生命体征数据"""
    patient_id: str
    timestamp: datetime
    heart_rate: int
    blood_pressure_systolic: int
    blood_pressure_diastolic: int
    temperature: float
    spo2: int

class TestDataFactory:
    """测试数据工厂"""

    @staticmethod
    def create_patient(patient_id=None) -> PatientData:
        """创建患者数据"""
        return PatientData(
            patient_id=patient_id or f"P{random.randint(100000, 999999)}",
            name=fake.name(),
            gender=random.choice(['M', 'F']),
            birth_date=fake.date_of_birth(minimum_age=18, maximum_age=90).isoformat(),
            phone=fake.phone_number(),
            address=fake.address()
        )

    @staticmethod
    def create_vital_signs(patient_id: str, timestamp=None) -> VitalSigns:
        """创建生命体征数据"""
        return VitalSigns(
            patient_id=patient_id,
            timestamp=timestamp or datetime.now(),
            heart_rate=random.randint(60, 100),
            blood_pressure_systolic=random.randint(110, 140),
            blood_pressure_diastolic=random.randint(70, 90),
            temperature=round(random.uniform(36.0, 37.5), 1),
            spo2=random.randint(95, 100)
        )

    @staticmethod
    def create_ecg_data(duration_seconds=60, sample_rate=500, heart_rate=72):
        """创建ECG数据"""
        import numpy as np
        num_samples = duration_seconds * sample_rate
        t = np.linspace(0, duration_seconds, num_samples)
        frequency = heart_rate / 60.0

        # 生成基础正弦波
        ecg = np.sin(2 * np.pi * frequency * t)

        # 添加噪声
        noise = np.random.normal(0, 0.1, num_samples)
        ecg += noise

        return ecg

    @staticmethod
    def create_batch_patients(count=100) -> List[PatientData]:
        """批量创建患者数据"""
        return [TestDataFactory.create_patient() for _ in range(count)]

    @staticmethod
    def create_patient_history(patient_id: str, days=30) -> List[VitalSigns]:
        """创建患者历史数据"""
        history = []
        start_date = datetime.now() - timedelta(days=days)

        for day in range(days):
            for hour in range(0, 24, 4):  # 每4小时一次测量
                timestamp = start_date + timedelta(days=day, hours=hour)
                history.append(
                    TestDataFactory.create_vital_signs(patient_id, timestamp)
                )

        return history

# 使用示例
def test_patient_data_processing():
    """测试患者数据处理"""
    # 创建测试数据
    patient = TestDataFactory.create_patient()
    vitals = TestDataFactory.create_vital_signs(patient.patient_id)

    # 执行测试
    processor = PatientDataProcessor()
    result = processor.process(patient, vitals)

    assert result is not None
    assert result.patient_id == patient.patient_id

测试数据库管理

# test_database.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from medical_device.models import Base, Patient, VitalSigns

@pytest.fixture(scope='session')
def test_engine():
    """创建测试数据库引擎"""
    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(engine)
    return engine

@pytest.fixture(scope='function')
def db_session(test_engine):
    """创建数据库会话"""
    Session = sessionmaker(bind=test_engine)
    session = Session()

    yield session

    # 清理
    session.rollback()
    session.close()

@pytest.fixture
def sample_patient(db_session):
    """创建样本患者"""
    patient = Patient(
        patient_id='P123456',
        name='张三',
        gender='M',
        birth_date='1980-01-01'
    )
    db_session.add(patient)
    db_session.commit()
    return patient

def test_patient_crud(db_session, sample_patient):
    """测试患者CRUD操作"""
    # Read
    patient = db_session.query(Patient).filter_by(
        patient_id='P123456'
    ).first()
    assert patient is not None
    assert patient.name == '张三'

    # Update
    patient.name = '李四'
    db_session.commit()

    updated_patient = db_session.query(Patient).filter_by(
        patient_id='P123456'
    ).first()
    assert updated_patient.name == '李四'

    # Delete
    db_session.delete(patient)
    db_session.commit()

    deleted_patient = db_session.query(Patient).filter_by(
        patient_id='P123456'
    ).first()
    assert deleted_patient is None

回归测试自动化

回归测试套件

# regression_test_suite.py
import pytest
from medical_device import ECGProcessor, PatientDataManager
from test_data_factory import TestDataFactory

class RegressionTestSuite:
    """回归测试套件"""

    @pytest.mark.regression
    class TestECGProcessing:
        """ECG处理回归测试"""

        def test_heart_rate_calculation_accuracy(self):
            """测试心率计算准确性 - 回归测试"""
            processor = ECGProcessor(sample_rate=500)

            # 测试已知的心率值
            test_cases = [
                (60, 60),   # 60 bpm
                (72, 72),   # 72 bpm
                (100, 100), # 100 bpm
                (120, 120)  # 120 bpm
            ]

            for expected_hr, input_hr in test_cases:
                ecg_data = TestDataFactory.create_ecg_data(
                    duration_seconds=60,
                    heart_rate=input_hr
                )
                calculated_hr = processor.calculate_heart_rate(ecg_data)
                assert abs(calculated_hr - expected_hr) <= 2, \
                    f"心率计算偏差过大: 期望{expected_hr}, 实际{calculated_hr}"

        def test_r_peak_detection_consistency(self):
            """测试R波检测一致性 - 回归测试"""
            processor = ECGProcessor(sample_rate=500)
            ecg_data = TestDataFactory.create_ecg_data(duration_seconds=60)

            # 多次运行应该产生相同结果
            results = [processor.detect_r_peaks(ecg_data) for _ in range(10)]

            # 验证结果一致性
            first_result = results[0]
            for result in results[1:]:
                assert len(result) == len(first_result), \
                    "R波检测结果不一致"

        def test_filter_performance(self):
            """测试滤波器性能 - 回归测试"""
            processor = ECGProcessor(sample_rate=500)
            ecg_data = TestDataFactory.create_ecg_data(duration_seconds=60)

            import time
            start_time = time.time()
            filtered_data = processor.apply_bandpass_filter(ecg_data)
            elapsed_time = time.time() - start_time

            # 性能回归检查
            assert elapsed_time < 0.1, \
                f"滤波器性能退化: {elapsed_time}秒 > 0.1秒"
            assert len(filtered_data) == len(ecg_data), \
                "滤波后数据长度改变"

    @pytest.mark.regression
    class TestPatientDataManagement:
        """患者数据管理回归测试"""

        def test_patient_search_functionality(self, db_session):
            """测试患者搜索功能 - 回归测试"""
            # 创建测试数据
            patients = TestDataFactory.create_batch_patients(100)
            for patient in patients:
                db_session.add(patient)
            db_session.commit()

            manager = PatientDataManager(db_session)

            # 测试各种搜索条件
            results = manager.search_patients(name='张')
            assert len(results) > 0, "按姓名搜索失败"

            results = manager.search_patients(gender='M')
            assert all(p.gender == 'M' for p in results), "按性别搜索失败"

        def test_data_export_format(self, db_session):
            """测试数据导出格式 - 回归测试"""
            patient = TestDataFactory.create_patient()
            db_session.add(patient)
            db_session.commit()

            manager = PatientDataManager(db_session)
            exported_data = manager.export_patient_data(patient.patient_id)

            # 验证导出格式
            required_fields = ['patient_id', 'name', 'gender', 'birth_date']
            for field in required_fields:
                assert field in exported_data, \
                    f"导出数据缺少字段: {field}"

# 运行回归测试
# pytest -m regression -v

视觉回归测试

# visual_regression_test.py
import pytest
from selenium import webdriver
from selenium.webdriver.common.by import By
from PIL import Image, ImageChops
import os

class VisualRegressionTest:
    """视觉回归测试"""

    @pytest.fixture
    def driver(self):
        """创建WebDriver"""
        options = webdriver.ChromeOptions()
        options.add_argument('--headless')
        driver = webdriver.Chrome(options=options)
        driver.set_window_size(1920, 1080)
        yield driver
        driver.quit()

    def capture_screenshot(self, driver, url, filename):
        """捕获截图"""
        driver.get(url)
        driver.save_screenshot(filename)

    def compare_images(self, baseline_path, current_path, threshold=0.01):
        """比较图像"""
        baseline = Image.open(baseline_path)
        current = Image.open(current_path)

        # 计算差异
        diff = ImageChops.difference(baseline, current)

        # 计算差异百分比
        histogram = diff.histogram()
        total_pixels = sum(histogram)
        different_pixels = sum(histogram[1:])
        diff_percentage = different_pixels / total_pixels

        return diff_percentage <= threshold

    @pytest.mark.visual
    def test_dashboard_layout(self, driver):
        """测试仪表板布局 - 视觉回归"""
        url = 'https://medical-device.com/dashboard'
        baseline_path = 'tests/visual/baseline/dashboard.png'
        current_path = 'tests/visual/current/dashboard.png'

        # 捕获当前截图
        self.capture_screenshot(driver, url, current_path)

        # 如果基线不存在，创建基线
        if not os.path.exists(baseline_path):
            os.makedirs(os.path.dirname(baseline_path), exist_ok=True)
            self.capture_screenshot(driver, url, baseline_path)
            pytest.skip("创建基线图像")

        # 比较图像
        is_similar = self.compare_images(baseline_path, current_path)
        assert is_similar, "仪表板布局发生变化"

API自动化测试

REST API测试

# test_api.py
import pytest
import requests
from requests.auth import HTTPBasicAuth

class TestMedicalDeviceAPI:
    """医疗设备API测试"""

    BASE_URL = "https://api.medical-device.com/v1"

    @pytest.fixture(scope='class')
    def auth_token(self):
        """获取认证令牌"""
        response = requests.post(
            f"{self.BASE_URL}/auth/login",
            json={
                "username": "test_user",
                "password": "test_password"
            }
        )
        assert response.status_code == 200
        return response.json()['token']

    @pytest.fixture
    def headers(self, auth_token):
        """请求头"""
        return {
            "Authorization": f"Bearer {auth_token}",
            "Content-Type": "application/json"
        }

    def test_get_patients_list(self, headers):
        """测试获取患者列表"""
        response = requests.get(
            f"{self.BASE_URL}/patients",
            headers=headers
        )

        assert response.status_code == 200
        data = response.json()
        assert 'patients' in data
        assert isinstance(data['patients'], list)

    def test_create_patient(self, headers):
        """测试创建患者"""
        patient_data = {
            "name": "测试患者",
            "gender": "M",
            "birth_date": "1980-01-01",
            "phone": "13800138000"
        }

        response = requests.post(
            f"{self.BASE_URL}/patients",
            json=patient_data,
            headers=headers
        )

        assert response.status_code == 201
        data = response.json()
        assert 'patient_id' in data
        assert data['name'] == patient_data['name']

        # 清理
        patient_id = data['patient_id']
        requests.delete(
            f"{self.BASE_URL}/patients/{patient_id}",
            headers=headers
        )

    def test_get_patient_by_id(self, headers):
        """测试根据ID获取患者"""
        # 先创建患者
        patient_data = {"name": "测试患者", "gender": "M", "birth_date": "1980-01-01"}
        create_response = requests.post(
            f"{self.BASE_URL}/patients",
            json=patient_data,
            headers=headers
        )
        patient_id = create_response.json()['patient_id']

        # 获取患者
        response = requests.get(
            f"{self.BASE_URL}/patients/{patient_id}",
            headers=headers
        )

        assert response.status_code == 200
        data = response.json()
        assert data['patient_id'] == patient_id
        assert data['name'] == patient_data['name']

        # 清理
        requests.delete(
            f"{self.BASE_URL}/patients/{patient_id}",
            headers=headers
        )

    def test_update_patient(self, headers):
        """测试更新患者信息"""
        # 创建患者
        patient_data = {"name": "原始姓名", "gender": "M", "birth_date": "1980-01-01"}
        create_response = requests.post(
            f"{self.BASE_URL}/patients",
            json=patient_data,
            headers=headers
        )
        patient_id = create_response.json()['patient_id']

        # 更新患者
        update_data = {"name": "更新后姓名"}
        response = requests.patch(
            f"{self.BASE_URL}/patients/{patient_id}",
            json=update_data,
            headers=headers
        )

        assert response.status_code == 200
        data = response.json()
        assert data['name'] == update_data['name']

        # 清理
        requests.delete(
            f"{self.BASE_URL}/patients/{patient_id}",
            headers=headers
        )

    def test_delete_patient(self, headers):
        """测试删除患者"""
        # 创建患者
        patient_data = {"name": "待删除患者", "gender": "M", "birth_date": "1980-01-01"}
        create_response = requests.post(
            f"{self.BASE_URL}/patients",
            json=patient_data,
            headers=headers
        )
        patient_id = create_response.json()['patient_id']

        # 删除患者
        response = requests.delete(
            f"{self.BASE_URL}/patients/{patient_id}",
            headers=headers
        )

        assert response.status_code == 204

        # 验证已删除
        get_response = requests.get(
            f"{self.BASE_URL}/patients/{patient_id}",
            headers=headers
        )
        assert get_response.status_code == 404

    @pytest.mark.parametrize("invalid_data", [
        {},  # 空数据
        {"name": ""},  # 空姓名
        {"name": "测试", "gender": "X"},  # 无效性别
        {"name": "测试", "birth_date": "invalid"},  # 无效日期
    ])
    def test_create_patient_validation(self, headers, invalid_data):
        """测试患者创建验证"""
        response = requests.post(
            f"{self.BASE_URL}/patients",
            json=invalid_data,
            headers=headers
        )

        assert response.status_code == 400
        data = response.json()
        assert 'error' in data

使用Tavern进行API测试

# tests/api/test_patients.tavern.yaml
---
test_name: 患者管理API测试

stages:
  - name: 登录获取令牌
    request:
      url: "{base_url}/auth/login"
      method: POST
      json:
        username: test_user
        password: test_password
    response:
      status_code: 200
      save:
        json:
          auth_token: token

  - name: 创建患者
    request:
      url: "{base_url}/patients"
      method: POST
      headers:
        Authorization: "Bearer {auth_token}"
      json:
        name: 测试患者
        gender: M
        birth_date: "1980-01-01"
        phone: "13800138000"
    response:
      status_code: 201
      json:
        name: 测试患者
        gender: M
      save:
        json:
          patient_id: patient_id

  - name: 获取患者信息
    request:
      url: "{base_url}/patients/{patient_id}"
      method: GET
      headers:
        Authorization: "Bearer {auth_token}"
    response:
      status_code: 200
      json:
        patient_id: "{patient_id}"
        name: 测试患者

  - name: 更新患者信息
    request:
      url: "{base_url}/patients/{patient_id}"
      method: PATCH
      headers:
        Authorization: "Bearer {auth_token}"
      json:
        name: 更新后姓名
    response:
      status_code: 200
      json:
        name: 更新后姓名

  - name: 删除患者
    request:
      url: "{base_url}/patients/{patient_id}"
      method: DELETE
      headers:
        Authorization: "Bearer {auth_token}"
    response:
      status_code: 204

  - name: 验证患者已删除
    request:
      url: "{base_url}/patients/{patient_id}"
      method: GET
      headers:
        Authorization: "Bearer {auth_token}"
    response:
      status_code: 404

测试报告和度量

自定义测试报告

# conftest.py
import pytest
from datetime import datetime
import json

class TestReport:
    """测试报告生成器"""

    def __init__(self):
        self.results = {
            'start_time': datetime.now().isoformat(),
            'tests': [],
            'summary': {
                'total': 0,
                'passed': 0,
                'failed': 0,
                'skipped': 0,
                'errors': 0
            }
        }

    def add_test_result(self, nodeid, outcome, duration, error=None):
        """添加测试结果"""
        self.results['tests'].append({
            'test': nodeid,
            'outcome': outcome,
            'duration': duration,
            'error': error
        })

        self.results['summary']['total'] += 1
        self.results['summary'][outcome] += 1

    def generate_report(self, filename='test_report.json'):
        """生成报告"""
        self.results['end_time'] = datetime.now().isoformat()

        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.results, f, indent=2, ensure_ascii=False)

@pytest.fixture(scope='session')
def test_report():
    """测试报告夹具"""
    return TestReport()

@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
    """钩子：生成测试报告"""
    outcome = yield
    report = outcome.get_result()

    if report.when == 'call':
        test_report = item.config._test_report
        test_report.add_test_result(
            nodeid=item.nodeid,
            outcome=report.outcome,
            duration=report.duration,
            error=str(report.longrepr) if report.failed else None
        )

def pytest_configure(config):
    """配置钩子"""
    config._test_report = TestReport()

def pytest_sessionfinish(session, exitstatus):
    """会话结束钩子"""
    test_report = session.config._test_report
    test_report.generate_report()

HTML测试报告

# 使用pytest-html生成HTML报告
# pytest --html=report.html --self-contained-html

# 自定义HTML报告
from py.xml import html

def pytest_html_report_title(report):
    """自定义报告标题"""
    report.title = "医疗设备软件测试报告"

def pytest_html_results_summary(prefix, summary, postfix):
    """自定义摘要"""
    prefix.extend([
        html.h2("测试环境"),
        html.p("操作系统: Ubuntu 20.04"),
        html.p("Python版本: 3.9"),
        html.p("测试日期: " + datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    ])

def pytest_html_results_table_header(cells):
    """自定义表头"""
    cells.insert(2, html.th("测试类型"))
    cells.insert(3, html.th("优先级"))

def pytest_html_results_table_row(report, cells):
    """自定义表格行"""
    # 添加测试类型
    test_type = "单元测试"
    if "integration" in report.nodeid:
        test_type = "集成测试"
    elif "e2e" in report.nodeid:
        test_type = "端到端测试"

    cells.insert(2, html.td(test_type))

    # 添加优先级
    priority = "中"
    if "critical" in report.keywords:
        priority = "高"
    elif "low" in report.keywords:
        priority = "低"

    cells.insert(3, html.td(priority))

测试最佳实践

1. 测试命名规范

# 好的测试命名
def test_ecg_processor_calculates_correct_heart_rate_for_normal_rhythm():
    """测试ECG处理器为正常心律计算正确的心率"""
    pass

def test_patient_manager_raises_error_when_patient_id_is_invalid():
    """测试当患者ID无效时患者管理器抛出错误"""
    pass

# 不好的测试命名
def test_1():
    pass

def test_ecg():
    pass

2. 测试独立性

# 好的做法 - 测试独立
class TestPatientManager:
    @pytest.fixture
    def manager(self):
        return PatientManager()

    def test_add_patient(self, manager):
        patient = manager.add_patient("张三")
        assert patient is not None

    def test_remove_patient(self, manager):
        patient = manager.add_patient("李四")
        result = manager.remove_patient(patient.id)
        assert result is True

# 不好的做法 - 测试相互依赖
class TestPatientManager:
    def test_add_patient(self):
        self.patient = manager.add_patient("张三")
        assert self.patient is not None

    def test_remove_patient(self):
        # 依赖于test_add_patient
        result = manager.remove_patient(self.patient.id)
        assert result is True

3. 使用测试夹具

@pytest.fixture(scope='session')
def database():
    """会话级数据库夹具"""
    db = create_test_database()
    yield db
    db.close()

@pytest.fixture(scope='module')
def test_data():
    """模块级测试数据"""
    return load_test_data()

@pytest.fixture(scope='function')
def clean_database(database):
    """函数级数据库清理"""
    yield database
    database.clear()

4. 参数化测试

@pytest.mark.parametrize("heart_rate,expected_category", [
    (40, "bradycardia"),
    (60, "normal"),
    (100, "normal"),
    (120, "tachycardia"),
])
def test_heart_rate_categorization(heart_rate, expected_category):
    """测试心率分类"""
    category = categorize_heart_rate(heart_rate)
    assert category == expected_category

总结

测试自动化是确保医疗设备软件质量的关键实践。通过建立完善的自动化测试体系，可以：

• ✅ 提高测试效率和覆盖率
• ✅ 支持持续集成和快速迭代
• ✅ 确保测试一致性和可重复性
• ✅ 降低长期测试成本
• ✅ 满足监管合规要求

相关资源

• 测试策略概述
• 性能测试
• 安全测试
• 硬件在环测试

参考资料

• pytest文档: https://docs.pytest.org/
• JUnit 5文档: https://junit.org/junit5/docs/current/user-guide/
• Selenium文档: https://www.selenium.dev/documentation/
• GitHub Actions文档: https://docs.github.com/en/actions
• GitLab CI/CD文档: https://docs.gitlab.com/ee/ci/

---

内容有用

报告问题 链接失效 建议改进

---

💬 讨论区

欢迎在这里分享您的想法、提出问题或参与讨论。需要 GitHub 账号登录。