"""Test Tool Discovery Engine for MVP 1 (TDD approach).""" from unittest.mock import patch import numpy as np import pytest # These imports will fail initially - that's the TDD approach try: from kg_services.tool_discovery import ( EmbeddingService, KnowledgeGraphService, MCPTool, ToolDiscoveryEngine, ToolMatch, ToolSearchCriteria, ) except ImportError: # Expected to fail initially in TDD ToolDiscoveryEngine = None MCPTool = None ToolSearchCriteria = None ToolMatch = None EmbeddingService = None KnowledgeGraphService = None @pytest.mark.unit class TestMCPTool: """Test MCP Tool data model.""" def test_mcp_tool_exists(self) -> None: """Test that MCPTool can be imported.""" assert MCPTool is not None, "MCPTool should be importable" def test_mcp_tool_creation(self) -> None: """Test MCPTool creation with required fields.""" if MCPTool is None: pytest.skip("MCPTool not implemented yet") tool = MCPTool( id="summarizer", name="Text Summarizer", description="Summarizes long text into concise key points", category="text_processing", capabilities=["summarization", "text_analysis"], input_types=["text"], output_types=["text"], ) assert tool.id == "summarizer" assert tool.name == "Text Summarizer" assert tool.description == "Summarizes long text into concise key points" assert tool.category == "text_processing" assert "summarization" in tool.capabilities assert "text" in tool.input_types assert "text" in tool.output_types def test_mcp_tool_validation(self) -> None: """Test MCPTool field validation.""" if MCPTool is None: pytest.skip("MCPTool not implemented yet") # Should require name and description with pytest.raises(Exception): # Pydantic ValidationError MCPTool(id="test") @pytest.mark.unit class TestToolDiscoveryEngine: """Test Tool Discovery Engine core functionality.""" def test_tool_discovery_engine_exists(self) -> None: """Test that ToolDiscoveryEngine can be imported.""" assert ( ToolDiscoveryEngine is not None ), "ToolDiscoveryEngine should be importable" def test_discovery_engine_initialization(self) -> None: """Test ToolDiscoveryEngine initialization.""" if ToolDiscoveryEngine is None: pytest.skip("ToolDiscoveryEngine not implemented yet") engine = ToolDiscoveryEngine() assert engine is not None assert hasattr(engine, "search_tools") assert hasattr(engine, "load_curated_tools") assert hasattr(engine, "get_tool_by_id") def test_curated_tools_loading(self) -> None: """Test loading of curated mini-KG with 3-5 tools.""" if ToolDiscoveryEngine is None: pytest.skip("ToolDiscoveryEngine not implemented yet") engine = ToolDiscoveryEngine() tools = engine.load_curated_tools() assert isinstance(tools, list) assert 3 <= len(tools) <= 5 # As specified in MVP 1 # Each tool should be an MCPTool instance for tool in tools: assert hasattr(tool, "id") assert hasattr(tool, "name") assert hasattr(tool, "description") def test_basic_tool_search(self) -> None: """Test basic tool search functionality.""" if ToolDiscoveryEngine is None or ToolSearchCriteria is None: pytest.skip("ToolDiscoveryEngine not implemented yet") engine = ToolDiscoveryEngine() # Mock curated tools mock_tools = [ { "id": "summarizer", "name": "Text Summarizer", "description": "Summarizes long text into key points", }, { "id": "sentiment", "name": "Sentiment Analyzer", "description": "Analyzes sentiment of text content", }, { "id": "image_gen", "name": "Image Generator", "description": "Generates images from text descriptions", }, ] criteria = ToolSearchCriteria( query="I want to summarize a news article", max_results=3 ) with patch.object(engine, "load_curated_tools", return_value=mock_tools): results = engine.search_tools(criteria) assert isinstance(results, list) assert len(results) <= 3 # Should find summarizer as most relevant if results: top_result = results[0] assert hasattr(top_result, "tool_id") or "id" in top_result def test_semantic_search_relevance(self) -> None: """Test that semantic search returns relevant tools.""" if ToolDiscoveryEngine is None or ToolSearchCriteria is None: pytest.skip("ToolDiscoveryEngine not implemented yet") engine = ToolDiscoveryEngine() # Test query for text summarization criteria = ToolSearchCriteria( query="compress this long document into bullet points", max_results=2 ) mock_tools = [ {"id": "summarizer", "description": "Summarizes text into key points"}, {"id": "translator", "description": "Translates text between languages"}, {"id": "image_gen", "description": "Creates images from descriptions"}, ] with patch.object(engine, "load_curated_tools", return_value=mock_tools): results = engine.search_tools(criteria) # Summarizer should be most relevant for summarization query if results: top_result = results[0] tool_id = getattr(top_result, "tool_id", None) or top_result.get("id") assert tool_id == "summarizer" def test_different_query_types(self) -> None: """Test discovery engine with different types of queries.""" if ToolDiscoveryEngine is None or ToolSearchCriteria is None: pytest.skip("ToolDiscoveryEngine not implemented yet") engine = ToolDiscoveryEngine() queries = [ "analyze the mood of this text", # Should find sentiment analyzer "create a picture from description", # Should find image generator "translate this to Spanish", # Should find translator ] for query in queries: criteria = ToolSearchCriteria(query=query, max_results=1) with patch.object(engine, "load_curated_tools", return_value=[]): results = engine.search_tools(criteria) assert isinstance(results, list) @pytest.mark.unit class TestEmbeddingService: """Test Embedding Service for semantic search.""" def test_embedding_service_exists(self) -> None: """Test that EmbeddingService can be imported.""" assert EmbeddingService is not None, "EmbeddingService should be importable" def test_embedding_generation(self) -> None: """Test embedding generation for text.""" if EmbeddingService is None: pytest.skip("EmbeddingService not implemented yet") service = EmbeddingService() text = "Summarize this document into key points" # Mock the embedding call to avoid actual API usage in tests with patch.object( service, "_get_openai_embedding", return_value=np.random.rand(384).tolist() ): embedding = service.get_embedding(text) assert isinstance(embedding, list | np.ndarray) assert len(embedding) > 0 def test_similarity_calculation(self) -> None: """Test cosine similarity calculation between embeddings.""" if EmbeddingService is None: pytest.skip("EmbeddingService not implemented yet") service = EmbeddingService() # Mock embeddings embedding1 = np.random.rand(384).tolist() embedding2 = np.random.rand(384).tolist() similarity = service.calculate_similarity(embedding1, embedding2) assert isinstance(similarity, float) assert -1 <= similarity <= 1 # Cosine similarity range @pytest.mark.unit class TestKnowledgeGraphService: """Test Knowledge Graph Service for tool metadata.""" def test_kg_service_exists(self) -> None: """Test that KnowledgeGraphService can be imported.""" assert ( KnowledgeGraphService is not None ), "KnowledgeGraphService should be importable" def test_tool_storage_and_retrieval(self) -> None: """Test storing and retrieving tool metadata.""" if KnowledgeGraphService is None: pytest.skip("KnowledgeGraphService not implemented yet") kg_service = KnowledgeGraphService() tool_data = { "id": "test_tool", "name": "Test Tool", "description": "A test MCP tool", "category": "testing", "capabilities": ["test_capability"], } # Store tool kg_service.store_tool(tool_data) # Retrieve tool retrieved = kg_service.get_tool("test_tool") assert retrieved is not None assert retrieved["id"] == "test_tool" assert retrieved["name"] == "Test Tool" def test_tool_search_by_capability(self) -> None: """Test searching tools by capabilities.""" if KnowledgeGraphService is None: pytest.skip("KnowledgeGraphService not implemented yet") kg_service = KnowledgeGraphService() # Mock tools with different capabilities tools = [ {"id": "tool1", "capabilities": ["summarization", "text_analysis"]}, {"id": "tool2", "capabilities": ["image_generation", "creativity"]}, {"id": "tool3", "capabilities": ["translation", "language_processing"]}, ] for tool in tools: kg_service.store_tool(tool) # Search by capability text_tools = kg_service.find_tools_by_capability("text_analysis") assert len(text_tools) >= 1 assert any(tool["id"] == "tool1" for tool in text_tools) @pytest.mark.integration class TestToolDiscoveryIntegration: """Test integration between discovery engine components.""" def test_end_to_end_tool_discovery(self) -> None: """Test complete tool discovery workflow.""" if ToolDiscoveryEngine is None: pytest.skip("ToolDiscoveryEngine not implemented yet") engine = ToolDiscoveryEngine() # User query from MVP 1 example query = "I want to summarize a news article" # Should return relevant tools with similarity scores results = engine.search_tools(ToolSearchCriteria(query=query, max_results=3)) assert isinstance(results, list) if results: top_result = results[0] # Should have similarity score and tool metadata assert hasattr(top_result, "similarity_score") or "score" in top_result assert hasattr(top_result, "tool") or "tool" in top_result def test_mvp1_demo_scenario(self) -> None: """Test the exact MVP 1 demo scenario.""" if ToolDiscoveryEngine is None: pytest.skip("ToolDiscoveryEngine not implemented yet") engine = ToolDiscoveryEngine() # Load the curated 3-5 tools tools = engine.load_curated_tools() assert 3 <= len(tools) <= 5 # Test the MVP 1 example query query = "I want to summarize a news article" criteria = ToolSearchCriteria(query=query, max_results=3) results = engine.search_tools(criteria) # Should return top 2-3 relevant tools as per MVP 1 spec assert len(results) <= 3 assert len(results) >= 1 # Should find at least one relevant tool # Results should include tool name and description for Gradio UI for result in results: tool = getattr(result, "tool", result) assert "name" in tool or hasattr(tool, "name") assert "description" in tool or hasattr(tool, "description")