张翔
332 lines
14 KiB
Python
332 lines
14 KiB
Python
"""
|
|
Caffeine缓存管理模块测试 - TDD Red阶段
|
|
|
|
测试基于Caffeine的本地缓存管理功能。
|
|
"""
|
|
|
|
import pytest
|
|
import allure
|
|
import time
|
|
from typing import Any, Optional
|
|
|
|
|
|
@allure.epic("核心框架")
|
|
@allure.feature("Caffeine缓存管理 - TDD Red阶段")
|
|
class TestCaffeineCache:
|
|
"""Caffeine缓存管理测试类 - TDD Red阶段（期望失败）"""
|
|
|
|
@allure.title("测试缓存基本操作 - TDD Red阶段")
|
|
@allure.description("验证缓存的基本CRUD操作 - 期望失败(Red)")
|
|
@allure.severity(allure.severity_level.CRITICAL)
|
|
@pytest.mark.smoke
|
|
def test_cache_basic_operations(self) -> None:
|
|
"""
|
|
TDD Red阶段: 测试缓存基本操作
|
|
|
|
预期结果:
|
|
- 可以put数据到缓存
|
|
- 可以从缓存get数据
|
|
- 可以delete缓存数据
|
|
- 可以检查key是否存在
|
|
"""
|
|
from core.caffeine_cache import CaffeineCache
|
|
|
|
with allure.step("Step 1: 创建缓存实例"):
|
|
cache = CaffeineCache()
|
|
allure.attach("✅ 缓存实例创建成功", "步骤1", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 2: 测试put操作"):
|
|
cache.put("key1", "value1")
|
|
allure.attach("✅ put操作成功", "步骤2", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 3: 测试get操作"):
|
|
value = cache.get("key1")
|
|
allure.attach(f"获取值: {value}", "步骤3", allure.attachment_type.TEXT)
|
|
assert value == "value1", f"缓存值不匹配，期望: value1, 实际: {value}"
|
|
|
|
with allure.step("Step 4: 测试exists操作"):
|
|
exists = cache.exists("key1")
|
|
allure.attach(f"key1存在: {exists}", "步骤4", allure.attachment_type.TEXT)
|
|
assert exists is True, "key1应该存在"
|
|
|
|
with allure.step("Step 5: 测试delete操作"):
|
|
cache.delete("key1")
|
|
value = cache.get("key1")
|
|
allure.attach(f"删除后获取值: {value}", "步骤5", allure.attachment_type.TEXT)
|
|
assert value is None, "删除后应该返回None"
|
|
|
|
@allure.title("测试缓存过期时间 - TDD Red阶段")
|
|
@allure.description("验证缓存数据在指定时间后过期 - 期望失败(Red)")
|
|
@allure.severity(allure.severity_level.CRITICAL)
|
|
@pytest.mark.smoke
|
|
def test_cache_expiration(self) -> None:
|
|
"""
|
|
TDD Red阶段: 测试缓存过期时间
|
|
|
|
预期结果:
|
|
- 设置过期时间的缓存数据在过期后自动失效
|
|
- 未过期的数据仍然有效
|
|
"""
|
|
from core.caffeine_cache import CaffeineCache
|
|
|
|
with allure.step("Step 1: 创建缓存实例"):
|
|
cache = CaffeineCache()
|
|
allure.attach("✅ 缓存实例创建成功", "步骤1", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 2: 设置带过期时间的缓存"):
|
|
cache.put("temp_key", "temp_value", expire_seconds=1)
|
|
allure.attach("✅ 设置1秒过期时间的缓存", "步骤2", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 3: 验证缓存立即生效"):
|
|
value = cache.get("temp_key")
|
|
assert value == "temp_value", f"缓存值不匹配，期望: temp_value, 实际: {value}"
|
|
allure.attach("✅ 缓存立即生效", "步骤3", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 4: 等待缓存过期"):
|
|
time.sleep(1.5) # 等待1.5秒，确保过期
|
|
allure.attach("⏱️ 等待1.5秒", "步骤4", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 5: 验证缓存已过期"):
|
|
value = cache.get("temp_key")
|
|
allure.attach(f"过期后获取值: {value}", "步骤5", allure.attachment_type.TEXT)
|
|
assert value is None, "过期后应该返回None"
|
|
|
|
@allure.title("测试缓存容量限制 - TDD Red阶段")
|
|
@allure.description("验证缓存在达到容量限制时的行为 - 期望失败(Red)")
|
|
@allure.severity(allure.severity_level.CRITICAL)
|
|
@pytest.mark.smoke
|
|
def test_cache_capacity_limit(self) -> None:
|
|
"""
|
|
TDD Red阶段: 测试缓存容量限制
|
|
|
|
预期结果:
|
|
- 当缓存达到容量限制时，最久未使用的数据被移除
|
|
- 最近使用的数据仍然保留
|
|
"""
|
|
from core.caffeine_cache import CaffeineCache
|
|
|
|
with allure.step("Step 1: 创建容量为3的缓存"):
|
|
cache = CaffeineCache(max_size=3)
|
|
allure.attach("✅ 创建容量为3的缓存", "步骤1", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 2: 添加3个缓存项"):
|
|
cache.put("key1", "value1")
|
|
cache.put("key2", "value2")
|
|
cache.put("key3", "value3")
|
|
allure.attach("✅ 添加3个缓存项", "步骤2", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 3: 访问key1使其最近使用"):
|
|
value = cache.get("key1")
|
|
assert value == "value1"
|
|
allure.attach("✅ 访问key1", "步骤3", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 4: 添加第4个缓存项"):
|
|
cache.put("key4", "value4")
|
|
allure.attach("✅ 添加第4个缓存项key4", "步骤4", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 5: 验证key2被移除（最久未使用）"):
|
|
value2 = cache.get("key2")
|
|
value1 = cache.get("key1")
|
|
allure.attach(f"key2值: {value2}", "步骤5", allure.attachment_type.TEXT)
|
|
allure.attach(f"key1值: {value1}", "步骤5", allure.attachment_type.TEXT)
|
|
assert value2 is None, "key2应该被移除"
|
|
assert value1 == "value1", "key1应该仍然存在"
|
|
|
|
@allure.title("测试缓存统计信息 - TDD Red阶段")
|
|
@allure.description("验证缓存统计信息正确收集 - 期望失败(Red)")
|
|
@allure.severity(allure.severity_level.NORMAL)
|
|
@pytest.mark.regression
|
|
def test_cache_statistics(self) -> None:
|
|
"""
|
|
TDD Red阶段: 测试缓存统计信息
|
|
|
|
预期结果:
|
|
- 缓存命中率统计正确
|
|
- 缓存大小统计正确
|
|
- 加载次数统计正确
|
|
"""
|
|
from core.caffeine_cache import CaffeineCache
|
|
|
|
with allure.step("Step 1: 创建启用统计的缓存"):
|
|
cache = CaffeineCache(record_stats=True)
|
|
allure.attach("✅ 创建启用统计的缓存", "步骤1", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 2: 执行缓存操作"):
|
|
cache.put("key1", "value1")
|
|
cache.get("key1") # 命中
|
|
cache.get("key1") # 命中
|
|
cache.get("key2") # 未命中
|
|
allure.attach("✅ 执行缓存操作", "步骤2", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 3: 获取统计信息"):
|
|
stats = cache.get_stats()
|
|
allure.attach(f"统计信息: {stats}", "步骤3", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 4: 验证统计信息"):
|
|
assert "hit_count" in stats, "统计信息应包含hit_count"
|
|
assert "miss_count" in stats, "统计信息应包含miss_count"
|
|
assert "hit_rate" in stats, "统计信息应包含hit_rate"
|
|
assert stats["hit_count"] == 2, "命中次数应为2"
|
|
assert stats["miss_count"] == 1, "未命中次数应为1"
|
|
|
|
@allure.title("测试缓存刷新机制 - TDD Red阶段")
|
|
@allure.description("验证缓存自动刷新机制 - 期望失败(Red)")
|
|
@allure.severity(allure.severity_level.NORMAL)
|
|
@pytest.mark.regression
|
|
def test_cache_refresh(self) -> None:
|
|
"""
|
|
TDD Red阶段: 测试缓存刷新机制
|
|
|
|
预期结果:
|
|
- 缓存支持手动刷新
|
|
- 刷新后数据更新
|
|
"""
|
|
from core.caffeine_cache import CaffeineCache
|
|
|
|
with allure.step("Step 1: 创建缓存实例"):
|
|
cache = CaffeineCache()
|
|
allure.attach("✅ 缓存实例创建成功", "步骤1", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 2: 添加初始数据"):
|
|
cache.put("refresh_key", "initial_value")
|
|
allure.attach("✅ 添加初始数据", "步骤2", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 3: 更新数据"):
|
|
cache.put("refresh_key", "updated_value")
|
|
allure.attach("✅ 更新数据", "步骤3", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 4: 验证数据已更新"):
|
|
value = cache.get("refresh_key")
|
|
allure.attach(f"更新后的值: {value}", "步骤4", allure.attachment_type.TEXT)
|
|
assert value == "updated_value", f"缓存值未更新，期望: updated_value, 实际: {value}"
|
|
|
|
@allure.title("测试缓存批量操作 - TDD Red阶段")
|
|
@allure.description("验证缓存批量操作功能 - 期望失败(Red)")
|
|
@allure.severity(allure.severity_level.NORMAL)
|
|
@pytest.mark.regression
|
|
def test_cache_batch_operations(self) -> None:
|
|
"""
|
|
TDD Red阶段: 测试缓存批量操作
|
|
|
|
预期结果:
|
|
- 支持批量put
|
|
- 支持批量get
|
|
- 支持批量delete
|
|
"""
|
|
from core.caffeine_cache import CaffeineCache
|
|
|
|
with allure.step("Step 1: 创建缓存实例"):
|
|
cache = CaffeineCache()
|
|
allure.attach("✅ 缓存实例创建成功", "步骤1", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 2: 批量添加数据"):
|
|
data = {
|
|
"batch_key1": "batch_value1",
|
|
"batch_key2": "batch_value2",
|
|
"batch_key3": "batch_value3",
|
|
}
|
|
cache.put_all(data)
|
|
allure.attach(f"批量添加: {data}", "步骤2", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 3: 批量获取数据"):
|
|
keys = ["batch_key1", "batch_key2", "batch_key3"]
|
|
values = cache.get_all(keys)
|
|
allure.attach(f"批量获取: {values}", "步骤3", allure.attachment_type.TEXT)
|
|
assert len(values) == 3, f"批量获取结果数量错误，期望: 3, 实际: {len(values)}"
|
|
|
|
with allure.step("Step 4: 批量删除数据"):
|
|
cache.delete_all(keys)
|
|
allure.attach("✅ 批量删除完成", "步骤4", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 5: 验证数据已删除"):
|
|
values = cache.get_all(keys)
|
|
allure.attach(f"删除后批量获取: {values}", "步骤5", allure.attachment_type.TEXT)
|
|
assert all(v is None for v in values.values()), "所有值应该为None"
|
|
|
|
@allure.title("测试缓存清空 - TDD Red阶段")
|
|
@allure.description("验证缓存清空功能 - 期望失败(Red)")
|
|
@allure.severity(allure.severity_level.NORMAL)
|
|
@pytest.mark.regression
|
|
def test_cache_clear(self) -> None:
|
|
"""
|
|
TDD Red阶段: 测试缓存清空
|
|
|
|
预期结果:
|
|
- 清空后所有数据被移除
|
|
- 缓存大小归零
|
|
"""
|
|
from core.caffeine_cache import CaffeineCache
|
|
|
|
with allure.step("Step 1: 创建缓存实例并添加数据"):
|
|
cache = CaffeineCache()
|
|
cache.put("key1", "value1")
|
|
cache.put("key2", "value2")
|
|
cache.put("key3", "value3")
|
|
allure.attach("✅ 添加3个缓存项", "步骤1", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 2: 清空缓存"):
|
|
cache.clear()
|
|
allure.attach("✅ 缓存已清空", "步骤2", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 3: 验证缓存为空"):
|
|
value1 = cache.get("key1")
|
|
value2 = cache.get("key2")
|
|
value3 = cache.get("key3")
|
|
allure.attach(f"key1: {value1}, key2: {value2}, key3: {value3}", "步骤3", allure.attachment_type.TEXT)
|
|
assert value1 is None, "key1应该为None"
|
|
assert value2 is None, "key2应该为None"
|
|
assert value3 is None, "key3应该为None"
|
|
|
|
@allure.title("测试缓存线程安全 - TDD Red阶段")
|
|
@allure.description("验证缓存在多线程环境下的安全性 - 期望失败(Red)")
|
|
@allure.severity(allure.severity_level.CRITICAL)
|
|
@pytest.mark.smoke
|
|
def test_cache_thread_safety(self) -> None:
|
|
"""
|
|
TDD Red阶段: 测试缓存线程安全
|
|
|
|
预期结果:
|
|
- 多线程并发操作不会导致数据不一致
|
|
- 不会出现竞态条件
|
|
"""
|
|
from core.caffeine_cache import CaffeineCache
|
|
import threading
|
|
|
|
with allure.step("Step 1: 创建缓存实例"):
|
|
cache = CaffeineCache()
|
|
allure.attach("✅ 缓存实例创建成功", "步骤1", allure.attachment_type.TEXT)
|
|
|
|
with allure.step("Step 2: 多线程并发写入"):
|
|
errors = []
|
|
|
|
def write_data(thread_id: int):
|
|
try:
|
|
for i in range(10):
|
|
cache.put(f"thread_{thread_id}_key_{i}", f"value_{i}")
|
|
except Exception as e:
|
|
errors.append(str(e))
|
|
|
|
threads = []
|
|
for i in range(5):
|
|
t = threading.Thread(target=write_data, args=(i,))
|
|
threads.append(t)
|
|
t.start()
|
|
|
|
for t in threads:
|
|
t.join()
|
|
|
|
allure.attach(f"线程数: 5, 每线程写入: 10条", "步骤2", allure.attachment_type.TEXT)
|
|
assert len(errors) == 0, f"并发写入出现错误: {errors}"
|
|
|
|
with allure.step("Step 3: 验证数据完整性"):
|
|
total_keys = 5 * 10 # 5个线程，每线程10条
|
|
count = 0
|
|
for thread_id in range(5):
|
|
for i in range(10):
|
|
value = cache.get(f"thread_{thread_id}_key_{i}")
|
|
if value is not None:
|
|
count += 1
|
|
|
|
allure.attach(f"成功写入: {count}/{total_keys} 条", "步骤3", allure.attachment_type.TEXT)
|
|
assert count == total_keys, f"数据不完整，期望: {total_keys}, 实际: {count}"
|