3、集成测试环境搭建：Docker部署Redis集群、测试数据准备、测试框架选型（pytest+redis-py）

集成测试环境是错误码断言工程的基石。本章将完整搭建一套可复用的本地集成测试环境，涵盖Redis集群的Docker化部署、测试数据的结构化准备，以及pytest+redis-py测试框架的初始化配置。环境搭建遵循“容器化、可重复、易清理”原则，确保每次测试运行在一致的初始状态下。

3.1 Docker部署Redis集群

搜索服务的Redis集群采用3主3从架构，每个节点运行在独立的Docker容器中。使用docker-compose统一编排，确保网络互通与持久化目录隔离。

3.1.1 集群配置文件

创建redis-cluster.yml，定义6个Redis节点服务：

version: '3.8'
services:
  redis-node-1:
    image: redis:7.0-alpine
    container_name: redis-node-1
    ports:
      - "7000:6379"
      - "17000:16379"
    volumes:
      - ./data/node-1:/data
      - ./conf/redis-cluster.conf:/usr/local/etc/redis/redis.conf
    command: redis-server /usr/local/etc/redis/redis.conf
    networks:
      - redis-net

  redis-node-2:
    image: redis:7.0-alpine
    container_name: redis-node-2
    ports:
      - "7001:6379"
      - "17001:16379"
    volumes:
      - ./data/node-2:/data
      - ./conf/redis-cluster.conf:/usr/local/etc/redis/redis.conf
    command: redis-server /usr/local/etc/redis/redis.conf
    networks:
      - redis-net

  # ... 节点3-6配置类似，端口依次为7002-7005 / 17002-17005

  redis-cluster-init:
    image: redis:7.0-alpine
    container_name: redis-cluster-init
    depends_on:
      - redis-node-1
      - redis-node-2
      - redis-node-3
      - redis-node-4
      - redis-node-5
      - redis-node-6
    command: >
      sh -c "sleep 5 && \
      redis-cli --cluster create \
      172.20.0.2:6379 172.20.0.3:6379 172.20.0.4:6379 \
      172.20.0.5:6379 172.20.0.6:6379 172.20.0.7:6379 \
      --cluster-replicas 1 --cluster-yes"
    networks:
      - redis-net

networks:
  redis-net:
    driver: bridge
    ipam:
      config:
        - subnet: 172.20.0.0/16

3.1.2 节点配置文件

每个节点共用同一份redis-cluster.conf，关键参数如下：

3.1.3 启动与验证

# 启动所有容器
docker-compose -f redis-cluster.yml up -d

# 等待集群初始化完成（约10秒）
sleep 10

# 验证集群状态
docker exec redis-node-1 redis-cli -p 6379 cluster info

# 预期输出包含 cluster_state:ok，且 cluster_size:3

3.2 测试数据准备

测试数据分为三类：基础种子数据（模拟搜索索引）、错误码触发数据（用于断言验证）、边界数据（测试极限场景）。数据通过redis-py客户端在pytest的conftest.py中统一注入。

3.2.1 数据分类与结构

3.2.2 数据准备脚本

在tests/conftest.py中定义seed_test_data函数，使用pytest的session作用域确保数据只加载一次：

import pytest
import redis
import json

@pytest.fixture(scope="session")
def redis_cluster():
    """创建Redis集群客户端连接"""
    startup_nodes = [
        {"host": "localhost", "port": 7000},
        {"host": "localhost", "port": 7001},
        {"host": "localhost", "port": 7002},
    ]
    client = redis.RedisCluster(
        startup_nodes=startup_nodes,
        decode_responses=True,
        skip_full_coverage_check=True
    )
    yield client
    # 清理：删除测试数据
    keys = client.keys("test:*")
    if keys:
        client.delete(*keys)

@pytest.fixture(scope="session", autouse=True)
def seed_test_data(redis_cluster):
    """注入测试种子数据"""
    # 1. 搜索索引数据（3条文档）
    docs = [
        {"id": "1001", "title": "Redis集群部署指南", "content": "步骤详解..."},
        {"id": "1002", "title": "错误码MOVED处理", "content": "重定向机制..."},
        {"id": "1003", "title": "pytest集成测试", "content": "断言技巧..."},
    ]
    for doc in docs:
        key = f"search:doc:{doc['id']}"
        redis_cluster.set(key, json.dumps(doc))

    # 2. 错误码触发数据
    error_cases = [
        {"code": "MOVED", "slot": 1234, "expected_node": "172.20.0.3:6379"},
        {"code": "ASK", "slot": 5678, "expected_node": "172.20.0.4:6379"},
        {"code": "CROSSSLOT", "keys": ["key1", "key2"]},
    ]
    for case in error_cases:
        key = f"test:error:{case['code']}"
        redis_cluster.set(key, json.dumps(case))

    # 3. 边界数据
    redis_cluster.set("test:boundary:empty", "")
    redis_cluster.set("test:boundary:large", "x" * 10240)
    redis_cluster.set("test:boundary:unicode", "中文测试🔍")

    return redis_cluster

3.3 测试框架选型：pytest + redis-py

选择pytest作为测试框架，redis-py作为Redis客户端，二者组合具备以下优势：

• pytest：支持fixture、参数化、插件生态（pytest-xdist并行、pytest-assume软断言）
• redis-py：原生支持RedisCluster，自动处理MOVED/ASK重定向，提供RedisClusterException异常类
• 集成优势：fixture管理连接生命周期，参数化覆盖多种错误码场景，断言直接捕获异常码

3.3.1 依赖安装

# requirements-dev.txt
pytest==7.4.0
redis==5.0.0
pytest-xdist==3.3.1   # 可选：并行测试
pytest-assume==2.4.3  # 可选：软断言

3.3.2 测试框架初始化结构

项目测试目录结构如下：

tests/
├── conftest.py              # 全局fixture：redis_cluster, seed_test_data
├── test_error_code_assert/  # 错误码断言测试模块
│   ├── __init__.py
│   ├── test_moved_assert.py
│   ├── test_ask_assert.py
│   └── test_crossslot_assert.py
├── test_data_preparation/   # 数据准备验证模块
│   ├── __init__.py
│   └── test_seed_data.py
└── pytest.ini               # pytest配置文件

pytest.ini 配置示例：

[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
markers =
    error_code: 错误码断言测试
    data_prep: 数据准备验证
    slow: 耗时较长的测试
log_cli = true
log_cli_level = INFO

3.3.3 第一个集成测试用例

验证种子数据是否成功注入：

# tests/test_data_preparation/test_seed_data.py
import json
import pytest

class TestSeedData:
    """验证测试数据准备是否正确"""

    def test_search_doc_exists(self, redis_cluster):
        """搜索索引数据应可正常读取"""
        doc_json = redis_cluster.get("search:doc:1001")
        assert doc_json is not None, "文档1001未找到"
        doc = json.loads(doc_json)
        assert doc["title"] == "Redis集群部署指南"

    def test_error_trigger_data(self, redis_cluster):
        """错误码触发数据应包含expected字段"""
        error_json = redis_cluster.get("test:error:MOVED")
        error_data = json.loads(error_json)
        assert "expected_node" in error_data
        assert error_data["expected_node"] == "172.20.0.3:6379"

    def test_boundary_empty_value(self, redis_cluster):
        """空值边界数据应返回空字符串"""
        value = redis_cluster.get("test:boundary:empty")
        assert value == ""

    @pytest.mark.slow
    def test_boundary_large_value(self, redis_cluster):
        """大值边界数据应完整返回"""
        value = redis_cluster.get("test:boundary:large")
        assert len(value) == 10240

3.4 环境验证清单

完成上述步骤后，执行以下验证命令确保环境就绪：

# 1. 验证Docker容器运行状态
docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Ports}}" | grep redis

# 2. 验证集群状态
docker exec redis-node-1 redis-cli -p 6379 cluster nodes | head -10

# 3. 运行数据准备验证测试
pytest tests/test_data_preparation/ -v

# 4. 运行空测试确认框架正常
pytest --collect-only