3. Testing框架核心机制：测试用例编写规范、断言库使用、Mock与Stub技术、Fixture与参数化

在容器化故障排查场景中，Testing 框架是保障压测脚本可靠性与可复现性的基石。本章深入剖析框架的四大核心机制，帮助你在混沌工程与故障注入场景中编写健壮的测试用例。

3.1 测试用例编写规范

容器化环境下的测试用例需遵循“原子性、可重复、可隔离”原则。每个用例应只验证一个故障场景或一个恢复动作。

3.2 断言库使用

断言是验证故障注入效果的“裁判”。在容器化压测中，断言不仅要检查状态码，还需验证延迟、资源使用、日志输出等指标。

3.2.1 常用断言模式

# 基础断言
assert response.status_code == 200
assert response.elapsed.total_seconds() < 5.0

# 容器状态断言
assert container.status == "running"
assert float(container.attrs['State']['StartedAt']) > threshold

# 日志内容断言
assert "connection refused" in container.logs().decode()

# 指标范围断言
assert 0.5 <= cpu_percent <= 0.8, f"CPU 使用率异常: {cpu_percent}"

3.2.2 断言库对比

3.3 Mock与Stub技术

在容器化环境中，Mock 和 Stub 用于模拟外部依赖（如数据库、消息队列、第三方 API），避免故障注入影响真实服务。

3.3.1 Mock vs Stub 对比

3.3.2 容器化Mock实践

from unittest.mock import patch

def test_container_health_check_retry():
    with patch('requests.get') as mock_get:
        # 模拟前两次失败，第三次成功
        mock_get.side_effect = [
            ConnectionError("timeout"),
            ConnectionError("timeout"),
            Mock(status_code=200)
        ]
        result = health_check_with_retry("http://container:8080/health")
        assert result is True
        assert mock_get.call_count == 3

3.4 Fixture与参数化

Fixture 是容器化测试的“基础设施”，负责容器的启动、配置、故障注入和清理。参数化则让一个测试用例覆盖多种故障参数。

3.4.1 Fixture 生命周期

import pytest
import docker

@pytest.fixture(scope="module")
def container():
    """启动一个 Nginx 容器作为测试目标"""
    client = docker.from_env()
    container = client.containers.run(
        "nginx:alpine",
        detach=True,
        ports={'80/tcp': None}
    )
    yield container
    # 清理：停止并删除容器
    container.remove(force=True)

def test_nginx_responds(container):
    response = requests.get(f"http://{container.attrs['NetworkSettings']['IPAddress']}")
    assert response.status_code == 200

3.4.2 参数化故障注入

@pytest.mark.parametrize("delay_ms,expected_timeout", [
    (100, False),   # 100ms 延迟，不应超时
    (500, False),   # 500ms 延迟，不应超时
    (2000, True),   # 2000ms 延迟，应超时
])
def test_network_delay(container, delay_ms, expected_timeout):
    # 注入网络延迟
    container.exec_run(f"tc qdisc add dev eth0 root netem delay {delay_ms}ms")
    try:
        response = requests.get(
            f"http://{container.attrs['NetworkSettings']['IPAddress']}",
            timeout=1
        )
        assert not expected_timeout
    except requests.Timeout:
        assert expected_timeout
    finally:
        # 清理网络规则
        container.exec_run("tc qdisc del dev eth0 root")

3.5 核心机制关系图

下图展示了 Testing 框架四大核心机制在容器化故障排查中的协作关系：

3.6 综合实践：故障注入测试模板

将上述机制整合为一个完整的容器化故障注入测试模板：

import pytest
import docker
import requests
from unittest.mock import patch

# Fixture：启动目标容器
@pytest.fixture(scope="module")
def target_container():
    client = docker.from_env()
    container = client.containers.run(
        "nginx:alpine",
        detach=True,
        ports={'80/tcp': None}
    )
    yield container
    container.remove(force=True)

# 参数化：多种故障类型
@pytest.mark.parametrize("fault_type,params", [
    ("delay", {"delay_ms": 200}),
    ("packet_loss", {"loss_percent": 10}),
    ("bandwidth", {"rate": "1mbit"}),
])
def test_network_fault_injection(target_container, fault_type, params):
    """验证不同网络故障对容器响应的影响"""
    # Arrange：注入故障
    if fault_type == "delay":
        cmd = f"tc qdisc add dev eth0 root netem delay {params['delay_ms']}ms"
    elif fault_type == "packet_loss":
        cmd = f"tc qdisc add dev eth0 root netem loss {params['loss_percent']}%"
    elif fault_type == "bandwidth":
        cmd = f"tc qdisc add dev eth0 root tbf rate {params['rate']} burst 32kbit latency 400ms"
    
    target_container.exec_run(cmd)
    
    # Act：发起请求
    ip = target_container.attrs['NetworkSettings']['IPAddress']
    try:
        response = requests.get(f"http://{ip}", timeout=5)
        # Assert：故障下仍应返回 200
        assert response.status_code == 200
    except requests.Timeout:
        # 超时也是预期行为之一
        assert fault_type == "delay" and params["delay_ms"] > 1000
    finally:
        # 清理：移除故障
        target_container.exec_run("tc qdisc del dev eth0 root")