test_tool_registry.py•11.1 kB
"""Unit tests for tool registry."""
import json
import pytest
from pathlib import Path
from unittest.mock import MagicMock
from src.registry.tool_registry import ToolRegistry
from src.tools.base_tool import BaseTool
class MockTool(BaseTool):
"""Mock tool for testing."""
def __init__(self, name: str = "mock_tool"):
self._name = name
def get_name(self) -> str:
return self._name
async def execute(self, params):
return {"result": "success"}
class TestToolRegistry:
"""Test suite for tool registry."""
def test_initialization(self, tmp_path):
"""Test registry initialization."""
metadata_dir = tmp_path / "metadata"
metadata_dir.mkdir()
registry = ToolRegistry(metadata_dir=metadata_dir)
assert registry.metadata_dir == metadata_dir
assert registry.tools == {}
assert registry.tool_metadata == {}
def test_load_metadata_empty_directory(self, tmp_path):
"""Test loading metadata from empty directory."""
metadata_dir = tmp_path / "metadata"
metadata_dir.mkdir()
registry = ToolRegistry(metadata_dir=metadata_dir)
metadata = registry.load_metadata()
assert metadata == {}
def test_load_metadata_nonexistent_directory(self, tmp_path):
"""Test loading metadata from non-existent directory."""
metadata_dir = tmp_path / "nonexistent"
registry = ToolRegistry(metadata_dir=metadata_dir)
metadata = registry.load_metadata()
assert metadata == {}
def test_load_metadata_single_file(self, tmp_path):
"""Test loading metadata from single JSON file."""
metadata_dir = tmp_path / "metadata"
metadata_dir.mkdir()
tool_meta = {
"name": "test_tool",
"description": "Test tool",
"input_schema": {
"type": "object",
"properties": {"param1": {"type": "string"}}
}
}
with open(metadata_dir / "test_tool.json", "w") as f:
json.dump(tool_meta, f)
registry = ToolRegistry(metadata_dir=metadata_dir)
metadata = registry.load_metadata()
assert len(metadata) == 1
assert "test_tool" in metadata
assert metadata["test_tool"]["description"] == "Test tool"
def test_load_metadata_multiple_files(self, tmp_path):
"""Test loading metadata from multiple JSON files."""
metadata_dir = tmp_path / "metadata"
metadata_dir.mkdir()
tools = ["tool1", "tool2", "tool3"]
for tool_name in tools:
tool_meta = {
"name": tool_name,
"description": f"Description for {tool_name}",
"input_schema": {"type": "object"}
}
with open(metadata_dir / f"{tool_name}.json", "w") as f:
json.dump(tool_meta, f)
registry = ToolRegistry(metadata_dir=metadata_dir)
metadata = registry.load_metadata()
assert len(metadata) == 3
for tool_name in tools:
assert tool_name in metadata
def test_load_metadata_invalid_json(self, tmp_path):
"""Test handling of invalid JSON file."""
metadata_dir = tmp_path / "metadata"
metadata_dir.mkdir()
# Create invalid JSON file
with open(metadata_dir / "invalid.json", "w") as f:
f.write("{ invalid json }")
registry = ToolRegistry(metadata_dir=metadata_dir)
metadata = registry.load_metadata()
# Should skip invalid file
assert metadata == {}
def test_load_metadata_missing_name(self, tmp_path):
"""Test handling of metadata without name field."""
metadata_dir = tmp_path / "metadata"
metadata_dir.mkdir()
tool_meta = {
"description": "Tool without name",
"input_schema": {"type": "object"}
}
with open(metadata_dir / "no_name.json", "w") as f:
json.dump(tool_meta, f)
registry = ToolRegistry(metadata_dir=metadata_dir)
metadata = registry.load_metadata()
# Should skip file without name
assert metadata == {}
def test_register_tool_success(self, tool_registry):
"""Test successful tool registration."""
tool = MockTool(name="verify_pan")
tool_registry.register_tool(tool)
assert "verify_pan" in tool_registry.tools
assert tool_registry.tools["verify_pan"] == tool
def test_register_tool_without_metadata(self, tmp_path):
"""Test registering tool without metadata."""
metadata_dir = tmp_path / "metadata"
metadata_dir.mkdir()
registry = ToolRegistry(metadata_dir=metadata_dir)
registry.initialize()
tool = MockTool(name="unknown_tool")
registry.register_tool(tool)
# Tool should not be registered without metadata
assert "unknown_tool" not in registry.tools
def test_get_tool_existing(self, tool_registry):
"""Test getting existing tool."""
tool = MockTool(name="verify_pan")
tool_registry.register_tool(tool)
retrieved = tool_registry.get_tool("verify_pan")
assert retrieved == tool
def test_get_tool_nonexistent(self, tool_registry):
"""Test getting non-existent tool."""
retrieved = tool_registry.get_tool("nonexistent")
assert retrieved is None
def test_list_tools_empty(self, tmp_path):
"""Test listing tools when none are registered."""
metadata_dir = tmp_path / "metadata"
metadata_dir.mkdir()
registry = ToolRegistry(metadata_dir=metadata_dir)
registry.initialize()
tools = registry.list_tools()
assert tools == []
def test_list_tools_with_registered_tools(self, tool_registry):
"""Test listing registered tools."""
tool = MockTool(name="verify_pan")
tool_registry.register_tool(tool)
tools = tool_registry.list_tools()
assert len(tools) == 1
assert tools[0]["name"] == "verify_pan"
assert "description" in tools[0]
assert "inputSchema" in tools[0]
def test_get_tool_metadata_existing(self, tool_registry):
"""Test getting metadata for existing tool."""
metadata = tool_registry.get_tool_metadata("verify_pan")
assert metadata is not None
assert metadata["name"] == "verify_pan"
assert "description" in metadata
assert "input_schema" in metadata
def test_get_tool_metadata_nonexistent(self, tool_registry):
"""Test getting metadata for non-existent tool."""
metadata = tool_registry.get_tool_metadata("nonexistent")
assert metadata is None
def test_initialize_returns_count(self, tmp_path):
"""Test that initialize returns metadata count."""
metadata_dir = tmp_path / "metadata"
metadata_dir.mkdir()
# Create 3 metadata files
for i in range(3):
tool_meta = {
"name": f"tool{i}",
"description": f"Tool {i}",
"input_schema": {"type": "object"}
}
with open(metadata_dir / f"tool{i}.json", "w") as f:
json.dump(tool_meta, f)
registry = ToolRegistry(metadata_dir=metadata_dir)
count = registry.initialize()
assert count == 3
def test_generate_mcp_schema(self, tool_registry):
"""Test MCP schema generation."""
metadata = {
"name": "test_tool",
"description": "Test description",
"input_schema": {
"type": "object",
"properties": {"param": {"type": "string"}}
}
}
schema = tool_registry._generate_mcp_schema(metadata)
assert schema["name"] == "test_tool"
assert schema["description"] == "Test description"
assert schema["inputSchema"] == metadata["input_schema"]
def test_multiple_tool_registration(self, tool_registry):
"""Test registering multiple tools."""
tool1 = MockTool(name="verify_pan")
tool2 = MockTool(name="check_pan_aadhaar_link")
tool_registry.register_tool(tool1)
# Create metadata for second tool
metadata_dir = tool_registry.metadata_dir
tool2_meta = {
"name": "check_pan_aadhaar_link",
"description": "Check PAN-Aadhaar link",
"input_schema": {"type": "object"}
}
with open(metadata_dir / "tool2.json", "w") as f:
json.dump(tool2_meta, f)
tool_registry.tool_metadata = tool_registry.load_metadata()
tool_registry.register_tool(tool2)
assert len(tool_registry.tools) == 2
assert "verify_pan" in tool_registry.tools
assert "check_pan_aadhaar_link" in tool_registry.tools
def test_registry_isolation(self, tmp_path):
"""Test that multiple registries are isolated."""
metadata_dir1 = tmp_path / "metadata1"
metadata_dir2 = tmp_path / "metadata2"
metadata_dir1.mkdir()
metadata_dir2.mkdir()
# Create different metadata in each directory
tool1_meta = {"name": "tool1", "description": "Tool 1", "input_schema": {}}
tool2_meta = {"name": "tool2", "description": "Tool 2", "input_schema": {}}
with open(metadata_dir1 / "tool1.json", "w") as f:
json.dump(tool1_meta, f)
with open(metadata_dir2 / "tool2.json", "w") as f:
json.dump(tool2_meta, f)
registry1 = ToolRegistry(metadata_dir=metadata_dir1)
registry2 = ToolRegistry(metadata_dir=metadata_dir2)
registry1.initialize()
registry2.initialize()
assert "tool1" in registry1.tool_metadata
assert "tool1" not in registry2.tool_metadata
assert "tool2" in registry2.tool_metadata
assert "tool2" not in registry1.tool_metadata
def test_non_json_files_ignored(self, tmp_path):
"""Test that non-JSON files are ignored."""
metadata_dir = tmp_path / "metadata"
metadata_dir.mkdir()
# Create JSON file
tool_meta = {"name": "tool1", "description": "Tool 1", "input_schema": {}}
with open(metadata_dir / "tool1.json", "w") as f:
json.dump(tool_meta, f)
# Create non-JSON files
with open(metadata_dir / "readme.txt", "w") as f:
f.write("This is not JSON")
with open(metadata_dir / "config.yaml", "w") as f:
f.write("key: value")
registry = ToolRegistry(metadata_dir=metadata_dir)
metadata = registry.load_metadata()
# Should only load the JSON file
assert len(metadata) == 1
assert "tool1" in metadata