""" Integration tests for OrganizeCommand. Tests Commands with full skill chains to verify end-to-end functionality. """ import pytest import json from unittest.mock import AsyncMock, patch, Mock from aioresponses import aioresponses from agent_core import CommandContext, SkillResult, Agent from commands.organize_command import OrganizeCommand class TestOrganizeCommand: """Integration tests for OrganizeCommand""" @pytest.fixture def command(self): """Create OrganizeCommand instance""" return OrganizeCommand() @pytest.fixture def context(self, sample_config): """Create command context for organize command""" return CommandContext( command_name="organize", args={ "date": "2024-01-15", "vault_path": sample_config["obsidian"]["vault_path"], "daily_folder": "Daily" }, config=sample_config ) @pytest.fixture def sample_journal_content(self): """Sample journal content for testing""" return """# 2024-01-15 Daily Journal ## 今天的经历 - 完成了项目的重要里程碑，团队协作非常顺利 - 与客户进行了产品演示，获得了积极反馈 - 参加了技术分享会议，学到了新的架构模式 ## 学到的东西 - 学会了新的Python异步编程技巧，提升了代码效率 - 理解了微服务架构的最佳实践 - 掌握了更好的错误处理和日志记录模式 ## 遇到的问题 - API调用偶尔超时，影响用户体验 - 配置文件格式需要改进，当前格式不够灵活 - 团队沟通中存在信息不对称问题 ## 今天的成就 - 成功部署了新版本到生产环境 - 解决了困扰团队一周的性能问题 - 获得了客户的正面评价 ## 明天的计划 - 优化API调用的重试机制 - 重构配置管理模块 - 组织团队同步会议 """ @pytest.mark.asyncio async def test_organize_command_full_workflow_success(self, command, context, sample_journal_content): """Test complete organize command workflow with all skills""" # Mock Claude API responses analyze_response = { "experiences": [ { "title": "项目里程碑完成", "description": "成功完成了项目的重要里程碑，团队协作非常顺利", "category": "项目管理", "importance": "high" }, { "title": "客户产品演示", "description": "与客户进行了产品演示，获得了积极反馈", "category": "客户关系", "importance": "high" } ], "lessons": [ { "title": "Python异步编程技巧", "description": "学会了新的Python异步编程技巧，提升了代码效率", "category": "技术学习", "application": "可以应用到当前项目的API调用优化中" }, { "title": "微服务架构最佳实践", "description": "理解了微服务架构的最佳实践", "category": "架构设计", "application": "用于指导下一个项目的架构设计" } ], "problems": [ { "title": "API调用超时", "description": "API调用偶尔超时，影响用户体验", "severity": "medium", "suggested_solution": "实现重试机制和超时处理" } ], "achievements": [ { "title": "生产环境部署", "description": "成功部署了新版本到生产环境", "impact": "提升了系统稳定性和性能" } ] } # Mock transformation responses experience_note = """# 项目里程碑完成 ## 经验描述 成功完成了项目的重要里程碑，团队协作非常顺利。这次经历展现了良好的项目管理能力和团队协作精神。 ## 关键要点 - 项目管理技能得到提升 - 团队协作效率显著改善 - 里程碑按时完成 ## 应用场景 这个经验可以应用到未来的项目管理中，特别是在设定和跟踪项目里程碑方面。 ## 相关标签 #项目管理 #里程碑 #团队协作 --- *创建时间: 2024-01-15* *来源: [[Daily/2024-01-15]]* """ lesson_note = """# Python异步编程技巧 ## 学习内容 学会了新的Python异步编程技巧，提升了代码效率。 ## 关键概念 - 异步编程模式 - 性能优化技巧 - 代码效率提升 ## 实际应用 可以应用到当前项目的API调用优化中，提升系统响应速度。 ## 相关标签 #技术学习 #Python #异步编程 --- *创建时间: 2024-01-15* *来源: [[Daily/2024-01-15]]* """ with aioresponses() as m: # Mock Obsidian API calls # 1. Read daily journal m.get( "https://localhost:27123/vault/Daily/2024-01-15.md", payload={ "content": sample_journal_content, "stat": {"ctime": 1642204800000, "mtime": 1642204800000, "size": len(sample_journal_content)} }, status=200 ) # 2. Write experience note m.put( "https://localhost:27123/vault/Knowledge/Experiences/项目里程碑完成.md", payload={ "path": "Knowledge/Experiences/项目里程碑完成.md", "stat": {"ctime": 1642204900000, "mtime": 1642204900000, "size": len(experience_note)} }, status=200 ) # 3. Write lesson note m.put( "https://localhost:27123/vault/Knowledge/Lessons/Python异步编程技巧.md", payload={ "path": "Knowledge/Lessons/Python异步编程技巧.md", "stat": {"ctime": 1642204900000, "mtime": 1642204900000, "size": len(lesson_note)} }, status=200 ) # 4. Additional notes for other categories (problems, achievements) m.put( "https://localhost:27123/vault/Knowledge/Problems/API调用超时.md", payload={"path": "Knowledge/Problems/API调用超时.md", "stat": {}}, status=200 ) m.put( "https://localhost:27123/vault/Knowledge/Achievements/生产环境部署.md", payload={"path": "Knowledge/Achievements/生产环境部署.md", "stat": {}}, status=200 ) # Mock Claude API with patch('skills.claude_skill.anthropic') as mock_anthropic: mock_client = Mock() mock_anthropic.Anthropic.return_value = mock_client # Mock analyze response mock_analyze_message = Mock() mock_analyze_message.content = [Mock(text=json.dumps(analyze_response, ensure_ascii=False))] # Mock transform responses (one for each item to be transformed) mock_transform_responses = [ Mock(content=[Mock(text=experience_note)]), Mock(content=[Mock(text=experience_note)]), # Second experience Mock(content=[Mock(text=lesson_note)]), Mock(content=[Mock(text=lesson_note)]), # Second lesson Mock(content=[Mock(text="# API调用超时\n\n问题描述...")]), # Problem note Mock(content=[Mock(text="# 生产环境部署\n\n成就描述...")]) # Achievement note ] # Set up responses: first analyze, then multiple transforms mock_client.messages.create.side_effect = [mock_analyze_message] + mock_transform_responses # Execute the organize command result = await command.execute(context) # Verify overall success assert result.success is True assert "organized successfully" in result.message.lower() or "completed" in result.message.lower() # Verify data structure assert "summary" in result.data assert "created_notes" in result.data # Verify created notes created_notes = result.data["created_notes"] assert len(created_notes) > 0 # Should have created notes for experiences, lessons, problems, achievements note_types = [note.get("type") for note in created_notes] expected_types = ["experience", "lesson", "problem", "achievement"] for expected_type in expected_types: assert any(expected_type in note_type for note_type in note_types if note_type) # Verify API calls were made assert mock_client.messages.create.call_count >= 2 # At least analyze + some transforms @pytest.mark.asyncio async def test_organize_command_journal_not_found(self, command, context): """Test organize command when daily journal doesn't exist""" with aioresponses() as m: # Mock journal file not found m.get( "https://localhost:27123/vault/Daily/2024-01-15.md", status=404, payload={"error": "File not found"} ) result = await command.execute(context) assert result.success is False assert "not found" in result.error.lower() or "missing" in result.error.lower() @pytest.mark.asyncio async def test_organize_command_claude_api_error(self, command, context, sample_journal_content): """Test organize command when Claude API fails""" with aioresponses() as m: # Mock successful journal read m.get( "https://localhost:27123/vault/Daily/2024-01-15.md", payload={ "content": sample_journal_content, "stat": {"ctime": 1642204800000, "mtime": 1642204800000, "size": len(sample_journal_content)} }, status=200 ) # Mock Claude API error with patch('skills.claude_skill.anthropic') as mock_anthropic: mock_client = Mock() mock_anthropic.Anthropic.return_value = mock_client mock_client.messages.create.side_effect = Exception("Claude API rate limit exceeded") result = await command.execute(context) assert result.success is False assert "api" in result.error.lower() or "claude" in result.error.lower() @pytest.mark.asyncio async def test_organize_command_partial_success(self, command, context, sample_journal_content): """Test organize command with partial success (some notes created, some failed)""" analyze_response = { "experiences": [ { "title": "Test Experience", "description": "Test description", "category": "Test", "importance": "medium" } ], "lessons": [], "problems": [], "achievements": [] } with aioresponses() as m: # Mock successful journal read m.get( "https://localhost:27123/vault/Daily/2024-01-15.md", payload={ "content": sample_journal_content, "stat": {"ctime": 1642204800000, "mtime": 1642204800000, "size": len(sample_journal_content)} }, status=200 ) # Mock successful experience note creation m.put( "https://localhost:27123/vault/Knowledge/Experiences/Test Experience.md", payload={ "path": "Knowledge/Experiences/Test Experience.md", "stat": {"ctime": 1642204900000, "mtime": 1642204900000, "size": 100} }, status=200 ) # Mock Claude API with patch('skills.claude_skill.anthropic') as mock_anthropic: mock_client = Mock() mock_anthropic.Anthropic.return_value = mock_client # Mock analyze response mock_analyze_message = Mock() mock_analyze_message.content = [Mock(text=json.dumps(analyze_response, ensure_ascii=False))] # Mock transform response mock_transform_message = Mock() mock_transform_message.content = [Mock(text="# Test Experience\n\nTransformed content")] mock_client.messages.create.side_effect = [mock_analyze_message, mock_transform_message] result = await command.execute(context) # Should succeed even with minimal content assert result.success is True assert len(result.data["created_notes"]) >= 1 @pytest.mark.asyncio async def test_organize_command_empty_journal(self, command, context): """Test organize command with empty journal content""" with aioresponses() as m: # Mock journal with empty content m.get( "https://localhost:27123/vault/Daily/2024-01-15.md", payload={ "content": "", "stat": {"ctime": 1642204800000, "mtime": 1642204800000, "size": 0} }, status=200 ) result = await command.execute(context) assert result.success is False assert "empty" in result.error.lower() or "content" in result.error.lower() class TestOrganizeCommandWithAgent: """Integration tests for OrganizeCommand within Agent context""" @pytest.fixture def agent(self, sample_config): """Create Agent with OrganizeCommand registered""" agent = Agent("test_agent", sample_config) agent.register_command(OrganizeCommand()) return agent @pytest.mark.asyncio async def test_agent_execute_organize_command(self, agent, sample_journal_content): """Test executing organize command through Agent""" with aioresponses() as m: # Mock Obsidian API m.get( "https://localhost:27123/vault/Daily/2024-01-15.md", payload={ "content": sample_journal_content, "stat": {"ctime": 1642204800000, "mtime": 1642204800000, "size": len(sample_journal_content)} }, status=200 ) # Mock note creation (simplified - just one note) m.put( "https://localhost:27123/vault/Knowledge/Experiences/Test.md", payload={"path": "Knowledge/Experiences/Test.md", "stat": {}}, status=200 ) # Mock Claude API with patch('skills.claude_skill.anthropic') as mock_anthropic: mock_client = Mock() mock_anthropic.Anthropic.return_value = mock_client # Minimal response for testing analyze_response = { "experiences": [{"title": "Test", "description": "Test", "category": "Test"}], "lessons": [], "problems": [], "achievements": [] } mock_analyze_message = Mock() mock_analyze_message.content = [Mock(text=json.dumps(analyze_response, ensure_ascii=False))] mock_transform_message = Mock() mock_transform_message.content = [Mock(text="# Test\n\nTest content")] mock_client.messages.create.side_effect = [mock_analyze_message, mock_transform_message] # Execute command through agent result = await agent.execute_command( "organize", args={"date": "2024-01-15"}, options={"verbose": True} ) assert result.success is True assert result.data is not None @pytest.mark.asyncio async def test_agent_execute_organize_by_alias(self, agent, sample_journal_content): """Test executing organize command by alias through Agent""" with aioresponses() as m: # Mock minimal successful workflow m.get( "https://localhost:27123/vault/Daily/2024-01-15.md", payload={ "content": sample_journal_content, "stat": {"ctime": 1642204800000, "mtime": 1642204800000, "size": len(sample_journal_content)} }, status=200 ) # Mock Claude API with minimal response with patch('skills.claude_skill.anthropic') as mock_anthropic: mock_client = Mock() mock_anthropic.Anthropic.return_value = mock_client analyze_response = {"experiences": [], "lessons": [], "problems": [], "achievements": []} mock_message = Mock() mock_message.content = [Mock(text=json.dumps(analyze_response))] mock_client.messages.create.return_value = mock_message # Execute by alias result = await agent.execute_command("org", args={"date": "2024-01-15"}) assert result.success is True @pytest.mark.asyncio async def test_agent_command_info(self, agent): """Test getting command information through Agent""" commands_info = agent.get_commands_info() assert "organize" in commands_info["commands"] organize_info = commands_info["commands"]["organize"] assert organize_info["name"] == "organize" assert organize_info["description"] == "分析和整理日记内容，提取经验和要点" assert "org" in organize_info["aliases"] assert "organize-journal" in organize_info["aliases"] # Verify skills are registered assert len(organize_info["skills"]) > 0 skill_names = list(organize_info["skills"].keys()) expected_skills = ["obsidian_read", "obsidian_write", "obsidian_append", "claude_analyze", "claude_transform"] for expected_skill in expected_skills: assert any(expected_skill in skill_name for skill_name in skill_names)