test_cache_optimization_new.py

# -*- coding: utf-8 -*-
"""
分词缓存优化测试脚本
验证 TokenizeCache 和优化后的 InvertedIndex 功能
"""

import sys
import os
import time

# 添加 src 目录到路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))

from inverted_index import InvertedIndex, TokenizeCache


def test_tokenize_cache():
    """测试 TokenizeCache 基本功能"""
    print("=== 测试 TokenizeCache 基本功能 ===")

    cache = TokenizeCache(max_size=5, ttl_seconds=2)

    # 测试缓存未命中
    result = cache.get("测试文本")
    assert result is None, "缓存应该未命中"
    print("✓ 缓存未命中测试通过")

    # 测试缓存存储和命中
    tokens = ["测试", "文本"]
    cache.put("测试文本", tokens)
    result = cache.get("测试文本")
    assert result == tokens, "缓存应该命中"
    print("✓ 缓存存储和命中测试通过")

    # 测试统计信息
    stats = cache.get_stats()
    assert stats['hits'] == 1, "命中次数应该为1"
    assert stats['misses'] == 1, "未命中次数应该为1"
    print("✓ 缓存统计信息测试通过")

    # 测试 TTL 过期
    print("等待缓存过期...")
    time.sleep(3)
    result = cache.get("测试文本")
    assert result is None, "缓存应该已过期"
    print("✓ TTL 过期测试通过")

    print("TokenizeCache 基本功能测试完成\n")


def test_inverted_index_with_cache():
    """测试集成缓存的 InvertedIndex"""
    print("=== 测试集成缓存的 InvertedIndex ===")

    # 测试数据
    test_mapping = {
        "前端开发规范": "React组件开发规范，包括组件命名、状态管理等",
        "后端API设计": "RESTful API设计规范，包括接口命名、错误处理等",
        "数据库设计": "数据库表设计规范，包括字段命名、索引优化等"
    }

    # 创建索引实例
    index = InvertedIndex(test_mapping, cache_size=10, cache_ttl=60)
    print("✓ InvertedIndex 初始化成功")

    # 测试分词功能（第一次调用，缓存未命中）
    tokens1 = index.tokenize("React组件开发")
    print(f"第一次分词结果: {tokens1}")

    # 测试分词功能（第二次调用，缓存命中）
    start_time = time.time()
    tokens2 = index.tokenize("React组件开发")
    end_time = time.time()

    assert tokens1 == tokens2, "两次分词结果应该相同"
    print(f"第二次分词结果: {tokens2}")
    print(f"第二次分词耗时: {(end_time - start_time) * 1000:.2f}ms")
    print("✓ 分词缓存功能测试通过")

    # 测试搜索功能
    search_results = index.search("React组件", max_results=2)
    assert len(search_results) > 0, "搜索应该有结果"
    print(f"搜索结果: {search_results}")
    print("✓ 搜索功能测试通过")

    # 测试缓存统计
    cache_stats = index.get_cache_statistics()
    print(f"缓存统计: {cache_stats}")
    assert cache_stats['hits'] >= 1, "应该有缓存命中"
    print("✓ 缓存统计功能测试通过")

    # 测试综合统计
    comprehensive_stats = index.get_comprehensive_statistics()
    print(f"综合统计: {comprehensive_stats}")
    assert 'index_statistics' in comprehensive_stats, "应该包含索引统计"
    assert 'cache_statistics' in comprehensive_stats, "应该包含缓存统计"
    assert 'performance_metrics' in comprehensive_stats, "应该包含性能指标"
    print("✓ 综合统计功能测试通过")

    print("InvertedIndex 缓存集成测试完成\n")


def test_performance_comparison():
    """性能对比测试"""
    print("=== 性能对比测试 ===")

    test_mapping = {
        f"规范{i}": f"这是第{i}个前端开发规范，包含组件设计、状态管理、样式规范等内容"
        for i in range(100)
    }

    index = InvertedIndex(test_mapping, cache_size=50, cache_ttl=300)

    # 测试文本列表
    test_texts = [
        "前端开发规范",
        "组件设计模式",
        "状态管理方案",
        "样式规范指南",
        "前端开发规范",  # 重复文本，测试缓存效果
        "组件设计模式",  # 重复文本
    ]

    # 第一轮：缓存未命中
    start_time = time.time()
    for text in test_texts:
        tokens = index.tokenize(text)
    first_round_time = time.time() - start_time

    # 第二轮：部分缓存命中
    start_time = time.time()
    for text in test_texts:
        tokens = index.tokenize(text)
    second_round_time = time.time() - start_time

    print(f"第一轮分词耗时: {first_round_time * 1000:.2f}ms")
    print(f"第二轮分词耗时: {second_round_time * 1000:.2f}ms")
    print(f"性能提升: {((first_round_time - second_round_time) / first_round_time * 100):.1f}%")

    # 显示缓存统计
    cache_stats = index.get_cache_statistics()
    print(f"最终缓存统计: {cache_stats}")

    print("性能对比测试完成\n")


def main():
    """主测试函数"""
    print("开始分词缓存优化测试...\n")

    try:
        test_tokenize_cache()
        test_inverted_index_with_cache()
        test_performance_comparison()

        print("🎉 所有测试通过！分词缓存优化实现成功！")

    except Exception as e:
        print(f"❌ 测试失败: {e}")
        import traceback
        traceback.print_exc()
        return False

    return True


if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)