"""Comprehensive tests for Natural Language Tools module using systematic MCP tool test pattern.
import logging
logging.basicConfig(level=logging.DEBUG)
Tests cover natural language processing capabilities including command interpretation,
intent recognition, conversation management, and language understanding with property-based
testing and comprehensive enterprise-grade validation using the proven pattern that achieved
100% success across 22+ tool suites.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, Mock, patch
import pytest
# Import FastMCP tool objects and extract underlying functions (systematic MCP pattern)
import src.server.tools.natural_language_tools as nl_tools
from hypothesis import assume, given
from hypothesis import strategies as st
if TYPE_CHECKING:
from collections.abc import Callable
# Extract underlying functions from FastMCP tool objects (systematic pattern)
km_process_natural_command = nl_tools.km_process_natural_command.fn
km_recognize_intent = nl_tools.km_recognize_intent.fn
km_generate_from_description = nl_tools.km_generate_from_description.fn
km_conversational_interface = nl_tools.km_conversational_interface.fn
km_nlp_performance_metrics = nl_tools.km_nlp_performance_metrics.fn
# Test data generators using systematic MCP pattern
@st.composite
def command_text_strategy(draw: Callable[..., Any]) -> Mock:
"""Generate valid command text inputs."""
commands = [
"Open the calculator application",
"Save the current document",
"Switch to dark mode",
"Create a new file named report.txt",
"Close all windows",
"Show me the clipboard history",
]
return draw(st.sampled_from(commands))
@st.composite
def intent_text_strategy(draw: Callable[..., Any]) -> Mock:
"""Generate text inputs for intent recognition."""
intents = [
"I want to automate my morning routine",
"Help me organize my files",
"Can you set up a hotkey for screenshot?",
"Schedule a backup every evening",
"Find all documents containing project data",
]
return draw(st.sampled_from(intents))
@st.composite
def conversation_text_strategy(draw: Callable[..., Any]) -> Mock:
"""Generate conversation text inputs."""
conversations = [
"Hello, how can I automate my workflow?",
"What's the best way to organize macros?",
"I need help with keyboard shortcuts",
"Can you explain how triggers work?",
"Thank you for the assistance",
]
return draw(st.sampled_from(conversations))
@st.composite
def language_code_strategy(draw: Callable[..., Any]) -> Mock:
"""Generate valid language codes."""
languages = ["en", "es", "fr", "de", "it", "pt", "ja", "ko", "zh", "auto"]
return draw(st.sampled_from(languages))
@st.composite
def confidence_threshold_strategy(draw: Callable[..., Any]) -> Mock:
"""Generate valid confidence thresholds."""
return draw(st.floats(min_value=0.1, max_value=1.0))
@st.composite
def processing_mode_strategy(draw: Callable[..., Any]) -> None:
"""Generate valid processing modes."""
modes = ["fast", "standard", "detailed", "conversational"]
return draw(st.sampled_from(modes))
class TestNaturalLanguageDependencies:
"""Test natural language module dependencies and imports."""
def test_natural_language_imports(self) -> None:
"""Test that natural language tools can be imported."""
assert km_process_natural_command is not None
assert callable(km_process_natural_command)
assert km_recognize_intent is not None
assert callable(km_recognize_intent)
assert km_generate_from_description is not None
assert callable(km_generate_from_description)
assert km_conversational_interface is not None
assert callable(km_conversational_interface)
assert km_nlp_performance_metrics is not None
assert callable(km_nlp_performance_metrics)
class TestNaturalLanguageParameterValidation:
"""Test parameter validation for natural language operations."""
@given(command_text_strategy())
def test_valid_command_text(self, command_text: list[Any] | str) -> None:
"""Test that command text inputs are properly validated."""
assert isinstance(command_text, str)
assert len(command_text) > 0
assert len(command_text) < 1000 # Reasonable length limit
@given(language_code_strategy())
def test_valid_language_codes(self, language_code: Any) -> None:
"""Test that language codes are properly validated."""
valid_codes = ["en", "es", "fr", "de", "it", "pt", "ja", "ko", "zh", "auto"]
assert language_code in valid_codes
@given(confidence_threshold_strategy())
def test_valid_confidence_thresholds(self, threshold: float) -> None:
"""Test that confidence thresholds are properly validated."""
assert 0.1 <= threshold <= 1.0
@given(processing_mode_strategy())
def test_valid_processing_modes(self, mode: str) -> None:
"""Test that processing modes are properly validated."""
valid_modes = ["fast", "standard", "detailed", "conversational"]
assert mode in valid_modes
class TestKMProcessNaturalCommandMocked:
"""Test km_process_natural_command function with mocked dependencies."""
@pytest.mark.asyncio
async def test_km_process_natural_command_success(self) -> None:
"""Test successful command interpretation."""
with (
patch(
"src.server.tools.natural_language_tools.command_processor",
) as mock_processor,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
# Setup mock command processor (aligned with actual source code)
mock_processed_command = Mock()
mock_processed_command.command_id = "cmd_123"
mock_processed_command.recognized_intent = Mock()
mock_processed_command.recognized_intent.intent = "application_control"
mock_processed_command.recognized_intent.category = Mock()
mock_processed_command.recognized_intent.category.value = "automation"
mock_processed_command.recognized_intent.confidence = 0.92
mock_processed_command.recognized_intent.parameters = {
"application": "calculator",
}
mock_processed_command.extracted_entities = []
mock_processed_command.automation_actions = ["open_application"]
mock_processed_command.confidence_score = 0.92
mock_processed_command.sentiment = None
mock_processed_command.alternatives = []
mock_result = Mock()
mock_result.is_left.return_value = False
mock_result.is_right.return_value = True
mock_result.get_right.return_value = mock_processed_command
mock_processor.process_command = AsyncMock(return_value=mock_result)
# Test data
test_command = "Open the calculator application"
# Execute function (using actual source code function name)
result = await km_process_natural_command(
natural_command=test_command,
language="en",
confidence_threshold=0.7,
)
# Verify result structure
# Debug what's in the result first
print(f"Result: {result}")
if not result.get("success", False):
print(f"Error: {result.get('error', 'No error info')}")
print(f"Error code: {result.get('error_code', 'No error code')}")
# Verify result structure (aligned with actual source code response)
assert result["success"] is True, f"Expected success=True but got {result}"
assert "command_id" in result
assert "original_command" in result
assert "recognized_intent" in result
assert "extracted_entities" in result
assert "automation_actions" in result
assert "confidence_score" in result
assert "processing_time_ms" in result
# Verify recognized intent structure
intent = result["recognized_intent"]
assert "intent" in intent
assert "category" in intent
assert "confidence" in intent
assert "parameters" in intent
assert 0.0 <= intent["confidence"] <= 1.0
@pytest.mark.asyncio
async def test_km_interpret_command_low_confidence(self) -> None:
"""Test command interpretation with low confidence results."""
with (
patch(
"src.server.tools.natural_language_tools.command_processor",
) as mock_processor,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
# Setup mock with low confidence (aligned with actual source code)
mock_processed_command = Mock()
mock_processed_command.command_id = "cmd_low"
mock_processed_command.recognized_intent = Mock()
mock_processed_command.recognized_intent.intent = "unknown"
mock_processed_command.recognized_intent.category = Mock()
mock_processed_command.recognized_intent.category.value = "general"
mock_processed_command.recognized_intent.confidence = 0.3
mock_processed_command.recognized_intent.parameters = {}
mock_processed_command.extracted_entities = []
mock_processed_command.automation_actions = []
mock_processed_command.confidence_score = 0.3
mock_processed_command.sentiment = None
mock_processed_command.alternatives = []
mock_result = Mock()
mock_result.is_left.return_value = False
mock_result.is_right.return_value = True
mock_result.get_right.return_value = mock_processed_command
mock_processor.process_command = AsyncMock(return_value=mock_result)
result = await km_process_natural_command(
natural_command="Unclear command text",
confidence_threshold=0.8,
)
assert result["success"] is True
intent = result["recognized_intent"]
assert intent["confidence"] < 0.8
@pytest.mark.asyncio
async def test_km_interpret_command_invalid_input(self) -> None:
"""Test command interpretation with invalid input."""
result = await km_process_natural_command(
natural_command="", # Empty command
language="en",
)
assert result["success"] is False
assert "error" in result
assert "error_code" in result
class TestKMRecognizeIntentMocked:
"""Test km_recognize_intent function with mocked dependencies."""
@pytest.mark.asyncio
async def test_km_recognize_intent_success(self) -> None:
"""Test successful intent recognition."""
with (
patch(
"src.server.tools.natural_language_tools.intent_classifier",
) as mock_classifier,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
# Setup mock intent classifier (aligned with actual source code)
mock_intent = Mock()
mock_intent.intent = "automation_request"
mock_intent.category = Mock()
mock_intent.category.value = "productivity"
mock_intent.confidence = 0.89
mock_intent.parameters = {
"task_type": "morning_routine",
"frequency": "daily",
}
mock_intent.context_requirements = ["user_schedule", "time_context"]
mock_intent.suggested_actions = ["create_macro", "set_trigger"]
mock_intent.entities = []
mock_result = Mock()
mock_result.is_left.return_value = False
mock_result.is_right.return_value = True
mock_result.get_right.return_value = [mock_intent] # Return list of intents
mock_classifier.recognize_intent = AsyncMock(return_value=mock_result)
# Test data
test_text = "I want to automate my morning routine"
# Execute function (aligned with actual source code parameters)
result = await km_recognize_intent(
user_input=test_text,
include_entities=True,
confidence_threshold=0.6,
)
# Verify result structure (aligned with actual source code response)
assert result["success"] is True
assert "input_text" in result
assert "recognized_intents" in result
assert "processing_time_ms" in result
# Verify intent recognition structure
intents = result["recognized_intents"]
assert isinstance(intents, list)
assert len(intents) > 0
# Check first intent structure
intent = intents[0]
assert "intent" in intent
assert "category" in intent
assert "confidence" in intent
assert "parameters" in intent
assert "context_requirements" in intent
assert "suggested_actions" in intent
assert 0.0 <= intent["confidence"] <= 1.0
@pytest.mark.asyncio
async def test_km_recognize_intent_no_entities(self) -> None:
"""Test intent recognition without entity extraction."""
with (
patch(
"src.server.tools.natural_language_tools.intent_classifier",
) as mock_classifier,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
mock_intent = Mock()
mock_intent.intent = "general_inquiry"
mock_intent.category = Mock()
mock_intent.category.value = "general"
mock_intent.confidence = 0.75
mock_intent.parameters = {}
mock_intent.context_requirements = []
mock_intent.suggested_actions = []
mock_intent.entities = []
mock_result = Mock()
mock_result.is_left.return_value = False
mock_result.is_right.return_value = True
mock_result.get_right.return_value = [mock_intent] # Return list of intents
mock_classifier.recognize_intent = AsyncMock(return_value=mock_result)
result = await km_recognize_intent(
user_input="General question about functionality",
include_entities=False,
)
assert result["success"] is True
intents = result["recognized_intents"]
assert len(intents) > 0
intent = intents[0]
assert "intent" in intent
# Should not include entities when include_entities=False
class TestKMProcessConversationMocked:
"""Test km_process_conversation function with mocked dependencies."""
@pytest.mark.asyncio
async def test_km_process_conversation_success(self) -> None:
"""Test successful conversation processing."""
with (
patch(
"src.server.tools.natural_language_tools.conversation_manager",
) as mock_manager,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
# Setup mock conversation manager (aligned with actual source code)
mock_conversation_response = Mock()
mock_conversation_response.response_text = "I can help you automate your workflow. What specific tasks would you like to automate?"
mock_conversation_response.response_type = "guidance"
mock_conversation_response.confidence = 0.91
mock_conversation_response.suggestions = [
"create_macro",
"setup_hotkey",
"organize_triggers",
]
mock_conversation_response.examples = []
mock_conversation_response.follow_up_questions = []
mock_conversation_response.automation_context = {}
mock_conversation_response.requires_action = False
mock_result = Mock()
mock_result.is_left.return_value = False
mock_result.is_right.return_value = True
mock_result.get_right.return_value = mock_conversation_response
mock_manager.process_conversation = AsyncMock(return_value=mock_result)
# Test data
test_message = "Hello, how can I automate my workflow?"
# Execute function (aligned with actual source code function name)
result = await km_conversational_interface(
conversation_mode="creation",
user_message=test_message,
conversation_id="conv_123",
include_suggestions=True,
)
# Verify result structure (aligned with actual source code response)
assert result["success"] is True
assert "conversation_id" in result
assert "mode" in result
assert "response" in result
assert "processing_time_ms" in result
# Verify conversation response structure
response = result["response"]
assert "text" in response
assert "type" in response
assert "confidence" in response
assert 0.0 <= response["confidence"] <= 1.0
@pytest.mark.asyncio
async def test_km_process_conversation_new_conversation(self) -> None:
"""Test conversation processing for new conversation."""
with (
patch(
"src.server.tools.natural_language_tools.conversation_manager",
) as mock_manager,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
mock_conversation_response = Mock()
mock_conversation_response.response_text = (
"Welcome! I'm here to help you with Keyboard Maestro automation."
)
mock_conversation_response.response_type = "welcome"
mock_conversation_response.confidence = 0.95
mock_conversation_response.suggestions = ["get_started", "view_examples"]
mock_conversation_response.examples = []
mock_conversation_response.follow_up_questions = []
mock_conversation_response.automation_context = {"session_type": "new_user"}
mock_conversation_response.requires_action = False
mock_result = Mock()
mock_result.is_left.return_value = False
mock_result.is_right.return_value = True
mock_result.get_right.return_value = mock_conversation_response
mock_manager.process_conversation = AsyncMock(return_value=mock_result)
result = await km_conversational_interface(
conversation_mode="guidance",
user_message="Hello",
)
assert result["success"] is True
response = result["response"]
assert "text" in response
assert "type" in response
class TestKMNLPMetricsMocked:
"""Test km_nlp_metrics function with mocked dependencies."""
@pytest.mark.asyncio
async def test_km_nlp_metrics_success(self) -> None:
"""Test successful NLP metrics retrieval."""
with (
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
patch(
"src.server.tools.natural_language_tools.intent_classifier",
) as mock_classifier,
):
mock_validate.return_value = None
mock_classifier.get_classification_stats.return_value = {
"total_classifications": 100,
"average_confidence": 0.85,
"cache_size": 50,
}
result = await km_nlp_performance_metrics()
assert result["success"] is True
assert "metrics" in result
metrics = result["metrics"]
assert "system_status" in metrics
assert "performance_metrics" in metrics
assert "usage_statistics" in metrics
assert "component_status" in metrics
@pytest.mark.asyncio
async def test_km_nlp_metrics_minimal(self) -> None:
"""Test NLP metrics with minimal data."""
with (
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
patch(
"src.server.tools.natural_language_tools.intent_classifier",
) as mock_classifier,
):
mock_validate.return_value = None
mock_classifier.get_classification_stats.return_value = {
"total_classifications": 0,
"cache_size": 0,
}
result = await km_nlp_performance_metrics()
assert result["success"] is True
metrics = result["metrics"]
assert "system_status" in metrics
# Performance and usage should be excluded or minimal
class TestNaturalLanguageErrorHandling:
"""Test error handling and edge cases for natural language operations."""
@pytest.mark.asyncio
async def test_command_interpretation_error(self) -> None:
"""Test handling of command interpretation errors."""
with (
patch(
"src.server.tools.natural_language_tools.command_processor",
) as mock_processor,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
mock_processor.process_command = AsyncMock(
side_effect=Exception("Processing error"),
)
result = await km_process_natural_command(
natural_command="Test command",
language="en",
)
assert result["success"] is False
assert "error" in result
@pytest.mark.asyncio
async def test_intent_recognition_error(self) -> None:
"""Test handling of intent recognition errors."""
with (
patch(
"src.server.tools.natural_language_tools.intent_classifier",
) as mock_classifier,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
mock_classifier.recognize_intent = AsyncMock(
side_effect=Exception("Classification error"),
)
result = await km_recognize_intent(user_input="Test input")
assert result["success"] is False
assert "error" in result
@pytest.mark.asyncio
async def test_conversation_processing_error(self) -> None:
"""Test handling of conversation processing errors."""
with (
patch(
"src.server.tools.natural_language_tools.conversation_manager",
) as mock_manager,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
mock_manager.process_conversation = AsyncMock(
side_effect=Exception("Conversation error"),
)
result = await km_conversational_interface(
conversation_mode="troubleshooting",
user_message="Test message",
)
assert result["success"] is False
assert "error" in result
class TestNaturalLanguageIntegration:
"""Test integration scenarios for natural language operations."""
@pytest.mark.asyncio
async def test_complete_nlp_workflow(self) -> None:
"""Test complete natural language processing workflow."""
with (
patch(
"src.server.tools.natural_language_tools.command_processor",
) as mock_processor,
patch(
"src.server.tools.natural_language_tools.intent_classifier",
) as mock_classifier,
patch(
"src.server.tools.natural_language_tools.conversation_manager",
) as mock_manager,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
# Setup mocks for integrated workflow (aligned with actual source code)
mock_command = Mock()
mock_command.command_id = "cmd_integration"
mock_command.recognized_intent = Mock()
mock_command.recognized_intent.intent = "create_automation"
mock_command.recognized_intent.category = Mock()
mock_command.recognized_intent.category.value = "automation"
mock_command.recognized_intent.confidence = 0.9
mock_command.recognized_intent.parameters = {"type": "automation_macro"}
mock_command.extracted_entities = []
mock_command.automation_actions = ["create_macro"]
mock_command.confidence_score = 0.9
mock_command.sentiment = None
mock_command.alternatives = []
mock_processor.process_command = AsyncMock(
return_value=Mock(
is_left=Mock(return_value=False),
is_right=Mock(return_value=True),
get_right=Mock(return_value=mock_command),
),
)
mock_intent = Mock()
mock_intent.intent = "create_automation"
mock_intent.category = Mock()
mock_intent.category.value = "automation"
mock_intent.confidence = 0.85
mock_intent.parameters = {"automation_type": "macro"}
mock_intent.context_requirements = ["productivity"]
mock_intent.suggested_actions = ["create_macro"]
mock_intent.entities = []
mock_classifier.recognize_intent = AsyncMock(
return_value=Mock(
is_left=Mock(return_value=False),
is_right=Mock(return_value=True),
get_right=Mock(return_value=[mock_intent]),
),
)
mock_conversation = Mock()
mock_conversation.response_text = "I can help you create that automation macro. Let me guide you through the process."
mock_conversation.response_type = "guidance"
mock_conversation.confidence = 0.92
mock_conversation.suggestions = ["define_trigger", "specify_actions"]
mock_conversation.examples = []
mock_conversation.follow_up_questions = []
mock_conversation.automation_context = {"task": "macro_creation"}
mock_conversation.requires_action = True
mock_manager.process_conversation = AsyncMock(
return_value=Mock(
is_left=Mock(return_value=False),
is_right=Mock(return_value=True),
get_right=Mock(return_value=mock_conversation),
),
)
# Execute integrated workflow
user_input = "I want to create a macro that automatically saves my work"
# Step 1: Command interpretation
command_result = await km_process_natural_command(
natural_command=user_input,
language="en",
)
# Step 2: Intent recognition
intent_result = await km_recognize_intent(
user_input=user_input,
include_entities=True,
)
# Step 3: Conversation processing
conversation_result = await km_conversational_interface(
conversation_mode="creation",
user_message=user_input,
include_suggestions=True,
)
# Step 4: Get metrics
metrics_result = await km_nlp_performance_metrics()
# Verify workflow integration
assert command_result["success"] is True
assert intent_result["success"] is True
assert conversation_result["success"] is True
assert metrics_result["success"] is True
# Verify basic response structure consistency
assert "recognized_intent" in command_result
assert "recognized_intents" in intent_result
assert "response" in conversation_result
assert "metrics" in metrics_result
class TestNaturalLanguageProperties:
"""Property-based tests for natural language operations."""
@given(
command_text_strategy(),
processing_mode_strategy(),
confidence_threshold_strategy(),
)
@pytest.mark.asyncio
async def test_command_interpretation_properties(
self,
command_text: Any,
processing_mode: Any,
confidence_threshold: Any,
) -> None:
"""Test properties of command interpretation operations."""
assume(len(command_text.strip()) > 0)
with (
patch(
"src.server.tools.natural_language_tools.command_processor",
) as mock_processor,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
mock_command = Mock()
mock_command.command_id = "test_cmd"
mock_command.recognized_intent = Mock()
mock_command.recognized_intent.intent = "test_intent"
mock_command.recognized_intent.category = Mock()
mock_command.recognized_intent.category.value = "test"
mock_command.recognized_intent.confidence = confidence_threshold
mock_command.recognized_intent.parameters = {}
mock_command.extracted_entities = []
mock_command.automation_actions = []
mock_command.confidence_score = confidence_threshold
mock_command.sentiment = None
mock_command.alternatives = []
mock_processor.process_command = AsyncMock(
return_value=Mock(
is_left=Mock(return_value=False),
is_right=Mock(return_value=True),
get_right=Mock(return_value=mock_command),
),
)
result = await km_process_natural_command(
natural_command=command_text,
language="en",
confidence_threshold=confidence_threshold,
)
# Property: All operations should return structured results
assert "success" in result
assert "processing_time_ms" in result
# Property: Successful operations should have required fields
if result["success"]:
assert "recognized_intent" in result
intent = result["recognized_intent"]
assert "confidence" in intent
assert 0.0 <= intent["confidence"] <= 1.0
@given(intent_text_strategy(), language_code_strategy())
@pytest.mark.asyncio
async def test_intent_recognition_properties(
self,
intent_text: str,
language_code: Any,
) -> None:
"""Test properties of intent recognition operations."""
assume(len(intent_text.strip()) > 0)
with (
patch(
"src.server.tools.natural_language_tools.intent_classifier",
) as mock_classifier,
patch(
"src.server.tools.natural_language_tools._validate_components",
) as mock_validate,
):
mock_validate.return_value = None
mock_intent = Mock()
mock_intent.intent = "test_intent"
mock_intent.category = Mock()
mock_intent.category.value = "test"
mock_intent.confidence = 0.8
mock_intent.parameters = {}
mock_intent.context_requirements = []
mock_intent.suggested_actions = []
mock_intent.entities = []
mock_classifier.recognize_intent = AsyncMock(
return_value=Mock(
is_left=Mock(return_value=False),
is_right=Mock(return_value=True),
get_right=Mock(return_value=[mock_intent]),
),
)
result = await km_recognize_intent(user_input=intent_text)
# Property: All intent operations should return structured results
assert "success" in result
# Property: Successful operations should have intent data
if result["success"]:
assert "recognized_intents" in result
intents = result["recognized_intents"]
assert isinstance(intents, list)
if len(intents) > 0:
intent = intents[0]
assert "confidence" in intent
assert 0.0 <= intent["confidence"] <= 1.0
