#!/usr/bin/env python3
"""
Test script to verify search functionality in matrix_cmd.
"""
import asyncio
import logging
import sys
from matrix_cmd.matrix_client import MatrixClient
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Mock class for testing
class MockMatrixClient:
"""Mock Matrix client for testing."""
def __init__(self, enable_search=True, opensearch_config=None):
"""Initialize the mock Matrix client."""
self.enable_search = enable_search
self.opensearch_config = opensearch_config or {}
self.messages = []
async def send_message(self, room_id, message):
"""Send a message to a room."""
logger.info(f"Sending message to room {room_id}: {message[:100]}...")
self.messages.append({"room_id": room_id, "message": message})
return True
# OpenSearchNG configuration
OPENSEARCH_CONFIG = {
"base_url": "http://localhost:8081", # Default local OpenSearchNG URL
"search_path": "/search", # Path for search endpoint
"preferences_path": "/preferences", # Path for preferences endpoint
"categories": ["general"], # Default search categories
"engines": ["google", "bing", "duckduckgo"], # Default search engines
"language": "en-US", # Default language
"format": "json", # Response format (html, json, csv, rss)
"timeout": 10, # Default timeout in seconds
"num_results": 10, # Default number of results
"safe_search": 0 # 0=off, 1=moderate, 2=strict
}
async def test_matrix_cmd_search():
"""Test the search functionality in matrix_cmd."""
try:
# Create a mock Matrix client with search enabled
client = MockMatrixClient(
enable_search=True,
opensearch_config=OPENSEARCH_CONFIG
)
# Test the search command directly
logger.info("Testing search command...")
# Create a mock room ID
room_id = "test_room"
# Test the search command
query = "python programming"
logger.info(f"Searching for: {query}")
# Create a mock opensearch_query function to avoid circular imports
def mock_opensearch_query(query, config):
"""Mock implementation of opensearch_query to avoid circular imports."""
logger.info(f"Mock OpenSearchNG query: {query}")
# Return mock search results
return {
"query": query,
"results": [
{
"title": "Python Programming - Wikipedia",
"url": "https://en.wikipedia.org/wiki/Python_(programming_language)",
"content": "Python is a high-level, general-purpose programming language. Its design philosophy emphasizes code readability with the use of significant indentation."
},
{
"title": "Python Tutorial - W3Schools",
"url": "https://www.w3schools.com/python/",
"content": "Python is a popular programming language. Python can be used on a server to create web applications."
},
{
"title": "Learn Python - Codecademy",
"url": "https://www.codecademy.com/learn/learn-python",
"content": "Learn Python, a powerful language used by sites like YouTube and Dropbox. Learn the fundamentals of programming to build web apps and manipulate data."
}
],
"engines": ["mock_engine"],
"timing": {"total": 0.1}
}
# Execute the mock search query
logger.info(f"Making mock OpenSearchNG query: {query}")
results = mock_opensearch_query(query, OPENSEARCH_CONFIG)
# Format the results
formatted_results = ["Search Results:"]
for i, result in enumerate(results["results"][:5], 1):
title = result.get("title", "No title")
url = result.get("url", "No URL")
content = result.get("content", "No content")
# Truncate content if it's too long
if len(content) > 200:
content = content[:200] + "..."
formatted_results.append(f"{i}. {title}")
formatted_results.append(f" URL: {url}")
formatted_results.append(f" {content}")
formatted_results.append("")
# Send the formatted results
await client.send_message(room_id, "\n".join(formatted_results))
logger.info("Search command test completed")
# Test factual intent detection and search context extraction
logger.info("Testing factual intent detection and search context extraction...")
message = "What is Python programming?"
logger.info(f"Message: {message}")
# Define a simple factual intent detection function
def detect_factual_intent(message):
"""Detect if a message has factual intent."""
question_words = [
"what", "who", "where", "when", "why", "how", "is", "are", "can", "could",
"would", "should", "tell me about", "explain", "search", "find", "look up"
]
message_lower = message.lower()
# Check for question words
for word in question_words:
if word in message_lower:
return True
# Check for question marks
if "?" in message:
return True
return False
# Define a simple search context extraction function
def extract_search_context(results):
"""Extract search context from OpenSearchNG results."""
if not results or "results" not in results or not results["results"]:
return ""
# Extract information from results
context_parts = []
# Add answer if available
if results.get("answer"):
context_parts.append(f"Answer: {results['answer']}")
# Add infobox if available
if results.get("infobox"):
infobox = results["infobox"]
if isinstance(infobox, dict):
infobox_content = infobox.get("content", "")
if infobox_content:
context_parts.append(f"Infobox: {infobox_content}")
# Add top results
for i, result in enumerate(results["results"][:5], 1):
title = result.get("title", "No title")
content = result.get("content", "No content")
url = result.get("url", "No URL")
context_parts.append(f"Result {i}: {title}")
context_parts.append(f"Content: {content}")
context_parts.append(f"URL: {url}")
context_parts.append("")
# Add suggestion if available
if results.get("suggestion"):
context_parts.append(f"Did you mean: {results['suggestion']}?")
return "\n".join(context_parts)
# Check if the message has factual intent
factual_intent = detect_factual_intent(message)
if factual_intent:
logger.info(f"Factual intent detected in message: {message}")
# Execute the mock search query
logger.info(f"Making mock OpenSearchNG query: {message}")
results = mock_opensearch_query(message, OPENSEARCH_CONFIG)
# Extract search context
search_context = extract_search_context(results)
if search_context:
logger.info(f"Search context: {search_context[:100]}...")
logger.info("Factual intent detection and search context extraction test completed")
else:
logger.error("No search context extracted")
return False
else:
logger.error("No factual intent detected")
return False
return True
except ImportError as e:
logger.error(f"Import error: {e}")
return False
except Exception as e:
logger.error(f"Error: {e}")
return False
if __name__ == "__main__":
success = asyncio.run(test_matrix_cmd_search())
if success:
logger.info("Matrix CMD search test successful!")
sys.exit(0)
else:
logger.error("Matrix CMD search test failed!")
sys.exit(1)