test_memory_driven_evolution.py

"""
Test Memory-Driven Evolution System

Tests for the memory-driven evolution system that generates
evidence-based evolution options from memory analysis.
"""

import sys
import tempfile
import shutil
from pathlib import Path
from datetime import datetime
from evolution.core.memory_driven_evolution import (
    EvolutionOption, MemoryAnalysisResult, EvolutionExecutionResult,
    MemoryDrivenEvolution, EvidenceSource
)
from evolution.core.three_layer_memory import (
    SoulLayerMemory, LongTermMemory, LogEntry, MemorySnapshot
)


def test_memory_analysis():
    """测试记忆分析"""
    test_dir = tempfile.mkdtemp(prefix="memory_analysis_test_")
    try:
        # 创建基础结构 - 使用 parents=True
        (Path(test_dir) / "soul").mkdir(exist_ok=True)
        (Path(test_dir) / "long_term" / "user_preferences").mkdir(parents=True, exist_ok=True)
        (Path(test_dir) / "logs" / "daily").mkdir(parents=True, exist_ok=True)

        # 添加一些测试数据
        (Path(test_dir) / "soul" / "agent_identity.md").write_text(
            "测试 AI 系统",
            encoding='utf-8'
        )

        mde = MemoryDrivenEvolution(test_dir)
        analysis = mde.analyze_memory_state()

        assert isinstance(analysis, MemoryAnalysisResult)
        assert isinstance(analysis.soul_issues, dict)
        assert isinstance(analysis.long_term_issues, dict)
        assert isinstance(analysis.log_issues, dict)
        assert isinstance(analysis.cross_layer_issues, dict)
        assert isinstance(analysis.snapshot, MemorySnapshot)
        assert 0.0 <= analysis.overall_health <= 1.0

        # 验证 soul_issues 的结构
        assert "health_score" in analysis.soul_issues
        assert "issues" in analysis.soul_issues
        assert 0.0 <= analysis.soul_issues["health_score"] <= 1.0

        print("✅ test_memory_analysis passed")
        return True
    except Exception as e:
        print(f"❌ test_memory_analysis failed: {e}")
        import traceback
        traceback.print_exc()
        return False
    finally:
        shutil.rmtree(test_dir)


def test_evolution_option_generation():
    """测试进化选项生成"""
    test_dir = tempfile.mkdtemp(prefix="option_gen_test_")
    try:
        # 创建基础结构
        (Path(test_dir) / "soul").mkdir(exist_ok=True)
        (Path(test_dir) / "long_term" / "user_preferences").mkdir(parents=True, exist_ok=True)
        (Path(test_dir) / "logs" / "daily").mkdir(parents=True, exist_ok=True)

        mde = MemoryDrivenEvolution(test_dir)
        options = mde.generate_evolution_options()

        assert isinstance(options, list)
        # 验证选项格式
        for option in options:
            assert isinstance(option, EvolutionOption)
            assert option.option_id
            assert option.type in [
                "soul_update", "memory_consolidation", "pattern_transfer",
                "optimization", "bugfix", "consistency_fix"
            ]
            assert 0.0 <= option.priority <= 1.0
            assert 0.0 <= option.confidence <= 1.0
            assert option.estimated_impact in ["low", "medium", "high"]
            assert isinstance(option.actions, list)
            assert isinstance(option.evidence_sources, list)
            assert option.reasoning  # 必须有推理过程

            # 验证证据来源
            for evidence in option.evidence_sources:
                assert isinstance(evidence, EvidenceSource)
                assert evidence.source_type
                assert evidence.location
                assert evidence.excerpt
                assert 0.0 <= evidence.confidence <= 1.0

        # 验证按优先级排序
        if len(options) > 1:
            for i in range(len(options) - 1):
                assert options[i].priority >= options[i + 1].priority

        print(f"✅ test_evolution_option_generation passed ({len(options)} options)")
        return True
    except Exception as e:
        print(f"❌ test_evolution_option_generation failed: {e}")
        import traceback
        traceback.print_exc()
        return False
    finally:
        shutil.rmtree(test_dir)


def test_consolidation_execution():
    """测试记忆整合执行"""
    test_dir = tempfile.mkdtemp(prefix="consolidation_test_")
    try:
        # 创建基础结构
        (Path(test_dir) / "soul").mkdir(exist_ok=True)
        (Path(test_dir) / "long_term" / "user_preferences").mkdir(parents=True, exist_ok=True)
        (Path(test_dir) / "logs" / "daily").mkdir(parents=True, exist_ok=True)

        mde = MemoryDrivenEvolution(test_dir)

        # 创建一个整合选项
        option = EvolutionOption(
            option_id="test_consolidation",
            type="memory_consolidation",
            source_layer="long_term",
            title="测试整合",
            description="测试记忆整合功能",
            priority=0.8,
            confidence=0.9,
            estimated_impact="medium",
            actions=["测试动作"],
            rollback_strategy="test"
        )

        result = mde.execute_evolution_option(option)

        assert isinstance(result, EvolutionExecutionResult)
        assert result.success is True
        assert result.option == option
        assert isinstance(result.changes_made, list)
        assert result.errors == []

        print("✅ test_consolidation_execution passed")
        return True
    except Exception as e:
        print(f"❌ test_consolidation_execution failed: {e}")
        import traceback
        traceback.print_exc()
        return False
    finally:
        shutil.rmtree(test_dir)


def test_soul_health_calculation():
    """测试人格层健康度计算"""
    test_dir = tempfile.mkdtemp(prefix="soul_health_test_")
    try:
        # 创建完整人格层 - 确保内容足够长（至少200字）
        soul_dir = Path(test_dir) / "soul"
        soul_dir.mkdir(exist_ok=True)

        long_identity = "测试身份描述" * 10
        long_work_style = "测试工作方式描述" * 10
        long_communication = "测试沟通风格描述" * 10

        for file, content in [
            ("agent_identity.md", long_identity),
            ("behavior_rules.md", long_work_style),
            ("communication_style.md", long_communication),
            ("constraints.md", "- 约束1\n- 约束2\n- 约束3\n- 约束4\n- 约束5\n" * 10)
        ]:
            (soul_dir / file).write_text(content, encoding='utf-8')

        (Path(test_dir) / "long_term" / "user_preferences").mkdir(parents=True, exist_ok=True)
        (Path(test_dir) / "logs" / "daily").mkdir(parents=True, exist_ok=True)

        mde = MemoryDrivenEvolution(test_dir)
        analysis = mde.analyze_memory_state()

        # 完整人格应该有高分（但需要考虑大小惩罚）
        print(f"Debug: health_score={analysis.soul_issues['health_score']:.2f}, total_size={analysis.snapshot.soul.total_size}")
        # 由于大小限制惩罚，健康度可能不是0.9，我们检查合理范围
        assert analysis.soul_issues["health_score"] >= 0.8
        # 由于文件大小可能超出5000字限制，会有一些问题
        # 我们检查问题数量不超过阈值
        assert len(analysis.soul_issues["issues"]) <= 2

        print("✅ test_soul_health_calculation passed")
        return True
    except Exception as e:
        print(f"❌ test_soul_health_calculation failed: {e}")
        import traceback
        traceback.print_exc()
        return False
    finally:
        shutil.rmtree(test_dir)


def test_evidence_source():
    """测试证据来源数据结构"""
    try:
        evidence = EvidenceSource(
            source_type="log_file",
            location=".evolution/logs/evolution.log",
            timestamp=datetime.now(),
            excerpt="测试摘录内容",
            line_range=(10, 20),
            confidence=0.95
        )

        assert evidence.source_type == "log_file"
        assert evidence.location.endswith(".log")
        assert evidence.excerpt == "测试摘录内容"
        assert evidence.line_range == (10, 20)
        assert evidence.confidence == 0.95

        # 测试不带行号的证据
        evidence_no_line = EvidenceSource(
            source_type="claude_mem",
            location="memory_id:123",
            timestamp=None,
            excerpt="记忆摘录",
            confidence=0.8
        )

        assert evidence_no_line.line_range is None

        print("✅ test_evidence_source passed")
        return True
    except Exception as e:
        print(f"❌ test_evidence_source failed: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_evolution_option_with_evidence():
    """测试带证据的进化选项"""
    try:
        evidence1 = EvidenceSource(
            source_type="log_file",
            location=".evolution/logs/test.log",
            timestamp=datetime.now(),
            excerpt="发现错误模式",
            confidence=0.9
        )

        evidence2 = EvidenceSource(
            source_type="test_file",
            location="tests/test.py",
            timestamp=None,
            excerpt="测试失败记录",
            confidence=0.85
        )

        option = EvolutionOption(
            option_id="fix_error_123",
            type="bugfix",
            source_layer="log",
            title="修复重复错误模式",
            description="检测到5个错误事件需要处理",
            priority=0.9,
            confidence=0.85,
            estimated_impact="high",
            actions=["分析错误根因", "修复错误模式", "增加错误处理"],
            rollback_strategy="恢复修复前的代码",
            evidence_sources=[evidence1, evidence2],
            reasoning="基于5个真实错误事件分析和测试失败记录"
        )

        assert len(option.evidence_sources) == 2
        assert option.reasoning
        assert "5个真实错误事件" in option.reasoning

        print("✅ test_evolution_option_with_evidence passed")
        return True
    except Exception as e:
        print(f"❌ test_evolution_option_with_evidence failed: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_get_evolution_options_interface():
    """测试对外接口"""
    test_dir = tempfile.mkdtemp(prefix="interface_test_")
    try:
        # 创建基础结构
        (Path(test_dir) / "soul").mkdir(exist_ok=True)
        (Path(test_dir) / "long_term" / "user_preferences").mkdir(parents=True, exist_ok=True)
        (Path(test_dir) / "logs" / "daily").mkdir(parents=True, exist_ok=True)

        mde = MemoryDrivenEvolution(test_dir)
        options = mde.get_evolution_options()

        assert isinstance(options, list)
        # 验证是 EvolutionOption 对象
        for option in options:
            assert isinstance(option, EvolutionOption)

        print("✅ test_get_evolution_options_interface passed")
        return True
    except Exception as e:
        print(f"❌ test_get_evolution_options_interface failed: {e}")
        import traceback
        traceback.print_exc()
        return False
    finally:
        shutil.rmtree(test_dir)


def test_prioritize_options():
    """测试选项优先级排序"""
    try:
        options = [
            EvolutionOption(
                option_id="opt1", type="test", source_layer="test",
                title="选项1", description="测试", priority=0.5,
                confidence=0.8, estimated_impact="low"
            ),
            EvolutionOption(
                option_id="opt2", type="test", source_layer="test",
                title="选项2", description="测试", priority=0.9,
                confidence=0.8, estimated_impact="high"
            ),
            EvolutionOption(
                option_id="opt3", type="test", source_layer="test",
                title="选项3", description="测试", priority=0.7,
                confidence=0.8, estimated_impact="medium"
            )
        ]

        mde = MemoryDrivenEvolution()
        prioritized = mde.prioritize_options(options)

        # 验证按优先级降序排列
        assert prioritized[0].priority >= prioritized[1].priority
        assert prioritized[1].priority >= prioritized[2].priority
        assert prioritized[0].option_id == "opt2"

        print("✅ test_prioritize_options passed")
        return True
    except Exception as e:
        print(f"❌ test_prioritize_options failed: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_transfer_execution():
    """测试模式迁移执行"""
    test_dir = tempfile.mkdtemp(prefix="transfer_test_")
    try:
        # 创建基础结构
        (Path(test_dir) / "soul").mkdir(exist_ok=True)
        (Path(test_dir) / "long_term" / "user_preferences").mkdir(parents=True, exist_ok=True)
        (Path(test_dir) / "logs" / "daily").mkdir(parents=True, exist_ok=True)

        mde = MemoryDrivenEvolution(test_dir)

        # 创建迁移选项
        option = EvolutionOption(
            option_id="test_transfer",
            type="pattern_transfer",
            source_layer="long_term",
            title="测试模式迁移",
            description="测试模式迁移功能",
            priority=0.7,
            confidence=0.75,
            estimated_impact="high",
            actions=["评估迁移适用性", "执行模式迁移", "验证迁移结果"],
            rollback_strategy="移除迁移的模式"
        )

        result = mde.execute_evolution_option(option)

        assert isinstance(result, EvolutionExecutionResult)
        assert result.success is True

        print("✅ test_transfer_execution passed")
        return True
    except Exception as e:
        print(f"❌ test_transfer_execution failed: {e}")
        import traceback
        traceback.print_exc()
        return False
    finally:
        shutil.rmtree(test_dir)


def test_soul_update_execution():
    """测试人格层更新执行"""
    test_dir = tempfile.mkdtemp(prefix="soul_update_test_")
    try:
        # 创建基础结构
        (Path(test_dir) / "soul").mkdir(exist_ok=True)
        (Path(test_dir) / "long_term" / "user_preferences").mkdir(parents=True, exist_ok=True)
        (Path(test_dir) / "logs" / "daily").mkdir(parents=True, exist_ok=True)

        mde = MemoryDrivenEvolution(test_dir)

        # 创建人格更新选项
        option = EvolutionOption(
            option_id="test_soul_update",
            type="soul_update",
            source_layer="soul",
            title="更新人格层定义",
            description="修复人格层问题",
            priority=0.9,
            confidence=0.8,
            estimated_impact="medium",
            actions=["审查并更新人格层文件", "确保大小在范围内", "补充缺失定义"],
            rollback_strategy="恢复到修改前的文件"
        )

        result = mde.execute_evolution_option(option)

        assert isinstance(result, EvolutionExecutionResult)
        assert result.success is True

        print("✅ test_soul_update_execution passed")
        return True
    except Exception as e:
        print(f"❌ test_soul_update_execution failed: {e}")
        import traceback
        traceback.print_exc()
        return False
    finally:
        shutil.rmtree(test_dir)


def run_all_tests():
    print("=" * 70)
    print("🧪 Memory-Driven Evolution Tests")
    print("=" * 70)
    print()

    tests = [
        test_memory_analysis,
        test_evolution_option_generation,
        test_consolidation_execution,
        test_soul_health_calculation,
        test_evidence_source,
        test_evolution_option_with_evidence,
        test_get_evolution_options_interface,
        test_prioritize_options,
        test_transfer_execution,
        test_soul_update_execution
    ]

    passed = 0
    for test in tests:
        if test():
            passed += 1

    print()
    print(f"Results: {passed}/{len(tests)} tests passed")

    if passed == len(tests):
        print("\n🎉 All tests passed!")
        return True
    else:
        print(f"\n⚠️  {len(tests) - passed} test(s) failed.")
        return False


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